diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 15e4d75e4..e061eb6b7 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -27,6 +27,7 @@ - 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)) - Fixed partial commits of a streamed answer copying every growing text, thinking and tool-call argument string in full on each commit (about 20 s of CPU and a 100 MB heap peak for a 500 KB tool call): they now append new text with Chord's `appendText()`, and apply the argument changes of their own parse of the streamed argument text ([#9265](https://github.com/earendil-works/pi/issues/9265)) +- Fixed stored assistant messages lacking `thinkingLevel`: a generation records the Pi thinking level its request asked for on the response, also after polling a deferred one, as pi-agent-core's agent loop does ## [1.4.0] - 2026-10-21 diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index 11bf946f2..210ae981e 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -81,6 +81,8 @@ export type GenerationCheckpoint = attempt: number; compacted?: TaskId; model: ModelRef; + /** Absent in checkpoints of earlier versions. */ + thinkingLevel?: ModelThinkingLevel; cutoff: EntryId; handle: DeferredHandle; pollAt: number; @@ -106,6 +108,7 @@ type Request = { readonly attempt: number; readonly compacted: TaskId | undefined; readonly model: ModelRef; + readonly thinkingLevel: ModelThinkingLevel | undefined; readonly cutoff: EntryId; /** Committed model context through `cutoff`, when the phase already derived it. */ readonly messages?: readonly Message[]; @@ -231,7 +234,7 @@ export const GenerationTask = defineTask { - const { attempt, compacted, model: ref, cutoff, handle, pollAt } = task.state.checkpoint; + const { attempt, compacted, model: ref, thinkingLevel, cutoff, handle, pollAt } = task.state.checkpoint; if (pollAt > runtime.now()) return runtime.sleepUntil(pollAt, task.state.checkpoint, context); const model = runtime.models.getModel(ref.provider, ref.modelId); if (model === undefined) return failNoModel(runtime, ref, context); @@ -258,7 +261,7 @@ export const GenerationTask = defineTask { @@ -522,7 +525,9 @@ async function classify( // An abort mark or close: the abort invocation or the reopened run handles the committed state. runtime.signal.throwIfAborted(); const conversationId = runtime.conversationId; - const { attempt, compacted, model: ref, cutoff } = request; + const { attempt, compacted, model: ref, thinkingLevel, cutoff } = request; + // Record the requested level, as pi-agent-core's agent loop does. + if (thinkingLevel !== undefined) message.thinkingLevel = thinkingLevel; if (message.stopReason === "deferred" && message.deferred !== undefined) { const handle = message.deferred; const pollAt = Math.max( @@ -536,6 +541,7 @@ async function classify( attempt, ...(compacted === undefined ? {} : { compacted }), model: ref, + ...(thinkingLevel === undefined ? {} : { thinkingLevel }), cutoff, handle, pollAt, diff --git a/packages/durable/test/harness-generation-recovery.test.ts b/packages/durable/test/harness-generation-recovery.test.ts index ef104ac4c..468723ce4 100644 --- a/packages/durable/test/harness-generation-recovery.test.ts +++ b/packages/durable/test/harness-generation-recovery.test.ts @@ -287,6 +287,43 @@ describe("generation recovery", () => { await opened.harness.close(context); }); + it("polls a checkpoint stored without a thinking level and records none", async () => { + const path = await sqlitePath(); + const setup = chatSetup({ deferred: { pollAfterMs: 60_000 } }); + let now = 1_000; + setup.now = () => now; + setup.faux.setResponses([fauxAssistantMessage("deferred answer")]); + let opened = await open(path, setup); + await opened.root.configure({ thinkingLevel: "high" }, context); + opened.harness.resume(); + setup.settings.stream = { deferred: true }; + const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id; + await waitFor(async () => { + const live = await opened.harness.snapshot(LiveDoc, opened.root.id, context); + return live?.generation?.deferred !== undefined; + }); + const taskId = await runTaskId(opened.harness); + expect(await checkpoint(opened.harness, taskId)).toMatchObject({ phase: "poll", thinkingLevel: "high" }); + await opened.harness.close(context); + // Earlier versions stored poll checkpoints without the field. + const database = new DatabaseSync(path); + database + .prepare("UPDATE tasks SET record = json_remove(record, '$.state.checkpoint.thinkingLevel') WHERE id = ?") + .run(taskId); + database.close(); + + opened = await open(path, setup); + expect(await checkpoint(opened.harness, taskId)).not.toHaveProperty("thinkingLevel"); + now = 61_000; + opened.harness.resume(); + const settled = await (await opened.harness.submission(id, context))!.wait(context); + if (settled.status !== "done" || settled.type !== "input") throw new Error(`Unexpected ${settled.status}`); + const answer = (await opened.root.commit((tx) => tx.entry(AssistantEntry, settled.answer), context))?.model?.[0]; + expect(textOf(answer)).toBe("deferred answer"); + expect(answer).not.toHaveProperty("thinkingLevel"); + await opened.harness.close(context); + }); + it("fails no_model when the pinned model is gone after reopen, in request and in poll", async () => { const path = await sqlitePath(); const setup = chatSetup({ deferred: { pollAfterMs: 60_000 } }); diff --git a/packages/durable/test/harness-generation.test.ts b/packages/durable/test/harness-generation.test.ts index 3cb077264..05a5f5359 100644 --- a/packages/durable/test/harness-generation.test.ts +++ b/packages/durable/test/harness-generation.test.ts @@ -385,6 +385,24 @@ describe("generation", () => { await harness.close(context); }); + it("records the requested thinking level on answers, streamed and polled", async () => { + const setup = chatSetup({ deferred: { pendingFetches: 1, pollAfterMs: 1 } }); + setup.faux.setResponses([fauxAssistantMessage("streamed"), fauxAssistantMessage("polled")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + harness.resume(); + const answer = async (content: string) => { + const settled = await (await root.submit({ type: "input", content }, context)).wait(context); + if (settled.status !== "done" || settled.type !== "input") throw new Error(`Unexpected ${settled.status}`); + return (await root.commit((tx) => tx.entry(AssistantEntry, settled.answer), context))?.model?.[0]; + }; + await root.configure({ thinkingLevel: "high" }, context); + expect(await answer("one")).toMatchObject({ thinkingLevel: "high" }); + setup.settings.stream = { deferred: true }; + expect(await answer("two")).toMatchObject({ thinkingLevel: "high" }); + expect(setup.faux.state.deferredFetchCount).toBe(2); + await harness.close(context); + }); + it("attributes model requests to pi by default and merges the Harness identity over it", async () => { const seen: (SimpleStreamOptions["identity"] | undefined)[] = []; const capture = (_context: unknown, options?: SimpleStreamOptions) => {