diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index eff3b87a9..15e4d75e4 100644 --- a/packages/durable/CHANGELOG.md +++ b/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)) diff --git a/packages/durable/README.md b/packages/durable/README.md index 452b7c740..0ce9885bb 100644 --- a/packages/durable/README.md +++ b/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 diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 1b95481b1..3af8ff115 100644 --- a/packages/durable/docs/spec.md +++ b/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 -(section 8.5) uses final behavior instead. At `final`, the current run's placed +selected steer IDs join that continuation. A handoff tool round (section 8.5) +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 -`handoff`. Failure, a run task's abort handler, fault, and orphan settle the -run's inputs `unanswered` and leave the inbox alone (`Conversation.abort()` -separately withdraws queued user items); the queued items stay visible in +Only successful run ends apply the final boundary: an answer or `terminate`. +Failure, a run task's abort handler, fault, and orphan settle the run's inputs +`unanswered` and leave the inbox alone (`Conversation.abort()` 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 -run the same way after appending a `pi.reset` entry with `head: "self"` and the -handoff text as a user message (section 8.1), exactly what `reset(handoff)` -writes. +`handoff` in the round, the last one in call order when several ask, appends a +`pi.reset` entry with `head: "self"` and the handoff text as a user message +(section 8.1), exactly what `reset(handoff)` writes, and the run continues: the +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 - `handoff`, it appends the handoff's `pi.reset` entry, if any, settles the - run's inputs `done` with the tool-calling answer, removes `run` and `tools`, - and applies the final boundary. +- When any result requests `handoff`, it first appends the `pi.reset` entry of + the last one in call order and advances the boundary's active range start to + it, so the boundary below places its writes and user items after the handoff + 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 - selected user items start a successor run (section 6). Otherwise selected - steer IDs join `pi.live.run`, and it creates the next generation, owned by the - conversation, and hands it the run. + boundary placed a queued reset, which wins over a handoff written before it, + the run's inputs settle `unanswered` with `reset` and selected user items + start a successor run (section 6). Otherwise selected steer IDs join + `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 diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index 087ec9a97..11bf946f2 100644 --- a/packages/durable/src/harness/generation.ts +++ b/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`, - * `handoff`, or a queued reset) or hand it to the next generation at the `postTools` boundary (spec §8.5). + * The round's tools are terminal: apply their controls, append a `handoff`'s reset, and then end the run at the final + * 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) { - const message = { role: "user", content: handoff, timestamp: now } as const; - const entry = await tx.appendEntry(ResetEntry, conversationId, { head: "self", model: [message] }); - boundary.head = entry.id; - } + 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 entry = await tx.appendEntry(ResetEntry, conversationId, { head: "self", model: [message] }); + 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); diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index fbe636dd7..748696f8c 100644 --- a/packages/durable/src/harness/types.ts +++ b/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; }; diff --git a/packages/durable/test/harness-inbox.test.ts b/packages/durable/test/harness-inbox.test.ts index 19a719286..396c94e59 100644 --- a/packages/durable/test/harness-inbox.test.ts +++ b/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 { @@ -239,11 +247,7 @@ describe("inbox", () => { const gate = deferred(); holdTool(setup, gate); const requests: string[][] = []; - const record: FauxResponseStep = (request) => { - requests.push(request.messages.map((message) => `${message.role}:${textOf(message) ?? ""}`)); - return answer("fresh"); - }; - setup.faux.setResponses([HOLD, record]); + 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); @@ -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" }); - expect(transcript(await allEntries(root)).slice(-4)).toEqual([ + const entries = await allEntries(root); + 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); });