Merge v3 and retain v3 packaging with the kache-compatible shell

host/js/README.md

@@ -1,7 +1,7 @@
 # Host JavaScript
 
-The console uses pi-durable in V8. Rust owns the SQLite connection.
-The console runs V8 on a dedicated current-thread Tokio runtime.
+The headless harness uses pi-durable in V8. Rust owns the SQLite connection.
+The harness runs V8 on a dedicated current-thread Tokio runtime.
 This satisfies Deno's runtime requirement without a change to the CLI runtime.
 The database lives at `<dir>/host/harness.sqlite`, outside the world.
 WAL preserves atomic transactions after SIGKILL. `synchronous=FULL` flushes each commit before SQLite reports success.
@@ -28,7 +28,13 @@ npm run --prefix host/js bundle
 CARGO_TARGET_DIR=target/harness nix develop -c cargo test -p sausage --test harness
 ```
 
-The lock file pins npm dependencies. The checked-in bundles let Cargo build without npm or network access for JavaScript.
+The lock file pins npm dependencies. The bundle command requires Nushell, GNU patch, and npm.
+It applies `patches/pi-durable-1.5.0.patch` before esbuild runs.
+This is Ultimate's patch at `7ec0b4f`, without the Node execution environment hunks (D65).
+The patch adds transactional `beforeInput`, input markers, scheduler hold/release, and the failed-run boundary fix.
+The step verifies an existing application before another bundle build. A mismatched or partially applied patch fails the command.
+`npm ci` restores the unpatched dependency. The next bundle command applies the patch again.
+The checked-in bundles let Cargo build without npm or network access for JavaScript.
 Cargo still needs its normal Rust dependencies and the V8 archive.
 
 For Nix, use a separate npm derivation with `importNpmLock` and the checked-in lock file.
@@ -52,3 +58,25 @@ A decided task admits one plain-language input and completes in one SQLite commi
 The input starts a new turn when the conversation is idle. Neither the input nor the task contains a secret.
 The task and its input survive SIGKILL. The input uses a stable request id to prevent duplicate admission.
 Hosts with a shared credentials handle can use `run_module_with_credentials`.
+
+## Channel input and shutdown
+
+`run_module_with_input` accepts a receiver from `harness::input::channel()` and runs without console input or output.
+Its caller supplies the world, process interface, shared credentials handle, and bundled module.
+`Input::Submit` contains a conversation id, JSON content, and a request id.
+The harness admits that input into the specified conversation. The request id preserves submission deduplication.
+Headless admission does not wait for a model answer.
+The harness groups queued follow-ups into one new run.
+`Input::Stop` waits for admitted foreground work, then closes the harness cleanly.
+The JavaScript `serve` function implements this channel protocol.
+The main module exports `storage`, including its `scanSubmissions` method.
+
+`run_console` connects stdin and stop signals to the same channel.
+The console submits to the durable root conversation (#1).
+EOF closes only the stdin client. `sausage run` stays active until SIGINT or SIGTERM.
+The signal client sends `Input::Stop`. The harness closes its watches, credential coordinator, and SQLite connection.
+The host then stops its process services and HTTP listener.
+Console output still streams assistant messages, without replay of the historical transcript.
+
+`harness_foundation.rs` runs the three Ultimate input tests and the scheduler phase-boundary test through bundled V8 modules.
+It also tests the channel with closed stdin, CLI EOF and SIGTERM, submission scans, and the patched bundle hooks.

host/js/apply-patch.nu

@@ -0,0 +1,15 @@
+# Apply the versioned dist patch, or verify that all its hunks already exist.
+let args = [-p1 -d node_modules/@earendil-works/pi-durable -i ../../../patches/pi-durable-1.5.0.patch --batch --fuzz=0]
+let forward = (^patch ...$args --forward --dry-run | complete)
+if $forward.exit_code == 0 {
+    ^patch ...$args --forward
+    if $env.LAST_EXIT_CODE != 0 { error make {msg: "pi-durable patch failed"} }
+} else {
+    let reverse = (^patch ...$args --force --reverse --dry-run | complete)
+    if $reverse.exit_code != 0 {
+        print $forward.stdout
+        print --stderr $forward.stderr
+        error make {msg: "pi-durable 1.5.0 patch does not match the installed package"}
+    }
+    print "pi-durable patch already applied"
+}

host/js/bundle.js

@@ -30494,8 +30494,8 @@ var CachedStorage = class {
   #submissions = new Lru(256);
   /** Per record with a read in flight: the reads, and how many published commits wrote it since the first began. */
   #reads = /* @__PURE__ */ new Map();
-  constructor(storage) {
-    this.#storage = storage;
+  constructor(storage2) {
+    this.#storage = storage2;
   }
   /** The backend's own scope for a publication (`Storage.publishing`), when it has one. */
   get publishing() {
@@ -30686,10 +30686,10 @@ var CachedStorage = class {
     return this.#reads.size;
   }
 };
-function scopeOf(storage) {
-  if (storage.publishing === void 0)
+function scopeOf(storage2) {
+  if (storage2.publishing === void 0)
     return void 0;
-  return (publish2, context) => storage.publishing(publish2, context);
+  return (publish2, context) => storage2.publishing(publish2, context);
 }
 var Lru = class {
   #map = /* @__PURE__ */ new Map();
@@ -30877,8 +30877,8 @@ function tableKeys(writes, keys) {
 var GatedStorage = class {
   #storage;
   #touch;
-  constructor(storage, touch) {
-    this.#storage = storage;
+  constructor(storage2, touch) {
+    this.#storage = storage2;
     this.#touch = touch;
   }
   commit(batches, receive, context) {
@@ -30954,24 +30954,24 @@ var ROOT_CONVERSATION_ID = 1;
 
 // node_modules/@earendil-works/pi-durable/dist/session/forks.js
 var SCAN_PAGE_SIZE = 256;
-async function prepareForkDocumentCopies(storage, parentConversationId, at, childConversationId, context) {
-  const entry2 = await storage.entry(parentConversationId, at, context);
+async function prepareForkDocumentCopies(storage2, parentConversationId, at, childConversationId, context) {
+  const entry2 = await storage2.entry(parentConversationId, at, context);
   if (entry2 === void 0)
     throw new Error(`Entry ${at} is not visible from conversation ${parentConversationId}`);
   const copies = [];
   const copiedAddresses = /* @__PURE__ */ new Set();
-  await collectCopies(storage, { kind: "conversation", conversationId: entry2.entry.conversationId }, entry2.commitSeq, "asOf", childConversationId, copies, copiedAddresses, context);
-  await collectCopies(storage, { kind: "conversation", conversationId: parentConversationId }, "current", "current", childConversationId, copies, copiedAddresses, context);
+  await collectCopies(storage2, { kind: "conversation", conversationId: entry2.entry.conversationId }, entry2.commitSeq, "asOf", childConversationId, copies, copiedAddresses, context);
+  await collectCopies(storage2, { kind: "conversation", conversationId: parentConversationId }, "current", "current", childConversationId, copies, copiedAddresses, context);
   return copies;
 }
-async function collectCopies(storage, scope, at, policy, childConversationId, copies, copiedAddresses, context) {
+async function collectCopies(storage2, scope, at, policy, childConversationId, copies, copiedAddresses, context) {
   let cursor;
   do {
-    const page2 = await storage.scanDocuments({ scope, at }, SCAN_PAGE_SIZE, cursor, context);
+    const page2 = await storage2.scanDocuments({ scope, at }, SCAN_PAGE_SIZE, cursor, context);
     for (const source of page2.items) {
       if (source.scope.kind !== "conversation" || source.fork !== policy)
         continue;
-      const id = await storage.mintId();
+      const id = await storage2.mintId();
       const identity = {
         id,
         kind: source.kind,
@@ -31140,6 +31140,9 @@ var Transaction = class {
     this.#assertOpen();
     return record;
   }
+  beforeInput(conversationId, input) {
+    return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
+  }
   appendEntry(first, second, third) {
     if (typeof first === "number")
       return this.#appendEntry(first, second);
@@ -31612,7 +31615,7 @@ var Transaction = class {
     return { observed: publications, stored: stored ?? publications };
   }
   async #assemble() {
-    const storage = this.#host.storage;
+    const storage2 = this.#host.storage;
     const plans = this.#plans;
     for (const document of this.#documents) {
       const plan = planDocument(document);
@@ -31667,7 +31670,7 @@ var Transaction = class {
         }
         let cursor;
         do {
-          const page2 = await storage.scanDocuments({ scope: { kind: "task", taskId }, at: "current" }, INTERNAL_SCAN_PAGE_SIZE, cursor, this.#context);
+          const page2 = await storage2.scanDocuments({ scope: { kind: "task", taskId }, at: "current" }, INTERNAL_SCAN_PAGE_SIZE, cursor, this.#context);
           for (const record of page2.items)
             retire(record);
           cursor = page2.next;
@@ -31691,7 +31694,7 @@ var Transaction = class {
       plan.conversationId = task.publicationConversationId;
     }
     for (const { id, change } of this.#submissionChanges) {
-      const current = this.#submissions.get(id) ?? await storage.submission(id, this.#context);
+      const current = this.#submissions.get(id) ?? await storage2.submission(id, this.#context);
       if (current === void 0)
         throw new Error(`Submission ${id} does not exist`);
       const next = applySubmissionChange(current, change);
@@ -32262,12 +32265,12 @@ var SessionImpl = class {
   get failed() {
     return this.#failed.promise;
   }
-  constructor(storage, now = Date.now, documentRetentionMs = () => void 0, setTimer, readOnly = false, persistWithinMs = () => void 0) {
+  constructor(storage2, now = Date.now, documentRetentionMs = () => void 0, setTimer, readOnly = false, persistWithinMs = () => void 0) {
     this.#failed.promise.catch(() => {
     });
     this.#readOnly = readOnly;
-    this.#cached = readOnly ? void 0 : new CachedStorage(storage);
-    this.#storage = new GuardedStorage(this.#cached ?? storage, this);
+    this.#cached = readOnly ? void 0 : new CachedStorage(storage2);
+    this.#storage = new GuardedStorage(this.#cached ?? storage2, this);
     this.#gated = new GatedStorage(this.#storage, (key) => this.touch(key));
     this.#writer = new Writer({
       commit: (batches, receive, context) => this.#storage.commit(batches, receive, context),
@@ -32322,6 +32325,7 @@ var SessionImpl = class {
         if (this.#documents.get(id)?.record.id === recordId)
           this.#dropDocument(id);
       },
+      beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
       conversationCreated: (tx, record) => this.conversationCreated(tx, record),
       taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
       latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -32646,6 +32650,9 @@ var SessionImpl = class {
    */
   report(_error, _info = LISTENER, _context) {
   }
+  async beforeInput(_tx, _id, _input, _context) {
+    return [];
+  }
   /**
    * Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
    * plain Session stages nothing; a Harness stages its built-in documents.
@@ -33561,8 +33568,8 @@ var GuardedStorage = class {
   /** Backend calls underway, also those off the Session line, which close waits for before it closes the backend. */
   #underway = /* @__PURE__ */ new Set();
   #closing = false;
-  constructor(storage, latch) {
-    this.#storage = storage;
+  constructor(storage2, latch) {
+    this.#storage = storage2;
     this.#latch = latch;
   }
   /** The backend's own scope for a publication (`Storage.publishing`), when it has one; not a call to guard. */
@@ -34072,21 +34079,21 @@ function defineTask(definition) {
 var SCAN_PAGE_SIZE2 = 256;
 var EXCLUDED_STOP_REASONS = /* @__PURE__ */ new Set(["aborted", "error", "deferred"]);
 var MISSING_RESULT_TEXT = "Tool result unavailable: history ends before this call completed.";
-async function captureContextBounds(session, storage, conversationId, context, at, visible = false) {
+async function captureContextBounds(session, storage2, conversationId, context, at, visible = false) {
   let tail;
   if (at === void 0) {
     tail = await session.latestEntryId(conversationId, context);
     if (tail === void 0)
       return void 0;
   } else {
-    if (!visible && await storage.entry(conversationId, at, context) === void 0) {
+    if (!visible && await storage2.entry(conversationId, at, context) === void 0) {
       throw new Error(`Entry ${at} is not visible from conversation ${conversationId}`);
     }
     tail = at;
   }
   return { head: await session.latestHeadMarker(conversationId, tail, context), tail };
 }
-async function readContextFrom(session, storage, conversationId, context, at, previous) {
+async function readContextFrom(session, storage2, conversationId, context, at, previous) {
   const visible = at !== void 0 && previous !== void 0 && rangeEntry(previous, at) !== void 0;
   const bounds = await session.readOnLine((line) => captureContextBounds(session, line, conversationId, context, at, visible));
   if (bounds === void 0)
@@ -34095,7 +34102,7 @@ async function readContextFrom(session, storage, conversationId, context, at, pr
   const current = at === void 0;
   let range;
   if (previous === void 0 || previous.bounds.head?.id !== bounds.head?.id) {
-    const entries2 = freezeJson(await scanRange(storage, conversationId, rangeQuery(bounds), context));
+    const entries2 = freezeJson(await scanRange(storage2, conversationId, rangeQuery(bounds), context));
     range = deriveRange(bounds, entries2, current);
   } else if (bounds.tail === previous.bounds.tail) {
     range = previous;
@@ -34103,7 +34110,7 @@ async function readContextFrom(session, storage, conversationId, context, at, pr
     range = deriveRange(bounds, previous.entries.filter((entry2) => entry2.id <= bounds.tail), false);
   } else {
     const minEntryId = idFromNumber(previous.bounds.tail + 1);
-    const added = await scanRange(storage, conversationId, { minEntryId, maxEntryId: bounds.tail }, context);
+    const added = await scanRange(storage2, conversationId, { minEntryId, maxEntryId: bounds.tail }, context);
     range = extendRange(previous, bounds, freezeJson(added), current);
   }
   if (range.current && session.knownLatestEntryId(conversationId) !== range.bounds.tail) {
@@ -34120,14 +34127,14 @@ async function readContextFrom(session, storage, conversationId, context, at, pr
     range
   };
 }
-async function readContext(session, storage, conversationId, context, at) {
+async function readContext(session, storage2, conversationId, context, at) {
   const bounds = await session.readOnLine((line) => captureContextBounds(session, line, conversationId, context, at));
-  return deriveContext(storage, conversationId, bounds, context);
+  return deriveContext(storage2, conversationId, bounds, context);
 }
-async function deriveContext(storage, conversationId, bounds, context) {
+async function deriveContext(storage2, conversationId, bounds, context) {
   if (bounds === void 0)
     return emptyView();
-  return deriveRange(bounds, await scanRange(storage, conversationId, rangeQuery(bounds), context), false).view;
+  return deriveRange(bounds, await scanRange(storage2, conversationId, rangeQuery(bounds), context), false).view;
 }
 function emptyView() {
   return { head: void 0, entries: [], contributions: [], messages: [] };
@@ -34245,11 +34252,11 @@ function rangeQuery(bounds) {
   const head = bounds.head;
   return head === void 0 ? { maxEntryId: bounds.tail } : { minEntryId: head.head, maxEntryId: bounds.tail };
 }
-async function scanRange(storage, conversationId, range, context) {
+async function scanRange(storage2, conversationId, range, context) {
   const entries2 = [];
   let cursor;
   do {
-    const page2 = await storage.scanEntries({ conversationId, ...range }, SCAN_PAGE_SIZE2, cursor, context);
+    const page2 = await storage2.scanEntries({ conversationId, ...range }, SCAN_PAGE_SIZE2, cursor, context);
     entries2.push(...page2.items);
     cursor = page2.next;
   } while (cursor !== void 0);
@@ -34403,10 +34410,14 @@ var InboxDoc = defineDoc({
   initial: () => ({ items: [] }),
   checkpointWhen: (value2) => value2.items.length === 0
 });
-async function prepareBoundary(tx, conversationId, modes) {
+async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
   const head = (await tx.latestHeadMarker(conversationId))?.head;
   const inbox = await tx.doc(InboxDoc, conversationId);
+  const first = inbox.items.find((item) => item.mode === "steer" || item.mode === "followUp");
+  const input = first === void 0 ? incoming : inputDraft(first.content, first.entry, now);
+  const entries2 = input !== void 0 && (at === "final" || hasQueuedReset(inbox.items)) ? await tx.beforeInput(conversationId, input) : [];
   return {
+    entries: entries2,
     conversationId,
     inbox,
     steeringMode: modes.steeringMode,
@@ -34442,6 +34453,9 @@ async function applyBoundary(tx, boundary, at, now) {
   }
   const placed = [];
   for (const index of users) {
+    if (placed.length === 0)
+      for (const draft of boundary.entries)
+        await tx.appendEntry(conversationId, draft);
     const item = items[index];
     const entry2 = await appendInput(tx, conversationId, item.content, item.entry, now);
     tx.placeSubmission(item.id, entry2.id);
@@ -34454,10 +34468,11 @@ async function applyBoundary(tx, boundary, at, now) {
   return { users: placed, first, reset };
 }
 function appendInput(tx, conversationId, content, entry2, now) {
+  return tx.appendEntry(conversationId, inputDraft(content, entry2, now));
+}
+function inputDraft(content, entry2, now) {
   const message = { role: "user", content, timestamp: now };
-  if (entry2 === void 0)
-    return tx.appendEntry(UserEntry, conversationId, { model: [message] });
-  return tx.appendEntry(conversationId, { ...pickInputEntry(entry2), model: [message] });
+  return { ...entry2 === void 0 ? { kind: UserEntry.kind } : pickInputEntry(entry2), input: true, model: [message] };
 }
 function pickInputEntry({ kind, data }) {
   const picked = data === void 0 ? { kind } : { kind, data };
@@ -35177,6 +35192,7 @@ var TaskScheduler = class {
   #unsubscribeRegistry = () => {
   };
   #enabled = false;
+  #held = false;
   #readOnly;
   #closing = false;
   #dirty = false;
@@ -35205,6 +35221,14 @@ var TaskScheduler = class {
     this.#context = options.context;
     this.#readOnly = options.readOnly === true;
   }
+  /** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
+  hold() {
+    this.#held = true;
+  }
+  release() {
+    this.#held = false;
+    this.#kick();
+  }
   /**
    * Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
    * a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -35394,13 +35418,13 @@ var TaskScheduler = class {
     return marked.result;
   }
   async waitForTask(id, context) {
-    const found = await this.#session.readOnLine(async (storage) => {
+    const found = await this.#session.readOnLine(async (storage2) => {
       if (this.#closing)
         throw closedError(this.#session);
       if (await this.#region(() => this.#liveRecord(id) !== void 0)) {
         return { promise: this.#taskWaiters.add(id, context) };
       }
-      const record = await storage.task(id, context);
+      const record = await storage2.task(id, context);
       if (record === void 0)
         throw new Error(`Task ${id} does not exist`);
       return { promise: Promise.resolve(record) };
@@ -36659,6 +36683,8 @@ var TaskScheduler = class {
         for (const record of plan.ready) {
           const id = record.id;
           const mode = record.abortRequested ? "abort" : "run";
+          if (mode === "run" && this.#held)
+            continue;
           const cutOff = record.restarts !== void 0 && record.restarts >= this.#restartLimit();
           if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
             const message = mode === "abort" ? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times` : `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
@@ -36844,7 +36870,7 @@ var TaskScheduler = class {
     const tasks = [];
     for (const record of this.#live.values())
       tasks.push({ record, state: this.#inspectTask(record, snapshot) });
-    const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
+    const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
     return { scheduling, tasks };
   }
   #inspectTask(record, snapshot) {
@@ -36999,6 +37025,10 @@ var TaskScheduler = class {
    * invocation continues.
    */
   #progress(tx, current, state2, extension, context) {
+    if (this.#held) {
+      tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
+      return false;
+    }
     state2.snapshot = this.#registry.snapshot();
     const next = state2.snapshot.task(current.kind);
     if (next === state2.task)
@@ -37366,7 +37396,7 @@ var TaskScheduler = class {
       watchDoc: ((...args) => this.#watchDoc(invocation, args)),
       snapshot: ((...args) => this.#read(invocation, () => sessionMethod(this.#session, "snapshot")(...args))),
       snapshotAsOf: ((...args) => this.#read(invocation, () => sessionMethod(this.#session, "snapshotAsOf")(...args))),
-      getTask: ((id, context) => this.#read(invocation, () => this.#session.readOnLine((storage) => storage.task(id, context)))),
+      getTask: ((id, context) => this.#read(invocation, () => this.#session.readOnLine((storage2) => storage2.task(id, context)))),
       waitForTask: ((id, context) => this.#read(invocation, () => this.waitForTask(id, withAbortSignal(invocation.controller.signal, context)))),
       ownedTasks: (context) => this.#read(invocation, () => this.#session.readOnLine(async () => {
         if (invocation.ended)
@@ -37386,7 +37416,7 @@ var TaskScheduler = class {
       })),
       abortOwned: (id, context) => this.#read(invocation, async () => {
         const bound = withAbortSignal(invocation.controller.signal, context);
-        const record = await this.#session.readOnLine((storage) => storage.task(id, bound));
+        const record = await this.#session.readOnLine((storage2) => storage2.task(id, bound));
         if (record?.owner !== invocation.taskId) {
           throw new Error(`Task ${id} is not owned by task ${invocation.taskId}`);
         }
@@ -37395,10 +37425,10 @@ var TaskScheduler = class {
         await this.abort(id, bound, true);
         await this.waitForTask(id, bound);
       }),
-      outcomes: ((ids, context) => this.#read(invocation, () => this.#session.readOnLine(async (storage) => {
+      outcomes: ((ids, context) => this.#read(invocation, () => this.#session.readOnLine(async (storage2) => {
         const outcomes = [];
         for (const id of ids) {
-          const state2 = (await storage.task(id, context))?.state;
+          const state2 = (await storage2.task(id, context))?.state;
           if (state2?.status !== "terminal")
             throw new Error(`Task ${id} is not terminal`);
           outcomes.push(state2.outcome);
@@ -37416,11 +37446,11 @@ var TaskScheduler = class {
         const [token, id, context] = args.length === 2 ? [void 0, args[0], args[1]] : [args[0], args[1], args[2]];
         return this.#read(invocation, async () => {
           const kept = this.#contexts.get(invocation.conversationId);
-          const entry2 = (kept === void 0 ? void 0 : rangeEntry(kept.range, id)) ?? (await this.#session.readOnLine((storage) => storage.entry(invocation.conversationId, id, context)))?.entry;
+          const entry2 = (kept === void 0 ? void 0 : rangeEntry(kept.range, id)) ?? (await this.#session.readOnLine((storage2) => storage2.entry(invocation.conversationId, id, context)))?.entry;
           return token === void 0 || entry2?.kind === token.kind ? entry2 : void 0;
         });
       }),
-      entries: (query, limit2, cursor, context) => this.#read(invocation, () => this.#session.readOnLine((storage) => storage.scanEntries(conversationEntries(invocation.conversationId, query), limit2, cursor, context))),
+      entries: (query, limit2, cursor, context) => this.#read(invocation, () => this.#session.readOnLine((storage2) => storage2.scanEntries(conversationEntries(invocation.conversationId, query), limit2, cursor, context))),
       context: (conversationId, context, options) => this.#read(invocation, async () => {
         const read2 = await readContextFrom(this.#session, this.#storage, conversationId, context, options?.at, this.#contexts.get(conversationId)?.range);
         const { view: view2 } = read2;
@@ -39732,7 +39762,7 @@ async function prepareFailure(tx, runtime) {
   const policy = settings.afterFailedRun;
   if (policy !== "continue")
     return { policy };
-  return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
+  return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
 }
 async function endFailedRun(tx, runtime, live, failure, end) {
   const conversationId = runtime.conversationId;
@@ -39900,7 +39930,7 @@ async function answer(runtime, message, context) {
   const afterRun = await asksAfterRun(runtime, context);
   const conversationId = runtime.conversationId;
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
     const live = await tx.doc(LiveDoc, conversationId);
     const entry2 = await appendAssistant(tx, conversationId, message);
     const result = { status: "terminal", outcome: { status: "completed", result: { entryId: entry2.id } } };
@@ -39973,7 +40003,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
 async function endAnsweredRun(runtime, answer2, context) {
   const conversationId = runtime.conversationId;
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
     const live = await tx.doc(LiveDoc, conversationId);
     const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
     endRun(tx, live, runtime.taskId, { status: "done", answer: answer2 });
@@ -40046,7 +40076,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
   const maxTurns = (await runtime.agent(context)).maxTurns;
   const afterRun = terminate && await asksAfterRun(runtime, context);
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== void 0 ? "final" : "postTools", runtime.now());
     if (added.length > 0)
       await addTools(tx, conversationId, added);
     const live = await tx.doc(LiveDoc, conversationId);
@@ -40354,19 +40384,19 @@ var Submissions = class {
   }
   /** Handle for an existing submission, or `undefined`. */
   async get(id, context) {
-    const record = await this.#session.readOnLine((storage) => storage.submission(id, context));
+    const record = await this.#session.readOnLine((storage2) => storage2.submission(id, context));
     return record === void 0 ? void 0 : new SubmissionHandle(record.id, this);
   }
   async status(id, context) {
-    const record = await this.#session.readOnLine((storage) => storage.submission(id, context));
+    const record = await this.#session.readOnLine((storage2) => storage2.submission(id, context));
     if (record === void 0)
       throw new Error(`Submission ${id} does not exist`);
     return record;
   }
   async wait(id, context) {
     this.#resume();
-    const found = await this.#session.readOnLine(async (storage) => {
-      const record = await storage.submission(id, context);
+    const found = await this.#session.readOnLine(async (storage2) => {
+      const record = await storage2.submission(id, context);
       if (record === void 0)
         throw new Error(`Submission ${id} does not exist`);
       if (isSettled(record))
@@ -40463,13 +40493,13 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
   if (draft.type === "continue") {
     if (busy)
       throw new ConversationBusy(conversationId);
-    const boundary2 = await prepareBoundary(tx, conversationId, queueModes);
+    const boundary2 = await prepareBoundary(tx, conversationId, queueModes, "final", now);
     const { id: id2 } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
     const { users, first } = await applyBoundary(tx, boundary2, "final", now);
     await startRun(tx, conversationId, live, { inputs: [...users, id2], first: live.runFailure?.first ?? first });
     return { id: id2, status: "placed" };
   }
-  const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes);
+  const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : void 0);
   if (boundary === void 0 || boundary.inbox.items.length > 0) {
     const { id: id2 } = await tx.createSubmission({
       ...fields,
@@ -40507,6 +40537,8 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
     const write = { ...fields, type: "write", status: "done", entry: entry3.id };
     return { id: (await tx.createSubmission(write)).id, status: "done" };
   }
+  for (const before of boundary.entries)
+    await tx.appendEntry(conversationId, before);
   const entry2 = await appendInput(tx, conversationId, draft.content, draft.entry, now);
   const input = { ...fields, type: "input", status: "placed", entry: entry2.id };
   const { id } = await tx.createSubmission(input);
@@ -41037,9 +41069,9 @@ var TaskGraphView = class {
   }
   /** Register an observer created from the current revision, atomically on the Session line. */
   #attach(create, context) {
-    return this.#session.readOnLine(async (storage) => {
+    return this.#session.readOnLine(async (storage2) => {
       const mount = this.#mount ?? {
-        ...await this.#build(storage, context),
+        ...await this.#build(storage2, context),
         observers: /* @__PURE__ */ new Set()
       };
       const detach = () => {
@@ -41076,16 +41108,16 @@ var TaskGraphView = class {
     });
   }
   /** Read the graph on the line, through the line's `storage`. */
-  async #build(storage, context) {
+  async #build(storage2, context) {
     const tasks = {};
     const live = /* @__PURE__ */ new Map();
     const records = [];
     for (const status of LIVE_STATUSES2) {
-      records.push(...await scanAll((cursor) => storage.scanTasks({ status }, SCAN_PAGE_SIZE4, cursor, context)));
+      records.push(...await scanAll((cursor) => storage2.scanTasks({ status }, SCAN_PAGE_SIZE4, cursor, context)));
     }
     records.sort((a, b) => a.id - b.id);
     for (const record of records) {
-      const owned = await scanAll((cursor) => storage.scanConversations({ ownerTaskId: record.id }, SCAN_PAGE_SIZE4, cursor, context));
+      const owned = await scanAll((cursor) => storage2.scanConversations({ ownerTaskId: record.id }, SCAN_PAGE_SIZE4, cursor, context));
       const conversations = owned.map((conversation) => conversation.id).sort((a, b) => a - b);
       const subject = blockedSubject(record);
       tasks[String(record.id)] = nodeOf2(record, conversations, this.#blocking.blocked(subject));
@@ -41262,9 +41294,9 @@ var ConversationViews = class {
    */
   async attach(id, window2, create, context) {
     const bounds = resolveWindow(window2);
-    return this.#session.readOnLine(async (storage) => {
-      const mount = this.#mounts.get(id) ?? await this.#build(storage, id, context);
-      const { entries: entries2, before } = await loadWindow(storage, id, bounds, context);
+    return this.#session.readOnLine(async (storage2) => {
+      const mount = this.#mounts.get(id) ?? await this.#build(storage2, id, context);
+      const { entries: entries2, before } = await loadWindow(storage2, id, bounds, context);
       const value2 = view(mount.conversation, entries2, before, mount.docs);
       let windowed;
       const detach = () => {
@@ -41284,8 +41316,8 @@ var ConversationViews = class {
     });
   }
   /** Hydrate a mount on the line, through the line's `storage`. */
-  async #build(storage, id, context) {
-    const conversation = await storage.conversation(id, context);
+  async #build(storage2, id, context) {
+    const conversation = await storage2.conversation(id, context);
     if (conversation === void 0)
       throw new Error(`Conversation ${id} does not exist`);
     const docs = {};
@@ -41303,25 +41335,25 @@ var ConversationViews = class {
 function view(conversation, entries2, before, docs) {
   return before === void 0 ? { conversation, entries: entries2, docs } : { conversation, entries: entries2, before, docs };
 }
-async function loadWindow(storage, conversationId, bounds, context) {
+async function loadWindow(storage2, conversationId, bounds, context) {
   const { from, min, max } = bounds;
   if (from !== void 0) {
-    const since = await scanUpTo(storage, { conversationId, minEntryId: from, order: "ascending" }, max + 1, context);
+    const since = await scanUpTo(storage2, { conversationId, minEntryId: from, order: "ascending" }, max + 1, context);
     if (since.length <= max) {
       const query = { conversationId, maxEntryId: idFromNumber(from - 1) };
-      const previous = (await storage.scanEntries(query, 1, void 0, context)).items[0];
+      const previous = (await storage2.scanEntries(query, 1, void 0, context)).items[0];
       return { entries: since, before: previous?.id };
     }
   }
-  const newest = await scanUpTo(storage, { conversationId }, min + 1, context);
+  const newest = await scanUpTo(storage2, { conversationId }, min + 1, context);
   const before = newest.length > min ? newest.pop().id : void 0;
   return { entries: newest.reverse(), before };
 }
-async function scanUpTo(storage, query, limit2, context) {
+async function scanUpTo(storage2, query, limit2, context) {
   const entries2 = [];
   let cursor;
   do {
-    const page2 = await storage.scanEntries(query, Math.min(SCAN_PAGE_SIZE5, limit2 - entries2.length), cursor, context);
+    const page2 = await storage2.scanEntries(query, Math.min(SCAN_PAGE_SIZE5, limit2 - entries2.length), cursor, context);
     entries2.push(...page2.items);
     cursor = page2.next;
   } while (cursor !== void 0 && entries2.length < limit2);
@@ -41441,7 +41473,7 @@ var ConversationImpl = class {
   }
   entries(query, limit2, cursor, context) {
     const scan = conversationEntries(this.id, query);
-    return this.#host.harness.readOnLine((storage) => storage.scanEntries(scan, limit2, cursor, context));
+    return this.#host.harness.readOnLine((storage2) => storage2.scanEntries(scan, limit2, cursor, context));
   }
   fork(at, options, context) {
     return this.#host.create({ kind: "fork", parentId: this.id, at, ownership: options.ownership }, options, context);
@@ -41482,7 +41514,7 @@ var HarnessImpl = class extends SessionImpl {
     const now = safeNow(options.now, report2);
     const hostSettings = currentSettings(options.settings);
     super(rawStorage, now, () => hostSettings()?.documentRetentionMs, options.setTimer, options.readOnly === true, () => hostSettings()?.progress?.persistWithinMs);
-    const storage = this.storage;
+    const storage2 = this.storage;
     this.#options = options;
     this.#report = report2;
     this.#hostSettings = hostSettings;
@@ -41490,7 +41522,7 @@ var HarnessImpl = class extends SessionImpl {
     this.#tasks = new TaskScheduler({
       telemetry: options.telemetry,
       session: this,
-      storage,
+      storage: storage2,
       registry: options.registry,
       models: currentModels(options.models, options.identity),
       agent: (id, snapshot, callContext) => this.resolveAgent(id, snapshot, callContext),
@@ -41518,7 +41550,7 @@ var HarnessImpl = class extends SessionImpl {
     });
     this.#host = {
       harness: this,
-      storage,
+      storage: storage2,
       tasks: this.#tasks,
       submissions: this.#submissions,
       views: new ConversationViews(this),
@@ -41526,6 +41558,15 @@ var HarnessImpl = class extends SessionImpl {
       create: (target, createOptions, context2) => this.#create(target, createOptions, context2)
     };
   }
+  async beforeInput(tx, conversationId, input, context) {
+    const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
+    const entries2 = [];
+    for (const { handlers } of agentHooks(agent, "pi.generation")) {
+      const result = await handlers.beforeInput?.(input, { conversationId, tx }, context);
+      entries2.push(...result?.entries ?? []);
+    }
+    return entries2;
+  }
   /** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
   async resolveAgent(id, snapshot, context) {
     const registry2 = snapshot ?? this.#options.registry.snapshot();
@@ -41548,12 +41589,18 @@ var HarnessImpl = class extends SessionImpl {
     this.assertUsable();
     this.#tasks.resume();
   }
+  hold() {
+    this.#tasks.hold();
+  }
+  release() {
+    this.#tasks.release();
+  }
   wake() {
     this.assertUsable();
     return this.#tasks.wake();
   }
   getTask(id, context) {
-    return this.readOnLine((storage) => storage.task(id, context));
+    return this.readOnLine((storage2) => storage2.task(id, context));
   }
   /** Internal, for tests: sizes of the scheduler's ownership indexes. */
   get [SCHEDULER_INDEX_SIZES]() {
@@ -41564,9 +41611,9 @@ var HarnessImpl = class extends SessionImpl {
     return this.#tasks.idleState;
   }
   inspect(context) {
-    return this.readOnLine(async (storage) => {
+    return this.readOnLine(async (storage2) => {
       const { scheduling, tasks } = await this.#tasks.inspect(this.#options.registry.snapshot());
-      const scan = (status) => scanAll((cursor) => storage.scanSubmissions({ status }, SCAN_PAGE_SIZE6, cursor, context));
+      const scan = (status) => scanAll((cursor) => storage2.scanSubmissions({ status }, SCAN_PAGE_SIZE6, cursor, context));
       const submissions = [...await scan("queued"), ...await scan("placed")].sort((a, b) => a.id - b.id);
       return { scheduling, tasks, submissions };
     });
@@ -41590,7 +41637,7 @@ var HarnessImpl = class extends SessionImpl {
   }
   /** Sum every conversation's committed `pi.usage`. Each document is read at its own point; totals only grow. */
   async usage(context) {
-    const conversations = await this.readOnLine((storage) => scanAll((cursor) => storage.scanConversations({}, SCAN_PAGE_SIZE6, cursor, context)));
+    const conversations = await this.readOnLine((storage2) => scanAll((cursor) => storage2.scanConversations({}, SCAN_PAGE_SIZE6, cursor, context)));
     const total = UsageDoc.definition.initial();
     for (const { id } of conversations) {
       const state2 = await this.snapshot(UsageDoc, id, context);
@@ -41615,7 +41662,7 @@ var HarnessImpl = class extends SessionImpl {
   }
   async conversation(id, context) {
     this.assertUsable();
-    const record = await this.readOnLine((storage) => storage.conversation(id, context));
+    const record = await this.readOnLine((storage2) => storage2.conversation(id, context));
     return record === void 0 ? void 0 : new ConversationImpl(record.id, this.#host);
   }
   createConversation(options, context) {
@@ -41746,7 +41793,7 @@ function boundConversation(id, binding, submissions, tasks) {
 }
 var Harness = {
   /** Open a Harness over storage. The registry may keep changing while the Harness runs. */
-  async open(storage, options, context) {
+  async open(storage2, options, context) {
     context.abortSignal?.throwIfAborted();
     const snapshot = options.registry.snapshot();
     const missing = BUILTIN_TASKS.filter((task) => snapshot.task(task.definition.name) === void 0);
@@ -41754,7 +41801,7 @@ var Harness = {
       const names2 = missing.map((task) => task.definition.name).join(", ");
       throw new Error(`Registry lacks built-in tasks ${names2}; create it with createRegistry()`);
     }
-    const harness2 = new HarnessImpl(storage, options, context);
+    const harness2 = new HarnessImpl(storage2, options, context);
     try {
       await harness2.openTasks(context);
     } catch (error) {
@@ -44397,7 +44444,7 @@ function environment(cwd = "/", conversation) {
         }
       };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, conversation);
+        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, conversation === void 0 ? void 0 : String(conversation));
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result;
         while (true) {
@@ -44482,6 +44529,29 @@ async function startRequests(harness2, context) {
   };
 }
 
+// input.ts
+async function serve(harness2, context, afterSubmit) {
+  while (true) {
+    const input = await Deno.core.ops.op_input();
+    if (input === null || input.type === "stop") {
+      await harness2.waitForIdle(context);
+      return;
+    }
+    const conversation = await harness2.conversation(input.conversation, context);
+    if (!conversation) throw new Error(`Unknown conversation: ${input.conversation}`);
+    const submission2 = await conversation.submit({
+      type: "input",
+      content: input.content,
+      requestId: input.requestId
+    }, context);
+    if (afterSubmit) {
+      const result = await submission2.wait(context);
+      await afterSubmit();
+      if (result.status !== "done") throw new Error(`Run ended: ${JSON.stringify(result)}`);
+    }
+  }
+}
+
 // main.ts
 var proxy = Deno.core.ops.op_model_proxy();
 if (!proxy?.base_url || !proxy?.api_key) throw new Error("The llm-proxy credential is required");
@@ -44516,9 +44586,11 @@ models.setProvider({
 var registry = createRegistry();
 registry.install(CodingTools);
 registry.install(CredentialRequests);
-var harness = await Harness.open(await openStorage(globalThis.storagePath), {
+var storage = await openStorage(globalThis.storagePath);
+var harness = await Harness.open(storage, {
   models,
   registry,
+  settings: { followUpMode: "all" },
   env: (target) => environment(target.cwd ?? "/", target.conversationId)
 }, BACKGROUND_CONTEXT);
 var root = await harness.root(BACKGROUND_CONTEXT, { agent: { model: { provider: "openai", modelId: model2.id }, thinkingLevel: "low", cwd: "/", extensions: [CodingTools, CredentialRequests] } });
@@ -44543,25 +44615,21 @@ var render = (view2) => {
 for (const entry2 of watch.value.entries) {
   if (entry2.kind === "pi.assistant") emitted.set(entry2.id, text(entry2.model?.[0]));
 }
-watch.start((event) => {
+if (globalThis.consoleEnabled) watch.start((event) => {
   if (event.kind === "value") render(event.value);
 });
 var stopRequests = await startRequests(harness, BACKGROUND_CONTEXT);
 harness.resume();
-await harness.waitForIdle(BACKGROUND_CONTEXT);
-while (true) {
-  const line = await Deno.core.ops.op_input();
-  if (line === null) break;
-  if (!line.trim()) continue;
-  const submission2 = await root.submit({ type: "input", content: line, requestId: crypto.randomUUID() }, BACKGROUND_CONTEXT);
-  const result = await submission2.wait(BACKGROUND_CONTEXT);
+await serve(harness, BACKGROUND_CONTEXT, globalThis.consoleEnabled ? async () => {
   await harness.waitForIdle(BACKGROUND_CONTEXT);
   const final = await root.watch(BACKGROUND_CONTEXT);
   render(final.value);
   await final.stop();
   Deno.core.ops.op_output("\n");
-  if (result.status !== "done") throw new Error(`Run ended: ${JSON.stringify(result)}`);
-}
+} : void 0);
 await stopRequests();
 await watch.stop();
 await harness.close(BACKGROUND_CONTEXT);
+export {
+  storage
+};

host/js/durable-test-bundle.js

@@ -8883,6 +8883,9 @@ var Transaction = class {
     this.#assertOpen();
     return record;
   }
+  beforeInput(conversationId, input) {
+    return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
+  }
   appendEntry(first, second, third) {
     if (typeof first === "number")
       return this.#appendEntry(first, second);
@@ -10065,6 +10068,7 @@ var SessionImpl = class {
         if (this.#documents.get(id)?.record.id === recordId)
           this.#dropDocument(id);
       },
+      beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
       conversationCreated: (tx, record) => this.conversationCreated(tx, record),
       taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
       latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -10389,6 +10393,9 @@ var SessionImpl = class {
    */
   report(_error, _info = LISTENER, _context) {
   }
+  async beforeInput(_tx, _id, _input, _context) {
+    return [];
+  }
   /**
    * Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
    * plain Session stages nothing; a Harness stages its built-in documents.
@@ -12193,10 +12200,14 @@ var InboxDoc = defineDoc({
   initial: () => ({ items: [] }),
   checkpointWhen: (value2) => value2.items.length === 0
 });
-async function prepareBoundary(tx, conversationId, modes) {
+async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
   const head = (await tx.latestHeadMarker(conversationId))?.head;
   const inbox = await tx.doc(InboxDoc, conversationId);
+  const first = inbox.items.find((item) => item.mode === "steer" || item.mode === "followUp");
+  const input = first === void 0 ? incoming : inputDraft(first.content, first.entry, now);
+  const entries2 = input !== void 0 && (at === "final" || hasQueuedReset(inbox.items)) ? await tx.beforeInput(conversationId, input) : [];
   return {
+    entries: entries2,
     conversationId,
     inbox,
     steeringMode: modes.steeringMode,
@@ -12232,6 +12243,9 @@ async function applyBoundary(tx, boundary, at, now) {
   }
   const placed = [];
   for (const index of users) {
+    if (placed.length === 0)
+      for (const draft of boundary.entries)
+        await tx.appendEntry(conversationId, draft);
     const item = items[index];
     const entry2 = await appendInput(tx, conversationId, item.content, item.entry, now);
     tx.placeSubmission(item.id, entry2.id);
@@ -12244,10 +12258,11 @@ async function applyBoundary(tx, boundary, at, now) {
   return { users: placed, first, reset };
 }
 function appendInput(tx, conversationId, content, entry2, now) {
+  return tx.appendEntry(conversationId, inputDraft(content, entry2, now));
+}
+function inputDraft(content, entry2, now) {
   const message = { role: "user", content, timestamp: now };
-  if (entry2 === void 0)
-    return tx.appendEntry(UserEntry, conversationId, { model: [message] });
-  return tx.appendEntry(conversationId, { ...pickInputEntry(entry2), model: [message] });
+  return { ...entry2 === void 0 ? { kind: UserEntry.kind } : pickInputEntry(entry2), input: true, model: [message] };
 }
 function pickInputEntry({ kind, data }) {
   const picked = data === void 0 ? { kind } : { kind, data };
@@ -12959,6 +12974,7 @@ var TaskScheduler = class {
   #unsubscribeRegistry = () => {
   };
   #enabled = false;
+  #held = false;
   #readOnly;
   #closing = false;
   #dirty = false;
@@ -12987,6 +13003,14 @@ var TaskScheduler = class {
     this.#context = options.context;
     this.#readOnly = options.readOnly === true;
   }
+  /** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
+  hold() {
+    this.#held = true;
+  }
+  release() {
+    this.#held = false;
+    this.#kick();
+  }
   /**
    * Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
    * a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -14441,6 +14465,8 @@ var TaskScheduler = class {
         for (const record of plan.ready) {
           const id = record.id;
           const mode = record.abortRequested ? "abort" : "run";
+          if (mode === "run" && this.#held)
+            continue;
           const cutOff = record.restarts !== void 0 && record.restarts >= this.#restartLimit();
           if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
             const message = mode === "abort" ? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times` : `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
@@ -14626,7 +14652,7 @@ var TaskScheduler = class {
     const tasks = [];
     for (const record of this.#live.values())
       tasks.push({ record, state: this.#inspectTask(record, snapshot) });
-    const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
+    const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
     return { scheduling, tasks };
   }
   #inspectTask(record, snapshot) {
@@ -14781,6 +14807,10 @@ var TaskScheduler = class {
    * invocation continues.
    */
   #progress(tx, current, state, extension, context) {
+    if (this.#held) {
+      tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
+      return false;
+    }
     state.snapshot = this.#registry.snapshot();
     const next = state.snapshot.task(current.kind);
     if (next === state.task)
@@ -17429,7 +17459,7 @@ async function prepareFailure(tx, runtime) {
   const policy = settings.afterFailedRun;
   if (policy !== "continue")
     return { policy };
-  return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
+  return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
 }
 async function endFailedRun(tx, runtime, live, failure, end) {
   const conversationId = runtime.conversationId;
@@ -17597,7 +17627,7 @@ async function answer(runtime, message, context) {
   const afterRun = await asksAfterRun(runtime, context);
   const conversationId = runtime.conversationId;
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
     const live = await tx.doc(LiveDoc, conversationId);
     const entry2 = await appendAssistant(tx, conversationId, message);
     const result = { status: "terminal", outcome: { status: "completed", result: { entryId: entry2.id } } };
@@ -17670,7 +17700,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
 async function endAnsweredRun(runtime, answer2, context) {
   const conversationId = runtime.conversationId;
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
     const live = await tx.doc(LiveDoc, conversationId);
     const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
     endRun(tx, live, runtime.taskId, { status: "done", answer: answer2 });
@@ -17743,7 +17773,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
   const maxTurns = (await runtime.agent(context)).maxTurns;
   const afterRun = terminate && await asksAfterRun(runtime, context);
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== void 0 ? "final" : "postTools", runtime.now());
     if (added.length > 0)
       await addTools(tx, conversationId, added);
     const live = await tx.doc(LiveDoc, conversationId);
@@ -18159,13 +18189,13 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
   if (draft.type === "continue") {
     if (busy)
       throw new ConversationBusy(conversationId);
-    const boundary2 = await prepareBoundary(tx, conversationId, queueModes);
+    const boundary2 = await prepareBoundary(tx, conversationId, queueModes, "final", now);
     const { id: id2 } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
     const { users, first } = await applyBoundary(tx, boundary2, "final", now);
     await startRun(tx, conversationId, live, { inputs: [...users, id2], first: live.runFailure?.first ?? first });
     return { id: id2, status: "placed" };
   }
-  const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes);
+  const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : void 0);
   if (boundary === void 0 || boundary.inbox.items.length > 0) {
     const { id: id2 } = await tx.createSubmission({
       ...fields,
@@ -18203,6 +18233,8 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
     const write = { ...fields, type: "write", status: "done", entry: entry3.id };
     return { id: (await tx.createSubmission(write)).id, status: "done" };
   }
+  for (const before of boundary.entries)
+    await tx.appendEntry(conversationId, before);
   const entry2 = await appendInput(tx, conversationId, draft.content, draft.entry, now);
   const input = { ...fields, type: "input", status: "placed", entry: entry2.id };
   const { id } = await tx.createSubmission(input);
@@ -19215,6 +19247,15 @@ var HarnessImpl = class extends SessionImpl {
       create: (target, createOptions, context2) => this.#create(target, createOptions, context2)
     };
   }
+  async beforeInput(tx, conversationId, input, context) {
+    const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
+    const entries2 = [];
+    for (const { handlers } of agentHooks(agent, "pi.generation")) {
+      const result = await handlers.beforeInput?.(input, { conversationId, tx }, context);
+      entries2.push(...result?.entries ?? []);
+    }
+    return entries2;
+  }
   /** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
   async resolveAgent(id, snapshot, context) {
     const registry2 = snapshot ?? this.#options.registry.snapshot();
@@ -19237,6 +19278,12 @@ var HarnessImpl = class extends SessionImpl {
     this.assertUsable();
     this.#tasks.resume();
   }
+  hold() {
+    this.#tasks.hold();
+  }
+  release() {
+    this.#tasks.release();
+  }
   wake() {
     this.assertUsable();
     return this.#tasks.wake();

host/js/environment.ts

@@ -10,7 +10,7 @@ const path = (cwd: string, input: string) => {
   }
   return '/' + parts.join('/');
 };
-export function environment(cwd = '/', conversation?: string) {
+export function environment(cwd = '/', conversation?: string | number) {
   const call = (kind: string, p: string, bytes: number[] = []) => ops.op_file(kind, path(cwd, p), bytes);
   const env: any = {
     id: 'world', cwd,
@@ -87,7 +87,7 @@ export function environment(cwd = '/', conversation?: string) {
         }
       };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, conversation);
+        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, conversation === undefined ? undefined : String(conversation));
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result;
         while (true) {

host/js/foundation-test.ts

@@ -0,0 +1,127 @@
+// Ultimate's input-placement cases, scheduler hold, and the headless channel contract.
+import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
+import { createModels } from '@earendil-works/pi-ai/models';
+import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
+import { Harness, createRegistry, defineDoc, defineExtension, defineTask, GenerationTask, hook, InboxDoc } from '@earendil-works/pi-durable';
+import { openStorage } from './sqlite';
+import { serve } from './input';
+const assert = (value: any, message: string) => { if (!value) throw new Error(message); };
+const equal = (a: any, b: any) => assert(JSON.stringify(a) === JSON.stringify(b), `Expected ${JSON.stringify(b)}, got ${JSON.stringify(a)}`);
+const until = async (condition: () => any) => {
+  const deadline = Date.now() + 10000;
+  while (!await condition()) {
+    if (Date.now() > deadline) throw new Error('Condition deadline');
+    await new Promise(resolve => setTimeout(resolve, 0));
+  }
+};
+const Placed = defineDoc<{ count: number }>({ kind: 'input-test.placed', version: 1,
+  checkpointWhen: () => false, scope: 'conversation', history: 'latest', fork: 'initial', initial: () => ({ count: 0 }) });
+const gate = Promise.withResolvers<void>();
+let held = false, reject = false, calls = 0, second = false;
+const Notes = defineExtension({ name: 'input-test', hooks: [hook(GenerationTask, {
+  beforeInput: async (_input, { tx, conversationId }) => {
+    calls++;
+    const state = await tx.doc(Placed, conversationId);
+    state.count++;
+    if (reject) throw new Error('reject placement');
+    return { entries: [{ kind: 'input-test.note', data: { count: state.count },
+      model: [{ role: 'user', content: `before ${state.count}`, timestamp: 1 }] }] };
+  },
+})] });
+const Park = defineTask<null, { phase: 'first' | 'second' }, null>({ name: 'hold-test', version: 1,
+  initial: () => ({ phase: 'first' }), phases: {
+    first: async (_task, runtime, ctx) => {
+      held = true;
+      await gate.promise;
+      await runtime.commit(() => ({ status: 'pending', checkpoint: { phase: 'second' } }), ctx);
+    },
+    second: async (_task, runtime, ctx) => {
+      second = true;
+      await runtime.commit(() => ({ status: 'terminal', outcome: { status: 'completed', result: null } }), ctx);
+    },
+  }, abort: (_task, runtime, ctx) => runtime.commit(() => ({ status: 'terminal', outcome: { status: 'aborted' } }), ctx),
+});
+const models = createModels();
+const faux = fauxProvider({ tokensPerSecond: 100000 });
+models.setProvider(faux.provider);
+const registry = createRegistry();
+registry.install(Notes);
+registry.install(defineExtension({ name: 'hold-test', tasks: [Park] }));
+const storage = await openStorage(globalThis.storagePath);
+const harness = await Harness.open(storage, { models, registry, settings: { followUpMode: 'all' } }, context);
+const scenario = globalThis.foundationCase;
+const root = await harness.root(context, { agent: { model: { provider: 'faux', modelId: 'faux-1' },
+  extensions: scenario === 'placement' ? [Notes] : [] } });
+try {
+  if (scenario === 'placement' || scenario === 'followups') {
+    faux.setResponses([async () => {
+      held = true;
+      await gate.promise;
+      return fauxAssistantMessage([fauxToolCall('unknown', {})], { stopReason: 'toolUse' });
+    }, fauxAssistantMessage('first answer'), fauxAssistantMessage('follow-ups answer'),
+    fauxAssistantMessage('extra'), fauxAssistantMessage('extra')]);
+    const first = await root.submit({ type: 'input', content: 'first', requestId: 'one' }, context);
+    await until(() => held);
+    const queued = [];
+    const followups = scenario === 'placement' ? ['next'] : ['one', 'two', 'three'];
+    for (const content of followups) queued.push(await root.submit({ type: 'input', content, whenBusy: 'followUp' }, context));
+    const steer = await root.submit({ type: 'input', content: 'steer', whenBusy: 'steer' }, context);
+    if (scenario === 'placement') equal(calls, 1);
+    gate.resolve();
+    await Promise.all([first, steer, ...queued].map(submission => submission.wait(context)));
+    const view = await root.context(context);
+    if (scenario === 'placement') {
+      equal(view.messages.filter(m => m.role === 'user').map(m => m.content), ['before 1', 'first', 'steer', 'before 2', 'next']);
+      const notes = view.entries.filter(e => e.kind === 'input-test.note');
+      equal(notes.length, 2);
+      assert(notes.every(e => e.edits === undefined), 'Hook notes must be placed, not edits');
+      equal(calls, 2);
+      assert(view.entries.filter(e => e.kind === 'pi.user').every(e => e.input === true), 'Inputs need their marker');
+      await root.submit({ type: 'input', content: 'first', requestId: 'one' }, context);
+      equal(calls, 2);
+      reject = true;
+      let rejected = false;
+      try { await root.submit({ type: 'input', content: 'rejected' }, context); }
+      catch (error) { rejected = String(error).includes('reject placement'); }
+      assert(rejected, 'Placement must reject');
+      equal((await harness.snapshot(Placed, root.id, context))?.count, 2);
+      equal((await root.context(context)).entries.filter(e => e.kind === 'input-test.note').length, 2);
+    } else {
+      equal(view.messages.flatMap(m => m.role === 'user' ? [m.content] : m.role === 'assistant'
+        ? m.content.flatMap(p => p.type === 'text' ? [`> ${p.text}`] : []) : []),
+      ['first', 'steer', '> first answer', 'one', 'two', 'three', '> follow-ups answer']);
+      equal(await harness.snapshot(InboxDoc, root.id, context), { items: [] });
+    }
+  } else if (scenario === 'thinking') {
+    faux.setResponses([fauxAssistantMessage('unset'), fauxAssistantMessage('high')]);
+    await (await root.submit({ type: 'input', content: 'one' }, context)).wait(context);
+    await root.configure({ thinkingLevel: 'high' }, context);
+    await (await root.submit({ type: 'input', content: 'two' }, context)).wait(context);
+    equal((await root.context(context)).messages.flatMap(m => m.role === 'assistant' ? [m.thinkingLevel] : []), ['off', 'high']);
+  } else if (scenario === 'hold') {
+    const task = await root.commit(tx => tx.createTask(Park, null, { ownership: { kind: 'conversation' } }), context);
+    harness.resume();
+    await until(() => held);
+    harness.hold();
+    gate.resolve();
+    await until(async () => (await harness.getTask(task, context))?.state.status === 'pending');
+    equal((await harness.getTask(task, context))?.state.checkpoint.phase, 'second');
+    harness.resume();
+    equal((await harness.inspect(context)).scheduling, 'paused');
+    assert(!second, 'Held scheduler started another phase');
+    harness.release();
+    await until(() => second);
+    await harness.waitForIdle(context);
+    equal((await harness.getTask(task, context))?.state.status, 'terminal');
+  } else if (scenario === 'headless') {
+    faux.setResponses([fauxAssistantMessage('channel answer')]);
+    await serve(harness, context);
+    const submissions = await storage.scanSubmissions({ conversationId: root.id }, 100, undefined, context);
+    equal(submissions.items.length, 1);
+    equal(submissions.items[0].requestId, 'channel-request');
+    assert(JSON.stringify(await root.context(context)).includes('channel answer'), 'Channel answer missing');
+  } else throw new Error('Unknown foundation case');
+} finally {
+  gate.resolve();
+  await harness.close(context);
+}

host/js/input.ts

@@ -0,0 +1,20 @@
+// Channel submissions target any conversation. EOF and Stop close the harness cleanly.
+export async function serve(harness: any, context: any, afterSubmit?: () => Promise<void>) {
+  while (true) {
+    const input = await Deno.core.ops.op_input();
+    if (input === null || input.type === 'stop') {
+      await harness.waitForIdle(context);
+      return;
+    }
+    const conversation = await harness.conversation(input.conversation, context);
+    if (!conversation) throw new Error(`Unknown conversation: ${input.conversation}`);
+    const submission = await conversation.submit({
+      type: 'input', content: input.content, requestId: input.requestId,
+    }, context);
+    if (afterSubmit) {
+      const result = await submission.wait(context);
+      await afterSubmit();
+      if (result.status !== 'done') throw new Error(`Run ended: ${JSON.stringify(result)}`);
+    }
+  }
+}

host/js/main.ts

@@ -7,6 +7,7 @@ import { CodingTools } from '@earendil-works/pi-durable/tools';
 import { openStorage } from './sqlite';
 import { environment } from './environment';
 import { CredentialRequests, startRequests } from './credentials';
+import { serve } from './input';
 const proxy = Deno.core.ops.op_model_proxy();
 if (!proxy?.base_url || !proxy?.api_key) throw new Error('The llm-proxy credential is required');
 const models = createModels();
@@ -27,8 +28,10 @@ models.setProvider({ ...provider, baseUrl: model.baseUrl, getModels: () => [mode
 const registry = createRegistry();
 registry.install(CodingTools);
 registry.install(CredentialRequests);
-const harness = await Harness.open(await openStorage(globalThis.storagePath), {
-  models, registry, env: (target: any) => environment(target.cwd ?? '/', target.conversationId),
+export const storage = await openStorage(globalThis.storagePath);
+const harness = await Harness.open(storage, {
+  models, registry, settings: { followUpMode: 'all' },
+  env: (target: any) => environment(target.cwd ?? '/', target.conversationId),
 }, context);
 const root = await harness.root(context, { agent: { model: { provider: 'openai', modelId: model.id }, thinkingLevel: 'low', cwd: '/', extensions: [CodingTools, CredentialRequests] } });
 const watch = await root.watch(context);
@@ -53,23 +56,16 @@ const render = (view: any) => {
 for (const entry of watch.value.entries) {
   if (entry.kind === 'pi.assistant') emitted.set(entry.id, text(entry.model?.[0]));
 }
-watch.start((event: any) => { if (event.kind === 'value') render(event.value); });
+if (globalThis.consoleEnabled) watch.start((event: any) => { if (event.kind === 'value') render(event.value); });
 const stopRequests = await startRequests(harness, context);
 harness.resume();
-await harness.waitForIdle(context);
-while (true) {
-  const line = await Deno.core.ops.op_input();
-  if (line === null) break;
-  if (!line.trim()) continue;
-  const submission = await root.submit({ type: 'input', content: line, requestId: crypto.randomUUID() }, context);
-  const result = await submission.wait(context);
+await serve(harness, context, globalThis.consoleEnabled ? async () => {
   await harness.waitForIdle(context);
   const final = await root.watch(context);
   render(final.value);
   await final.stop();
   Deno.core.ops.op_output('\n');
-  if (result.status !== 'done') throw new Error(`Run ended: ${JSON.stringify(result)}`);
-}
+} : undefined);
 await stopRequests();
 await watch.stop();
 await harness.close(context);

host/js/p2-test.ts

@@ -0,0 +1,59 @@
+// P2 rehearsal: real bash tools pause for two passkey approvals and resume durably.
+import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
+import { createModels } from '@earendil-works/pi-ai/models';
+import { fauxProvider, fauxAssistantMessage, fauxToolCall, fauxText } from '@earendil-works/pi-ai/providers/faux';
+import { Harness, createRegistry } from '@earendil-works/pi-durable';
+import { CodingTools } from '@earendil-works/pi-durable/tools';
+import { environment } from './environment';
+import { openStorage } from './sqlite';
+import { CredentialRequests, startRequests } from './credentials';
+const models = createModels();
+const faux = fauxProvider({ tokensPerSecond: 100000 });
+models.setProvider(faux.provider);
+const bash = (command: string, id: string) => fauxAssistantMessage([fauxToolCall('bash', { command }, { id })], { stopReason: 'toolUse' });
+const curl = `curl -i http://127.0.0.1:${globalThis.p2Http}/whoami`;
+const push = 'cd /repo && git push origin main';
+faux.setResponses([
+  bash(curl, 'curl-denied'), fauxAssistantMessage([fauxText('Wait for HTTP approval.')]),
+  bash(curl, 'curl-approved'), bash(push, 'ssh-denied'), fauxAssistantMessage([fauxText('Wait for SSH approval.')]),
+  bash(push, 'ssh-approved'), fauxAssistantMessage([fauxText('P2 complete.')]),
+]);
+const registry = createRegistry();
+registry.install(CodingTools); registry.install(CredentialRequests);
+const harness = await Harness.open(await openStorage(globalThis.storagePath), {
+  models, registry, env: (target: any) => environment('/', target.conversationId),
+}, context);
+const root = await harness.root(context, { agent: {
+  model: { provider: 'faux', modelId: 'faux-1' }, extensions: [CodingTools, CredentialRequests], cwd: '/',
+} });
+const stop = await startRequests(harness, context);
+harness.resume();
+await (await root.submit({ type: 'input', content: 'Rehearse P2', requestId: 'p2' }, context)).wait(context);
+await Deno.core.ops.op_file('write', '/p2-http-idle', []);
+const watch = await root.watch(context);
+let sshIdle = false;
+await new Promise<void>((resolve, reject) => {
+  const check = async (event: any) => {
+    if (event.kind !== 'value') return;
+    const transcript = JSON.stringify(event.value.entries);
+    if (!sshIdle && transcript.includes('Wait for SSH approval.')) {
+      sshIdle = true;
+      await harness.waitForIdle(context);
+      await Deno.core.ops.op_file('write', '/p2-ssh-idle', []);
+    }
+    const approvals = event.value.entries.filter((entry: any) => JSON.stringify(entry).includes('Your request for'));
+    if (approvals.length > 2) reject(new Error('Duplicate approval wake-up'));
+    if (transcript.includes('P2 complete.')) resolve();
+  };
+  watch.start(check); check({ kind: 'value', value: watch.value });
+});
+await harness.waitForIdle(context);
+const transcript = JSON.stringify(watch.value.entries);
+for (const expected of ['403', 'demo-api', 'demo-ssh', 'Bearer accepted', 'P2 complete.']) {
+  if (!transcript.includes(expected)) throw new Error(`Missing rehearsal result: ${expected}`);
+}
+if (watch.value.entries.filter((entry: any) => JSON.stringify(entry).includes('Your request for')).length !== 2) {
+  throw new Error('Expected exactly two approval inputs');
+}
+await Deno.core.ops.op_file('write', '/p2-transcript', [...new TextEncoder().encode(transcript)]);
+await watch.stop(); await stop(); await harness.close(context);

host/js/package.json

@@ -1,7 +1,7 @@
 {
   "private": true,
   "type": "module",
-  "scripts": { "bundle": "esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js" },
+  "scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js" },
   "dependencies": {
     "@earendil-works/pi-durable": "1.5.0",
     "@earendil-works/pi-ai": "1.5.0",

host/js/patches/pi-durable-1.5.0.patch

@@ -0,0 +1,387 @@
+diff --git a/dist/harness/generation.js b/dist/harness/generation.js
+index ac42dfe741b20726ab8eb3d8139b1296d664210d..1e7f1ac26d17b83cb15bf00408fa70b0aee4e747 100644
+--- a/dist/harness/generation.js
++++ b/dist/harness/generation.js
+@@ -458,7 +458,7 @@ async function prepareFailure(tx, runtime) {
+     const policy = settings.afterFailedRun;
+     if (policy !== "continue")
+         return { policy };
+-    return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
++    return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
+ }
+ /**
+  * End a failed run, settling its inputs with `settlement`, and do with the inputs queued behind it what
+@@ -668,7 +668,7 @@ async function answer(runtime, message, context) {
+     const conversationId = runtime.conversationId;
+     await runtime.commit(async (tx) => {
+         // Queue modes are read on the Session line, when the boundary is decided.
+-        const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
++        const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
+         const live = await tx.doc(LiveDoc, conversationId);
+         const entry = await appendAssistant(tx, conversationId, message);
+         const result = { status: "terminal", outcome: { status: "completed", result: { entryId: entry.id } } };
+@@ -762,7 +762,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
+ async function endAnsweredRun(runtime, answer, context) {
+     const conversationId = runtime.conversationId;
+     await runtime.commit(async (tx) => {
+-        const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
++        const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
+         const live = await tx.doc(LiveDoc, conversationId);
+         const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
+         endRun(tx, live, runtime.taskId, { status: "done", answer });
+@@ -868,7 +868,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
+     const maxTurns = (await runtime.agent(context)).maxTurns;
+     const afterRun = terminate && (await asksAfterRun(runtime, context));
+     await runtime.commit(async (tx) => {
+-        const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
++        const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== undefined ? "final" : "postTools", runtime.now());
+         if (added.length > 0)
+             await addTools(tx, conversationId, added);
+         const live = await tx.doc(LiveDoc, conversationId);
+diff --git a/dist/harness/harness.js b/dist/harness/harness.js
+index e286f00f872ebec79367a594938040c6a37a879f..44979b05c2bcf30580841797175c099c8c621462 100644
+--- a/dist/harness/harness.js
++++ b/dist/harness/harness.js
+@@ -5,7 +5,7 @@ import { ResetEntry } from "../entries.js";
+ import { contained, SessionImpl } from "../session/session.js";
+ import { hostTimers } from "../timers.js";
+ import { ROOT_CONVERSATION_ID } from "../types.js";
+-import { AgentDoc, configure, createAgent, resolveAgent, resolveSettings } from "./agent.js";
++import { AgentDoc, agentHooks, configure, createAgent, resolveAgent, resolveSettings } from "./agent.js";
+ import { createCompaction } from "./compaction.js";
+ import { readContext } from "./context.js";
+ import { InboxDoc, withdrawQueuedInputs } from "./inbox.js";
+@@ -157,6 +157,15 @@ class HarnessImpl extends SessionImpl {
+             create: (target, createOptions, context) => this.#create(target, createOptions, context),
+         };
+     }
++    async beforeInput(tx, conversationId, input, context) {
++        const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
++        const entries = [];
++        for (const { handlers } of agentHooks(agent, "pi.generation")) {
++            const result = await handlers.beforeInput?.(input, { conversationId, tx }, context);
++            entries.push(...(result?.entries ?? []));
++        }
++        return entries;
++    }
+     /** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
+     async resolveAgent(id, snapshot, context) {
+         const registry = snapshot ?? this.#options.registry.snapshot();
+@@ -179,6 +188,12 @@ class HarnessImpl extends SessionImpl {
+         this.assertUsable();
+         this.#tasks.resume();
+     }
++    hold() {
++        this.#tasks.hold();
++    }
++    release() {
++        this.#tasks.release();
++    }
+     wake() {
+         this.assertUsable();
+         return this.#tasks.wake();
+diff --git a/dist/harness/inbox.d.ts b/dist/harness/inbox.d.ts
+index 1197339369188857a1ad62b528cea65e3833f27e..a61d9712ce55e80441847442772da6f3e354fb2b 100644
+--- a/dist/harness/inbox.d.ts
++++ b/dist/harness/inbox.d.ts
+@@ -23,6 +23,7 @@ export declare const InboxDoc: import("../types.ts").ConversationDocToken<InboxS
+ export type QueueModes = Pick<Settings, "steeringMode" | "followUpMode">;
+ /** What a boundary reads before the commit's first table write, and the newest head it has seen so far. */
+ export type Boundary = {
++    readonly entries: readonly EntryDraft[];
+     readonly conversationId: ConversationId;
+     readonly inbox: Draft<InboxState>;
+     readonly steeringMode: QueueMode;
+@@ -43,7 +44,7 @@ export type BoundaryResult = {
+  * Read what a boundary needs. Table reads must precede the commit's first table write, so callers prepare the
+  * boundary at the start of their commit.
+  */
+-export declare function prepareBoundary(tx: Tx, conversationId: ConversationId, modes: QueueModes): Promise<Boundary>;
++export declare function prepareBoundary(tx: Tx, conversationId: ConversationId, modes: QueueModes, at: "postTools" | "final", now: number, incoming?: EntryDraft): Promise<Boundary>;
+ /**
+  * Place the queued items a boundary selects (spec §6): every write, the first or all steers, and at `final` the first
+  * or all follow-ups. A selected reset turns a `postTools` boundary into `final`. Writes are placed first and user
+@@ -53,6 +54,7 @@ export declare function prepareBoundary(tx: Tx, conversationId: ConversationId,
+  */
+ export declare function applyBoundary(tx: Tx, boundary: Boundary, at: "postTools" | "final", now: number): Promise<BoundaryResult>;
+ /** Append the entry an input is placed as: `entry`'s kind and data, or `pi.user`, with the user message of `content`. */
++export declare function inputDraft(content: UserInput, entry: InputEntry | undefined, now: number): EntryDraft;
+ export declare function appendInput(tx: Tx, conversationId: ConversationId, content: UserInput, entry: InputEntry | undefined, now: number): Promise<EntryRecord>;
+ /** The kind and data of an input's entry as strict JSON, without other fields an untyped caller might pass. */
+ export declare function pickInputEntry({ kind, data }: InputEntry): JsonRepresentation<InputEntry>;
+diff --git a/dist/harness/inbox.js b/dist/harness/inbox.js
+index 5ff8c1703808c5ffd703164e6c339150e1f1e131..ec4cf6d5adc599e5fe428514a7cd7e0dac1a7c9d 100644
+--- a/dist/harness/inbox.js
++++ b/dist/harness/inbox.js
+@@ -14,10 +14,15 @@ export const InboxDoc = defineDoc({
+  * Read what a boundary needs. Table reads must precede the commit's first table write, so callers prepare the
+  * boundary at the start of their commit.
+  */
+-export async function prepareBoundary(tx, conversationId, modes) {
++export async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
+     const head = (await tx.latestHeadMarker(conversationId))?.head;
+     const inbox = await tx.doc(InboxDoc, conversationId);
++    const first = inbox.items.find(item => item.mode === "steer" || item.mode === "followUp");
++    const input = first === undefined ? incoming : inputDraft(first.content, first.entry, now);
++    const entries = input !== undefined && (at === "final" || hasQueuedReset(inbox.items))
++        ? await tx.beforeInput(conversationId, input) : [];
+     return {
++        entries,
+         conversationId,
+         inbox,
+         steeringMode: modes.steeringMode,
+@@ -61,6 +66,9 @@ export async function applyBoundary(tx, boundary, at, now) {
+     }
+     const placed = [];
+     for (const index of users) {
++        if (placed.length === 0)
++            for (const draft of boundary.entries)
++                await tx.appendEntry(conversationId, draft);
+         const item = items[index];
+         const entry = await appendInput(tx, conversationId, item.content, item.entry, now);
+         tx.placeSubmission(item.id, entry.id);
+@@ -74,10 +82,11 @@ export async function applyBoundary(tx, boundary, at, now) {
+ }
+ /** Append the entry an input is placed as: `entry`'s kind and data, or `pi.user`, with the user message of `content`. */
+ export function appendInput(tx, conversationId, content, entry, now) {
++    return tx.appendEntry(conversationId, inputDraft(content, entry, now));
++}
++export function inputDraft(content, entry, now) {
+     const message = { role: "user", content, timestamp: now };
+-    if (entry === undefined)
+-        return tx.appendEntry(UserEntry, conversationId, { model: [message] });
+-    return tx.appendEntry(conversationId, { ...pickInputEntry(entry), model: [message] });
++    return { ...(entry === undefined ? { kind: UserEntry.kind } : pickInputEntry(entry)), input: true, model: [message] };
+ }
+ /** The kind and data of an input's entry as strict JSON, without other fields an untyped caller might pass. */
+ export function pickInputEntry({ kind, data }) {
+diff --git a/dist/harness/scheduler.d.ts b/dist/harness/scheduler.d.ts
+index f6189edb71ac0557060b623625fc4be4bec5e904..269a3c4ab1676da22db5a723f30d2974f84b6bea 100644
+--- a/dist/harness/scheduler.d.ts
++++ b/dist/harness/scheduler.d.ts
+@@ -107,6 +107,8 @@ export declare class TaskScheduler {
+      * Idempotent; the kick does nothing once closing.
+      */
+     resume(): void;
++    hold(): void;
++    release(): void;
+     /**
+      * Why a live task cannot run here, if it cannot: its definition is missing, too old, or failed to migrate. A running
+      * task, an abort-marked one, and one holding its outcome are never blocked.
+diff --git a/dist/harness/scheduler.js b/dist/harness/scheduler.js
+index 365ea3ab42dd4406ab11a03d6702b238dfd0776b..3e8170fdb790e492e95a390fa6ea4693eb3d2f97 100644
+--- a/dist/harness/scheduler.js
++++ b/dist/harness/scheduler.js
+@@ -179,6 +179,7 @@ export class TaskScheduler {
+     #cascadePending = false;
+     #unsubscribeRegistry = () => { };
+     #enabled = false;
++    #held = false;
+     #readOnly;
+     #closing = false;
+     #dirty = false;
+@@ -207,6 +208,14 @@ export class TaskScheduler {
+         this.#context = options.context;
+         this.#readOnly = options.readOnly === true;
+     }
++    /** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
++    hold() {
++        this.#held = true;
++    }
++    release() {
++        this.#held = false;
++        this.#kick();
++    }
+     /**
+      * Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
+      * a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
+@@ -1771,6 +1780,8 @@ export class TaskScheduler {
+                 for (const record of plan.ready) {
+                     const id = record.id;
+                     const mode = record.abortRequested ? "abort" : "run";
++                    if (mode === "run" && this.#held)
++                        continue;
+                     // Work cut off `restartLimit` times in a row would run forever: an abort handler, or a phase of a kind that
+                     // does not end itself at the limit, ends the task `faulted` instead.
+                     // The settings are read only for a task that was cut off at all.
+@@ -1973,7 +1984,7 @@ export class TaskScheduler {
+         const tasks = [];
+         for (const record of this.#live.values())
+             tasks.push({ record, state: this.#inspectTask(record, snapshot) });
+-        const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
++        const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
+         return { scheduling, tasks };
+     }
+     #inspectTask(record, snapshot) {
+@@ -2151,6 +2162,11 @@ export class TaskScheduler {
+      * invocation continues.
+      */
+     #progress(tx, current, state, extension, context) {
++        // Held: park at this phase boundary, as a handover does; reservation starts it again after `release()`.
++        if (this.#held) {
++            tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
++            return false;
++        }
+         state.snapshot = this.#registry.snapshot();
+         const next = state.snapshot.task(current.kind);
+         if (next === state.task)
+diff --git a/dist/harness/submissions.js b/dist/harness/submissions.js
+index 6e882a8c252dda942b8b45e5b1837cf2fee28341..89a85e8a52bf67044ff15c3c856a650dea4c7c27 100644
+--- a/dist/harness/submissions.js
++++ b/dist/harness/submissions.js
+@@ -3,7 +3,7 @@ import { withTelemetry } from "@earendil-works/chord/context";
+ import { createTypedSpanStarter, NOOP_TELEMETRY_CONTEXT, } from "@earendil-works/pi-telemetry";
+ import { ConversationBusy, ConversationIdle, ReadAfterWrite } from "../errors.js";
+ import { startRun } from "./generation.js";
+-import { appendInput, applyBoundary, hasQueuedReset, InboxDoc, isDisplayOnly, isStale, pickInputEntry, prepareBoundary, removeInboxItem, } from "./inbox.js";
++import { appendInput, inputDraft, applyBoundary, hasQueuedReset, InboxDoc, isDisplayOnly, isStale, pickInputEntry, prepareBoundary, removeInboxItem, } from "./inbox.js";
+ import { LiveDoc } from "./live.js";
+ import { DURABLE_TELEMETRY_SCHEMA } from "./telemetry.js";
+ import { closedError, Waiters } from "./util.js";
+@@ -205,7 +205,7 @@ export async function admitSubmission(tx, conversationId, draft, now, queueModes
+         // Never queued: whether a run can continue depends on how the current one ends. Rejected before any write.
+         if (busy)
+             throw new ConversationBusy(conversationId);
+-        const boundary = await prepareBoundary(tx, conversationId, queueModes);
++        const boundary = await prepareBoundary(tx, conversationId, queueModes, "final", now);
+         const { id } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
+         // Items an earlier failed run left queued (`afterFailedRun: "keep"`) are placed first and join the run. Its history
+         // starts with the failed run's, so `afterRun` hooks see what that run did.
+@@ -214,7 +214,7 @@ export async function admitSubmission(tx, conversationId, draft, now, queueModes
+         return { id, status: "placed" };
+     }
+     // A boundary reads the table, so it is prepared before the first table write; a busy one needs none.
+-    const boundary = busy ? undefined : await prepareBoundary(tx, conversationId, queueModes);
++    const boundary = busy ? undefined : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : undefined);
+     if (boundary === undefined || boundary.inbox.items.length > 0) {
+         const { id } = await tx.createSubmission({
+             ...fields,
+@@ -253,6 +253,8 @@ export async function admitSubmission(tx, conversationId, draft, now, queueModes
+         const write = { ...fields, type: "write", status: "done", entry: entry.id };
+         return { id: (await tx.createSubmission(write)).id, status: "done" };
+     }
++    for (const before of boundary.entries)
++        await tx.appendEntry(conversationId, before);
+     const entry = await appendInput(tx, conversationId, draft.content, draft.entry, now);
+     const input = { ...fields, type: "input", status: "placed", entry: entry.id };
+     const { id } = await tx.createSubmission(input);
+diff --git a/dist/harness/types.d.ts b/dist/harness/types.d.ts
+index eb64e474162b158a43563e6c68d8ffea8c18b74a..cce695c6d87c79dadea54560bafcca62cf073ca1 100644
+--- a/dist/harness/types.d.ts
++++ b/dist/harness/types.d.ts
+@@ -816,6 +816,9 @@ export interface Harness extends Session {
+      * never do.
+      */
+     resume(): void;
++    /** Start no task run until `release()`, whatever `resume()` says; running invocations and abort handlers go on. */
++    hold(): void;
++    release(): void;
+     /**
+      * When this Session next needs a running process, from the scheduler's memory: `active` while some task can run now
+      * or is running, `at` for the earliest sleep. Answers also while scheduling is paused. A host that wakes the
+@@ -888,6 +891,8 @@ export interface HookApi extends DocumentReader {
+ export type HookResult<T> = T | undefined | Promise<T | undefined>;
+ /** Hooks of the built-in generation task. */
+ export interface GenerationHooks {
++    /** Before a new run’s first input is placed. Use only tx for storage access; effects commit with the input. */
++    beforeInput(input: EntryDraft, api: { readonly conversationId: ConversationId; readonly tx: Tx }, context: Context): HookResult<{ readonly entries?: readonly EntryDraft[] }>;
+     /** Before every request attempt, including recovery; the result is used for that request only. */
+     beforeRequest(request: {
+         readonly messages: readonly Message[];
+diff --git a/dist/session/session.d.ts b/dist/session/session.d.ts
+index 8201d7cfefb710516a3d4d89ba208adc2cbebd19..37c9f362401b5bc81f2f24233aa3dd7f57e29204 100644
+--- a/dist/session/session.d.ts
++++ b/dist/session/session.d.ts
+@@ -1,5 +1,5 @@
+ import { type Context, type JsonValue, type WatchEnd } from "@earendil-works/chord";
+-import type { CommitChange, CommitObservation, CommitObserver, CommitOptions, ConversationDocFamilyToken, ConversationDocToken, ConversationId, ConversationRecord, DocumentCommitChange, DocumentRecord, DocumentWatch, EntryId, JsonObject, Observation, Observer, ReportInfo, RewindableConversationDocFamilyToken, RewindableConversationDocToken, Seq, Session, SessionDocFamilyToken, SessionDocToken, SessionEnd, SetTimer, Storage, StorageWrite, TaskDocFamilyToken, TaskDocToken, TaskId, Tx } from "../types.ts";
++import type { CommitChange, CommitObservation, CommitObserver, CommitOptions, ConversationDocFamilyToken, ConversationDocToken, EntryDraft, ConversationId, ConversationRecord, DocumentCommitChange, DocumentRecord, DocumentWatch, EntryId, JsonObject, Observation, Observer, ReportInfo, RewindableConversationDocFamilyToken, RewindableConversationDocToken, Seq, Session, SessionDocFamilyToken, SessionDocToken, SessionEnd, SetTimer, Storage, StorageWrite, TaskDocFamilyToken, TaskDocToken, TaskId, Tx } from "../types.ts";
+ import { type HeadMarker } from "./conversation-tails.ts";
+ import { Transaction, type TransactionScope } from "./transaction.ts";
+ /**
+@@ -171,6 +171,7 @@ export declare class SessionImpl implements Session {
+      * listener. A plain Session drops them.
+      */
+     report(_error: unknown, _info?: ReportInfo, _context?: Context): void;
++    protected beforeInput(tx: Transaction, id: ConversationId, input: EntryDraft, context: Context): Promise<readonly EntryDraft[]>;
+     /**
+      * Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
+      * plain Session stages nothing; a Harness stages its built-in documents.
+diff --git a/dist/session/session.js b/dist/session/session.js
+index e9ae9ebfc2c45765ccda141d7e60cdc9a15a0894..e1e5c69deed9f0d09cf67c26cdb903fe4593f9c6 100644
+--- a/dist/session/session.js
++++ b/dist/session/session.js
+@@ -224,6 +224,7 @@ export class SessionImpl {
+                 if (this.#documents.get(id)?.record.id === recordId)
+                     this.#dropDocument(id);
+             },
++            beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
+             conversationCreated: (tx, record) => this.conversationCreated(tx, record),
+             taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
+             latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
+@@ -570,6 +571,9 @@ export class SessionImpl {
+      * listener. A plain Session drops them.
+      */
+     report(_error, _info = LISTENER, _context) { }
++    async beforeInput(_tx, _id, _input, _context) {
++        return [];
++    }
+     /**
+      * Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
+      * plain Session stages nothing; a Harness stages its built-in documents.
+diff --git a/dist/session/transaction.d.ts b/dist/session/transaction.d.ts
+index 5508fa2d021d2153f4f104b4ef69b99ef98a6a74..1c94a4a6e29b8371d49f7d6cb4d2e4e7cd6aec50 100644
+--- a/dist/session/transaction.d.ts
++++ b/dist/session/transaction.d.ts
+@@ -48,6 +48,7 @@ export interface TransactionHost {
+     install(document: LoadedDocument): void;
+     /** Remove a retired incarnation if it is still the cached occupant of its address. */
+     evict(addressId: string, recordId: DocumentId): void;
++    beforeInput(tx: Transaction, conversationId: ConversationId, input: EntryDraft, context: Context): Promise<readonly EntryDraft[]>;
+     /** Stage writes that belong to every newly created or forked conversation, in its creating transaction. */
+     conversationCreated(tx: Transaction, record: ConversationRecord): Promise<void>;
+     /** The newest entry ID visible from a conversation, known from the Session's commits. */
+@@ -112,6 +113,7 @@ export declare class Transaction implements Tx {
+     forkConversation(parentConversationId: ConversationId, at: EntryId, options: {
+         readonly ownership: ConversationOwnership;
+     }): Promise<ConversationRecord>;
++    beforeInput(conversationId: ConversationId, input: EntryDraft): Promise<readonly EntryDraft[]>;
+     appendEntry(conversationId: ConversationId, value: EntryDraft): Promise<EntryRecord>;
+     appendEntry<D extends JsonValue>(token: Entry<D>, conversationId: ConversationId, value: TypedEntryDraft<NoInfer<D>>): Promise<TypedEntry<D>>;
+     createTask<I, S extends {
+diff --git a/dist/session/transaction.js b/dist/session/transaction.js
+index 7fbe75b11bb2a9ab28f4f23800143231172d71b7..a516853e2f9c4a4437dbc2bb8f1c8d2117386e9a 100644
+--- a/dist/session/transaction.js
++++ b/dist/session/transaction.js
+@@ -167,6 +167,9 @@ export class Transaction {
+         this.#assertOpen();
+         return record;
+     }
++    beforeInput(conversationId, input) {
++        return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
++    }
+     appendEntry(first, second, third) {
+         if (typeof first === "number")
+             return this.#appendEntry(first, second);
+diff --git a/dist/types.d.ts b/dist/types.d.ts
+index b43143c5e7998b36034ffe9577f74f3d88ab932b..a9ca10ae41c48ad0ebb813e62caa3699dec5c51e 100644
+--- a/dist/types.d.ts
++++ b/dist/types.d.ts
+@@ -409,6 +409,8 @@ export type ContextEdit = {
+ });
+ /** Immutable transcript event with separate model-facing and application-facing payloads. */
+ export type EntryRecord = {
++    /** Placed by the harness from an input submission, not a write. */
++    readonly input?: true;
+     readonly id: EntryId;
+     readonly conversationId: ConversationId;
+     /** Application-defined entry discriminator. */
+@@ -891,6 +893,8 @@ export type CommitPublication = {
+  * Session may freeze them (its caches and context ranges keep them), so mutating one can throw.
+  */
+ export interface Tx {
++    /** Prepare entries before a new run’s first input; called before table writes, in the placement transaction. */
++    beforeInput(conversationId: ConversationId, input: EntryDraft): Promise<readonly EntryDraft[]>;
+     conversation(id: ConversationId): Promise<ConversationRecord | undefined>;
+     entry(id: EntryId): Promise<EntryRecord | undefined>;
+     /** Undefined when the entry is absent or has another kind. */

host/js/test-bundle.js

@@ -8940,6 +8940,9 @@ var Transaction = class {
     this.#assertOpen();
     return record;
   }
+  beforeInput(conversationId, input) {
+    return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
+  }
   appendEntry(first, second, third) {
     if (typeof first === "number")
       return this.#appendEntry(first, second);
@@ -10122,6 +10125,7 @@ var SessionImpl = class {
         if (this.#documents.get(id)?.record.id === recordId)
           this.#dropDocument(id);
       },
+      beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
       conversationCreated: (tx, record) => this.conversationCreated(tx, record),
       taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
       latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -10446,6 +10450,9 @@ var SessionImpl = class {
    */
   report(_error, _info = LISTENER, _context) {
   }
+  async beforeInput(_tx, _id, _input, _context) {
+    return [];
+  }
   /**
    * Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
    * plain Session stages nothing; a Harness stages its built-in documents.
@@ -12250,10 +12257,14 @@ var InboxDoc = defineDoc({
   initial: () => ({ items: [] }),
   checkpointWhen: (value2) => value2.items.length === 0
 });
-async function prepareBoundary(tx, conversationId, modes) {
+async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
   const head = (await tx.latestHeadMarker(conversationId))?.head;
   const inbox = await tx.doc(InboxDoc, conversationId);
+  const first = inbox.items.find((item) => item.mode === "steer" || item.mode === "followUp");
+  const input = first === void 0 ? incoming : inputDraft(first.content, first.entry, now);
+  const entries2 = input !== void 0 && (at === "final" || hasQueuedReset(inbox.items)) ? await tx.beforeInput(conversationId, input) : [];
   return {
+    entries: entries2,
     conversationId,
     inbox,
     steeringMode: modes.steeringMode,
@@ -12289,6 +12300,9 @@ async function applyBoundary(tx, boundary, at, now) {
   }
   const placed = [];
   for (const index of users) {
+    if (placed.length === 0)
+      for (const draft of boundary.entries)
+        await tx.appendEntry(conversationId, draft);
     const item = items[index];
     const entry2 = await appendInput(tx, conversationId, item.content, item.entry, now);
     tx.placeSubmission(item.id, entry2.id);
@@ -12301,10 +12315,11 @@ async function applyBoundary(tx, boundary, at, now) {
   return { users: placed, first, reset };
 }
 function appendInput(tx, conversationId, content, entry2, now) {
+  return tx.appendEntry(conversationId, inputDraft(content, entry2, now));
+}
+function inputDraft(content, entry2, now) {
   const message = { role: "user", content, timestamp: now };
-  if (entry2 === void 0)
-    return tx.appendEntry(UserEntry, conversationId, { model: [message] });
-  return tx.appendEntry(conversationId, { ...pickInputEntry(entry2), model: [message] });
+  return { ...entry2 === void 0 ? { kind: UserEntry.kind } : pickInputEntry(entry2), input: true, model: [message] };
 }
 function pickInputEntry({ kind, data }) {
   const picked = data === void 0 ? { kind } : { kind, data };
@@ -13016,6 +13031,7 @@ var TaskScheduler = class {
   #unsubscribeRegistry = () => {
   };
   #enabled = false;
+  #held = false;
   #readOnly;
   #closing = false;
   #dirty = false;
@@ -13044,6 +13060,14 @@ var TaskScheduler = class {
     this.#context = options.context;
     this.#readOnly = options.readOnly === true;
   }
+  /** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
+  hold() {
+    this.#held = true;
+  }
+  release() {
+    this.#held = false;
+    this.#kick();
+  }
   /**
    * Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
    * a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -14498,6 +14522,8 @@ var TaskScheduler = class {
         for (const record of plan.ready) {
           const id = record.id;
           const mode = record.abortRequested ? "abort" : "run";
+          if (mode === "run" && this.#held)
+            continue;
           const cutOff = record.restarts !== void 0 && record.restarts >= this.#restartLimit();
           if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
             const message = mode === "abort" ? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times` : `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
@@ -14683,7 +14709,7 @@ var TaskScheduler = class {
     const tasks = [];
     for (const record of this.#live.values())
       tasks.push({ record, state: this.#inspectTask(record, snapshot) });
-    const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
+    const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
     return { scheduling, tasks };
   }
   #inspectTask(record, snapshot) {
@@ -14838,6 +14864,10 @@ var TaskScheduler = class {
    * invocation continues.
    */
   #progress(tx, current, state, extension, context) {
+    if (this.#held) {
+      tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
+      return false;
+    }
     state.snapshot = this.#registry.snapshot();
     const next = state.snapshot.task(current.kind);
     if (next === state.task)
@@ -17571,7 +17601,7 @@ async function prepareFailure(tx, runtime) {
   const policy = settings.afterFailedRun;
   if (policy !== "continue")
     return { policy };
-  return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
+  return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
 }
 async function endFailedRun(tx, runtime, live, failure, end) {
   const conversationId = runtime.conversationId;
@@ -17739,7 +17769,7 @@ async function answer(runtime, message, context) {
   const afterRun = await asksAfterRun(runtime, context);
   const conversationId = runtime.conversationId;
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
     const live = await tx.doc(LiveDoc, conversationId);
     const entry2 = await appendAssistant(tx, conversationId, message);
     const result2 = { status: "terminal", outcome: { status: "completed", result: { entryId: entry2.id } } };
@@ -17812,7 +17842,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
 async function endAnsweredRun(runtime, answer2, context) {
   const conversationId = runtime.conversationId;
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
     const live = await tx.doc(LiveDoc, conversationId);
     const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
     endRun(tx, live, runtime.taskId, { status: "done", answer: answer2 });
@@ -17885,7 +17915,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
   const maxTurns = (await runtime.agent(context)).maxTurns;
   const afterRun = terminate && await asksAfterRun(runtime, context);
   await runtime.commit(async (tx) => {
-    const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+    const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== void 0 ? "final" : "postTools", runtime.now());
     if (added.length > 0)
       await addTools(tx, conversationId, added);
     const live = await tx.doc(LiveDoc, conversationId);
@@ -18301,13 +18331,13 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
   if (draft.type === "continue") {
     if (busy)
       throw new ConversationBusy(conversationId);
-    const boundary2 = await prepareBoundary(tx, conversationId, queueModes);
+    const boundary2 = await prepareBoundary(tx, conversationId, queueModes, "final", now);
     const { id: id2 } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
     const { users, first } = await applyBoundary(tx, boundary2, "final", now);
     await startRun(tx, conversationId, live, { inputs: [...users, id2], first: live.runFailure?.first ?? first });
     return { id: id2, status: "placed" };
   }
-  const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes);
+  const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : void 0);
   if (boundary === void 0 || boundary.inbox.items.length > 0) {
     const { id: id2 } = await tx.createSubmission({
       ...fields,
@@ -18345,6 +18375,8 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
     const write = { ...fields, type: "write", status: "done", entry: entry3.id };
     return { id: (await tx.createSubmission(write)).id, status: "done" };
   }
+  for (const before of boundary.entries)
+    await tx.appendEntry(conversationId, before);
   const entry2 = await appendInput(tx, conversationId, draft.content, draft.entry, now);
   const input = { ...fields, type: "input", status: "placed", entry: entry2.id };
   const { id } = await tx.createSubmission(input);
@@ -19360,6 +19392,15 @@ var HarnessImpl = class extends SessionImpl {
       create: (target, createOptions, context2) => this.#create(target, createOptions, context2)
     };
   }
+  async beforeInput(tx, conversationId, input, context) {
+    const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
+    const entries2 = [];
+    for (const { handlers } of agentHooks(agent, "pi.generation")) {
+      const result2 = await handlers.beforeInput?.(input, { conversationId, tx }, context);
+      entries2.push(...result2?.entries ?? []);
+    }
+    return entries2;
+  }
   /** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
   async resolveAgent(id, snapshot, context) {
     const registry2 = snapshot ?? this.#options.registry.snapshot();
@@ -19382,6 +19423,12 @@ var HarnessImpl = class extends SessionImpl {
     this.assertUsable();
     this.#tasks.resume();
   }
+  hold() {
+    this.#tasks.hold();
+  }
+  release() {
+    this.#tasks.release();
+  }
   wake() {
     this.assertUsable();
     return this.#tasks.wake();
@@ -21291,7 +21338,7 @@ function environment(cwd = "/", conversation) {
         }
       };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, conversation);
+        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, conversation === void 0 ? void 0 : String(conversation));
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result2;
         while (true) {

host/src/api/server.rs

@@ -125,12 +125,7 @@ pub async fn run_with_userland(
         tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()?
-            .block_on(crate::harness::run_module_with_credentials(
-                world,
-                Arc::new(runtime),
-                cred,
-                include_str!("../../js/bundle.js"),
-            ))
+            .block_on(crate::harness::run_console(world, Arc::new(runtime), cred))
     })
     .await
     .map_err(anyhow::Error::from)

host/src/harness/environment.rs

@@ -129,19 +129,6 @@ pub async fn op_exec_next(
     Ok(value)
 }
 
-#[op2]
-#[string]
-pub async fn op_input() -> Result<Option<String>, JsErrorBox> {
-    use tokio::io::AsyncBufReadExt;
-    let mut reader = tokio::io::BufReader::with_capacity(1, tokio::io::stdin());
-    let mut line = String::new();
-    let n = reader
-        .read_line(&mut line)
-        .await
-        .map_err(|e| JsErrorBox::generic(e.to_string()))?;
-    Ok((n != 0).then_some(line))
-}
-
 #[op2(fast)]
 pub fn op_output(#[string] text: &str) {
     print!("{text}");

host/src/harness/input.rs

@@ -0,0 +1,74 @@
+//! Channel input for a headless harness. The console and stop signals are clients.
+use deno_core::{OpState, op2};
+use deno_error::JsErrorBox;
+use serde::Serialize;
+use serde_json::Value;
+use std::{cell::RefCell, rc::Rc};
+use tokio::sync::mpsc;
+
+#[derive(Serialize)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub enum Input {
+    Submit {
+        conversation: u64,
+        content: Value,
+        #[serde(rename = "requestId")]
+        request_id: String,
+    },
+    Stop,
+}
+
+pub type Sender = mpsc::UnboundedSender<Input>;
+pub type Receiver = mpsc::UnboundedReceiver<Input>;
+
+pub fn channel() -> (Sender, Receiver) {
+    mpsc::unbounded_channel()
+}
+
+pub(super) struct Inbox(pub tokio::sync::Mutex<Receiver>);
+
+#[op2]
+#[serde]
+pub async fn op_input(state: Rc<RefCell<OpState>>) -> Result<serde_json::Value, JsErrorBox> {
+    let inbox = state.borrow().borrow::<Rc<Inbox>>().clone();
+    serde_json::to_value(inbox.0.lock().await.recv().await)
+        .map_err(|error| JsErrorBox::generic(error.to_string()))
+}
+
+pub(super) fn console(sender: Sender) {
+    std::thread::spawn(move || {
+        use std::io::BufRead;
+        for line in std::io::stdin().lock().lines() {
+            let Ok(line) = line else { break };
+            if line.trim().is_empty() {
+                continue;
+            }
+            if sender
+                .send(Input::Submit {
+                    conversation: 1,
+                    content: Value::String(line),
+                    request_id: uuid::Uuid::new_v4().to_string(),
+                })
+                .is_err()
+            {
+                break;
+            }
+        }
+        // EOF closes only this client. The signal client keeps the harness alive.
+    });
+}
+
+pub(super) fn stop_signal(
+    sender: Sender,
+) -> std::io::Result<impl std::future::Future<Output = ()> + Send> {
+    // Register before V8 starts, so a stop cannot arrive before the listener.
+    let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
+    let mut interrupt = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
+    Ok(async move {
+        tokio::select! {
+            _ = interrupt.recv() => {}
+            _ = terminate.recv() => {}
+        }
+        let _ = sender.send(Input::Stop);
+    })
+}

host/src/harness/mod.rs

@@ -1,6 +1,7 @@
 //! Durable V8 harness, host-side SQLite storage, and world tool operations.
 mod credentials;
 mod environment;
+pub mod input;
 use crate::{exec::Exec, world::World};
 use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
 use deno_error::JsErrorBox;
@@ -85,7 +86,7 @@ deno_core::extension!(
         environment::op_file,
         environment::op_exec,
         environment::op_exec_next,
-        environment::op_input,
+        input::op_input,
         environment::op_output
     ]
 );
@@ -105,12 +106,7 @@ async fn run_with_shared_credentials(
         tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()?
-            .block_on(run_module_with_credentials(
-                world,
-                exec,
-                cred,
-                include_str!("../../js/bundle.js"),
-            ))
+            .block_on(run_console(world, exec, cred))
     })
     .await?
     .map_err(|error: anyhow::Error| {
@@ -139,6 +135,19 @@ pub async fn run_module_with_credentials(
     exec: Arc<dyn Exec>,
     cred: Arc<crate::cred::store::Cred>,
     source: &str,
+) -> anyhow::Result<()> {
+    let (_, receiver) = input::channel();
+    run_module_with_input(world, exec, cred, receiver, source).await
+}
+
+/// Run without the console. Submissions and stop requests arrive through the channel.
+/// The caller must use a current-thread Tokio runtime.
+pub async fn run_module_with_input(
+    world: World,
+    exec: Arc<dyn Exec>,
+    cred: Arc<crate::cred::store::Cred>,
+    receiver: input::Receiver,
+    source: &str,
 ) -> anyhow::Result<()> {
     let proxy = credentials::proxy(&cred)?;
     let mut runtime = JsRuntime::new(RuntimeOptions {
@@ -196,6 +205,12 @@ pub async fn run_module_with_credentials(
         .borrow_mut()
         .put(credentials::Notifications::new(&cred));
     runtime.op_state().borrow_mut().put(cred);
+    runtime
+        .op_state()
+        .borrow_mut()
+        .put(std::rc::Rc::new(input::Inbox(tokio::sync::Mutex::new(
+            receiver,
+        ))));
     runtime
         .op_state()
         .borrow_mut()
@@ -237,6 +252,25 @@ pub async fn run_module_with_credentials(
     })
 }
 
+/// Run the headless harness with console input and signal clients.
+/// The caller must use a current-thread Tokio runtime.
+pub async fn run_console(
+    world: World,
+    exec: Arc<dyn Exec>,
+    cred: Arc<crate::cred::store::Cred>,
+) -> anyhow::Result<()> {
+    let (sender, receiver) = input::channel();
+    input::console(sender.clone());
+    let signals = tokio::spawn(input::stop_signal(sender)?);
+    let source = concat!(
+        "globalThis.consoleEnabled = true;\n",
+        include_str!("../../js/bundle.js")
+    );
+    let result = run_module_with_input(world, exec, cred, receiver, source).await;
+    signals.abort();
+    result
+}
+
 /// Open a world and resume its durable root conversation.
 pub async fn run(path: &Path) -> anyhow::Result<()> {
     run_with_userland(path, None).await

host/tests/harness_foundation.rs

@@ -0,0 +1,172 @@
+//! Patched pi-durable input placement, hold boundaries, and headless channel lifecycle.
+use sausage::{
+    cred::Cred,
+    harness::{
+        input::{self, Input},
+        run_module, run_module_with_input,
+    },
+    world::World,
+};
+use std::{
+    process::{Command, Stdio},
+    sync::Arc,
+    time::{Duration, Instant},
+};
+
+const BUNDLE: &str = include_str!("../js/foundation-test-bundle.js");
+
+#[test]
+fn bundle_has_patched_hooks() {
+    for bundle in [BUNDLE, include_str!("../js/bundle.js")] {
+        assert!(bundle.contains("handlers.beforeInput"));
+        assert!(bundle.contains("input: true"));
+        assert!(bundle.contains("#held"));
+    }
+    assert!(!include_str!("../js/patches/pi-durable-1.5.0.patch").contains("dist/env/node"));
+}
+
+async fn probe(case: &str) -> anyhow::Result<()> {
+    let dir = tempfile::tempdir()?;
+    let world = World::open(dir.path())?;
+    let exec = sausage::process::exec(World::open(dir.path())?)?;
+    run_module(
+        world,
+        exec,
+        &format!("globalThis.foundationCase = {case:?};\n{BUNDLE}"),
+    )
+    .await
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn input_placement() -> anyhow::Result<()> {
+    probe("placement").await
+}
+#[tokio::test(flavor = "current_thread")]
+async fn followups_start_one_run() -> anyhow::Result<()> {
+    probe("followups").await
+}
+#[tokio::test(flavor = "current_thread")]
+async fn response_thinking_level() -> anyhow::Result<()> {
+    probe("thinking").await
+}
+#[tokio::test(flavor = "current_thread")]
+async fn hold_parks_at_phase_boundary() -> anyhow::Result<()> {
+    probe("hold").await
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn headless_child() -> anyhow::Result<()> {
+    if std::env::var_os("HEADLESS_CHILD").is_none() {
+        return Ok(());
+    }
+    let dir = tempfile::tempdir()?;
+    let world = World::open(dir.path())?;
+    let cred = Arc::new(Cred::open(&world.host_dir)?);
+    let exec = sausage::process::exec(World::open(dir.path())?)?;
+    let (sender, receiver) = input::channel();
+    sender.send(Input::Submit {
+        conversation: 1,
+        content: serde_json::json!("channel input"),
+        request_id: "channel-request".into(),
+    })?;
+    sender.send(Input::Stop)?;
+    tokio::time::timeout(
+        Duration::from_secs(30),
+        run_module_with_input(
+            world,
+            exec,
+            cred,
+            receiver,
+            &format!("globalThis.foundationCase = 'headless';\n{BUNDLE}"),
+        ),
+    )
+    .await??;
+    Ok(())
+}
+
+#[test]
+fn channel_with_closed_stdin() -> anyhow::Result<()> {
+    let mut child = Command::new(std::env::current_exe()?)
+        .args(["--exact", "headless_child", "--nocapture"])
+        .env("HEADLESS_CHILD", "1")
+        .stdin(Stdio::null())
+        .spawn()?;
+    let deadline = Instant::now() + Duration::from_secs(40);
+    loop {
+        if let Some(status) = child.try_wait()? {
+            assert!(status.success());
+            break;
+        }
+        if Instant::now() > deadline {
+            child.kill()?;
+            child.wait()?;
+            anyhow::bail!("Headless channel did not stop");
+        }
+        std::thread::yield_now();
+    }
+    Ok(())
+}
+
+#[test]
+fn cli_stays_up_after_stdin_eof() -> anyhow::Result<()> {
+    let dir = tempfile::tempdir()?;
+    let mut child = Command::new(env!("CARGO_BIN_EXE_sausage"))
+        .arg("run")
+        .arg(dir.path())
+        .args(["--api-listen", "127.0.0.1:0"])
+        .env("PI_PROXY_BASE_URL", "https://proxy.invalid")
+        .env("PI_PROXY_API_KEY", "faux-token")
+        .env_remove("SAUSAGE_USERLAND")
+        .stdin(Stdio::null())
+        .spawn()?;
+    let deadline = Instant::now() + Duration::from_secs(30);
+    loop {
+        if let Some(status) = child.try_wait()? {
+            anyhow::bail!("CLI exited at EOF: {status}");
+        }
+        if let Ok(db) = rusqlite::Connection::open(dir.path().join("host/harness.sqlite"))
+            && db
+                .query_row("SELECT count(*) FROM conversations", [], |r| {
+                    r.get::<_, i64>(0)
+                })
+                .unwrap_or(0)
+                > 0
+        {
+            break;
+        }
+        if Instant::now() > deadline {
+            child.kill()?;
+            child.wait()?;
+            anyhow::bail!("CLI did not initialize");
+        }
+        std::thread::yield_now();
+    }
+    // EOF shutdown previously took one credential tick. Check beyond that tick.
+    let alive_until = Instant::now() + Duration::from_secs(2);
+    while Instant::now() < alive_until {
+        if let Some(status) = child.try_wait()? {
+            anyhow::bail!("CLI exited after EOF: {status}");
+        }
+        std::thread::yield_now();
+    }
+    assert!(
+        Command::new("kill")
+            .args(["-TERM", &child.id().to_string()])
+            .status()?
+            .success()
+    );
+    let deadline = Instant::now() + Duration::from_secs(10);
+    loop {
+        if let Some(status) = child.try_wait()? {
+            assert!(status.success());
+            break;
+        }
+        if Instant::now() > deadline {
+            child.kill()?;
+            child.wait()?;
+            anyhow::bail!("CLI did not stop cleanly");
+        }
+        std::thread::yield_now();
+    }
+    Ok(())
+}

host/tests/p2.rs

@@ -0,0 +1,523 @@
+//! P2 exit rehearsal: native host pages, software passkey, durable bash tools, and real SSH Git push.
+use anyhow::{Result, ensure};
+use base64::{Engine, engine::general_purpose::STANDARD};
+use sausage::{
+    api::{Api, Listener, Options},
+    cred::{Cred, entry::Envelope, webauthn::Challenge},
+    world::World,
+};
+use serde_json::{Value, json};
+use std::{
+    io::{Read, Write},
+    net::{SocketAddr, TcpListener, TcpStream},
+    path::{Path, PathBuf},
+    process::{Child, Command, Stdio},
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use webauthn_authenticator_rs::{WebauthnAuthenticator, softpasskey::SoftPasskey};
+use webauthn_rs::prelude::{CreationChallengeResponse, RequestChallengeResponse};
+
+fn command(cwd: &Path, args: &[&str]) -> Result<()> {
+    ensure!(
+        Command::new(args[0])
+            .args(&args[1..])
+            .current_dir(cwd)
+            .stdout(Stdio::null())
+            .status()?
+            .success(),
+        "fixture command failed"
+    );
+    Ok(())
+}
+struct Sshd(Child);
+impl Drop for Sshd {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+struct Browser {
+    address: SocketAddr,
+    origin: String,
+    auth: WebauthnAuthenticator<SoftPasskey>,
+}
+impl Browser {
+    fn raw(&self, method: &str, path: &str, body: Value) -> Result<String> {
+        let bytes = serde_json::to_vec(&body)?;
+        let mut connection = TcpStream::connect(self.address)?;
+        connection.set_read_timeout(Some(Duration::from_secs(10)))?;
+        write!(
+            connection,
+            "{method} {path} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+            bytes.len()
+        )?;
+        connection.write_all(&bytes)?;
+        let mut response = String::new();
+        connection.read_to_string(&mut response)?;
+        let error = serde_json::from_str::<Value>(
+            response
+                .split_once("\r\n\r\n")
+                .map(|(_, body)| body)
+                .unwrap_or(""),
+        )
+        .ok();
+        ensure!(
+            response.starts_with("HTTP/1.1 200"),
+            "host page request failed: {path}, kind {}",
+            error
+                .as_ref()
+                .and_then(|value| value["error"].as_str())
+                .unwrap_or("invalid-response")
+        );
+        Ok(response
+            .split_once("\r\n\r\n")
+            .ok_or_else(|| anyhow::anyhow!("invalid HTTP response"))?
+            .1
+            .to_owned())
+    }
+    fn json(&self, method: &str, path: &str, body: Value) -> Result<Value> {
+        Ok(serde_json::from_str(&self.raw(method, path, body)?)?)
+    }
+    fn assertion(&mut self, path: &str, body: Value) -> Result<Value> {
+        let start: Challenge<RequestChallengeResponse> =
+            serde_json::from_value(self.json("POST", path, body)?)?;
+        let assertion = self
+            .auth
+            .do_authentication(url::Url::parse(&self.origin)?, start.options)?;
+        Ok(json!({"challenge_id":start.id,"assertion":assertion}))
+    }
+    fn approve(&mut self, id: &str) -> Result<()> {
+        ensure!(
+            self.raw("GET", &format!("/approve/{id}"), Value::Null)?
+                .contains("/app.js"),
+            "approval page missing"
+        );
+        let path = format!("/cred/requests/{id}");
+        let gate = self.assertion(&format!("{path}/challenge"), json!({"decision":"approve"}))?;
+        self.json("POST", &format!("{path}/decide"), gate)?;
+        Ok(())
+    }
+    fn create(&mut self, config: Value) -> Result<Value> {
+        let name = config["name"].as_str().unwrap().to_owned();
+        let request = self.json("POST", "/cred/requests", json!({"kind":"edit","key":format!("setup-{name}"),"request":{"item":null,"base_version":null,"change":[{"op":"create_item","config":config}]}}))?;
+        self.approve(request["request"].as_str().unwrap())?;
+        let items = self.json("GET", "/cred/items", Value::Null)?;
+        Ok(items
+            .as_array()
+            .unwrap()
+            .iter()
+            .find(|item| item["name"] == name)
+            .unwrap()
+            .clone())
+    }
+    async fn enter(&mut self, item: &Value, secret: &str) -> Result<()> {
+        let path = format!(
+            "/cred/items/{}/fields/{}",
+            item["id"].as_str().unwrap(),
+            item["fields"][0]["field_id"].as_str().unwrap()
+        );
+        ensure!(
+            self.raw(
+                "GET",
+                &format!(
+                    "/entry/{}/{}",
+                    item["id"].as_str().unwrap(),
+                    item["fields"][0]["field_id"].as_str().unwrap()
+                ),
+                Value::Null
+            )?
+            .contains("/app.js"),
+            "secret-entry page missing"
+        );
+        let public = self.json("GET", "/cred/entry/key", Value::Null)?;
+        let script = self.raw("GET", "/app.js", Value::Null)?;
+        let mut runtime = deno_core::JsRuntime::new(deno_core::RuntimeOptions {
+            extensions: vec![
+                deno_webidl::deno_webidl::init(),
+                deno_web::deno_web::init(
+                    Arc::new(deno_web::BlobStore::default()),
+                    None,
+                    false,
+                    Default::default(),
+                ),
+                deno_crypto::deno_crypto::init(None),
+            ],
+            ..Default::default()
+        });
+        runtime.execute_script("browser.js", "globalThis.crypto = Deno.core.loadExtScript('ext:deno_crypto/00_crypto.js').crypto; globalThis.TextEncoder = Deno.core.loadExtScript('ext:deno_web/08_text_encoding.js').TextEncoder; globalThis.document = {querySelector: () => ({addEventListener() {}})};")?;
+        runtime.execute_script("app.js", script)?;
+        runtime.execute_script("entry.js", format!("encrypt({},{}).then(value => globalThis.envelope = JSON.stringify(value.envelope))", public["public_key"], serde_json::to_string(secret)?))?;
+        runtime.run_event_loop(Default::default()).await?;
+        let result = runtime.execute_script("result.js", "envelope")?;
+        let text = {
+            deno_core::scope!(scope, &mut runtime);
+            deno_core::v8::Local::new(scope, result).to_rust_string_lossy(scope)
+        };
+        let envelope: Envelope = serde_json::from_str(&text)?;
+        let mut gate = self.assertion(
+            &format!("{path}/challenge"),
+            json!({"ciphertext_sha256":envelope.sha256()}),
+        )?;
+        gate["envelope"] = serde_json::to_value(envelope)?;
+        self.json("POST", &path, gate)?;
+        Ok(())
+    }
+}
+fn components() -> Result<PathBuf> {
+    let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland");
+    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("p2-userland");
+    if !userland.join("cwd-init/init.o").is_file() {
+        command(&userland, &["nu", "--no-config-file", "cwd-init/build.nu"])?;
+    }
+    let status = Command::new("cargo")
+        .current_dir(&userland)
+        .args([
+            "build",
+            "--release",
+            "--locked",
+            "--target",
+            "wasm32-wasip2",
+            "-j",
+            "2",
+            "-p",
+            "git",
+            "-p",
+            "curl",
+        ])
+        .arg("--target-dir")
+        .arg(&target)
+        .status()?;
+    ensure!(status.success(), "guest build failed");
+    let release = target.join("wasm32-wasip2/release");
+    if !release.join("sh.wasm").is_file() {
+        ensure!(
+            Command::new("nu")
+                .arg("brush/build.nu")
+                .env("CARGO_TARGET_DIR", &target)
+                .current_dir(&userland)
+                .status()?
+                .success(),
+            "shell build failed"
+        );
+        std::fs::copy(release.join("brush.wasm"), release.join("sh.wasm"))?;
+    }
+    Ok(release)
+}
+fn no_secrets(path: &Path, secrets: &[&[u8]]) -> Result<()> {
+    for entry in std::fs::read_dir(path)? {
+        let entry = entry?;
+        if entry.file_type()?.is_dir() {
+            no_secrets(&entry.path(), secrets)?;
+        } else {
+            let bytes = std::fs::read(entry.path())?;
+            for secret in secrets {
+                ensure!(
+                    !bytes.windows(secret.len()).any(|value| value == *secret),
+                    "secret escaped into a world file"
+                );
+            }
+        }
+    }
+    Ok(())
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn request_approve_fetch_and_ssh_push() -> Result<()> {
+    let components = components()?;
+    let dir = tempfile::tempdir()?;
+    let world = World::open(dir.path())?;
+    let root = dir.path().join("world");
+    std::fs::create_dir_all(root.join("bin"))?;
+    for (source, destination) in [
+        ("git.wasm", "git.wasm"),
+        ("curl.wasm", "curl.wasm"),
+        ("sh.wasm", "sh.wasm"),
+    ] {
+        std::fs::copy(components.join(source), root.join("bin").join(destination))?;
+    }
+    let token = format!("p2-private-{}", uuid::Uuid::new_v4());
+    let mut seed = [0; 32];
+    openssl::rand::rand_bytes(&mut seed)?;
+    let key = STANDARD.encode(seed);
+    let public = ed25519_dalek::SigningKey::from_bytes(&seed)
+        .verifying_key()
+        .to_bytes();
+    let mut blob = Vec::from(11u32.to_be_bytes());
+    blob.extend_from_slice(b"ssh-ed25519");
+    blob.extend_from_slice(&32u32.to_be_bytes());
+    blob.extend_from_slice(&public);
+    let fixture = dir.path().join("sshd");
+    std::fs::create_dir_all(&fixture)?;
+    std::fs::write(
+        fixture.join("authorized_keys"),
+        format!("ssh-ed25519 {}\n", STANDARD.encode(blob)),
+    )?;
+    command(
+        &fixture,
+        &[
+            "ssh-keygen",
+            "-q",
+            "-t",
+            "ed25519",
+            "-N",
+            "",
+            "-f",
+            "host_key",
+        ],
+    )?;
+    let reserve = TcpListener::bind("127.0.0.1:0")?;
+    let ssh_port = reserve.local_addr()?.port();
+    drop(reserve);
+    let user = std::env::var("USER")?;
+    std::fs::write(
+        fixture.join("config"),
+        format!(
+            "Port {ssh_port}\nListenAddress 127.0.0.1\nHostKey {}\nPidFile {}\nAuthorizedKeysFile {}\nStrictModes no\nUsePAM no\nPasswordAuthentication no\nKbdInteractiveAuthentication no\nPermitRootLogin no\nAllowUsers {user}\n",
+            fixture.join("host_key").display(),
+            fixture.join("pid").display(),
+            fixture.join("authorized_keys").display()
+        ),
+    )?;
+    let ssh_binary = std::env::split_paths(&std::env::var_os("PATH").unwrap())
+        .map(|path| path.join("sshd"))
+        .find(|path| path.is_file())
+        .unwrap()
+        .canonicalize()?;
+    let _sshd = Sshd(
+        Command::new(ssh_binary)
+            .args(["-D", "-e", "-f"])
+            .arg(fixture.join("config"))
+            .stdout(Stdio::null())
+            .stderr(std::fs::File::create(fixture.join("log"))?)
+            .spawn()?,
+    );
+    let deadline = Instant::now() + Duration::from_secs(10);
+    while TcpStream::connect(("127.0.0.1", ssh_port)).is_err() {
+        ensure!(Instant::now() < deadline, "sshd did not start");
+        tokio::task::yield_now().await;
+    }
+    std::fs::create_dir_all(root.join(".ssh"))?;
+    let host_public = std::fs::read_to_string(fixture.join("host_key.pub"))?;
+    std::fs::write(
+        root.join(".ssh/known_hosts"),
+        format!("[127.0.0.1]:{ssh_port} {host_public}"),
+    )?;
+    let repo = root.join("repo");
+    std::fs::create_dir_all(&repo)?;
+    command(&repo, &["git", "init", "-b", "main"])?;
+    std::fs::write(repo.join("demo"), "P2 push\n")?;
+    command(&repo, &["git", "add", "demo"])?;
+    command(
+        &repo,
+        &[
+            "git",
+            "-c",
+            "user.name=P2",
+            "-c",
+            "user.email=p2@example.com",
+            "commit",
+            "-m",
+            "rehearsal",
+        ],
+    )?;
+    command(&fixture, &["git", "init", "--bare", "destination.git"])?;
+    let url = format!(
+        "ssh://{user}@127.0.0.1:{ssh_port}{}/destination.git",
+        fixture.display()
+    );
+    command(&repo, &["git", "remote", "add", "origin", &url])?;
+    let server = Arc::new(
+        tiny_http::Server::http("127.0.0.1:0")
+            .map_err(|_| anyhow::anyhow!("HTTP fixture bind failed"))?,
+    );
+    let http_port = server.server_addr().to_ip().unwrap().port();
+    let serving = server.clone();
+    let expected = format!("Bearer {token}");
+    let thread = std::thread::spawn(move || {
+        if let Ok(request) = serving.recv() {
+            let valid = request.headers().iter().any(|header| {
+                header.field.equiv("Authorization") && header.value.as_str() == expected
+            });
+            request
+                .respond(tiny_http::Response::from_string(if valid {
+                    "Bearer accepted"
+                } else {
+                    "missing bearer"
+                }))
+                .unwrap();
+            valid
+        } else {
+            false
+        }
+    });
+    let reserve = TcpListener::bind("127.0.0.1:0")?;
+    let address = reserve.local_addr()?;
+    drop(reserve);
+    let origin = format!("http://localhost:{}", address.port());
+    let cred = Arc::new(Cred::open(&world.host_dir)?);
+    let api = Arc::new(Api::open(
+        cred.clone(),
+        &world.host_dir,
+        "localhost",
+        &origin,
+    )?);
+    let runtime = sausage::process::Runtime::new_with_hooks(
+        World {
+            root: world.root.clone(),
+            host_dir: world.host_dir.clone(),
+            processes: world.processes.clone(),
+        },
+        cred.clone(),
+        sausage::process::default_script(),
+        Some(api.clone()),
+    )?;
+    let listener = Listener::start_with_api(
+        api,
+        Options {
+            api_listen: address,
+            api_rp_id: "localhost".into(),
+            api_origin: origin.clone(),
+        },
+    )?;
+    runtime
+        .network_policy()
+        .deny_host_port(listener.address().port());
+    let mut browser = Browser {
+        address: listener.address(),
+        origin,
+        auth: WebauthnAuthenticator::new(SoftPasskey::new(true)),
+    };
+    ensure!(
+        browser
+            .raw("GET", "/register", Value::Null)?
+            .contains("/app.js"),
+        "registration page missing"
+    );
+    let start: Challenge<CreationChallengeResponse> = serde_json::from_value(browser.json(
+        "POST",
+        "/cred/passkeys/register/start",
+        json!({"label":"Surma rehearsal"}),
+    )?)?;
+    let registration = browser
+        .auth
+        .do_registration(url::Url::parse(&browser.origin)?, start.options)?;
+    browser.json(
+        "POST",
+        "/cred/passkeys/register/finish",
+        json!({"challenge_id":start.id,"response":registration}),
+    )?;
+    let http_item = browser.create(json!({"name":"demo-api","rules":[{"scheme":"http","hosts":["127.0.0.1"],"port":http_port,"path_prefixes":["/"]}],"transform_js":sausage::cred::summary::script("bearer")?,"fields":[{"name":"token","field_type":"text","public":false}]}))?;
+    browser.enter(&http_item, &token).await?;
+    let ssh_item = browser.create(json!({"name":"demo-ssh","rules":[{"scheme":"ssh","port":22,"hosts":["127.0.0.1"],"path_prefixes":["/"]}],"transform_js":"","fields":[{"name":"key","field_type":"ssh_private_key","key_type":"ed25519-raw","public":false}]}))?;
+    browser.enter(&ssh_item, &key).await?;
+    let source = format!(
+        "globalThis.p2Http = {http_port};\n{}",
+        include_str!("../js/p2-test-bundle.js")
+    );
+    let approve = async {
+        for item in [&http_item, &ssh_item] {
+            let deadline = Instant::now() + Duration::from_secs(90);
+            let request = loop {
+                let requests = browser.json("GET", "/cred/requests?state=pending", Value::Null)?;
+                if let Some(request) = requests.as_array().unwrap().iter().find(|request| {
+                    request["payload"]["items"]
+                        .as_array()
+                        .is_some_and(|items| items.contains(&item["id"]))
+                }) {
+                    break request["id"].as_str().unwrap().to_owned();
+                }
+                ensure!(
+                    Instant::now() < deadline,
+                    "conversation did not file a request"
+                );
+                tokio::task::yield_now().await;
+            };
+            let marker = if item["name"] == "demo-api" {
+                "p2-http-idle"
+            } else {
+                "p2-ssh-idle"
+            };
+            while !root.join(marker).exists() {
+                ensure!(
+                    Instant::now() < deadline,
+                    "conversation did not finish its turn"
+                );
+                tokio::task::yield_now().await;
+            }
+            browser.approve(&request)?;
+        }
+        Ok::<_, anyhow::Error>(())
+    };
+    let (harness, approvals) = tokio::time::timeout(Duration::from_secs(180), async {
+        tokio::try_join!(
+            sausage::harness::run_module_with_credentials(
+                world,
+                Arc::new(runtime),
+                cred.clone(),
+                &source
+            ),
+            approve
+        )
+    })
+    .await??;
+    let _ = (harness, approvals);
+    server.unblock();
+    ensure!(
+        thread.join().unwrap(),
+        "server did not receive the bearer token"
+    );
+    let local = Command::new("git")
+        .args(["rev-parse", "main"])
+        .current_dir(&repo)
+        .output()?;
+    let remote = Command::new("git")
+        .args(["rev-parse", "main"])
+        .current_dir(fixture.join("destination.git"))
+        .output()?;
+    ensure!(
+        local.status.success() && remote.status.success() && local.stdout == remote.stdout,
+        "SSH push did not update remote main"
+    );
+    let requests = browser.json("GET", "/cred/requests", Value::Null)?;
+    for item in [&http_item, &ssh_item] {
+        ensure!(
+            requests
+                .as_array()
+                .unwrap()
+                .iter()
+                .filter(|request| request["payload"]["items"]
+                    .as_array()
+                    .is_some_and(|items| items.contains(&item["id"])))
+                .count()
+                == 1,
+            "duplicate automatic item request"
+        );
+    }
+    no_secrets(&root, &[token.as_bytes(), key.as_bytes(), &seed])?;
+    for file in ["harness.sqlite", "harness.sqlite-wal"] {
+        if let Ok(bytes) = std::fs::read(dir.path().join("host").join(file)) {
+            for secret in [token.as_bytes(), key.as_bytes(), &seed] {
+                ensure!(
+                    !bytes.windows(secret.len()).any(|value| value == secret),
+                    "secret escaped into harness storage"
+                );
+            }
+        }
+    }
+    let audit = cred.audit_list()?;
+    ensure!(
+        audit
+            .iter()
+            .any(|entry| entry.event == "fetch.use"
+                && entry.item.as_deref() == http_item["id"].as_str()),
+        "HTTP use missing from audit"
+    );
+    ensure!(
+        audit
+            .iter()
+            .any(|entry| entry.event == "sign" && entry.item.as_deref() == ssh_item["id"].as_str()),
+        "SSH use missing from audit"
+    );
+    Ok(())
+}