fix(durable): record the requested thinking level on assistant messages

This commit is contained in:
Mario Zechner 2026-10-25 08:04:10 +00:00
parent 0eea54f11c
commit d5b3de22d0
4 changed files with 66 additions and 4 deletions

View file

@ -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

View file

@ -81,6 +81,8 @@ export type GenerationCheckpoint =
attempt: number;
compacted?: TaskId<CompactionResult>;
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<CompactionResult> | 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<GenerationInput, GenerationCheckpoint,
...(thinkingLevel === "off" ? {} : { reasoning: thinkingLevel }),
};
const message = await streamResponse(runtime, model, messages, options, attempt, context);
const request = { attempt, compacted, model: ref, cutoff, messages: view.messages };
const request = { attempt, compacted, model: ref, thinkingLevel, cutoff, messages: view.messages };
await classify(runtime, request, message, context);
},
/** After the backoff's sleep; a run before `until` (an early wake, or a record of an earlier version) sleeps again. */
@ -250,7 +253,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
},
/** After the sleep until `pollAt`; a run before it (an early wake, or a record of an earlier version) sleeps again. */
poll: async (task, runtime, context) => {
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<GenerationInput, GenerationCheckpoint,
signal: runtime.signal,
telemetry: context.telemetry,
});
const request = { attempt, compacted, model: ref, cutoff, pollAt };
const request = { attempt, compacted, model: ref, thinkingLevel, cutoff, pollAt };
await classify(runtime, request, message, context);
},
tools: async (task, runtime, context) => {
@ -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,

View file

@ -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 } });

View file

@ -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) => {