diff --git a/AGENTS.md b/AGENTS.md
index ce79b8a..3c6cd9c 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -16,7 +16,9 @@ Runner must not import server, HTTP, WebSocket, gRPC, broker, scheduler, or opti
 ## Public contracts
 
 - `Job` is a cold, single-use execution and awaitable lifetime node. It settles only after its body and actual descendants finish.
+- `FlexJob` is a Job with synchronous intermediate publication into a bounded single-delivery channel. Its default buffer retains the latest value; overflow is explicit configuration. Receivers never control producer progress, and cancelling a receive never cancels the Job. Body completion closes publication; final results still join all descendants. Cancellation stays linked until the Job settles, including before start and while descendants drain. The channel owns no Job lifetime or cancellation source.
 - `HandoffJob` is a Job whose body offers one value and suspends until resumed. The handoff is rendezvous state released by the Job's cancellation and closure; it owns no lifetime, signal, or queue.
+- `HandoffJob` and `Handoff` are deprecated but retain their existing one-shot, two-way behavior until removal. Do not turn them into FlexJob compatibility wrappers.
 - `Supervisor` manages a supplied ordinary Job. It does not create hidden startup, background, or shutdown owners.
 - `Supervisor.start(options)` follows `Job.start(options)` ownership and context rules. Use `{ parent: undefined }` to explicitly select an independent root; Supervisor state is its Job state.
 - Applications own initialization order and error reporting. Do not add readiness handshakes, automatic logging, or reporting callbacks to the execution core.
diff --git a/README.md b/README.md
index 2913427..49e2d41 100644
--- a/README.md
+++ b/README.md
@@ -190,7 +190,76 @@ const result = await supervisor.run(plan);
 - Completion joins every member and descendant, including cancelled finalizers. A failed group rejects with its genuine failure, or an `AggregateError` for independent ones.
 - Leaves capture no ambient environment: each uses its own seed over the owner's context.
 
-## HandoffJob
+## FlexJob
+
+A `FlexJob<Result, Update>` is an ordinary Job whose body can publish intermediate values through a bounded channel. `publish(value)` is synchronous: it delivers to a waiting receiver or stores the value, then the body continues without waiting for consumption or an acknowledgement. `await job` and `job.result()` retain the ordinary Job contract and wait for the body, descendants, and finalizers.
+
+```ts
+import { FlexJob } from "@tiberjs/runner";
+
+type Update = { phase: "started" | "processed" };
+
+const job = new FlexJob<string, Update>(async (publish) => {
+  publish({ phase: "started" });
+  await doWork();
+  publish({ phase: "processed" });
+  return "finished";
+});
+
+job.start();
+const update = await job.receive(); // { done: false, value: Update }, or channel completion
+const result = await job.result(); // { ok: true, value: "finished" }, or { ok: false, error }
+```
+
+### Buffering and overflow
+
+The default capacity is **one**, with `overflow: "drop-oldest"`: an unread value is replaced by the latest publication. A publication stays available even if no receiver was waiting when it was sent. Unlike a Promise, each value is consumed once and the channel can deliver subsequent values. This is useful for latest-state observations; it does not preserve every event.
+
+Use `FlexJob.withBuffer(capacity, body, options?)` to retain multiple unread values in FIFO order. Overflow behavior is explicit configuration:
+
+```ts
+const job = FlexJob.withBuffer<string, Update>(
+  10,
+  async (publish) => {
+    publish({ phase: "started" });
+    await doWork();
+    publish({ phase: "processed" });
+    return "finished";
+  },
+  { overflow: "drop-oldest" },
+);
+```
+
+| `overflow`                | When the buffer is full                                              |
+| ------------------------- | -------------------------------------------------------------------- |
+| `"drop-oldest"` (default) | Remove the oldest unread value and retain the new one                |
+| `"drop-newest"`           | Discard the new value and retain existing unread values              |
+| `"error"`                 | Throw `RangeError` from `publish()`; an uncaught error fails the Job |
+
+Capacity must be an integer from 1 through 4294967295. Buffer storage grows as values are published, rather than preallocating the entire capacity. None of the policies waits for a receiver. If every event matters, choose adequate capacity and `"error"` so overflow is reported rather than silently dropping events.
+
+`new FlexJob(body, options?)` also accepts `{ capacity, overflow, seed }`. `withBuffer()` takes the same options except `capacity`, which is its first argument. `seed` is an ordinary `ExecutionSeed`; start, ownership, context, failure propagation, and Supervisor submission follow `Job` rules.
+
+### Receiving, completion, and cancellation
+
+- `receive({ signal? })` returns `{ done: false, value }` for an update or `{ done: true, value: undefined }` after successful channel completion. An update may itself be `undefined`; use `done` to distinguish it from completion.
+- Each update goes to **one** receiver. Concurrent receives consume successive values in request order; this is a queue, not broadcast or replay for each subscriber. A receive can be registered before the cold Job starts, but never starts the Job itself.
+- Aborting a receive's signal rejects only that receive with its reason and removes its cancellation listener. It does not cancel the Job, discard a queued value, or stop other receivers. An already-aborted signal rejects without consuming a value.
+- Returning from the body closes publication and preserves buffered values for draining. Once drained, receives report channel completion. Captured `publish` callbacks cannot publish after the body ends, including from descendants that outlive the body.
+- A body failure or Job cancellation discards buffered progress and rejects pending receives. Cancellation stays connected from construction until the whole Job settles, including before start and while descendants outlive the body. Receivers are released even while uncooperative work has not stopped; the Job still waits for its actual lifetime.
+- Channel completion is not proof of whole-Job success. Descendants can still fail after the body returns. Always observe `job.result()` or `await job` for the final outcome and composed failures. Later receives reflect a failed final outcome; an earlier receive cannot be revised.
+- A Job cannot receive from itself or an ancestor; that throws `LifecycleDependencyError` synchronously.
+
+To bound an observation without terminating the producer, pass a separate observation signal to `receive()`:
+
+```ts
+const update = await job.receive({ signal: AbortSignal.timeout(5_000) });
+// If this receive times out, the Job keeps running and publishing under its existing owner.
+```
+
+## HandoffJob (deprecated)
+
+`HandoffJob` and `Handoff` are deprecated. Prefer FlexJob for intermediate publications. Existing handoffs retain their one-shot, two-way behavior; FlexJob does not provide `resume()` responses or suspend publication until an acknowledgement. Applications migrating an acknowledgement-dependent handoff must explicitly keep that exchange in their body rather than simply replacing `offer()` with `publish()`.
 
 Sometimes a Job must hand a value to someone else _before_ it is done — an HTTP exchange publishes its response, then keeps owning cleanup until delivery is acknowledged. `HandoffJob` models that as a one-shot, two-way rendezvous inside an ordinary Job lifetime.
 
@@ -217,12 +286,13 @@ await exchange; // body and descendants have settled
 
 ## API summary
 
-| Export                                                                                    | Role                                                                           |
-| ----------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------ |
-| `Job`, `HandoffJob`                                                                       | An execution and its lifetime; the handoff variant offers one value mid-flight |
-| `execute`, `fork`, `timeout`                                                              | Start a child: awaited boundary, background child, deadline-bounded boundary   |
-| `Supervisor`, `TaskGroup`                                                                 | Submit work to a supplied owner; declare nested groups of cold Jobs            |
-| `signal`, `deadline`, `use`, `hasContext`, `requireContext`, `withContext`                | Ambient environment of the running Job                                         |
-| `contextKey`, `provide`, `ContextFrame`                                                   | Typed context bindings                                                         |
-| `currentState`, `currentAttachment`, `peekState`, `runWith`                               | Runtime state access for integrations                                          |
-| `combinedError`, `LifecycleStateError`, `LifecycleDependencyError`, `MissingContextError` | Errors                                                                         |
+| Export                                                                                    | Role                                                                              |
+| ----------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------- |
+| `Job`, `FlexJob`                                                                          | An execution and its lifetime; FlexJob also publishes bounded intermediate values |
+| `HandoffJob`, `Handoff`                                                                   | Deprecated one-shot, two-way handoff                                              |
+| `execute`, `fork`, `timeout`                                                              | Start a child: awaited boundary, background child, deadline-bounded boundary      |
+| `Supervisor`, `TaskGroup`                                                                 | Submit work to a supplied owner; declare nested groups of cold Jobs               |
+| `signal`, `deadline`, `use`, `hasContext`, `requireContext`, `withContext`                | Ambient environment of the running Job                                            |
+| `contextKey`, `provide`, `ContextFrame`                                                   | Typed context bindings                                                            |
+| `currentState`, `currentAttachment`, `peekState`, `runWith`                               | Runtime state access for integrations                                             |
+| `combinedError`, `LifecycleStateError`, `LifecycleDependencyError`, `MissingContextError` | Errors                                                                            |
diff --git a/src/index.ts b/src/index.ts
index 2e4f9d6..2b03825 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -25,6 +25,9 @@ export {
 
 export { Job } from "./job/job.js";
 export type { JobStartOptions, JobState, JobResult } from "./job/job.js";
+export { FlexJob } from "./job/flex-job.js";
+export type { FlexJobOptions, Publish, ReceiveOptions } from "./job/flex-job.js";
+export type { OverflowPolicy } from "./job/channel.js";
 export { HandoffJob } from "./job/handoff-job.js";
 export type { Handoff } from "./job/handoff.js";
 export { TaskGroup } from "./supervisor/task-group.js";
diff --git a/src/job/channel.ts b/src/job/channel.ts
new file mode 100644
index 0000000..099c969
--- /dev/null
+++ b/src/job/channel.ts
@@ -0,0 +1,126 @@
+import { addAbortListener } from "node:events";
+
+export type OverflowPolicy = "drop-oldest" | "drop-newest" | "error";
+
+interface Receiver<T> {
+  resolve(value: IteratorResult<T, void>): void;
+  reject(error: unknown): void;
+  registration?: Disposable;
+}
+
+/** A bounded, single-delivery queue; it owns no execution or cancellation source. */
+export class Channel<T> {
+  private readonly values: (T | undefined)[] = [];
+  private head = 0;
+  private size = 0;
+  private receivers: Set<Receiver<T>> | undefined;
+  private state: "open" | "closed" | "failed" = "open";
+  private error: unknown;
+
+  constructor(
+    private readonly capacity: number,
+    private readonly overflow: OverflowPolicy,
+  ) {
+    if (!Number.isInteger(capacity) || capacity < 1 || capacity > 0xffff_ffff) {
+      throw new RangeError("FlexJob buffer capacity must be an integer between 1 and 4294967295.");
+    }
+
+    if (overflow !== "drop-oldest" && overflow !== "drop-newest" && overflow !== "error") {
+      throw new TypeError("Unknown FlexJob overflow policy.");
+    }
+  }
+
+  publish(value: T): void {
+    if (this.state !== "open") {
+      throw new TypeError("Cannot publish to a closed channel.");
+    }
+
+    const receiver = this.receivers?.values().next().value;
+    if (receiver) {
+      this.receivers!.delete(receiver);
+      receiver.registration?.[Symbol.dispose]();
+      receiver.resolve({ done: false, value });
+      return;
+    }
+
+    if (this.size === this.capacity) {
+      switch (this.overflow) {
+        case "drop-newest":
+          return;
+        case "error":
+          throw new RangeError("FlexJob publication buffer is full.");
+        case "drop-oldest":
+          this.values[this.head] = value;
+          this.head = (this.head + 1) % this.capacity;
+          return;
+      }
+    }
+
+    this.values[(this.head + this.size) % this.capacity] = value;
+    this.size++;
+  }
+
+  receive(signal?: AbortSignal): Promise<IteratorResult<T, void>> {
+    if (signal?.aborted) {
+      return Promise.reject(signal.reason);
+    }
+    if (this.state === "failed") {
+      return Promise.reject(this.error);
+    }
+
+    if (this.size > 0) {
+      const value = this.values[this.head] as T;
+      this.values[this.head] = undefined;
+      this.head = (this.head + 1) % this.capacity;
+      this.size--;
+      return Promise.resolve({ done: false, value });
+    }
+    if (this.state === "closed") {
+      return Promise.resolve({ done: true, value: undefined });
+    }
+
+    const { promise, resolve, reject } = Promise.withResolvers<IteratorResult<T, void>>();
+    const receiver: Receiver<T> = { resolve, reject };
+    (this.receivers ??= new Set()).add(receiver);
+
+    if (signal) {
+      receiver.registration = addAbortListener(signal, () => {
+        this.receivers!.delete(receiver);
+        receiver.registration?.[Symbol.dispose]();
+        reject(signal.reason);
+      });
+    }
+
+    return promise;
+  }
+
+  /** Stop publication; successful completion preserves values not yet received. */
+  close(): void {
+    if (this.state !== "open") {
+      return;
+    }
+
+    this.state = "closed";
+    for (const receiver of this.receivers ?? []) {
+      receiver.registration?.[Symbol.dispose]();
+      receiver.resolve({ done: true, value: undefined });
+    }
+    this.receivers?.clear();
+  }
+
+  /** Failure discards stale progress and releases all receivers, including late ones. */
+  fail(error: unknown): void {
+    this.state = "failed";
+    this.error = error;
+
+    this.values.length = 0;
+    this.head = 0;
+    this.size = 0;
+
+    for (const receiver of this.receivers ?? []) {
+      receiver.registration?.[Symbol.dispose]();
+      receiver.reject(error);
+    }
+    this.receivers?.clear();
+  }
+}
diff --git a/src/job/flex-job.ts b/src/job/flex-job.ts
new file mode 100644
index 0000000..7c0ef78
--- /dev/null
+++ b/src/job/flex-job.ts
@@ -0,0 +1,91 @@
+import { addAbortListener } from "node:events";
+import type { ExecutionSeed } from "../execution/context/execution-context.js";
+import { Channel } from "./channel.js";
+import type { OverflowPolicy } from "./channel.js";
+import { Job } from "./job.js";
+
+/** Synchronously send an intermediate value without waiting for a receiver. */
+export type Publish<T> = (value: T) => void;
+
+export interface FlexJobOptions {
+  readonly capacity?: number;
+  readonly overflow?: OverflowPolicy;
+  readonly seed?: ExecutionSeed;
+}
+
+export interface ReceiveOptions {
+  /** Cancels only this receive, never the Job or another receiver. */
+  readonly signal?: AbortSignal;
+}
+
+type FlexBody<Result, Update> = (publish: Publish<Update>) => Result | PromiseLike<Result>;
+
+/** A Job with a bounded channel for intermediate values, independent of its final result. */
+export class FlexJob<Result, Update> extends Job<Result> {
+  private readonly channel: Channel<Update>;
+
+  constructor(body: FlexBody<Result, Update>, options: FlexJobOptions = {}) {
+    if (typeof body !== "function") {
+      throw new TypeError("FlexJob requires a body.");
+    }
+
+    const channel = new Channel<Update>(options.capacity ?? 1, options.overflow ?? "drop-oldest");
+    super(() => this.runBody(body), options.seed);
+    this.channel = channel;
+
+    // Cancellation must reach the channel before start and while descendants outlive the body.
+    const signal = this.signal;
+    const registration = addAbortListener(signal, () => {
+      channel.fail(this.failed ? this.failure : signal.reason);
+    });
+
+    // Preparation and descendant failures can settle a Job without an active publication body.
+    void super.result().then((result) => {
+      registration[Symbol.dispose]();
+
+      if (result.ok) {
+        channel.close();
+      } else {
+        channel.fail(result.error);
+      }
+    });
+  }
+
+  private async runBody(body: FlexBody<Result, Update>): Promise<Result> {
+    // Activation may have added external sources since the constructor observed this signal.
+    const signal = this.signal;
+
+    const publish: Publish<Update> = (value) => {
+      this.recheckCancellation();
+      signal.throwIfAborted();
+      this.channel.publish(value);
+    };
+
+    try {
+      return await body(publish);
+    } catch (error) {
+      this.channel.fail(error);
+      throw error;
+    } finally {
+      this.channel.close();
+    }
+  }
+
+  static withBuffer<Result, Update>(
+    capacity: number,
+    body: FlexBody<Result, Update>,
+    options: Omit<FlexJobOptions, "capacity"> = {},
+  ): FlexJob<Result, Update> {
+    return new FlexJob(body, { ...options, capacity });
+  }
+
+  /** Consume the next update or channel completion; an optional signal cancels only this wait. */
+  receive(options: ReceiveOptions = {}): Promise<IteratorResult<Update, void>> {
+    const dependency = this.dependency("receive");
+    if (dependency) {
+      throw dependency;
+    }
+
+    return this.channel.receive(options.signal);
+  }
+}
diff --git a/src/job/handoff-job.ts b/src/job/handoff-job.ts
index b7121a5..22e47d9 100644
--- a/src/job/handoff-job.ts
+++ b/src/job/handoff-job.ts
@@ -2,7 +2,10 @@ import type { ExecutionSeed } from "../execution/context/execution-context.js";
 import { HandoffState, type Handoff } from "./handoff.js";
 import { Job } from "./job.js";
 
-/** A Job whose body offers one value mid-execution and suspends until the consumer resumes it. */
+/**
+ * A Job whose body offers one value mid-execution and suspends until the consumer resumes it.
+ * @deprecated Use FlexJob for intermediate publications. FlexJob does not wait for a resume response.
+ */
 export class HandoffJob<Result, Offered, Resumed> extends Job<Result> {
   private readonly rendezvous: HandoffState<Offered, Resumed>;
 
diff --git a/src/job/handoff.ts b/src/job/handoff.ts
index da9488c..287916d 100644
--- a/src/job/handoff.ts
+++ b/src/job/handoff.ts
@@ -1,3 +1,4 @@
+/** @deprecated Use FlexJob's Publish for intermediate values; it does not provide resume responses. */
 export interface Handoff<Offered, Resumed> {
   /** Hand over one value and suspend until resumed. Rejects with the Job's cancellation reason. */
   offer(value: Offered): Promise<Resumed>;
diff --git a/tests/flex-job.test.ts b/tests/flex-job.test.ts
new file mode 100644
index 0000000..ce5bbd8
--- /dev/null
+++ b/tests/flex-job.test.ts
@@ -0,0 +1,678 @@
+import { getEventListeners } from "node:events";
+import { setImmediate as nextTurn } from "node:timers/promises";
+import { expect, expectTypeOf, test, vi } from "vitest";
+import {
+  contextKey,
+  currentState,
+  execute,
+  FlexJob,
+  fork,
+  Job,
+  LifecycleDependencyError,
+  provide,
+  Supervisor,
+  use,
+} from "../src/index.js";
+import type { Publish } from "../src/index.js";
+
+test("construction is cold and publication never waits for a receiver", async () => {
+  let ran = false;
+  const job = new FlexJob<string, number>((publish) => {
+    ran = true;
+
+    expectTypeOf(publish).toEqualTypeOf<Publish<number>>();
+    expect(publish(1)).toBeUndefined();
+    publish(2);
+    publish(3);
+    return "done";
+  });
+
+  expect(ran).toBe(false);
+  expect(job.state).toBe("created");
+
+  job.start();
+  expect(await job).toBe("done");
+  expect(await job.receive()).toEqual({ done: false, value: 3 });
+  expect(await job.receive()).toEqual({ done: true, value: undefined });
+  expect(await job.result()).toEqual({ ok: true, value: "done" });
+  expectTypeOf(job.receive).returns.toEqualTypeOf<Promise<IteratorResult<number, void>>>();
+});
+
+test("a pending receive takes an undefined publication without confusing it with completion", async () => {
+  const job = new FlexJob<number, undefined>((publish) => {
+    publish(undefined);
+    return 42;
+  });
+
+  const received = job.receive();
+  job.start();
+
+  expect(await received).toEqual({ done: false, value: undefined });
+  expect(await job).toBe(42);
+  expect(await job.receive()).toEqual({ done: true, value: undefined });
+});
+
+test("buffered updates are consumed once in FIFO order across competing receivers", async () => {
+  const proceed = Promise.withResolvers<void>();
+  const job = FlexJob.withBuffer<void, number>(3, async (publish) => {
+    await proceed.promise;
+
+    publish(1);
+    publish(2);
+    publish(3);
+    publish(4);
+    publish(5);
+  }).start();
+
+  const first = job.receive();
+  const second = job.receive();
+  proceed.resolve();
+
+  expect(await first).toEqual({ done: false, value: 1 });
+  expect(await second).toEqual({ done: false, value: 2 });
+  await job;
+
+  for (const value of [3, 4, 5]) {
+    expect(await job.receive()).toEqual({ done: false, value });
+  }
+  expect(await job.receive()).toEqual({ done: true, value: undefined });
+});
+
+test("drop-oldest retains the newest values across repeated publication and consumption", async () => {
+  const proceed = Promise.withResolvers<void>();
+  const job = FlexJob.withBuffer<void, number>(2, async (publish) => {
+    publish(1);
+    publish(2);
+    publish(3);
+
+    await proceed.promise;
+
+    publish(4);
+    publish(5);
+    publish(6);
+  }).start();
+
+  expect(await job.receive()).toEqual({ done: false, value: 2 });
+
+  proceed.resolve();
+  await job;
+
+  expect(await job.receive()).toEqual({ done: false, value: 5 });
+  expect(await job.receive()).toEqual({ done: false, value: 6 });
+  expect(await job.receive()).toEqual({ done: true, value: undefined });
+});
+
+test("drop-newest preserves unread updates when its buffer is full", async () => {
+  const job = FlexJob.withBuffer<void, number>(
+    2,
+    (publish) => {
+      for (const value of [1, 2, 3, 4]) {
+        publish(value);
+      }
+    },
+    { overflow: "drop-newest" },
+  ).start();
+
+  await job;
+  expect(await job.receive()).toEqual({ done: false, value: 1 });
+  expect(await job.receive()).toEqual({ done: false, value: 2 });
+  expect(await job.receive()).toEqual({ done: true, value: undefined });
+});
+
+test("overflow errors fail the Job unless the body handles them", async () => {
+  let overflow: unknown;
+  const job = FlexJob.withBuffer<void, number>(
+    1,
+    (publish) => {
+      publish(1);
+
+      try {
+        publish(2);
+      } catch (error) {
+        overflow = error;
+        throw error;
+      }
+    },
+    { overflow: "error" },
+  ).start();
+
+  expect(await job.result()).toEqual({ ok: false, error: overflow });
+  expect(overflow).toBeInstanceOf(RangeError);
+  await expect(job.receive()).rejects.toBe(overflow);
+
+  const recovered = FlexJob.withBuffer<string, number>(
+    1,
+    (publish) => {
+      publish(1);
+      expect(() => publish(2)).toThrowError(RangeError);
+      return "recovered";
+    },
+    { overflow: "error" },
+  ).start();
+
+  expect(await recovered).toBe("recovered");
+  expect(await recovered.receive()).toEqual({ done: false, value: 1 });
+});
+
+test("buffer configuration is validated at construction", async () => {
+  const body = (publish: Publish<number>) => publish(42);
+
+  for (const capacity of [1, 0xffff_ffff]) {
+    const job = FlexJob.withBuffer(capacity, body).start();
+    await job;
+    expect(await job.receive()).toEqual({ done: false, value: 42 });
+    expect(await job.receive()).toEqual({ done: true, value: undefined });
+  }
+
+  for (const capacity of [0, -1, 1.5, NaN, Infinity, 2 ** 32]) {
+    expect(() => FlexJob.withBuffer(capacity, body)).toThrowError(RangeError);
+  }
+  expect(() => new FlexJob(body, { overflow: "unknown" as "error" })).toThrowError(TypeError);
+  expect(() => new FlexJob(undefined as unknown as () => void)).toThrowError(TypeError);
+});
+
+test("aborting one receive keeps the producer and other receivers running", async () => {
+  const proceed = Promise.withResolvers<void>();
+  const observation = new AbortController();
+  const reason = new Error("stop observing");
+  const job = new FlexJob<string, number>(async (publish) => {
+    await proceed.promise;
+
+    publish(42);
+    return "done";
+  }).start();
+
+  const abandoned = job.receive({ signal: observation.signal });
+  const rejected = abandoned.catch((error: unknown) => error);
+  const remaining = job.receive();
+  expect(getEventListeners(observation.signal, "abort")).toHaveLength(1);
+
+  observation.abort(reason);
+  expect(await rejected).toBe(reason);
+  expect(getEventListeners(observation.signal, "abort")).toHaveLength(0);
+  expect(job.signal.aborted).toBe(false);
+
+  proceed.resolve();
+  expect(await remaining).toEqual({ done: false, value: 42 });
+  expect(await job).toBe("done");
+});
+
+test("already-aborted receives do not consume buffered values", async () => {
+  const job = new FlexJob<void, number>((publish) => publish(42)).start();
+  await job;
+
+  const reason = new Error("already stopped");
+  await expect(job.receive({ signal: AbortSignal.abort(reason) })).rejects.toBe(reason);
+  expect(await job.receive()).toEqual({ done: false, value: 42 });
+});
+
+test("receive registrations are removed on delivery, completion, and failure", async () => {
+  for (const ending of ["value", "complete", "fail"] as const) {
+    const observation = new AbortController();
+    const proceed = Promise.withResolvers<void>();
+    const failure = new Error("body failed");
+    const job = new FlexJob<void, number>(async (publish) => {
+      await proceed.promise;
+      if (ending === "value") {
+        publish(42);
+      }
+      if (ending === "fail") {
+        throw failure;
+      }
+    }).start();
+
+    const received = job.receive({ signal: observation.signal });
+    const checked = received.then(
+      (value) => ({ ok: true, value }),
+      (error: unknown) => ({ ok: false, error }),
+    );
+
+    proceed.resolve();
+    expect(await checked).toEqual(
+      ending === "fail"
+        ? { ok: false, error: failure }
+        : {
+            ok: true,
+            value:
+              ending === "value" ? { done: false, value: 42 } : { done: true, value: undefined },
+          },
+    );
+    await job.result();
+
+    expect(getEventListeners(observation.signal, "abort")).toHaveLength(0);
+    expect(getEventListeners(job.signal, "abort")).toHaveLength(0);
+  }
+});
+
+test("body completion closes publication but final results still join descendants", async () => {
+  const releaseChild = Promise.withResolvers<void>();
+  let publishLater: Publish<number> | undefined;
+  let settled = false;
+
+  const job = new FlexJob<string, number>((publish) => {
+    publishLater = publish;
+    fork(() => releaseChild.promise);
+    publish(42);
+    return "done";
+  }).start();
+
+  const completion = job.then((value) => {
+    settled = true;
+    return value;
+  });
+
+  try {
+    expect(await job.receive()).toEqual({ done: false, value: 42 });
+    expect(await job.receive()).toEqual({ done: true, value: undefined });
+
+    await nextTurn();
+    expect(settled).toBe(false);
+    expect(() => publishLater!(43)).toThrowError(TypeError);
+  } finally {
+    releaseChild.resolve();
+    await job.result();
+  }
+
+  expect(await completion).toBe("done");
+  expect(getEventListeners(job.signal, "abort")).toHaveLength(0);
+});
+
+test("body failures preserve error identity and discard stale updates", async () => {
+  const failure = new Error("body failed");
+  const job = new FlexJob<void, number>((publish) => {
+    publish(1);
+    throw failure;
+  }).start();
+
+  expect(await job.result()).toEqual({ ok: false, error: failure });
+  await expect(job.receive()).rejects.toBe(failure);
+});
+
+test("an undefined thrown error remains distinguishable from channel completion", async () => {
+  const job = new FlexJob<void, number>(() => {
+    throw undefined;
+  }).start();
+
+  expect(await job.result()).toEqual({ ok: false, error: undefined });
+  await expect(job.receive()).rejects.toBeUndefined();
+});
+
+test("cancellation releases receivers before uncooperative work finishes", async () => {
+  const finishBody = Promise.withResolvers<void>();
+  const reason = new Error("stop");
+  let publishLater: Publish<number> | undefined;
+
+  const job = new FlexJob<void, number>(async (publish) => {
+    publishLater = publish;
+    await finishBody.promise;
+  }).start();
+
+  const received = job.receive();
+  const rejected = received.catch((error: unknown) => error);
+
+  job.cancel(reason);
+  expect(await rejected).toBe(reason);
+  expect(job.state).toBe("running");
+  expect(() => publishLater!(42)).toThrow(reason);
+
+  finishBody.resolve();
+  expect(await job.result()).toEqual({ ok: false, error: reason });
+  await expect(job.receive()).rejects.toBe(reason);
+});
+
+test.each(["direct", "external", "parent"] as const)(
+  "%s cancellation after body return discards progress before descendants finish",
+  async (source) => {
+    const releaseParent = Promise.withResolvers<void>();
+    const releaseChild = Promise.withResolvers<void>();
+    const external = new AbortController();
+    const reason = new Error("stop");
+    let settled = false;
+
+    const parent = new Job(() => releaseParent.promise).start({ parent: undefined });
+    const job = FlexJob.withBuffer<string, number>(
+      2,
+      (publish) => {
+        fork(() => releaseChild.promise);
+        publish(1);
+        publish(2);
+        return "body returned";
+      },
+      { seed: { signal: external.signal } },
+    ).start({ parent });
+    const completion = job.result().then((result) => {
+      settled = true;
+      return result;
+    });
+
+    try {
+      await nextTurn();
+      expect(job.state).toBe("closing");
+
+      switch (source) {
+        case "direct":
+          job.cancel(reason);
+          break;
+        case "external":
+          external.abort(reason);
+          break;
+        case "parent":
+          parent.cancel(reason);
+          break;
+      }
+
+      await nextTurn();
+      expect(settled).toBe(false);
+      await expect(job.receive()).rejects.toBe(reason);
+      await expect(job.receive()).rejects.toBe(reason);
+    } finally {
+      releaseChild.resolve();
+      releaseParent.resolve();
+      await Promise.all([completion, parent.result()]);
+    }
+
+    expect(await completion).toEqual({ ok: false, error: reason });
+    expect(getEventListeners(job.signal, "abort")).toHaveLength(0);
+    expect(getEventListeners(external.signal, "abort")).toHaveLength(0);
+  },
+);
+
+test("parent cancellation reaches a FlexJob channel without cancelling a receive separately", async () => {
+  const release = Promise.withResolvers<void>();
+  const ready = Promise.withResolvers<{ child: FlexJob<void, number> }>();
+  const reason = new Error("owner stopped");
+  const owner = new Job(async () => {
+    const child = new FlexJob<void, number>(() => release.promise).start();
+    ready.resolve({ child });
+    await release.promise;
+  }).start();
+
+  const { child } = await ready.promise;
+  const received = child.receive();
+  const rejected = received.catch((error: unknown) => error);
+
+  owner.cancel(reason);
+  expect(await rejected).toBe(reason);
+
+  release.resolve();
+  expect(await owner.result()).toEqual({ ok: false, error: reason });
+});
+
+test("an external source is observed even when the body does not read its signal", async () => {
+  const release = Promise.withResolvers<void>();
+  const external = new AbortController();
+  const reason = new Error("external stopped");
+  const job = new FlexJob<void, number>(() => release.promise, {
+    seed: { signal: external.signal },
+  }).start();
+
+  const received = job.receive();
+  const rejected = received.catch((error: unknown) => error);
+
+  external.abort(reason);
+  expect(await rejected).toBe(reason);
+
+  release.resolve();
+  expect(await job.result()).toEqual({ ok: false, error: reason });
+  expect(getEventListeners(external.signal, "abort")).toHaveLength(0);
+});
+
+test("cold cancellation releases receivers without starting or closing the Job", async () => {
+  const reason = new Error("stopped before start");
+  let ran = false;
+  let rejected = false;
+  let rejection: unknown;
+
+  const job = new FlexJob<void, number>(() => {
+    ran = true;
+  });
+  const received = job.receive().catch((error: unknown) => {
+    rejected = true;
+    rejection = error;
+  });
+
+  try {
+    job.cancel(reason);
+    await nextTurn();
+    expect(rejected).toBe(true);
+    expect(rejection).toBe(reason);
+    expect(job.state).toBe("created");
+    expect(ran).toBe(false);
+  } finally {
+    await job.close(reason);
+    await received;
+  }
+
+  expect(getEventListeners(job.signal, "abort")).toHaveLength(0);
+});
+
+test("cold close and failures before the body starts release receivers", async () => {
+  const reason = new Error("stopped before start");
+  const cold = new FlexJob<void, number>(() => {
+    throw new Error("must not run");
+  });
+  const received = cold.receive();
+  const rejected = received.catch((error: unknown) => error);
+
+  await cold.close(reason);
+  expect(await rejected).toBe(reason);
+
+  const skipped = new FlexJob<void, number>(
+    () => {
+      throw new Error("must not run");
+    },
+    {
+      seed: { signal: AbortSignal.abort(reason) },
+    },
+  );
+  const skippedReceive = skipped.receive();
+  const skippedRejection = skippedReceive.catch((error: unknown) => error);
+
+  skipped.start();
+  expect(await skipped.result()).toEqual({ ok: false, error: reason });
+  expect(await skippedRejection).toBe(reason);
+
+  const invalid = new FlexJob<void, number>(() => {}, { seed: { deadline: NaN } });
+  const invalidReceive = invalid.receive();
+  const invalidRejection = invalidReceive.catch((error: unknown) => error);
+
+  invalid.start();
+  expect((await invalid.result()).ok).toBe(false);
+  expect(await invalidRejection).toBeInstanceOf(RangeError);
+});
+
+test("publication runs under its Job's context and descendants retain ownership", async () => {
+  const Tenant = contextKey<string>("tenant");
+
+  await execute({ values: [provide(Tenant, "outer")] }, async () => {
+    const owner = currentState().job;
+    let bodyOwner: Job<unknown> | undefined;
+
+    const job = FlexJob.withBuffer<string, string>(
+      2,
+      async (publish) => {
+        bodyOwner = currentState().job;
+        publish(use(Tenant)!);
+
+        const child = fork(() => {
+          expect(currentState().job.parent).toBe(bodyOwner);
+          return use(Tenant)!;
+        });
+
+        publish(await child);
+        return "done";
+      },
+      { seed: { values: [provide(Tenant, "inner")] } },
+    ).start();
+
+    expect(job.parent).toBe(owner);
+    expect(await job).toBe("done");
+    expect(bodyOwner).toBe(job);
+    expect(await job.receive()).toEqual({ done: false, value: "inner" });
+    expect(await job.receive()).toEqual({ done: false, value: "inner" });
+    expect(use(Tenant)).toBe("outer");
+  });
+});
+
+test("receiving from oneself or an ancestor is rejected synchronously", async () => {
+  let job: FlexJob<void, number>;
+  job = new FlexJob<void, number>(async () => {
+    expect(() => job.receive()).toThrowError(LifecycleDependencyError);
+    await fork(() => {
+      expect(() => job.receive()).toThrowError(LifecycleDependencyError);
+    });
+  });
+
+  job.start();
+  await job;
+});
+
+test("a descendant failure discards unread progress before remaining descendants finish", async () => {
+  const failChild = Promise.withResolvers<void>();
+  const releaseSibling = Promise.withResolvers<void>();
+  const failure = new Error("descendant failed");
+  let settled = false;
+
+  const job = new FlexJob<string, number>((publish) => {
+    fork(async () => {
+      await failChild.promise;
+      throw failure;
+    });
+    fork(() => releaseSibling.promise);
+    publish(42);
+    return "body returned";
+  }).start();
+  const completion = job.result().then((result) => {
+    settled = true;
+    return result;
+  });
+
+  try {
+    await nextTurn();
+    expect(job.state).toBe("closing");
+
+    failChild.resolve();
+    await nextTurn();
+    expect(job.failed).toBe(true);
+    expect(settled).toBe(false);
+    await expect(job.receive()).rejects.toBe(failure);
+    await expect(job.receive()).rejects.toBe(failure);
+  } finally {
+    failChild.resolve();
+    releaseSibling.resolve();
+    await completion;
+  }
+
+  expect(await completion).toEqual({ ok: false, error: failure });
+  expect(getEventListeners(job.signal, "abort")).toHaveLength(0);
+});
+
+test("body await-using cleanup failure joins an independent descendant failure", async () => {
+  const failChild = Promise.withResolvers<void>();
+  const cleanupStarted = Promise.withResolvers<void>();
+  const finishCleanup = Promise.withResolvers<void>();
+  const operation = new Error("operation failed");
+  const cleanup = new Error("cleanup failed");
+  let settled = false;
+
+  const job = new FlexJob<void, number>(async (publish) => {
+    await using _resource = {
+      async [Symbol.asyncDispose]() {
+        cleanupStarted.resolve();
+        await finishCleanup.promise;
+        throw cleanup;
+      },
+    };
+
+    const child = fork(async () => {
+      await failChild.promise;
+      throw operation;
+    });
+    publish(42);
+    await child.result();
+  }).start();
+  const completion = job.result().then((result) => {
+    settled = true;
+    return result;
+  });
+
+  try {
+    expect(await job.receive()).toEqual({ done: false, value: 42 });
+
+    failChild.resolve();
+    await cleanupStarted.promise;
+    await nextTurn();
+    expect(settled).toBe(false);
+    await expect(job.receive()).rejects.toBe(operation);
+  } finally {
+    failChild.resolve();
+    finishCleanup.resolve();
+    await completion;
+  }
+
+  const result = await completion;
+  expect(result.ok).toBe(false);
+  if (result.ok) {
+    throw new Error("expected failure");
+  }
+
+  expect(result.error).toBeInstanceOf(AggregateError);
+  const errors = (result.error as AggregateError).errors;
+  expect(errors).toHaveLength(2);
+  expect(errors[0]).toBe(operation);
+  expect(errors[1]).toBe(cleanup);
+  await expect(job.receive()).rejects.toBe(result.error);
+});
+
+test("a deadline releases observation while completion still waits for work", async () => {
+  vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+  try {
+    const release = Promise.withResolvers<void>();
+    const job = new FlexJob<void, number>(() => release.promise, {
+      seed: { deadline: Date.now() + 20 },
+    }).start();
+
+    const received = job.receive().catch((error: unknown) => error);
+    await vi.advanceTimersByTimeAsync(20);
+    const reason = await received;
+    expect(reason).toBe(job.signal.reason);
+    expect(job.signal.aborted).toBe(true);
+    expect(job.state).toBe("running");
+
+    release.resolve();
+    expect(await job.result()).toEqual({ ok: false, error: reason });
+  } finally {
+    vi.useRealTimers();
+  }
+});
+
+test("isolated Supervisor submissions preserve FlexJob result and channel failure", async () => {
+  const finishOwner = Promise.withResolvers<void>();
+  const owner = new Job(() => finishOwner.promise);
+  await using supervisor = new Supervisor(owner);
+  supervisor.start({ parent: undefined });
+
+  const failure = new Error("isolated failure");
+  const job = new FlexJob<void, number>(() => {
+    throw failure;
+  });
+
+  await expect(supervisor.run(job)).rejects.toBe(failure);
+  expect(job.parent).toBe(owner);
+  expect(owner.signal.aborted).toBe(false);
+  expect(await job.result()).toEqual({ ok: false, error: failure });
+  await expect(job.receive()).rejects.toBe(failure);
+
+  finishOwner.resolve();
+});
+
+test("unobserved intermediate failures do not create unhandled promise rejections", async () => {
+  const failure = new Error("unobserved");
+  const job = new FlexJob<void, number>(() => {
+    throw failure;
+  }).start();
+
+  expect(await job.result()).toEqual({ ok: false, error: failure });
+  await nextTurn();
+  await expect(job.receive()).rejects.toBe(failure);
+});