fix(durable): continue runs after tool handoffs

packages/durable/CHANGELOG.md

@@ -22,6 +22,7 @@
 
 ### Fixed
 
+- Fixed a tool round with `control.handoff` ending the run on the tool-calling answer: the handoff's `pi.reset` entry is written and the same run continues with a new generation in the new context, so the submission settles with the eventual answer. Queued writes and selected steers are placed after the handoff text, follow-ups wait for the run's final boundary, and a queued reset still ends the run `unanswered` with reason `reset`. `terminate` remains the way to end a round without another model request.
 - Fixed a conversation watcher (`watchEvents()`, conversation views) holding a copy of a streamed text or tool call per partial commit, through Chord append ops that kept the whole string alive; a 140 KB answer followed by a 140 KB tool call held well over 100 MB ([#9265](https://github.com/earendil-works/pi/issues/9265))
 - Fixed a nested call's error summary keeping the call's whole output alive through a slice of it ([#9265](https://github.com/earendil-works/pi/issues/9265))
 - Fixed task span error messages and codemode call previews keeping the whole text they were cut from alive ([#9265](https://github.com/earendil-works/pi/issues/9265))

packages/durable/README.md

@@ -538,7 +538,9 @@ await root.reset(undefined, context);                                  // start
 await root.reset("We were fixing the flaky login test. Continue.", context); // start from a handoff note
 ```
 
-While busy, the reset is queued like a write. When it is placed during a tool round, the current run ends. A tool can request the same with `control: { handoff: "..." }`.
+While busy, the reset is queued like a write. When it is placed during a tool round, the current run ends: its inputs settle `unanswered` with reason `reset`, and queued inputs run in the new context.
+
+A tool can return `control: { handoff: "..." }` to write the same reset with its text, but the run continues: the next model request starts in the new context, and the submission settles with that eventual answer. Queued writes and steers land after the handoff text, follow-ups wait for the run's end, and a queued `reset()` still ends the run. When every call of the round also returns `terminate: true`, the run ends after the handoff instead.
 
 ## Compaction
 

packages/durable/docs/spec.md

@@ -2730,8 +2730,11 @@ A reset placed at a `final` boundary follows the answer, so the inputs are
 already `done`.
 
 At ordinary `postTools`, generation continues even with no queued trigger;
-selected steer IDs join that continuation. A terminating or handoff tool round
+selected steer IDs join that continuation. A handoff tool round (section 8.5)
-(section 8.5) uses final behavior instead. At `final`, the current run's placed
+first appends its reset and then applies this ordinary `postTools` boundary, so
+queued writes and selected steers land in the new context and follow-ups wait
+for the run's final boundary. A terminating tool round uses final behavior
+instead. At `final`, the current run's placed
 input submissions settle first, unless an `onYield` continuation keeps them open
 (below); selected user IDs start one successor generation with those IDs as the
 new run's inputs. Writes never trigger generation by
@@ -2739,10 +2742,9 @@ themselves. A final boundary without user triggers leaves the conversation idle,
 except for an `onYield` continuation (section 8.3), which applies only when the
 boundary selected no user item and no reset.
 
-Only successful run ends apply the final boundary: an answer, `terminate`, or
+Only successful run ends apply the final boundary: an answer or `terminate`.
-`handoff`. Failure, a run task's abort handler, fault, and orphan settle the
+Failure, a run task's abort handler, fault, and orphan settle the run's inputs
-run's inputs `unanswered` and leave the inbox alone (`Conversation.abort()`
+`unanswered` and leave the inbox alone (`Conversation.abort()` separately withdraws queued user items); the queued items stay visible in
-separately withdraws queued user items); the queued items stay visible in
 `pi.inbox` until the next input or queued write runs a boundary, or their withdrawal.
 
 Selected and stale items are removed positionally while retained item order is
@@ -3703,10 +3705,13 @@ generation's `tools` phase (section 8.5) adds the named tools to the
 conversation's stored tools filter; they take effect at the next preparation. The round
 terminates only when every result of the round requests `terminate`, as in the
 pi agent loop; the `tools` phase then uses a final boundary. Any
-`handoff` in the round, the last one in call order when several ask, ends the
+`handoff` in the round, the last one in call order when several ask, appends a
-run the same way after appending a `pi.reset` entry with `head: "self"` and the
+`pi.reset` entry with `head: "self"` and the handoff text as a user message
-handoff text as a user message (section 8.1), exactly what `reset(handoff)`
+(section 8.1), exactly what `reset(handoff)` writes, and the run continues: the
-writes.
+next generation answers in the new context, and the run's inputs settle with
+that eventual answer or error, not with the tool-calling answer. A later round
+may hand off again. A round that also terminates writes the reset and then ends
+the run as `terminate` does.
 
 A result's `usage` is stored on the tool-result message and added to the
 conversation's `pi.usage.tools[toolName]` in the result commit (section 8.6). A
@@ -4516,15 +4521,20 @@ and then commits once:
   no effect for others. Names are written without consulting the registry, so a
   tool that is not registered now is offered once it is again. The next
   preparation offers the tool when it resolves and the model may call it.
-- When every result of the round requests `terminate`, or any requests
+- When any result requests `handoff`, it first appends the `pi.reset` entry of
-  `handoff`, it appends the handoff's `pi.reset` entry, if any, settles the
+  the last one in call order and advances the boundary's active range start to
-  run's inputs `done` with the tool-calling answer, removes `run` and `tools`,
+  it, so the boundary below places its writes and user items after the handoff
-  and applies the final boundary.
+  text, and a queued head write targeting an entry before it is stale.
+- When every result of the round requests `terminate`, it settles the run's
+  inputs `done` with the tool-calling answer, removes `run` and `tools`, and
+  applies the final boundary.
 - Otherwise it removes `tools` and applies the `postTools` boundary. When that
-  boundary placed a reset, the run's inputs settle `unanswered` with `reset` and
+  boundary placed a queued reset, which wins over a handoff written before it,
-  selected user items start a successor run (section 6). Otherwise selected
+  the run's inputs settle `unanswered` with `reset` and selected user items
-  steer IDs join `pi.live.run`, and it creates the next generation, owned by the
+  start a successor run (section 6). Otherwise selected steer IDs join
-  conversation, and hands it the run.
+  `pi.live.run`, and it creates the next generation, owned by the conversation,
+  and hands it the run; after a handoff, that generation prepares and requests
+  in the new context.
 
 It completes with `{ entryId }` of the tool-calling answer. Its abort handler
 runs only after its tool tasks are terminal (section 5.5). For every call still

packages/durable/src/harness/generation.ts

@@ -690,8 +690,9 @@ async function startToolRound(
 }
 
 /**
- * The round's tools are terminal: apply their controls and either end the run at the final boundary (`terminate`,
+ * The round's tools are terminal: apply their controls, append a `handoff`'s reset, and then end the run at the final
- * `handoff`, or a queued reset) or hand it to the next generation at the `postTools` boundary (spec §8.5).
+ * boundary (`terminate`), end it at a queued reset, or hand it to the next generation at the `postTools` boundary, in
+ * the handoff's new context when there is one (spec §8.5).
  */
 async function finishToolRound(
 	runtime: Runtime,
@@ -725,12 +726,13 @@ async function finishToolRound(
 		if (added.length > 0) await addTools(tx, conversationId, added);
 		const live = await tx.doc(LiveDoc, conversationId);
 		const now = runtime.now();
-		if (terminate || handoff !== undefined) {
+		if (handoff !== undefined) {
-			if (handoff !== undefined) {
+			// The handoff's reset precedes every queued item, so queued writes and selected steers land in its context.
-				const message = { role: "user", content: handoff, timestamp: now } as const;
+			const message = { role: "user", content: handoff, timestamp: now } as const;
-				const entry = await tx.appendEntry(ResetEntry, conversationId, { head: "self", model: [message] });
+			const entry = await tx.appendEntry(ResetEntry, conversationId, { head: "self", model: [message] });
-				boundary.head = entry.id;
+			boundary.head = entry.id;
-			}
+		}
+		if (terminate) {
 			const { users } = await applyBoundary(tx, boundary, "final", now);
 			endRun(tx, live, runtime.taskId, { status: "done", answer: assistant });
 			if (users.length > 0) await startRun(tx, conversationId, live, users);

packages/durable/src/harness/types.ts

@@ -142,6 +142,10 @@ export type ToolControl = {
 	readonly addTools?: readonly string[];
 	/** End the run without another model request when every result of the round sets it to `true`; `false` is a no-op. */
 	readonly terminate?: boolean;
+	/**
+	 * Start a new context with this text as its user message, as `reset(handoff)` writes, and continue the run there.
+	 * The last handoff in call order wins.
+	 */
 	readonly handoff?: string;
 };
 

packages/durable/test/harness-inbox.test.ts

@@ -82,6 +82,14 @@ function queuedNote(text: string) {
 	return { kind: "note", model: [{ role: "user", content: text, timestamp: 0 }] } as const;
 }
 
+/** A faux response that records the request's messages as `role:text` and answers `text`. */
+function recordRequest(requests: string[][], text: string): FauxResponseStep {
+	return (request) => {
+		requests.push(request.messages.map((message) => `${message.role}:${textOf(message) ?? ""}`));
+		return answer(text);
+	};
+}
+
 const HOLD = fauxAssistantMessage([fauxToolCall("hold", {}, { id: "c1" })], { stopReason: "toolUse" });
 
 async function status(submission: Submission): Promise<SubmissionRecord> {
@@ -239,11 +247,7 @@ describe("inbox", () => {
 		const gate = deferred();
 		holdTool(setup, gate);
 		const requests: string[][] = [];
-		const record: FauxResponseStep = (request) => {
+		setup.faux.setResponses([HOLD, recordRequest(requests, "fresh")]);
-			requests.push(request.messages.map((message) => `${message.role}:${textOf(message) ?? ""}`));
-			return answer("fresh");
-		};
-		setup.faux.setResponses([HOLD, record]);
 		const { harness, root } = await openChat(new MemoryStorage(), setup);
 		const input = await root.submit({ type: "input", content: "a" }, context);
 		await toolRunning(harness, root);
@@ -348,11 +352,164 @@ describe("inbox", () => {
 		await harness.close(context);
 	});
 
-	it("ends the run with a pi.reset entry when a tool requests a handoff", async () => {
+	it("continues the run in a new context when a tool requests a handoff", async () => {
 		const setup = chatSetup();
 		const gate = deferred();
 		gate.resolve();
 		holdTool(setup, gate, { output: [], control: { handoff: "continue here" } });
+		const requests: string[][] = [];
+		setup.faux.setResponses([HOLD, recordRequest(requests, "after handoff")]);
+		const { harness, root } = await openChat(new MemoryStorage(), setup);
+		const settled = await (await root.submit({ type: "input", content: "a" }, context)).wait(context);
+		const entries = await allEntries(root);
+		expect(settled).toMatchObject({ status: "done", answer: entries.at(-1)!.id });
+		expect(transcript(entries)).toEqual([
+			"pi.user:a",
+			"pi.assistant",
+			"pi.tool-result",
+			"pi.reset:continue here",
+			"pi.assistant:after handoff",
+		]);
+		const reset = entries.find((entry) => ResetEntry.is(entry))!;
+		expect(reset.head).toBe(reset.id);
+		expect(requests).toEqual([["system:", "user:continue here"]]);
+		expect(setup.faux.state.callCount).toBe(2);
+		expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({});
+		await harness.close(context);
+	});
+
+	it("places queued writes and steers after the handoff text and adds the steers to the continuing run", async () => {
+		const setup = chatSetup();
+		const gate = deferred();
+		holdTool(setup, gate, { output: [], control: { handoff: "continue here" } });
+		const requests: string[][] = [];
+		setup.faux.setResponses([HOLD, recordRequest(requests, "after handoff")]);
+		const { harness, root } = await openChat(new MemoryStorage(), setup);
+		const input = await root.submit({ type: "input", content: "a" }, context);
+		await toolRunning(harness, root);
+		const steer = await root.submit({ type: "input", content: "s", whenBusy: "steer" }, context);
+		const write = await root.submit({ type: "write", entry: queuedNote("w") }, context);
+		gate.resolve();
+		await input.wait(context);
+		const entries = await allEntries(root);
+		expect(transcript(entries)).toEqual([
+			"pi.user:a",
+			"pi.assistant",
+			"pi.tool-result",
+			"pi.reset:continue here",
+			"note:w",
+			"pi.user:s",
+			"pi.assistant:after handoff",
+		]);
+		expect(requests).toEqual([["system:", "user:continue here", "user:w", "user:s"]]);
+		for (const submission of [input, steer]) {
+			expect(await status(submission)).toMatchObject({ status: "done", answer: entries.at(-1)!.id });
+		}
+		expect(await status(write)).toMatchObject({ status: "done" });
+		await harness.close(context);
+	});
+
+	it("holds a follow-up queued across a handoff until the continuing run's final boundary", async () => {
+		const setup = chatSetup();
+		const gate = deferred();
+		holdTool(setup, gate, { output: [], control: { handoff: "continue here" } });
+		const after = gated(answer("after handoff"));
+		setup.faux.setResponses([HOLD, after.step, answer("follow-up")]);
+		const { harness, root } = await openChat(new MemoryStorage(), setup);
+		const input = await root.submit({ type: "input", content: "a" }, context);
+		await toolRunning(harness, root);
+		const followUp = await root.submit({ type: "input", content: "f" }, context);
+		gate.resolve();
+		await after.reached;
+		expect((await status(input)).status).toBe("placed");
+		expect(await inbox(harness, root)).toEqual([[followUp.id, "followUp"]]);
+		after.release();
+		await followUp.wait(context);
+		const entries = await allEntries(root);
+		expect(transcript(entries)).toEqual([
+			"pi.user:a",
+			"pi.assistant",
+			"pi.tool-result",
+			"pi.reset:continue here",
+			"pi.assistant:after handoff",
+			"pi.user:f",
+			"pi.assistant:follow-up",
+		]);
+		const answers = entries.filter((entry) => entry.kind === "pi.assistant");
+		expect(await status(input)).toMatchObject({ status: "done", answer: answers[1]!.id });
+		expect(await status(followUp)).toMatchObject({ status: "done", answer: answers[2]!.id });
+		await harness.close(context);
+	});
+
+	it("ends a handed-off run at a queued reset and runs earlier follow-ups after it", async () => {
+		const setup = chatSetup();
+		const gate = deferred();
+		holdTool(setup, gate, { output: [], control: { handoff: "continue here" } });
+		const requests: string[][] = [];
+		setup.faux.setResponses([HOLD, recordRequest(requests, "fresh")]);
+		const { harness, root } = await openChat(new MemoryStorage(), setup);
+		const input = await root.submit({ type: "input", content: "a" }, context);
+		await toolRunning(harness, root);
+		const followUp = await root.submit({ type: "input", content: "f" }, context);
+		await root.reset(undefined, context);
+		gate.resolve();
+		await followUp.wait(context);
+		expect(await status(input)).toMatchObject({ status: "unanswered", reason: "reset" });
+		expect(transcript(await allEntries(root))).toEqual([
+			"pi.user:a",
+			"pi.assistant",
+			"pi.tool-result",
+			"pi.reset:continue here",
+			"pi.reset",
+			"pi.user:f",
+			"pi.assistant:fresh",
+		]);
+		expect(requests).toEqual([["system:", "user:f"]]);
+		expect(setup.faux.state.callCount).toBe(2);
+		await harness.close(context);
+	});
+
+	it("hands off again in a later round of the same run", async () => {
+		const setup = chatSetup();
+		let handoffs = 0;
+		addTool(
+			setup.registry,
+			defineTool({
+				name: "hold",
+				description: "Hands off",
+				parameters: Type.Object({}),
+				execute: async () => ({ output: [], control: { handoff: `h${++handoffs}` } }),
+			}),
+		);
+		setup.faux.setResponses([HOLD, HOLD, answer("done")]);
+		const { harness, root } = await openChat(new MemoryStorage(), setup);
+		const settled = await (await root.submit({ type: "input", content: "a" }, context)).wait(context);
+		const entries = await allEntries(root);
+		expect(transcript(entries)).toEqual([
+			"pi.user:a",
+			"pi.assistant",
+			"pi.tool-result",
+			"pi.reset:h1",
+			"pi.assistant",
+			"pi.tool-result",
+			"pi.reset:h2",
+			"pi.assistant:done",
+		]);
+		expect(settled).toMatchObject({ status: "done", answer: entries.at(-1)!.id });
+		const messages = (await root.context(context)).messages;
+		expect(messages.map((message) => `${message.role}:${textOf(message) ?? ""}`)).toEqual([
+			"system:",
+			"user:h2",
+			"assistant:done",
+		]);
+		await harness.close(context);
+	});
+
+	it("ends the run after the handoff when the round also terminates", async () => {
+		const setup = chatSetup();
+		const gate = deferred();
+		gate.resolve();
+		holdTool(setup, gate, { output: [], control: { handoff: "continue here", terminate: true } });
 		setup.faux.setResponses([HOLD]);
 		const { harness, root } = await openChat(new MemoryStorage(), setup);
 		const settled = await (await root.submit({ type: "input", content: "a" }, context)).wait(context);
@@ -360,7 +517,6 @@ describe("inbox", () => {
 		const calling = entries.find((entry) => entry.kind === "pi.assistant")!;
 		expect(settled).toMatchObject({ status: "done", answer: calling.id });
 		expect(transcript(entries)).toEqual(["pi.user:a", "pi.assistant", "pi.tool-result", "pi.reset:continue here"]);
-		expect(entries.at(-1)!.head).toBe(entries.at(-1)!.id);
 		expect(setup.faux.state.callCount).toBe(1);
 		expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({});
 		await harness.close(context);
@@ -630,7 +786,7 @@ describe("inbox", () => {
 		await harness.close(context);
 	});
 
-	it("writes the last handoff in call order and then runs queued follow-ups in the new context", async () => {
+	it("writes only the last handoff in call order and runs queued follow-ups after the continuing run", async () => {
 		const setup = chatSetup();
 		const firstGate = deferred();
 		const secondGate = deferred();
@@ -651,7 +807,7 @@ describe("inbox", () => {
 			[fauxToolCall("hold", {}, { id: "c1" }), fauxToolCall("later", {}, { id: "c2" })],
 			{ stopReason: "toolUse" },
 		);
-		setup.faux.setResponses([round, answer("follow-up")]);
+		setup.faux.setResponses([round, answer("after handoff"), answer("follow-up")]);
 		const { harness, root } = await openChat(new MemoryStorage(), setup);
 		const input = await root.submit({ type: "input", content: "a" }, context);
 		await toolRunning(harness, root);
@@ -660,13 +816,19 @@ describe("inbox", () => {
 		await waitFor(async () => (await harness.snapshot(LiveDoc, root.id, context))?.tools?.[1]?.status === "done");
 		firstGate.resolve();
 		await f.wait(context);
-		expect(await status(input)).toMatchObject({ status: "done" });
+		const entries = await allEntries(root);
-		expect(transcript(await allEntries(root)).slice(-4)).toEqual([
+		expect(transcript(entries)).toEqual([
+			"pi.user:a",
+			"pi.assistant",
+			"pi.tool-result",
 			"pi.tool-result",
 			"pi.reset:two",
+			"pi.assistant:after handoff",
 			"pi.user:f",
 			"pi.assistant:follow-up",
 		]);
+		const answers = entries.filter((entry) => entry.kind === "pi.assistant");
+		expect(await status(input)).toMatchObject({ status: "done", answer: answers[1]!.id });
 		await harness.close(context);
 	});