diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index b682090a9..52c08ec13 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Persisted a distinct provider session UUID per conversation and forwarded it for prompt-cache and session affinity ([#10424](https://github.com/earendil-works/pi/issues/10424)) + ## [1.0.1] - 2026-10-03 ## [1.0.0] - 2026-10-01 diff --git a/packages/durable/README.md b/packages/durable/README.md index fef41c069..521ffcaee 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -81,7 +81,7 @@ Every async call takes a Chord `Context`, which carries cancellation. `BACKGROUN - **Conversation**: a transcript. `root()` creates the root conversation on first use; you can create more and fork them. A `Conversation` handle holds no state; compare handles by `id`. - **Entry**: one immutable transcript record, such as a user message (`pi.user`), a model response (`pi.assistant`), a tool result (`pi.tool-result`), a system prompt change (`pi.system`), a reset (`pi.reset`), or your own kind. The model sees the entries from the newest reset onward. - **Commit**: an atomic write. `conversation.commit((tx) => ...)` can append entries, edit documents, and create tasks together; either all of it is stored or none of it. -- **Document**: typed JSON state stored next to the transcript and changed in commits. Built-in ones hold each conversation's agent choices (`pi.agent`), the running generation and tools (`pi.live`), queued submissions (`pi.inbox`), and spend (`pi.usage`). +- **Document**: typed JSON state stored next to the transcript and changed in commits. Built-in ones hold each conversation's agent choices (`pi.agent`), provider-facing session identity (`pi.provider`), running generation and tools (`pi.live`), queued submissions (`pi.inbox`), and spend (`pi.usage`). - **Task**: a durable state machine that saves a checkpoint at every step, so a restarted process continues from the last one. Every task has an owner: its conversation, or another task. The Harness runs answers as built-in tasks: `pi.generation` calls the model and owns the `pi.tool` tasks of its tool calls, waits for them, and hands the run to the next generation. - **Submission**: something you hand to a conversation, either user input or an entry to write, which you can wait for. - **Turn and run**: a turn is one model response and its tool calls; a run is the turns from an input to its final answer. A conversation is busy while a run is going. @@ -110,7 +110,9 @@ const root = await harness.root(context); // the same root as last time harness.resume(); // continue any run the last process left unfinished ``` -Work interrupted by a crash or close stays pending. `resume()` starts the task scheduler; submitting or waiting starts it too. A retried submission with the same `requestId` returns the existing submission instead of submitting twice: +Work interrupted by a crash or close stays pending. `resume()` starts the task scheduler; submitting or waiting starts it too. Each conversation has its own persisted UUIDv7 in `pi.provider`, forwarded to pi-ai as `sessionId` for provider prompt-cache and session affinity. It survives reopen, retries, reset, compaction, and model changes; a child or fork receives a fresh identity. A legacy conversation receives and persists one before its first generation or compaction request. + +A retried submission with the same `requestId` returns the existing submission instead of submitting twice: ```typescript const submission = await root.submit({ type: "input", content: "Hello", requestId: "greeting-1" }, context); @@ -265,7 +267,7 @@ const view = await root.viewState(context); view.subscribe((value) => { // value.entries: the active transcript // value.docs["pi.live"]: the running generation (streamed partial, retry, deferred) and tool calls (output, details) - // value.docs["pi.inbox"], value.docs["pi.usage"], value.docs["pi.agent"] + // value.docs["pi.inbox"], value.docs["pi.usage"], value.docs["pi.agent"], value.docs["pi.provider"] render(value); }); // later: view.dispose(); @@ -392,7 +394,7 @@ const other = await harness.createConversation({ ownership: { kind: "ownerless" const fork = await root.fork(entryId, { ownership: { kind: "ownerless" } }, context); ``` -A fork sees its parent's entries up to `entryId` and continues independently. It keeps the parent's agent as of that entry. Both take `agent` and `init`, applied in the creating commit. +A fork sees its parent's entries up to `entryId` and continues independently. It keeps the parent's agent as of that entry but receives a fresh provider session identity. Both take `agent` and `init`, applied in the creating commit. ## Abort and Subagents diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index e30cc9368..862129117 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -654,7 +654,8 @@ raw `tx.createConversation()` and `tx.forkConversation()`, for example inside a tool commit. It runs inside `tx.createConversation()` and `tx.forkConversation()`, before they return, so a `configure()` later in the same callback overrides its copy. It creates empty `pi.live`, `pi.inbox`, and -`pi.usage`, and handles `pi.agent`: +`pi.usage`, creates `pi.provider` with a fresh provider-facing UUIDv7, and handles +`pi.agent`: - A fork keeps the `asOf` copy of its parent's `pi.agent` (section 3.7), whatever its ownership. @@ -733,6 +734,29 @@ A change neither starts generation nor appends a system entry. Request preparation later compares the desired prompt and tools with transcript history and appends the required positional system baseline or delta (section 7.4). +The built-in provider document is final at version 1: + +| field | value | +|---|---| +| kind | `pi.provider` | +| version | `1` (no migration) | +| scope/history/fork | conversation, `latest`, `initial` | +| schema | `{ sessionId: string }` | +| `initial()` | `{ sessionId: uuidv7() }` | +| checkpoint | complete base on every change | +| view mount | `docs["pi.provider"]` | + +The UUID is a provider-facing conversation identity, not the numeric Durable +`ConversationId` and not an enclosing application's Session ID. Every new, +task-owned, raw-created, and forked conversation receives its own UUID in its +creating commit; a fork never copies its parent's UUID. Generation requests and +compaction summarization pass it to pi-ai as `options.sessionId`. Reset, +compaction, model changes, and reopen do not change it. A legacy conversation without the document creates and persists it +on the Session line before its first generation or compaction provider request. +Concurrent callers therefore observe one winner. Provider behavior still +applies: for example, Codex suppresses cache/session identity when +`cacheRetention` is `"none"`. + A missing tool implementation never fails a request. Request preparation offers only the agent's resolved tools. If the replayed tool state still offers a tool that no longer resolves, the appended system delta lists it in `toolsRemoved`; @@ -3397,8 +3421,9 @@ The run's inputs live in `pi.live.run`, not in the task input. - `request` first converts a committed partial left in `pi.live` by an interrupted attempt into an aborted `pi.assistant` entry. It then streams the model context through `cutoff` with the invocation signal, the thinking level - as `reasoning` (omitted for `off`), and the pinned `streamOptions`, - committing throttled partials. Before streaming, the `beforeRequest` chain may + as `reasoning` (omitted for `off`), the conversation's persisted provider + `sessionId`, and the pinned `streamOptions`, committing throttled partials. + Before streaming, the `beforeRequest` chain may replace the messages for this request only. Recovery resends the same committed messages with the same pinned model, thinking level, and stream options, and reruns `beforeRequest`. @@ -3680,9 +3705,11 @@ Phases: system prompt, and a user message with the serialized text in `` tags followed by the built-in summarization prompt, and `Additional focus: ` when the input has instructions. It uses the - pinned thinking level as `reasoning`, and the pinned stream options without - `deferred`, with `cacheRetention: "none"` and the pinned `maxTokens`. - Recovery resends the same request. The response is classified in one commit + pinned thinking level as `reasoning`, the conversation's persisted provider + `sessionId`, and the pinned stream options without `deferred`, with + `cacheRetention: "none"` and the pinned `maxTokens`. Providers such as Codex + may suppress that identity when caching is disabled. Recovery resends the same + request. The response is classified in one commit that adds its usage to `pi.usage` (section 8.6): - `stop` with non-empty text and no tool call: the text is the summary, placed as below. @@ -3911,7 +3938,7 @@ type ConversationView = { readonly conversation: ConversationRecord; /** Raw active entries, as `ContextView.entries` (section 2.1): the head marker, then the non-head entries from its head. */ readonly entries: readonly EntryRecord[]; - /** `pi.agent`, `pi.live`, `pi.inbox`, and `pi.usage`, keyed by kind; absent documents are absent. */ + /** `pi.agent`, `pi.live`, `pi.inbox`, `pi.provider`, and `pi.usage`, keyed by kind; absent documents are absent. */ readonly docs: Readonly>; }; ``` diff --git a/packages/durable/src/harness/compaction.ts b/packages/durable/src/harness/compaction.ts index 5f36642cf..89db279d8 100644 --- a/packages/durable/src/harness/compaction.ts +++ b/packages/durable/src/harness/compaction.ts @@ -16,6 +16,7 @@ import type { } from "../types.ts"; import { orderToolResults } from "./context.ts"; import { addCompactionStatus, compactionStatus, LiveDoc, type LiveState, removeCompactionStatus } from "./live.ts"; +import { ensureProviderSessionId } from "./provider.ts"; import { admitSubmission } from "./submissions.ts"; import type { CompactionHooks, @@ -167,6 +168,7 @@ export const CompactionTask = defineTask extends SessionImpl implements /** * The built-in creation hook, in every commit that creates or forks a conversation: empty `pi.live`, `pi.inbox`, and - * `pi.usage`, the conversation's `pi.agent` (see `createAgent()`), then `HarnessOptions.conversationCreated`. + * `pi.usage`, a fresh `pi.provider`, the conversation's `pi.agent` (see `createAgent()`), then + * `HarnessOptions.conversationCreated`. */ protected override async conversationCreated(tx: Transaction, record: ConversationRecord): Promise { await tx.doc(LiveDoc, record.id); await tx.doc(InboxDoc, record.id); await tx.doc(UsageDoc, record.id); + await tx.doc(ProviderDoc, record.id); await createAgent(tx, record); await this.#options.conversationCreated?.(tx, record); } diff --git a/packages/durable/src/harness/provider.ts b/packages/durable/src/harness/provider.ts new file mode 100644 index 000000000..b838982a6 --- /dev/null +++ b/packages/durable/src/harness/provider.ts @@ -0,0 +1,39 @@ +import type { Context } from "@earendil-works/chord"; +import { uuidv7 } from "@earendil-works/pi-ai/utils/uuid"; +import { defineDoc } from "../documents.ts"; +import type { TaskRuntime } from "../types.ts"; + +/** Stable provider-facing identity of one conversation. */ +export type ProviderState = { + sessionId: string; +}; + +/** Built-in provider state; every fork starts with a fresh identity instead of copying its parent. */ +export const ProviderDoc = defineDoc({ + kind: "pi.provider", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ sessionId: uuidv7() }), + checkpointWhen: () => true, +}); + +/** + * Return the persisted identity without writing in the normal path. A legacy conversation without `pi.provider` gets + * one migration commit whose `tx.doc()` runs `initial()` before the provider request starts. + */ +export async function ensureProviderSessionId( + runtime: TaskRuntime, + context: Context, +): Promise { + const existing = await runtime.snapshot(ProviderDoc, runtime.conversationId, context); + if (existing !== undefined) return existing.sessionId; + let created: string | undefined; + await runtime.commit(async (tx) => { + created = (await tx.doc(ProviderDoc, runtime.conversationId)).sessionId; + return undefined; + }, context); + if (created === undefined) throw new Error(`Conversation ${runtime.conversationId} has no provider session ID`); + return created; +} diff --git a/packages/durable/src/harness/view.ts b/packages/durable/src/harness/view.ts index bb942e164..1b706e401 100644 --- a/packages/durable/src/harness/view.ts +++ b/packages/durable/src/harness/view.ts @@ -18,6 +18,7 @@ import { AgentDoc } from "./agent.ts"; import { activeEntries, captureContextBounds } from "./context.ts"; import { InboxDoc } from "./inbox.ts"; import { LiveDoc } from "./live.ts"; +import { ProviderDoc } from "./provider.ts"; import { UsageDoc } from "./usage.ts"; import { closedError } from "./util.ts"; @@ -26,7 +27,7 @@ export type ConversationView = { readonly conversation: ConversationRecord; /** Raw active entries, as `ContextView.entries`: the head marker, then the non-head entries from its head. */ readonly entries: readonly EntryRecord[]; - /** `pi.agent`, `pi.live`, `pi.inbox`, and `pi.usage`, keyed by kind; absent documents are absent. */ + /** Built-in conversation documents keyed by kind; absent documents are absent. */ readonly docs: Readonly>; }; @@ -44,7 +45,13 @@ export type ViewObserver = { closeSession(): void; }; -const MOUNTED = [AgentDoc, LiveDoc, InboxDoc, UsageDoc] as unknown as readonly ConversationDocToken[]; +const MOUNTED = [ + AgentDoc, + LiveDoc, + InboxDoc, + ProviderDoc, + UsageDoc, +] as unknown as readonly ConversationDocToken[]; const MOUNTED_KINDS: ReadonlySet = new Set(MOUNTED.map((token) => token.definition.kind)); /** One conversation's mount: its current revision, the document incarnations it shows, and its observers. */ diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 5f2cbb88a..e92c6a654 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -38,6 +38,7 @@ export { export { Harness } from "./harness/harness.ts"; export { InboxDoc, type InboxItem, type InboxState } from "./harness/inbox.ts"; export { type CompactionStatus, LiveDoc, type LiveState, type ToolSlot } from "./harness/live.ts"; +export { ProviderDoc, type ProviderState } from "./harness/provider.ts"; export { createRegistry } from "./harness/registry.ts"; export type { TaskGraph, TaskGraphNode, TaskGraphState, TaskGraphWatch } from "./harness/task-graph.ts"; export { ToolTask, type ToolTaskCheckpoint, type ToolTaskInput, type ToolTaskResult } from "./harness/tool.ts"; diff --git a/packages/durable/test/harness-compaction.test.ts b/packages/durable/test/harness-compaction.test.ts index 9fc6b289e..f7cc07e40 100644 --- a/packages/durable/test/harness-compaction.test.ts +++ b/packages/durable/test/harness-compaction.test.ts @@ -24,6 +24,7 @@ import { type Harness, LiveDoc, MemoryStorage, + ProviderDoc, StorageRejected, type TaskId, UsageDoc, @@ -139,6 +140,8 @@ async function open( ): Promise { addSection(setup.registry, "preamble", () => "You are helpful.", { tag: false }); const { harness, root } = await openChat(options.storage ?? new MemoryStorage(), setup); + // These tests exercise compaction accounting and thresholds, not the faux provider's cache simulation. + setup.settings.stream = { cacheRetention: "none" }; setup.settings.compaction = options.policy ?? MANUAL; setup.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 1 }; harness.resume(); @@ -386,7 +389,11 @@ describe("manual compaction", () => { expect(prompt).not.toContain("u3 "); expect(prompt).toContain("## Goal"); expect(prompt.endsWith("\n\nAdditional focus: focus on files")).toBe(true); - expect(request.options).toMatchObject({ cacheRetention: "none", maxTokens: 800 }); + expect(request.options).toMatchObject({ + cacheRetention: "none", + maxTokens: 800, + sessionId: (await chat.harness.snapshot(ProviderDoc, chat.root.id, context))!.sessionId, + }); expect(request.options?.deferred).toBeUndefined(); // The summarizer's spend is in the ledger, and nothing counts it again later. @@ -400,6 +407,20 @@ describe("manual compaction", () => { await chat.harness.close(context); }); + // Regression coverage for #10424. + it("creates provider state before a legacy conversation's summarization request", async () => { + const chat = await open(); + await history(chat); + await chat.root.commit((tx) => tx.retireDoc(ProviderDoc, chat.root.id), context); + expect(await chat.harness.snapshot(ProviderDoc, chat.root.id, context)).toBeUndefined(); + chat.faux.summaries.push(summary()); + expect((await result(chat, await chat.root.compact(undefined, context))).status).toBe("completed"); + const stored = await chat.harness.snapshot(ProviderDoc, chat.root.id, context); + expect(stored?.sessionId).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/u); + expect(chat.faux.summaryRequests[0]!.options?.sessionId).toBe(stored?.sessionId); + await chat.harness.close(context); + }); + it("keeps working while busy and places the summary at the next final boundary", async () => { const chat = await open(); await history(chat); @@ -810,8 +831,14 @@ describe("compaction outcomes", () => { }); expect(chat.faux.summaryRequests).toHaveLength(2); expect(twice).toBe(2 * once); + const sessionId = (await chat.harness.snapshot(ProviderDoc, chat.root.id, context))!.sessionId; for (const request of chat.faux.summaryRequests) { - expect(request.options).toMatchObject({ reasoning: "high", timeoutMs: 1234, cacheRetention: "none" }); + expect(request.options).toMatchObject({ + reasoning: "high", + timeoutMs: 1234, + cacheRetention: "none", + sessionId, + }); expect(request.options?.deferred).toBeUndefined(); } await chat.harness.close(context); diff --git a/packages/durable/test/harness-conversations.test.ts b/packages/durable/test/harness-conversations.test.ts index a12306a05..88be495be 100644 --- a/packages/durable/test/harness-conversations.test.ts +++ b/packages/durable/test/harness-conversations.test.ts @@ -17,6 +17,7 @@ import { Harness, LiveDoc, MemoryStorage, + ProviderDoc, ROOT_CONVERSATION_ID, } from "@earendil-works/pi-durable"; import { afterEach, describe, expect, it } from "vitest"; @@ -25,6 +26,7 @@ import { addTool, openHarness, tool, user } from "./harness-support.ts"; import { ControlledStorage, context } from "./session-support.ts"; const directories = new Set(); +const UUID_V7 = /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/u; async function sqlitePath(): Promise { const directory = await mkdtemp(join(tmpdir(), "pi-durable-harness-")); @@ -81,7 +83,7 @@ describe("Harness root and conversations", () => { }); expect(root.id).toBe(ROOT_CONVERSATION_ID); expect(storage.commits).toHaveLength(1); - // Conversation, pi.live, pi.inbox, pi.usage, pi.agent, and the init note. + // Conversation, five built-in documents, and the init note. expect(storage.commits[0]!.map((write) => write.type)).toEqual([ "conversation", "document.create", @@ -89,8 +91,12 @@ describe("Harness root and conversations", () => { "document.create", "document.create", "document.create", + "document.create", ]); expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({}); + expect(await harness.snapshot(ProviderDoc, root.id, context)).toEqual({ + sessionId: expect.stringMatching(UUID_V7), + }); expect(await harness.snapshot(AgentDoc, root.id, context)).toEqual({ thinkingLevel: "high" }); const agent = await root.agent(context); expect(agent.thinkingLevel).toBe("high"); @@ -111,6 +117,14 @@ describe("Harness root and conversations", () => { const entry = await append(root, "hello"); const child = await first.harness.createConversation({ ownership: { kind: "ownerless" } }, context); const fork = await root.fork(entry.id, { ownership: { kind: "ownerless" } }, context); + const providerSessionIds = await Promise.all( + [root, child, fork].map( + async (conversation) => (await first.harness.snapshot(ProviderDoc, conversation.id, context))?.sessionId, + ), + ); + // Regression coverage for #10424: a fork must not inherit its parent's provider identity. + for (const sessionId of providerSessionIds) expect(sessionId).toMatch(UUID_V7); + expect(new Set(providerSessionIds).size).toBe(3); await first.harness.close(context); await expect(root.agent(context)).rejects.toThrow(); @@ -125,6 +139,13 @@ describe("Harness root and conversations", () => { }); expect(await allEntries(reopened)).toEqual(["hello"]); expect((await first.harness.conversation(child.id, context))?.id).toBe(child.id); + await expect( + Promise.all( + [root.id, child.id, fork.id].map( + async (id) => (await first.harness.snapshot(ProviderDoc, id, context))?.sessionId, + ), + ), + ).resolves.toEqual(providerSessionIds); const reopenedFork = await first.harness.conversation(fork.id, context); expect(await allEntries(reopenedFork!)).toEqual(["hello"]); expect((await reopenedFork!.agent(context)).model).toEqual({ provider: "anthropic", modelId: "claude" }); @@ -368,6 +389,12 @@ describe("Harness agent", () => { }, context); expect(await harness.snapshot(AgentDoc, ids.plain, context)).toEqual({}); expect(await harness.snapshot(LiveDoc, ids.plain, context)).toEqual({}); + expect(await harness.snapshot(ProviderDoc, ids.plain, context)).toEqual({ + sessionId: expect.stringMatching(UUID_V7), + }); + expect((await harness.snapshot(ProviderDoc, ids.owned, context))?.sessionId).not.toBe( + (await harness.snapshot(ProviderDoc, root.id, context))?.sessionId, + ); expect(await harness.snapshot(AgentDoc, ids.owned, context)).toEqual({ model: { provider: "faux", modelId: "m" }, instructions: "Main role.", @@ -387,6 +414,9 @@ describe("Harness agent", () => { ); expect(await harness.snapshot(AgentDoc, fork, context)).toEqual({}); expect(await harness.snapshot(LiveDoc, fork, context)).toEqual({}); + expect((await harness.snapshot(ProviderDoc, fork, context))?.sessionId).not.toBe( + (await harness.snapshot(ProviderDoc, ids.plain, context))?.sessionId, + ); await harness.close(context); }); diff --git a/packages/durable/test/harness-generation.test.ts b/packages/durable/test/harness-generation.test.ts index 0370816fa..c02aa03c1 100644 --- a/packages/durable/test/harness-generation.test.ts +++ b/packages/durable/test/harness-generation.test.ts @@ -19,6 +19,7 @@ import { LiveDoc, type LiveState, MemoryStorage, + ProviderDoc, type RegistrySnapshot, type TaskId, UserEntry, @@ -370,14 +371,84 @@ describe("generation", () => { await root.configure({ thinkingLevel: null }, context); setup.settings.stream = { timeoutMs: 99 }; await (await root.submit({ type: "input", content: "two" }, context)).wait(context); - expect(seen[0]).toMatchObject({ timeoutMs: 1234, headers: { "x-test": "1" }, reasoning: "high" }); + const sessionId = (await harness.snapshot(ProviderDoc, root.id, context))!.sessionId; + expect(seen[0]).toMatchObject({ + timeoutMs: 1234, + headers: { "x-test": "1" }, + reasoning: "high", + sessionId, + }); expect(seen[0]!.signal).toBeInstanceOf(AbortSignal); expect(seen[1]!.reasoning).toBeUndefined(); - expect(seen[1]).toMatchObject({ timeoutMs: 99 }); + expect(seen[1]).toMatchObject({ timeoutMs: 99, sessionId }); expect(seen[1]!.headers).toBeUndefined(); await harness.close(context); }); + // Regression coverage for #10424. + it("keeps provider session IDs request-local across concurrent conversations", async () => { + const setup = chatSetup(); + const seen = new Map(); + const capture = (request: { messages: readonly Message[] }, options?: SimpleStreamOptions) => { + const message = request.messages.findLast((candidate) => candidate.role === "user"); + const content = message?.role === "user" ? message.content : undefined; + const text = typeof content === "string" ? content : ""; + const values = seen.get(text) ?? []; + values.push(options?.sessionId ?? ""); + seen.set(text, values); + return fauxAssistantMessage(`answer:${text}`); + }; + setup.faux.setResponses([capture, capture, capture, capture]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const child = await harness.createConversation( + { + ownership: { kind: "ownerless" }, + agent: { model: { provider: "faux", modelId: "faux-1" } }, + }, + context, + ); + harness.resume(); + await Promise.all( + [root, child].map(async (conversation, index) => { + await (await conversation.submit({ type: "input", content: `first-${index}` }, context)).wait(context); + }), + ); + await Promise.all( + [root, child].map(async (conversation, index) => { + await (await conversation.submit({ type: "input", content: `second-${index}` }, context)).wait(context); + }), + ); + const rootId = (await harness.snapshot(ProviderDoc, root.id, context))!.sessionId; + const childId = (await harness.snapshot(ProviderDoc, child.id, context))!.sessionId; + expect(rootId).not.toBe(childId); + expect(seen.get("first-0")).toEqual([rootId]); + expect(seen.get("second-0")).toEqual([rootId]); + expect(seen.get("first-1")).toEqual([childId]); + expect(seen.get("second-1")).toEqual([childId]); + await harness.close(context); + }); + + // Regression coverage for #10424. + it("creates and persists provider state before a legacy conversation's request", async () => { + const setup = chatSetup(); + let sent: string | undefined; + setup.faux.setResponses([ + (_request, options) => { + sent = options?.sessionId; + return fauxAssistantMessage("ok"); + }, + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await root.commit((tx) => tx.retireDoc(ProviderDoc, root.id), context); + expect(await harness.snapshot(ProviderDoc, root.id, context)).toBeUndefined(); + harness.resume(); + await (await root.submit({ type: "input", content: "legacy" }, context)).wait(context); + const stored = await harness.snapshot(ProviderDoc, root.id, context); + expect(stored?.sessionId).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/u); + expect(sent).toBe(stored?.sessionId); + await harness.close(context); + }); + it("reads settings through getters at every decision", async () => { const setup = chatSetup(); let timeoutMs = 111; diff --git a/packages/durable/test/harness-view.test.ts b/packages/durable/test/harness-view.test.ts index 4d5a0e521..ce05d4c2a 100644 --- a/packages/durable/test/harness-view.test.ts +++ b/packages/durable/test/harness-view.test.ts @@ -9,6 +9,7 @@ import { InboxDoc, LiveDoc, MemoryStorage, + ProviderDoc, UsageDoc, } from "@earendil-works/pi-durable"; import { describe, expect, it } from "vitest"; @@ -18,7 +19,7 @@ import { aborted, deferred } from "./task-support.ts"; type Frame = { value: ConversationView; ops: readonly Op[] }; -const MOUNTED = ["pi.agent", "pi.inbox", "pi.live", "pi.usage"]; +const MOUNTED = ["pi.agent", "pi.inbox", "pi.live", "pi.provider", "pi.usage"]; /** Start a watch of `conversation` that records its acquisition revision and every delivered frame. */ async function record(conversation: Conversation) { @@ -42,7 +43,7 @@ async function fresh(conversation: Conversation): Promise { /** The view as committed state defines it, read without any mount: the active entries and the built-in documents. */ async function committed(harness: Harness, conversation: Conversation, record: ConversationView["conversation"]) { const docs: Record = {}; - for (const token of [AgentDoc, LiveDoc, InboxDoc, UsageDoc] as const) { + for (const token of [AgentDoc, LiveDoc, InboxDoc, ProviderDoc, UsageDoc] as const) { const value = await harness.snapshot(token as typeof LiveDoc, conversation.id, context); if (value !== undefined) docs[token.definition.kind] = value; } diff --git a/packages/durable/test/provider-session-cache-e2e.test.ts b/packages/durable/test/provider-session-cache-e2e.test.ts new file mode 100644 index 000000000..7a74a2065 --- /dev/null +++ b/packages/durable/test/provider-session-cache-e2e.test.ts @@ -0,0 +1,124 @@ +import { type AssistantMessage, type Model, type Models, normalizeContext, uuidv7 } from "@earendil-works/pi-ai"; +import { builtinModels } from "@earendil-works/pi-ai/providers/all"; +import { + AssistantEntry, + type ConversationId, + createRegistry, + type EntryId, + Harness, + MemoryStorage, + ProviderDoc, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { streamSimple as streamSimpleOpenAICodexResponses } from "../../ai/src/api/openai-codex-responses.ts"; +import { resolveApiKey } from "../../ai/test/oauth.ts"; +import { context } from "./session-support.ts"; + +const codexToken = await resolveApiKey("openai-codex"); + +function cacheProbe(nonce: string): string { + const lines = [ + "This is a real automated prompt-cache test. The repeated context below is intentional.", + "Read it silently, then reply with exactly: CACHE PROBE READY", + ]; + for (let index = 0; index < 180; index++) { + lines.push( + `${nonce} immutable cache record ${String(index).padStart(3, "0")}: alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu nu xi omicron pi rho sigma tau upsilon phi chi psi omega.`, + ); + } + return lines.join("\n"); +} + +function assistant(harness: Harness, conversationId: ConversationId, answer: EntryId) { + return harness + .commit((tx) => tx.entry(AssistantEntry, answer), context) + .then((record) => { + if (record?.conversationId !== conversationId) throw new Error("Answer belongs to another conversation"); + return record.model?.[0] as AssistantMessage | undefined; + }); +} + +function cacheHitRate(usage: AssistantMessage["usage"]): number { + const promptTokens = usage.input + usage.cacheRead + usage.cacheWrite; + return promptTokens === 0 ? 0 : usage.cacheRead / promptTokens; +} + +describe("Durable provider session cache e2e", () => { + // Live regression coverage for #10424; skipped without ~/.pi/agent/auth.json credentials. + it.skipIf(!codexToken)( + "reuses the Codex prompt cache across Durable turns", + { retry: 2, timeout: 120_000 }, + async () => { + const base = builtinModels(); + const forwarded: string[] = []; + const models = new Proxy(base, { + get(target, property) { + if (property === "streamSimple") { + return ( + model: Parameters[0], + request: Parameters[1], + options?: Parameters[2], + ) => { + forwarded.push(options?.sessionId ?? ""); + return streamSimpleOpenAICodexResponses( + model as Model<"openai-codex-responses">, + normalizeContext(request), + { + ...options, + apiKey: codexToken, + transport: "sse", + }, + ); + }; + } + const value = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }) as Models; + const harness = await Harness.open( + new MemoryStorage(), + { + models, + registry: createRegistry(), + settings: { + stream: { transport: "sse" }, + compaction: { enabled: false }, + }, + }, + context, + ); + try { + const root = await harness.root(context, { + agent: { model: { provider: "openai-codex", modelId: "gpt-5.5" } }, + }); + harness.resume(); + const usages: AssistantMessage["usage"][] = []; + let firstHit: number | undefined; + for (let turn = 0; turn < 8; turn++) { + const content = turn === 0 ? cacheProbe(uuidv7()) : `Reply exactly: CACHE PROBE FOLLOW-UP ${turn}`; + const settled = await (await root.submit({ type: "input", content }, context)).wait(context); + if (settled.status !== "done" || settled.type !== "input") { + throw new Error(`Unexpected submission status: ${settled.status}`); + } + const message = await assistant(harness, root.id, settled.answer); + expect(message?.stopReason, message?.errorMessage).not.toBe("error"); + usages.push(message!.usage); + if (firstHit === undefined && message!.usage.cacheRead > 0) firstHit = usages.length - 1; + if (firstHit !== undefined && usages.length - firstHit >= 3) break; + } + const sessionId = (await harness.snapshot(ProviderDoc, root.id, context))!.sessionId; + expect(forwarded).toEqual(Array.from({ length: usages.length }, () => sessionId)); + const firstPrompt = usages[0]!.input + usages[0]!.cacheRead + usages[0]!.cacheWrite; + expect(firstPrompt).toBeGreaterThanOrEqual(4_000); + expect(usages[0]!.cacheRead).toBe(0); + expect(firstHit).toBe(1); + if (firstHit === undefined) throw new Error("Codex reported no prompt-cache hit"); + const cached = usages.slice(firstHit); + expect(cached).toHaveLength(3); + for (const usage of cached) expect(cacheHitRate(usage)).toBeGreaterThanOrEqual(0.9); + } finally { + await harness.close(context); + } + }, + ); +}); diff --git a/scripts/check-entry-graphs.mjs b/scripts/check-entry-graphs.mjs index c404c49cb..e37fc95b1 100644 --- a/scripts/check-entry-graphs.mjs +++ b/scripts/check-entry-graphs.mjs @@ -43,8 +43,8 @@ const BUDGETS = { }, "packages/durable": { ".": { - // The built-in tool task validates arguments with pi-ai's TypeBox-based validation, so TypeBox is allowed. - maxFiles: 60, + // Tool argument validation reaches TypeBox; provider-session creation reaches pi-ai's lean UUID utility. + maxFiles: 62, forbid: ["packages/ai/src/index.ts", "packages/ai/src/utils/typebox-helpers.ts"], }, },