fix(durable): persist provider session identities

Assign each conversation its own UUIDv7 and forward it to model requests for prompt-cache affinity. Lazily backfill legacy conversations and cover real Codex cache reuse.

Closes #10424
This commit is contained in:
Mario Zechner 2026-10-04 00:54:10 +02:00
parent 83692682f0
commit 70eceaade6
15 changed files with 363 additions and 23 deletions

View file

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

View file

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

View file

@ -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
`<conversation>` tags followed by the built-in summarization prompt, and
`Additional focus: <instructions>` 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<Record<string, JsonObject>>;
};
```

View file

@ -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<CompactionInput, CompactionCheckpoint,
cacheRetention: "none",
maxTokens,
signal: runtime.signal,
sessionId: await ensureProviderSessionId(runtime, context),
...(thinkingLevel === "off" ? {} : { reasoning: thinkingLevel }),
};
const message = await runtime.models.completeSimple(model, { messages }, options);

View file

@ -31,6 +31,7 @@ import { applyBoundary, prepareBoundary } from "./inbox.ts";
import { assignJson } from "./json.ts";
import { endRun, LiveDoc, type LiveState, type ToolSlot } from "./live.ts";
import { planSystemEntries, renderSections, replaySections } from "./prompt.ts";
import { ensureProviderSessionId } from "./provider.ts";
import { appendToolResult, harnessError, ToolTask, type ToolTaskResult } from "./tool.ts";
import type {
CompactionPolicy,
@ -201,6 +202,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
const options: SimpleStreamOptions = {
...streamOptions,
signal: runtime.signal,
sessionId: await ensureProviderSessionId(runtime, context),
...(thinkingLevel === "off" ? {} : { reasoning: thinkingLevel }),
};
const message = await streamResponse(runtime, model, messages, options, attempt, context);

View file

@ -25,6 +25,7 @@ import { createCompaction } from "./compaction.ts";
import { readContext } from "./context.ts";
import { InboxDoc, withdrawQueuedInputs } from "./inbox.ts";
import { LiveDoc, settleSchedulerOutcome } from "./live.ts";
import { ProviderDoc } from "./provider.ts";
import { BUILTIN_TASKS } from "./registry.ts";
import { type InvocationBinding, TaskScheduler } from "./scheduler.ts";
import { Submissions } from "./submissions.ts";
@ -349,12 +350,14 @@ class HarnessImpl<Tool extends ToolRegistration> 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<void> {
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);
}

View file

@ -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<ProviderState>({
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<I, S, R, H extends object>(
runtime: TaskRuntime<I, S, R, H>,
context: Context,
): Promise<string> {
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;
}

View file

@ -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<Record<string, JsonObject>>;
};
@ -44,7 +45,13 @@ export type ViewObserver = {
closeSession(): void;
};
const MOUNTED = [AgentDoc, LiveDoc, InboxDoc, UsageDoc] as unknown as readonly ConversationDocToken<JsonObject>[];
const MOUNTED = [
AgentDoc,
LiveDoc,
InboxDoc,
ProviderDoc,
UsageDoc,
] as unknown as readonly ConversationDocToken<JsonObject>[];
const MOUNTED_KINDS: ReadonlySet<string> = new Set(MOUNTED.map((token) => token.definition.kind));
/** One conversation's mount: its current revision, the document incarnations it shows, and its observers. */

View file

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

View file

@ -24,6 +24,7 @@ import {
type Harness,
LiveDoc,
MemoryStorage,
ProviderDoc,
StorageRejected,
type TaskId,
UsageDoc,
@ -139,6 +140,8 @@ async function open(
): Promise<Chat> {
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);

View file

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

View file

@ -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<string, string[]>();
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;

View file

@ -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<ConversationView> {
/** 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<string, unknown> = {};
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;
}

View file

@ -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<Models["streamSimple"]>[0],
request: Parameters<Models["streamSimple"]>[1],
options?: Parameters<Models["streamSimple"]>[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);
}
},
);
});

View file

@ -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"],
},
},