feat: add FlexJob with bounded intermediate publication

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.

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                                                                            |

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";

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();
+  }
+}

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);
+  }
+}

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>;
 

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>;