delivery: unify background completion inbox contract

docs/design/COMPLETION_INBOX.md

@@ -0,0 +1,77 @@
+# Background Completion Inbox
+
+- Status: validated
+- Created: 2026-09-04
+- Verified: 2026-09-04
+- Source boundary: OpenPI implementation and tests in the pull request that closes issue #160
+- Related issue: https://github.com/openpi-dev/openpi/issues/160
+- Related pull request: https://github.com/openpi-dev/openpi/pull/382
+- Supersedes: the three producer-local consumption maps, not their execution state machines
+
+## Boundary
+
+Direct Subagent, Background Terminal, and Workflow keep independent execution
+lifecycles and canonical terminal records. The shared completion inbox is a
+small in-process delivery mechanism. It does not own execution status, result
+bytes, artifacts, cancellation, Goals, Tasks, model judgment, or UI state.
+
+The four relevant projections stay distinct:
+
+1. `SubagentSnapshot`, `TerminalSnapshot`, and `WorkflowDetails` are canonical
+   execution facts.
+2. `CompletionEnvelope` carries only delivery identity, owner, producer,
+   terminal reference, wake policy, and an in-process payload.
+3. Pi's existing `followUp` and `nextTurn` messages create model-visible
+   context.
+4. Existing TUI and Web surfaces render producer state; they do not infer
+   completion from the inbox.
+
+This uses Pi's SessionManager identity and message-delivery APIs rather than
+adding a second Session mailbox, scheduler, or agent runtime.
+
+## Envelope and ownership
+
+Every envelope has a stable `deliveryId`, `{sessionId, epoch}` owner,
+`producer`, `producerId`, `terminalRef`, and producer-selected `wake` policy.
+All producers observing the same Pi SessionManager object share one
+process-local epoch. Replacing that object, or changing its Session id, creates
+a new epoch. A claim with a missing or mismatched owner becomes an inspectable
+dead letter and is never redirected to another transcript.
+
+Workflow alone has durable producer state. Its delivery owner is persisted in
+`WorkflowDetails`. On restart, restoring a pending terminal artifact is the
+explicit revival boundary: the same Session id is rebound to the current
+process-local epoch and persisted; a different Session remains pending in
+canonical Workflow state and is dead-lettered by the inbox. Direct Subagents
+and Background Terminals do not survive Pi's `session_shutdown`, so their
+process-local inbox entries are cleared with their existing runtime lifecycle.
+
+## Consumption and receipts
+
+Pending and in-flight maps form one atomic consumption gate. Explicit
+`wait`/`status` consumption and automatic delivery race on that gate; the first
+claim wins. A transport failure restores the exact in-flight envelopes ahead
+of newer work. Independent batches can be acknowledged or retried without
+overwriting each other.
+
+The contract deliberately does not claim distributed exactly-once delivery:
+
+- Direct and Background transports consume after synchronous acceptance and
+  restore on synchronous rejection.
+- Workflow keeps its stable per-run receipt and durable at-least-once recovery.
+  If transport succeeds but receipt persistence fails, the same delivery id
+  may replay so the parent can identify it.
+- Partial Workflow receipts retry only unacknowledged siblings.
+
+Wake behavior remains producer-owned: Direct always follows up at its parent
+boundary, Background preserves its idle-follow-up versus busy-next-turn
+policy, and Workflow preserves its client-aware delivery adapter.
+
+## Validation
+
+Focused tests cover atomic explicit/automatic consumption, independent
+in-flight batches, retry ordering, stable producer identities, partial
+receipts, persistence failure, same-Session revival, Session id and epoch
+switches, owner loss, and dead letters. Existing integration tests retain the
+producer-specific wake, shutdown, process-restart, and canonical-state
+behavior. Repository gates are `bun run check` and `bun run test` on Node 22.

docs/design/README.md

@@ -13,5 +13,6 @@ These records predate [`Decision 0001`](../decisions/0001-documentation-and-evid
 - [`WORKFLOW_INVOCATION_GRAPH.md`](WORKFLOW_INVOCATION_GRAPH.md) — durable invocation facts, same-run handoff refs, reusable operators, and derived graph semantics
 - [`CHILD_TOOL_ACTIVITY.md`](CHILD_TOOL_ACTIVITY.md) — shared compact and Pi-native expanded evidence projection for Direct Subagent and Workflow child transcripts
 - [`OPENPI_WEB_ARCHITECTURE.md`](OPENPI_WEB_ARCHITECTURE.md) — draft architecture, protocol boundaries, delivery phases, and visual direction for the local Web workbench
+- [`COMPLETION_INBOX.md`](COMPLETION_INBOX.md) — shared owner, epoch, consumption, retry, and receipt contract for background completions
 
 开发与热更新流程见 [`docs/development/OPENPI_WEB_DEVELOPMENT.md`](../development/OPENPI_WEB_DEVELOPMENT.md)。

extensions/background-terminals/index.ts

@@ -32,6 +32,7 @@ import {
   OPENPI_TOOL_SURFACE,
   patchOwnedTools,
 } from "../shared/tool-surface.ts";
+import { completionOwnerFor } from "../shared/completion-inbox.ts";
 import {
   projectBackgroundTerminalCapability,
   registerWebCapability,
@@ -105,7 +106,12 @@ export default function (pi: ExtensionAPI) {
   let ui: ExtensionUIContext | undefined;
   let unsubStatus: (() => void) | undefined;
   let startReservations = 0;
-  const resultDelivery = createDeferredResultDelivery<TerminalSnapshot>();
+  const resultDelivery = createDeferredResultDelivery<TerminalSnapshot>({
+    owner: () =>
+      sessionContext
+        ? completionOwnerFor(sessionContext.sessionManager)
+        : undefined,
+  });
   const hideLifecycleTools = () =>
     patchOwnedTools(pi, "background", {
       disable: OPENPI_TOOL_SURFACE.background.deferred,
@@ -245,6 +251,7 @@ export default function (pi: ExtensionAPI) {
   const flushResults = (wake: boolean) => {
     const snaps = resultDelivery.drain(MAX_RUNNING);
     if (!deliverResults(snaps, wake)) resultDelivery.restore(snaps);
+    else resultDelivery.acknowledge(snaps);
   };
 
   const idleResultBatcher = createIdleResultBatcher({

extensions/background-terminals/src/result-delivery.ts

@@ -1,44 +1,64 @@
 import type { ConsumableResultDeliveryQueue } from "../../shared/result-delivery.ts";
+import {
+  type CompletionOwner,
+  createCompletionInbox,
+} from "../../shared/completion-inbox.ts";
 
 /**
- * Deferred one-shot delivery map (same semantics as subagents'): a settled
- * terminal's result is held here until it is either drained into a follow-up
- * message or consumed by a tool call (bg_kill / bg_status) that already
- * returned the settlement itself. Keyed by id, so double delivery is
- * structurally impossible — whoever drains first wins.
+ * Deferred one-shot delivery adapter (same semantics as subagents'): a
+ * settled terminal's result is held in the shared inbox until it is either
+ * drained into a follow-up message or consumed by a tool call (bg_kill /
+ * bg_status) that already returned the settlement itself. Stable ids make
+ * double delivery structurally impossible — whoever claims first wins.
  */
-export function createDeferredResultDelivery<T extends { id: string }>() {
-  const pending = new Map<string, T>();
+export function createDeferredResultDelivery<T extends { id: string }>(
+  options: { readonly owner?: () => CompletionOwner | undefined } = {},
+) {
+  const inbox = createCompletionInbox<T>();
+  const owner = options.owner ?? (() => ({ sessionId: "test", epoch: 0 }));
 
   const queue = {
     defer(result: T) {
-      pending.set(result.id, result);
-      return pending.size;
+      const currentOwner = owner();
+      inbox.defer(
+        {
+          deliveryId: `background:${result.id}`,
+          owner: currentOwner ?? { sessionId: "unowned", epoch: 0 },
+          producer: "background",
+          producerId: result.id,
+          terminalRef: { kind: "terminal-snapshot", id: result.id },
+          wake: "producer-policy",
+          payload: result,
+        },
+        currentOwner,
+      );
+      return inbox.size();
     },
     consume(ids: Iterable<string>) {
-      for (const id of ids) pending.delete(id);
+      inbox.consume("background", ids);
     },
     drain(maxResults = Number.POSITIVE_INFINITY) {
-      const results: T[] = [];
-      for (const [id, result] of pending) {
-        if (results.length >= maxResults) break;
-        results.push(result);
-        pending.delete(id);
-      }
-      return results;
+      return inbox
+        .claim(owner(), maxResults)
+        .map((envelope) => envelope.payload);
     },
     restore(results: readonly T[]) {
-      const current = [...pending.values()];
-      pending.clear();
-      for (const result of results) pending.set(result.id, result);
-      for (const result of current) pending.set(result.id, result);
+      inbox.retryClaimed(
+        "background",
+        results.map((result) => result.id),
+        owner(),
+      );
+    },
+    acknowledge(results: readonly T[]) {
+      inbox.acknowledge(results.map((result) => `background:${result.id}`));
     },
     size() {
-      return pending.size;
+      return inbox.size();
     },
     clear() {
-      pending.clear();
+      inbox.clear();
     },
+    inspectDeadLetters: inbox.inspectDeadLetters,
   };
   return queue satisfies ConsumableResultDeliveryQueue<T>;
 }

extensions/shared/completion-inbox.ts

@@ -0,0 +1,188 @@
+export type CompletionProducer = "subagent" | "workflow" | "background";
+export type CompletionWakePolicy =
+  | "follow-up"
+  | "next-turn"
+  | "producer-policy";
+
+export interface CompletionOwner {
+  readonly sessionId: string;
+  readonly epoch: number;
+}
+
+export interface CompletionSessionIdentity {
+  getSessionId(): string;
+}
+
+/** Transport metadata only; producer state remains the terminal authority. */
+export interface CompletionEnvelope<T> {
+  readonly deliveryId: string;
+  readonly owner: CompletionOwner;
+  readonly producer: CompletionProducer;
+  readonly producerId: string;
+  readonly terminalRef: unknown;
+  readonly wake: CompletionWakePolicy;
+  readonly payload: T;
+}
+
+export interface CompletionDeadLetter {
+  readonly deliveryId: string;
+  readonly producer: CompletionProducer;
+  readonly producerId: string;
+  readonly failure: "owner-unavailable" | "stale-owner";
+}
+
+function sameOwner(left: CompletionOwner, right: CompletionOwner) {
+  return left.sessionId === right.sessionId && left.epoch === right.epoch;
+}
+
+let nextOwnerEpoch = 1;
+const ownerBySessionIdentity = new WeakMap<object, CompletionOwner>();
+
+/**
+ * Bind one process-local generation to a Pi SessionManager identity.
+ *
+ * All OpenPI producers observing the same manager share an owner. A replaced
+ * manager, or a manager whose Session id changes, receives a new epoch so late
+ * callbacks cannot target the replacement transcript.
+ */
+export function completionOwnerFor(
+  identity: CompletionSessionIdentity,
+): CompletionOwner {
+  const sessionId = identity.getSessionId();
+  const existing = ownerBySessionIdentity.get(identity);
+  if (existing?.sessionId === sessionId) return existing;
+  const owner = { sessionId, epoch: nextOwnerEpoch++ };
+  ownerBySessionIdentity.set(identity, owner);
+  return owner;
+}
+
+/**
+ * One atomic consumption gate shared by all background producers.
+ *
+ * Claim removes before transport; a failed transport retries the exact
+ * envelopes. A successful transport leaves them consumed. No execution facts
+ * or result bytes are stored here.
+ */
+export function createCompletionInbox<T>() {
+  const pending = new Map<string, CompletionEnvelope<T>>();
+  const inFlight = new Map<string, CompletionEnvelope<T>>();
+  const deadLetters: CompletionDeadLetter[] = [];
+
+  const reject = (
+    envelope: CompletionEnvelope<T>,
+    failure: CompletionDeadLetter["failure"],
+  ) => {
+    deadLetters.push({
+      deliveryId: envelope.deliveryId,
+      producer: envelope.producer,
+      producerId: envelope.producerId,
+      failure,
+    });
+    return false;
+  };
+
+  const admit = (
+    envelope: CompletionEnvelope<T>,
+    owner: CompletionOwner | undefined,
+  ) => {
+    if (!owner) return reject(envelope, "owner-unavailable");
+    if (!sameOwner(envelope.owner, owner)) {
+      return reject(envelope, "stale-owner");
+    }
+    pending.set(envelope.deliveryId, envelope);
+    return true;
+  };
+
+  /** Restore a failed attempt ahead of completions that arrived meanwhile. */
+  const retry = (
+    envelopes: readonly CompletionEnvelope<T>[],
+    owner: CompletionOwner | undefined,
+  ) => {
+    const current = [...pending.values()];
+    pending.clear();
+    for (const envelope of envelopes) {
+      inFlight.delete(envelope.deliveryId);
+      admit(envelope, owner);
+    }
+    for (const envelope of current) admit(envelope, owner);
+  };
+
+  return {
+    defer(envelope: CompletionEnvelope<T>, owner: CompletionOwner | undefined) {
+      return admit(envelope, owner);
+    },
+
+    /** Explicit status/wait and automatic delivery atomically race here. */
+    consume(producer: CompletionProducer, producerIds: Iterable<string>) {
+      const ids = new Set(producerIds);
+      for (const [deliveryId, envelope] of pending) {
+        if (envelope.producer === producer && ids.has(envelope.producerId)) {
+          pending.delete(deliveryId);
+          inFlight.delete(deliveryId);
+        }
+      }
+    },
+
+    consumeDeliveryIds(deliveryIds: Iterable<string>) {
+      for (const deliveryId of deliveryIds) {
+        pending.delete(deliveryId);
+        inFlight.delete(deliveryId);
+      }
+    },
+
+    claim(
+      owner: CompletionOwner | undefined,
+      maximum = Number.POSITIVE_INFINITY,
+    ) {
+      const claimed: CompletionEnvelope<T>[] = [];
+      for (const [deliveryId, envelope] of pending) {
+        if (claimed.length >= maximum) break;
+        pending.delete(deliveryId);
+        if (!owner) {
+          reject(envelope, "owner-unavailable");
+          continue;
+        }
+        if (!sameOwner(envelope.owner, owner)) {
+          reject(envelope, "stale-owner");
+          continue;
+        }
+        inFlight.set(deliveryId, envelope);
+        claimed.push(envelope);
+      }
+      return claimed;
+    },
+
+    retry,
+
+    retryClaimed(
+      producer: CompletionProducer,
+      producerIds: Iterable<string>,
+      owner: CompletionOwner | undefined,
+    ) {
+      const ids = new Set(producerIds);
+      const envelopes = [...inFlight.values()].filter(
+        (envelope) =>
+          envelope.producer === producer && ids.has(envelope.producerId),
+      );
+      retry(envelopes, owner);
+    },
+
+    acknowledge(deliveryIds: Iterable<string>) {
+      for (const deliveryId of deliveryIds) inFlight.delete(deliveryId);
+    },
+
+    size() {
+      return pending.size;
+    },
+
+    inspectDeadLetters() {
+      return [...deadLetters];
+    },
+
+    clear() {
+      pending.clear();
+      inFlight.clear();
+      deadLetters.length = 0;
+    },
+  };
+}