refactor: single-result Job plus HandoffJob for one-shot handoff

AGENTS.md

@@ -16,6 +16,7 @@ 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.
+- `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.
 - `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

@@ -30,7 +30,6 @@ const result = await execute(async () => {
 - `fork()` starts a child of the current Job. `new Job(body).start()` does the same inside an execution, or starts a root outside one.
 - `start({ parent })` selects an explicit owner and inherits that owner's context. `start({ parent: undefined })` starts an independent root.
 - `cancel(reason)` requests cooperative cancellation. It does not mean the work has finished.
-- `value()` observes a value published by the body without waiting for descendants. It rejects if the Job settles without invoking its publisher.
 - `join()` and `await job` observe the complete result. Awaiting a cold Job rejects; awaiting never starts it.
 - `finish()` stops admission of direct children and joins without cancellation. `close(reason)` stops admission, cancels, and joins. Repeated `close()` calls share their result.
 - `close()` ignores expected cancellation but rejects genuine execution or finalizer failures. Jobs support `await using`.
@@ -55,27 +54,7 @@ if (result.ok) {
 }
 ```
 
-Use `ok` to distinguish success from failure: both a successful value and a thrown error may be `undefined`. `result()` and `value()` reject self/ancestor observation synchronously.
-
-The body receives a publisher that may be called once while the body is running.
-`Job<Result, Published = Result>` keeps the published value's type separate from the
-body's eventual result. The publisher accepts a value or `PromiseLike` value.
-Cancellation does not disable publication; `result()` and `await job` still report the
-Job's final success or failure independently.
-
-```ts
-const release = Promise.withResolvers<void>();
-const job = new Job<number, string>(async (publish) => {
-  publish("ready");
-  await release.promise;
-  return 42;
-});
-
-job.start();
-console.log(await job.value()); // "ready"; the Job is still running.
-release.resolve();
-console.log(await job); // 42; the body and descendants have settled.
-```
+Use `ok` to distinguish success from failure: both a successful value and a thrown error may be `undefined`. `result()` rejects self/ancestor observation synchronously.
 
 `cancelChildren(reason)` cancels direct children, lets cancellation cascade through
 their descendants, and resolves after those child lifetimes settle. It does not cancel
@@ -87,6 +66,30 @@ are preserved together in an `AggregateError`.
 
 Use native `try/finally`, `using`, or `await using` for resources. Body-local cleanup has its ordinary lexical lifetime. If a resource must outlive an entire subtree, acquire it outside the awaited Job or `execute()` call.
 
+### HandoffJob
+
+A `HandoffJob` body offers one value and suspends until the consumer resumes it. The Job still settles only after its body and descendants finish.
+
+```ts
+import { HandoffJob } from "@tiberjs/runner";
+
+const exchange = new HandoffJob<void, Response, "delivered" | "aborted">(async (handoff) => {
+  const response = await prepare();
+  const outcome = await handoff.offer(response); // suspended until resume()
+  await release(outcome);
+});
+
+exchange.start();
+const response = await exchange.receive(); // the Job is still running
+exchange.resume(await deliver(response));
+await exchange; // body and descendants have settled
+```
+
+- `offer()` and `resume()` each accept one call; a second throws `TypeError`.
+- Cancellation rejects a pending `offer()` with its reason. An offer after cancellation still reaches `receive()` but rejects for the body. A later `resume()` is discarded.
+- A body that returns without offering fails. A Job that closes without offering rejects `receive()` with its failure.
+- `receive()` rejects self/ancestor observation synchronously.
+
 ## Supervision
 
 A `Supervisor` manages an ordinary Job supplied by the caller. It directs submissions to that Job and applies failure policy. The application decides when initialization has finished and when to submit subsequent work.

src/execution/state.ts

@@ -7,7 +7,7 @@ import type { Job } from "../job/job.js";
  */
 export interface RuntimeState {
   readonly context: ExecutionContext;
-  readonly job: Job<unknown, unknown>;
+  readonly job: Job<unknown>;
 }
 
 const storage = new AsyncLocalStorage<RuntimeState>();

src/index.ts

@@ -24,6 +24,8 @@ export {
 } from "./execution/context/access.js";
 
 export { Job } from "./job/job.js";
-export type { JobPublisher, JobStartOptions, JobState, JobResult } from "./job/job.js";
+export type { JobStartOptions, JobState, JobResult } from "./job/job.js";
+export { HandoffJob } from "./job/handoff-job.js";
+export type { Handoff } from "./job/handoff.js";
 export { TaskGroup } from "./supervisor/task-group.js";
 export type { TaskGroupOptions, GroupMember, GroupResults } from "./supervisor/task-group.js";

src/job/handoff-job.ts

@@ -0,0 +1,34 @@
+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. */
+export class HandoffJob<Result, Offered, Resumed> extends Job<Result> {
+  private readonly rendezvous: HandoffState<Offered, Resumed>;
+
+  constructor(
+    body: (handoff: Handoff<Offered, Resumed>) => Result | PromiseLike<Result>,
+    seed?: ExecutionSeed,
+  ) {
+    if (typeof body !== "function") {
+      throw new TypeError("HandoffJob requires a body.");
+    }
+    const rendezvous = new HandoffState<Offered, Resumed>();
+    super(() => body(rendezvous), seed);
+    this.rendezvous = rendezvous;
+    this.attach(rendezvous);
+  }
+
+  /** The offered value, or the Job's failure if it closes without offering. */
+  receive(): Promise<Offered> {
+    const dependency = this.dependency("receive");
+    if (dependency) {
+      throw dependency;
+    }
+    return this.rendezvous.receive();
+  }
+
+  resume(value: Resumed): void {
+    this.rendezvous.resume(value);
+  }
+}

src/job/handoff.ts

@@ -0,0 +1,64 @@
+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>;
+}
+
+export interface OwnedHandoff {
+  readonly offered: boolean;
+  release(reason: unknown): void;
+  settle(failure: unknown): void;
+}
+
+export class HandoffState<Offered, Resumed> implements Handoff<Offered, Resumed>, OwnedHandoff {
+  private readonly value = Promise.withResolvers<Offered>();
+  private readonly answer = Promise.withResolvers<Resumed>();
+  offered = false;
+  private resumed = false;
+  private received = false;
+  private released = false;
+  private releaseReason: unknown;
+
+  offer(value: Offered): Promise<Resumed> {
+    if (this.offered) {
+      throw new TypeError("Handoff.offer() may only be called once.");
+    }
+    this.offered = true;
+    this.value.resolve(value);
+    if (this.released) {
+      this.answer.reject(this.releaseReason);
+    }
+    return this.answer.promise;
+  }
+
+  receive(): Promise<Offered> {
+    this.received = true;
+    return this.value.promise;
+  }
+
+  resume(value: Resumed): void {
+    if (this.resumed) {
+      throw new TypeError("Handoff.resume() may only be called once.");
+    }
+    this.resumed = true;
+    this.answer.resolve(value);
+  }
+
+  release(reason: unknown): void {
+    if (this.released) {
+      return;
+    }
+    this.released = true;
+    this.releaseReason = reason;
+    if (this.offered) {
+      this.answer.reject(reason);
+    }
+  }
+
+  settle(failure: unknown): void {
+    this.offered = true;
+    this.value.reject(failure);
+    if (!this.received) {
+      void this.value.promise.catch(() => {});
+    }
+  }
+}

src/job/job.ts

@@ -6,9 +6,10 @@ import { isCancellation } from "./abort.js";
 import { CancellationBindings } from "./cancellation-bindings.js";
 import { FailureSet } from "./failure-set.js";
 import { peekState, runWith, withoutExecution } from "../execution/state.js";
+import type { OwnedHandoff } from "./handoff.js";
 
 export interface JobStartOptions {
-  readonly parent?: Job<unknown, unknown>;
+  readonly parent?: Job<unknown>;
   readonly context?: ExecutionContext;
   readonly propagation?: "propagate" | "isolate";
 }
@@ -18,55 +19,34 @@ export type JobResult<T> =
   | { readonly ok: true; readonly value: T }
   | { readonly ok: false; readonly error: unknown };
 
-export type JobPublisher<T> = (value: T | PromiseLike<T>) => void;
-
 const CHILD_FAILED = new DOMException("A child job failed", "AbortError");
 
-/** A cold, single-use execution that may publish before its full lifetime settles. */
-export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
+/** A cold, single-use execution that settles once its body and descendants finish. */
+export class Job<T> implements PromiseLike<T>, AsyncDisposable {
   private readonly controller = new AbortController();
   private readonly settled = Promise.withResolvers<JobResult<unknown>>();
-  private publication?: PromiseWithResolvers<JobResult<unknown>>;
-  private publicationResult?: JobResult<unknown>;
-  private publishing = false;
-  private children: Set<Job<unknown, unknown>> | undefined;
-  private owner: Job<unknown, unknown> | undefined;
+  private children: Set<Job<unknown>> | undefined;
+  private owner: Job<unknown> | undefined;
   private executionContext: ExecutionContext | undefined;
   private phase: JobState = "created";
   private propagation: "propagate" | "isolate" = "propagate";
-  private supervisor: Supervisor<unknown, unknown> | undefined;
+  private supervisor: Supervisor<unknown> | undefined;
   private propagateFailureToParent = false;
   private failures: FailureSet | undefined;
   private cancellation: CancellationBindings | undefined;
   private closing: Promise<void> | undefined;
+  private handoff: OwnedHandoff | undefined;
 
   constructor(
-    private readonly body: (publish: JobPublisher<Published>) => T | PromiseLike<T>,
+    private readonly body: () => T | PromiseLike<T>,
     private readonly seed?: ExecutionSeed,
   ) {
     if (typeof body !== "function") {
       throw new TypeError("Job requires a body.");
     }
   }
 
-  /**
-   * Publish one value while the body is running, independently of Job cancellation.
-   */
-  private readonly publish = (value: unknown | PromiseLike<unknown>): void => {
-    if (this.phase !== "running") {
-      throw new LifecycleStateError("Job", "publish", this.phase);
-    }
-    if (this.publishing) {
-      throw new TypeError("Job.publish() may only be called once.");
-    }
-    this.publishing = true;
-    void Promise.resolve(value).then(
-      (published) => this.resolvePublication({ ok: true, value: published }),
-      (error: unknown) => this.resolvePublication({ ok: false, error }),
-    );
-  };
-
-  get parent(): Job<unknown, unknown> | undefined {
+  get parent(): Job<unknown> | undefined {
     return this.owner;
   }
 
@@ -111,7 +91,7 @@ export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
     return this.children?.size ?? 0;
   }
 
-  owns(other: Job<unknown, unknown> | undefined): boolean {
+  owns(other: Job<unknown> | undefined): boolean {
     for (let node = other; node; node = node.owner) {
       if (node === this) {
         return true;
@@ -121,13 +101,21 @@ export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
   }
 
   /** Bind an optional manager before activation; ownership remains on this Job. */
-  manage(supervisor: Supervisor<unknown, unknown>): void {
+  manage(supervisor: Supervisor<unknown>): void {
     if (this.phase !== "created" || this.supervisor) {
       throw new LifecycleStateError("Job", "manage", this.phase);
     }
     this.supervisor = supervisor;
   }
 
+  /** @internal Bind a HandoffJob's rendezvous before activation. */
+  attach(handoff: OwnedHandoff): void {
+    if (this.phase !== "created" || this.handoff) {
+      throw new LifecycleStateError("Job", "attach", this.phase);
+    }
+    this.handoff = handoff;
+  }
+
   start(options: JobStartOptions = {}): this {
     if (this.phase !== "created") {
       throw new LifecycleStateError("Job", "start", this.phase);
@@ -230,7 +218,11 @@ export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
       value = await runWith({ job: this, context: this.context }, async () => {
         try {
           this.signal.throwIfAborted();
-          return await this.body(this.publish);
+          const value = await this.body();
+          if (this.handoff && !this.handoff.offered) {
+            throw new TypeError("HandoffJob body completed without offering a value.");
+          }
+          return value;
         } catch (error) {
           // Capture before a caller can cancel with this same value.
           this.recordFailure(error);
@@ -294,7 +286,7 @@ export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
   }
 
   cancel(reason?: unknown): void {
-    const jobs: Job<unknown, unknown>[] = [this];
+    const jobs: Job<unknown>[] = [this];
     const reasons: unknown[] = [reason];
     while (jobs.length > 0) {
       const job = jobs.pop()!;
@@ -303,14 +295,15 @@ export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
         continue;
       }
       job.controller.abort(received);
+      job.handoff?.release(job.signal.reason);
       for (const child of job.children ?? []) {
         jobs.push(child);
         reasons.push(job.signal.reason);
       }
     }
   }
 
-  private dependency(operation: string): LifecycleDependencyError | undefined {
+  protected dependency(operation: string): LifecycleDependencyError | undefined {
     return this.owns(peekState()?.job)
       ? new LifecycleDependencyError("Job", operation, "Job")
       : undefined;
@@ -348,29 +341,6 @@ export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
     return this.settled.promise as Promise<JobResult<T>>;
   }
 
-  /**
-   * Observe the body-published value without joining the Job's descendants.
-   *
-   * A Job that closes without publishing rejects with its failure, or with a
-   * lifecycle error when the Job itself succeeded.
-   */
-  value(): Promise<Published> {
-    const dependency = this.dependency("value");
-    if (dependency) {
-      throw dependency;
-    }
-    const publication = (this.publication ??= Promise.withResolvers<JobResult<unknown>>());
-    if (this.publicationResult) {
-      publication.resolve(this.publicationResult);
-    }
-    return publication.promise.then((result) => {
-      if (!result.ok) {
-        throw result.error;
-      }
-      return result.value as Published;
-    });
-  }
-
   joinChildren(): Promise<void> {
     const current = peekState()?.job;
     if (current !== this && this.owns(current)) {
@@ -440,24 +410,14 @@ export class Job<T, Published = T> implements PromiseLike<T>, AsyncDisposable {
     return promise;
   }
 
-  private resolvePublication(result: JobResult<unknown>): void {
-    this.publicationResult = result;
-    this.publication?.resolve(result);
-  }
-
   private complete(result: JobResult<unknown>): void {
     this.phase = "closed";
-    if (!this.publishing) {
-      this.publishing = true;
-      this.resolvePublication(
-        result.ok
-          ? { ok: false, error: new LifecycleStateError("Job", "value", this.phase) }
-          : result,
-      );
-    }
     this.cancellation?.[Symbol.dispose]();
     this.cancellation = undefined;
     this.owner?.children?.delete(this);
+    if (this.handoff && !this.handoff.offered && !result.ok) {
+      this.handoff.settle(result.error);
+    }
     this.settled.resolve(result);
   }