From d58edc04f7f34a3b5250cd58d8037137ae28d3f4 Mon Sep 17 00:00:00 2001 From: Alain Schlesser Date: Fri, 16 Oct 2026 16:15:55 +0200 Subject: [PATCH] fix(backends): preserve provider transcript bytes with UI replay metadata --- .changeset/preserve-provider-transcripts.md | 12 +++ api-report/ui-sdk.txt | 2 + docs/decisions/governed-writable-workers.md | 12 +-- docs/decisions/policy-write-boundary.md | 2 +- docs/hosting/agent-workers.md | 2 +- docs/plans/host-owned-web-search.md | 2 +- packages/ui-backend-claude/README.md | 10 +++ packages/ui-backend-claude/src/history.ts | 17 ++-- .../ui-backend-claude/src/worker-state.ts | 28 ++++--- .../tests/fixtures/transcript-prefix-probe.ts | 69 +++++++++++++++++ .../tests/transcript-prefix.test.ts | 29 +++++++ .../tests/worker-state.test.ts | 14 +++- .../ui-backend-claude/tests/worker.test.ts | 26 +++++-- packages/ui-backend-pi/README.md | 13 ++++ packages/ui-backend-pi/src/history.ts | 45 ++++++++--- packages/ui-backend-pi/src/worker-session.ts | 77 +++++++++++-------- .../tests/direct-application-replay.test.ts | 72 +++++++++++++++++ .../fixtures/direct-application-worker.ts | 32 ++++++++ .../tests/fixtures/transcript-prefix-probe.ts | 61 +++++++++++++++ .../tests/transcript-prefix.test.ts | 26 +++++++ .../tests/worker-session.test.ts | 77 ++++++++++++++++++- packages/ui-sdk/src/internal.ts | 3 + packages/ui-sdk/src/server/session-replay.ts | 73 ++++++++++++++++++ packages/ui-sdk/tests/session-replay.test.ts | 55 +++++++++++++ packages/ui-server/src/ws/message-sources.ts | 2 +- 25 files changed, 681 insertions(+), 80 deletions(-) create mode 100644 .changeset/preserve-provider-transcripts.md create mode 100644 packages/ui-backend-claude/tests/fixtures/transcript-prefix-probe.ts create mode 100644 packages/ui-backend-claude/tests/transcript-prefix.test.ts create mode 100644 packages/ui-backend-pi/tests/direct-application-replay.test.ts create mode 100644 packages/ui-backend-pi/tests/fixtures/direct-application-worker.ts create mode 100644 packages/ui-backend-pi/tests/fixtures/transcript-prefix-probe.ts create mode 100644 packages/ui-backend-pi/tests/transcript-prefix.test.ts create mode 100644 packages/ui-sdk/src/server/session-replay.ts create mode 100644 packages/ui-sdk/tests/session-replay.test.ts diff --git a/.changeset/preserve-provider-transcripts.md b/.changeset/preserve-provider-transcripts.md new file mode 100644 index 000000000..b68e7b198 --- /dev/null +++ b/.changeset/preserve-provider-transcripts.md @@ -0,0 +1,12 @@ +--- +"@schlessera/brain-backend-claude": patch +"@schlessera/brain-backend-pi": patch +"@schlessera/brain-ui-sdk": patch +"@schlessera/brain-ui-server": patch +--- + +Preserve runtime-written provider transcript entries across turns. Store promoted +answer-link display mappings and authoritative Pi application receipts in host-owned +replay metadata so history retains promoted links and receipts without changing +prior model text or tool results. Recover only missing results for runtime-written +Pi calls after the worker exits; direct server applications create no provider entry. diff --git a/api-report/ui-sdk.txt b/api-report/ui-sdk.txt index 8dfbd2657..493859b77 100644 --- a/api-report/ui-sdk.txt +++ b/api-report/ui-sdk.txt @@ -885,6 +885,7 @@ export "./internal" (./src/internal.ts) type PrepareWorkerSandboxOptions type PreparedWorkerOutput type QuestionObservation + type ReplayReceipt type Reservation type RestrictedEnvelope STALE_NATIVE_WRITE @@ -1017,6 +1018,7 @@ export "./internal" (./src/internal.ts) runMaintenanceWorker runMaintenanceWorkerSync sessionPromptContext + sessionReplay startInferenceRelay turnTempFullMessage turnTempLimit diff --git a/docs/decisions/governed-writable-workers.md b/docs/decisions/governed-writable-workers.md index 337a43374..9fba726b4 100644 --- a/docs/decisions/governed-writable-workers.md +++ b/docs/decisions/governed-writable-workers.md @@ -698,9 +698,10 @@ against its own budget, and a turn that cannot reserve it is refused as reports usage and purges paths through the same descriptor (`export function purgeTurnTemp`, `packages/ui-sdk/src/server/worker-turn-temp.ts:412-420`). The restricted envelope gets the same sized `/tmp` and promotes nothing. -After the worker exits, the backend -rewrites the answer text the turn appended to its saved transcript, and only -that text. It looks at Markdown link and image destinations and reference +After the worker exits, the backend observes the answer text the turn +appended and retains promoted-link display mappings in host-owned storage. +The runtime transcript stays byte-identical; #2049 supersedes the original +transcript rewrite described by #1655. It looks at Markdown link and image destinations and reference definitions outside code. A file is promoted once per turn however often it is linked, up to 128 MiB per file, 256 MiB and 64 files per turn (`export function createTurnTempPromoter`, `packages/ui-sdk/src/server/worker-turn-temp.ts:123-134`). @@ -709,8 +710,9 @@ file links accept, so the rewritten link is brain-relative and is pinned like any other scratch link. Each result reaches the server as a `promotion` or `promotion_refused` activity event, which it stores as the governance event of the same name. `scratch_unavailable` is the refusal for a turn whose -`.brain/scratch/` was not usable. The stored answer is rewritten before the -turn's history boundary is read. A reader still watching the live stream +`.brain/scratch/` was not usable. History readers apply the retained mapping +by native session/message/part identity and the original text's digest before +normalization. It survives reload and archival without entering runtime input. A reader still watching the live stream holds the text as it streamed, so the server also sends the turn's clients a `message_links` frame for each promoted file, and the client rewrites the same destinations with the function the backend used. The link changes in diff --git a/docs/decisions/policy-write-boundary.md b/docs/decisions/policy-write-boundary.md index 122c6e585..c6fd9928a 100644 --- a/docs/decisions/policy-write-boundary.md +++ b/docs/decisions/policy-write-boundary.md @@ -474,7 +474,7 @@ host (`export function launchAgentWorker`, 3. After the worker exits, its PID namespace is gone, and with it every process that could write the tmpfs. The held descriptor still reaches the detached mount, so the server walks it then (`persist(root, rewrite) {`, - `packages/ui-backend-claude/src/worker-state.ts:226-265`) with the existing + `packages/ui-backend-claude/src/worker-state.ts:226-276`) with the existing rules: only this brain's native transcripts and `memory/*.md` notes, every directory checked with `lstat` before it is entered, every file opened with `O_NOFOLLOW` and refused on the descriptor when it is not a regular file, diff --git a/docs/hosting/agent-workers.md b/docs/hosting/agent-workers.md index 6be897008..ac1067a9d 100644 --- a/docs/hosting/agent-workers.md +++ b/docs/hosting/agent-workers.md @@ -259,7 +259,7 @@ write verified scratch/workspace copies through the aliases below. | Place | Paths | Lifetime | Link it for the reader? | | --- | --- | --- | --- | -| Turn temp | `$BRAIN_WORKER_SCRATCH`, `/tmp` and the XDG directories in it | Deleted when the turn ends | Never: the reader cannot open it. A file the answer links there anyway is copied into `.brain/scratch/turns//` and the saved link rewritten. | +| Turn temp | `$BRAIN_WORKER_SCRATCH`, `/tmp` and the XDG directories in it | Deleted when the turn ends | Never: the reader cannot open it. A file the answer links there anyway is copied into `.brain/scratch/turns//` and a host-owned display mapping makes history show the promoted link; runtime transcript bytes stay unchanged. | | Persistent scratch/workspaces | Governed: `.brain/scratch/` and `workspaces/`; strict: `/tmp/brain-worker-areas/scratch` and `/tmp/brain-worker-areas/workspaces` | Governed writes persist directly; strict writes persist after validated publication. Scratch is pruned after 7 days, oldest first past 1 GB | Yes: link the canonical brain destinations, not the strict aliases | | The brain | Everything else | Canonical | Yes, as documents | diff --git a/docs/plans/host-owned-web-search.md b/docs/plans/host-owned-web-search.md index 3aa7e236e..e64f2f194 100644 --- a/docs/plans/host-owned-web-search.md +++ b/docs/plans/host-owned-web-search.md @@ -216,7 +216,7 @@ runtime authority or a public plugin seam. Both candidates need them: Current source confirms why a bridge is needed: ordinary initialization copies only auth/models/settings (`const init`, -`packages/ui-backend-pi/src/worker-session.ts:111-113`), then selects private Pi +`packages/ui-backend-pi/src/worker-session.ts:113-115`), then selects private Pi state. The worker RPC schema has an existing bounded, grant-checked bridge but no search method (`export const piWorkerMessage`, `packages/ui-backend-pi/src/worker-protocol.ts:64-70`). Resource discovery remains diff --git a/packages/ui-backend-claude/README.md b/packages/ui-backend-claude/README.md index 63a92eca6..c6001b1bc 100644 --- a/packages/ui-backend-claude/README.md +++ b/packages/ui-backend-claude/README.md @@ -318,3 +318,13 @@ Per-item server context lives in the user message so the selected runtime cannot coalesce it into the static system block. See the [assembly facts and optional cache measurement procedure](../../docs/hosting/autonomous-prompt-cache.md). Cache savings remain unclaimed without an authorized provider measurement. + +## Promoted links in history + +Turn-temp promotion leaves the native Claude JSONL bytes unchanged. The backend +retains display mappings outside the runtime state under the host Claude config +root's `session-replay/` directory. History applies them by native message UUID, +text-part index and exact original text before folding the transcript into UI +messages. Cold archival and resume continue to copy the runtime transcript as +written; display metadata never becomes provider input. Older transcripts keep +their existing bytes. diff --git a/packages/ui-backend-claude/src/history.ts b/packages/ui-backend-claude/src/history.ts index e39e0e6e8..d540673a1 100644 --- a/packages/ui-backend-claude/src/history.ts +++ b/packages/ui-backend-claude/src/history.ts @@ -1,4 +1,6 @@ import { claudePromptContext } from "./prompt-context.js"; +import { sessionReplay } from "@schlessera/brain-ui-sdk/internal"; +import { dirname } from "node:path"; import { FIRST_PARTY_FAILURE_CLASSES } from "@schlessera/brain-ui-sdk/protocol"; import { listSessions as sdkListSessions, @@ -11,7 +13,7 @@ import type { TurnFailure, } from "@schlessera/brain-ui-sdk"; import { subscriptionAuthAction } from "@schlessera/brain-ui-sdk"; -import { listArchivedSessions, restoreArchivedSession } from "./transcript-archive.js"; +import { listArchivedSessions, restoreArchivedSession, transcriptPaths } from "./transcript-archive.js"; import { isSessionId } from "./worker-state.js"; import { statusFromText } from "./stream-adapter.js"; @@ -78,7 +80,8 @@ export function createHistory(deps: HistoryDeps): History { sessionId: string ): Promise { if (isSessionId(sessionId)) restoreArchivedSession(sessionId, deps); - return buildSessionHistory(sessionId, getSessionMessagesFn, dir); + const root = dirname(dirname(transcriptPaths(dir, deps.env).live)); + return buildSessionHistory(sessionId, getSessionMessagesFn, dir, sessionReplay(root, dir, sessionId).texts()); } return { listSessions, getHistory }; @@ -94,7 +97,8 @@ export function createHistory(deps: HistoryDeps): History { export async function buildSessionHistory( sessionId: string, getSessionMessagesFn: typeof sdkGetSessionMessages, - dir: string + dir: string, + displayText: (id: string, part: number, text: string) => string = (_id, _part, text) => text ): Promise { const history = await getSessionMessagesFn(sessionId, { dir }); @@ -197,14 +201,15 @@ export async function buildSessionHistory( } }; - for (const block of content as any[]) { + for (const [index, block] of (content as any[]).entries()) { if (block.type === "thinking" && block.thinking) { thinking += block.thinking; pushText("thinking", block.thinking); } if (block.type === "text" && block.text) { - text += block.text; - pushText("text", block.text); + const displayed = displayText(entry.uuid, index, block.text); + text += displayed; + pushText("text", displayed); } if (block.type === "tool_use") { const result = toolResults.get(block.id); diff --git a/packages/ui-backend-claude/src/worker-state.ts b/packages/ui-backend-claude/src/worker-state.ts index 0ffc0777c..532ad4ec8 100644 --- a/packages/ui-backend-claude/src/worker-state.ts +++ b/packages/ui-backend-claude/src/worker-state.ts @@ -10,7 +10,7 @@ import { dirname, join, resolve } from "node:path"; import { createHash } from "node:crypto"; import { BackendRequestError, type TurnFailure } from "@schlessera/brain-ui-sdk/server"; import { FIRST_PARTY_FAILURE_CLASSES, type StateRoomObservation } from "@schlessera/brain-ui-sdk/protocol"; -import { mib, resolveWorkerStateLimits, WORKER_STATE_BYTES_ENV, WorkerHostError, type WorkerNoteLimit, type WorkerStateLimits } from "@schlessera/brain-ui-sdk/internal"; +import { sessionReplay, mib, resolveWorkerStateLimits, WORKER_STATE_BYTES_ENV, WorkerHostError, type WorkerNoteLimit, type WorkerStateLimits } from "@schlessera/brain-ui-sdk/internal"; import { processStateLedger, type Reservation, type StateLedger } from "./worker-state-budget.js"; const hash = (bytes: Uint8Array) => createHash("sha256").update(bytes).digest("hex"); @@ -104,7 +104,7 @@ export interface ClaudeWorkerState { populate(root: string): void; /** * Copies validated results out of the private tmpfs, after the worker's - * namespace is gone; `rewrite` maps the answer text this run appended (#1655). + * namespace is gone; `rewrite` promotes newly appended answer links for display. */ persist(root: string, rewrite?: (text: string) => string): void; /** Releases the admission reservation; idempotent, called on every exit path. */ @@ -220,8 +220,8 @@ function plan(brainPath: string, env: Record, if (options.resume && transcripts.length > 0) observation = { usedBytes: total, allowanceBytes: allowance, observedAt: Date.now() }; }, /** - * `rewrite` maps the answer text of the transcript lines this run - * appended to a session (#1655); earlier lines are saved as they were. + * `rewrite` observes this turn’s answer links for promotion. All transcript + * bytes stay exactly as the runtime wrote them (#2049). */ persist(root, rewrite) { // Only this brain's native transcript files and its memory notes, never @@ -247,7 +247,17 @@ function plan(brainPath: string, env: Record, if (total > allowance) throw new WorkerHostError("runtime transcripts and memory exceed their byte bound"); const prefix = initial.has(name) ? { length: initialLength.get(name)!, hash: initial.get(name)! } : undefined; // Only a session's own transcript holds the answer; subagent files stay as written. - const bytes = rewrite && !file.includes("/") ? rewriteAppendedAnswers(read, prefix, rewrite) : read; + // Promotion observes only newly appended answer parts. Its display output + // belongs to host metadata; copy out the exact runtime bytes. + if (rewrite && !file.includes("/")) { + const display = sessionReplay(saved, brain, id); + rewriteAppendedAnswers(read, prefix, (text, messageId, part) => { + const next = rewrite(text); + if (messageId !== undefined) display.retainText(messageId, part!, text, next); + return text; + }); + } + const bytes = read; if (initial.get(name) === hash(bytes)) continue; write(name, bytes, "transcript"); } @@ -282,19 +292,19 @@ export function workerStateLimit(limits: WorkerStateLimits): WorkerNoteLimit { * returned as it is: nothing but this run's own lines may be rewritten. */ export function rewriteAppendedAnswers(bytes: Buffer, prefix: { length: number; hash: string } | undefined, - rewrite: (text: string) => string): Buffer { + rewrite: (text: string, messageId?: string, part?: number) => string): Buffer { const start = prefix?.length ?? 0; if (prefix && (bytes.length < start || hash(bytes.subarray(0, start)) !== prefix.hash)) return bytes; let changed = false; const lines = bytes.subarray(start).toString("utf8").split("\n").map(line => { if (!line.includes("\"assistant\"")) return line; - let record: { type?: unknown; message?: { content?: unknown } }; + let record: { type?: unknown; uuid?: string; message?: { content?: unknown } }; try { record = JSON.parse(line); } catch { return line; } if (record?.type !== "assistant" || !Array.isArray(record.message?.content)) return line; let touched = false; - for (const part of record.message.content as Array<{ type?: unknown; text?: unknown }>) { + for (const [index, part] of (record.message.content as Array<{ type?: unknown; text?: unknown }>).entries()) { if (part?.type !== "text" || typeof part.text !== "string") continue; - const next = rewrite(part.text); + const next = rewrite(part.text, record.uuid, index); if (next !== part.text) { part.text = next; touched = true; } } if (!touched) return line; diff --git a/packages/ui-backend-claude/tests/fixtures/transcript-prefix-probe.ts b/packages/ui-backend-claude/tests/fixtures/transcript-prefix-probe.ts new file mode 100644 index 000000000..ab4f8733c --- /dev/null +++ b/packages/ui-backend-claude/tests/fixtures/transcript-prefix-probe.ts @@ -0,0 +1,69 @@ +/** Narrow #2049 proof; follows the offline CLI pattern from #2053/#2047. */ +import { cpSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { query, getSessionMessages } from "@anthropic-ai/claude-agent-sdk"; +import { createClaudeSdkTurn } from "../../src/sdk-options.js"; +import { createHistory } from "../../src/history.js"; +import { createFixtureWorkerStartupView, prepareFixtureWorkerSandbox } from "../../../ui-sdk/tests/helpers/worker-startup-view-fixture.js"; +import { replaceNativeStartupAcquisitionForFixture } from "../../../ui-server/tests/helpers/fixture-native-startup.js"; +const restore = replaceNativeStartupAcquisitionForFixture(); +const scratch = process.argv[2]!; +const cacheEnabled = process.argv[3] === "cache-on"; +process.env.HOME = join(scratch, "home"); process.env.CLAUDE_CONFIG_DIR = join(scratch, "claude"); +mkdirSync(process.env.HOME, { recursive: true }); +const brain = join(scratch, "ithaca"); +cpSync(resolve("packages/core/fixtures/corpus"), brain, { recursive: true }); +const config = join(brain, "brain.config.ts"); +writeFileSync(config, readFileSync(config, "utf8").replace('"@schlessera/brain"', JSON.stringify(resolve("packages/core/src/index.ts")))); +writeFileSync(join(brain, ".gitignore"), ".brain/\n"); +mkdirSync(join(brain, ".brain/scratch"), { recursive: true }); +mkdirSync(join(brain, "docs"), { recursive: true }); +writeFileSync(join(brain, "docs/ithaca-chart.md"), "Odysseus verifies the Ithaca chart.\n"); +// Disable runtime cache-breakpoint annotations in this byte comparison only; +// moving cache_control markers are transport metadata, not transcript content. +const requests: any[] = []; +let step = 0; +const link = "[Ithaca chart](/tmp/ithaca-chart.pdf)"; +const server = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(request) { + if (!new URL(request.url).pathname.endsWith("/messages")) return Response.json({}); + const body = await request.json() as any; + if (!body.tools?.some((tool: any) => tool.name === "Read")) return Response.json({}); + requests.push(body); + const index = step++; + const content: Array<{ type: "text"; text: string } | { type: "tool_use"; id: string; name: string; input: Record }> = index === 0 + ? [{ type: "tool_use", id: "toolu_chart", name: "Bash", input: { command: "printf 'Odysseus charts Ithaca.' > /tmp/ithaca-chart.pdf" } }] + : index === 1 ? [{ type: "text", text: `Before checking: ${link}` }, { type: "tool_use", id: "toolu_read", name: "Read", input: { file_path: join(brain, "docs/ithaca-chart.md") } }] + : [{ type: "text", text: index === 2 ? `After checking: ${link}` : "Odysseus follows the chart." }]; + const events: any[] = [{ type: "message_start", message: { id: `msg_odysseus_${index}`, type: "message", role: "assistant", model: body.model, + content: [], stop_reason: null, stop_sequence: null, usage: { input_tokens: 3, output_tokens: 0 } } }]; + content.forEach((block, i) => events.push({ type: "content_block_start", index: i, content_block: block.type === "tool_use" ? { ...block, input: {} } : { type: "text", text: "" } }, + { type: "content_block_delta", index: i, delta: block.type === "tool_use" ? { type: "input_json_delta", partial_json: JSON.stringify(block.input) } : { type: "text_delta", text: block.text } }, + { type: "content_block_stop", index: i })); + events.push({ type: "message_delta", delta: { stop_reason: index < 2 ? "tool_use" : "end_turn", stop_sequence: null }, usage: { output_tokens: 5 } }, { type: "message_stop" }); + return new Response(events.map(event => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(""), { headers: { "content-type": "text/event-stream" } }); +} }); +let sessionId: string | undefined; +try { + for (let turnIndex = 0; turnIndex < 2; turnIndex++) { + const controller = new AbortController(), view = createFixtureWorkerStartupView(brain); + const sandbox = prepareFixtureWorkerSandbox(brain, { home: process.env.HOME, profile: "strict" }); + const turn = createClaudeSdkTurn({ backend: { brainPath: brain }, req: { prompt: turnIndex ? "Use the chart" : "Prepare the chart", sessionId, + signal: controller.signal, turnId: `odysseus-${turnIndex}`, workerStartupView: view, workerSandbox: sandbox, + bridge: { emit() {}, async requestPermission() { return { behavior: "allow" }; } } }, + profile: { id: "fixture", label: "Fixture", model: "claude-sonnet-4-6", billing: "api", requiredEnvKeys: [], buildEnv: () => ({ + ANTHROPIC_BASE_URL: server.url.origin, ANTHROPIC_API_KEY: "offline-odysseus", ...(!cacheEnabled ? { DISABLE_PROMPT_CACHING: "1" } : {}), CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: "1", CLAUDE_CODE_DISABLE_AUTO_MEMORY: "1" }) }, + abortController: controller, allowedTools: ["Read", "Bash"], confirmPatterns: [], turnLock: { acquire() {}, release() {} } as never, log() {} }); + const deadline = setTimeout(() => controller.abort(), 45000); + try { + let success = false; + for await (const message of query({ prompt: turn.prompt, options: turn.options })) { + if (message.type === "assistant") for (const block of message.message.content) if (block.type === "text") turn.observeAnswerText(block.text); + if (message.type === "result") { if (message.subtype !== "success") throw new Error(JSON.stringify(message)); sessionId = message.session_id; success = true; } + } + if (!success) throw new Error("No successful runtime result"); + } finally { clearTimeout(deadline); await turn.finishWorker(); await view.cleanup(); } + } + const history = await createHistory({ brainPath: brain, getSessionMessagesFn: getSessionMessages }).getHistory(sessionId!); + const transcript = readFileSync(join(process.env.CLAUDE_CONFIG_DIR!, "projects", brain.replace(/[^a-zA-Z0-9]/g, "-"), `${sessionId}.jsonl`), "utf8"); + writeFileSync(join(scratch, "requests.json"), JSON.stringify({ requests, history, transcript })); +} finally { restore(); server.stop(true); } diff --git a/packages/ui-backend-claude/tests/transcript-prefix.test.ts b/packages/ui-backend-claude/tests/transcript-prefix.test.ts new file mode 100644 index 000000000..d4b331988 --- /dev/null +++ b/packages/ui-backend-claude/tests/transcript-prefix.test.ts @@ -0,0 +1,29 @@ +/** Exact provider-history regression using the installed runtimes, no live provider. */ +import { expect, test } from "bun:test"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { tmpdir } from "node:os"; +const root = resolve(import.meta.dir, "../../.."); +for (const cacheEnabled of [false, true]) test(`CLAUDE promoted links preserve previous request messages (cache ${cacheEnabled ? "on" : "off"})`, async () => { + const scratch = mkdtempSync(join(tmpdir(), "transcript-prefix-claude-")); + try { + const child = Bun.spawn(["bwrap", "--unshare-net", "--bind", "/", "/", "--proc", "/proc", "--dev", "/dev", process.execPath, + "--preload", join(root, "scripts/test-network-child-preload.ts"), join(import.meta.dir, "fixtures/transcript-prefix-probe.ts"), scratch, cacheEnabled ? "cache-on" : "cache-off"], + { cwd: root, env: { PATH: process.env.PATH!, LANG: "C.UTF-8", TMPDIR: tmpdir() }, stdout: "pipe", stderr: "pipe" }); + const [out, err, code] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text(), child.exited]); + expect(code, err || out).toBe(0); + const { requests, history, transcript } = JSON.parse(readFileSync(join(scratch, "requests.json"), "utf8")); + expect(requests, "REAL_TOOL_ROUND_TRIP_AND_RESUMED_REQUEST").toHaveLength(4); + const previous = requests[2].messages, resumed = requests[3].messages; + expect(JSON.stringify(previous), "INTERMEDIATE_PROVIDER_LINK_BEFORE_TOOL").toContain("Before checking: [Ithaca chart](/tmp/ithaca-chart.pdf)"); + // Cache breakpoints can move; preserve every content byte and property order. + // The cache-off arm compares raw serialized objects without normalization. + const bytes = (value: unknown) => JSON.stringify(value, (key, item) => cacheEnabled && key === "cache_control" ? undefined : item); + expect(bytes(resumed.slice(0, previous.length)), "CLAUDE_BYTE_IDENTICAL_MESSAGES_PREFIX").toBe(bytes(previous)); + expect(transcript, "SAVED_RUNTIME_ANSWER_UNCHANGED").toContain("After checking: [Ithaca chart](/tmp/ithaca-chart.pdf)"); + const displayed = JSON.stringify(history); + expect(displayed, "RELOADED_DISPLAY_HAS_BOTH_PROMOTED_LINKS").toContain("Before checking: [Ithaca chart](.brain/scratch/turns/odysseus-0/ithaca-chart.pdf)"); + expect(displayed).toContain("After checking: [Ithaca chart](.brain/scratch/turns/odysseus-0/ithaca-chart.pdf)"); + expect(displayed).not.toContain("[Ithaca chart](/tmp/"); + } finally { rmSync(scratch, { recursive: true, force: true }); } +}, 180_000); diff --git a/packages/ui-backend-claude/tests/worker-state.test.ts b/packages/ui-backend-claude/tests/worker-state.test.ts index 7ff61414c..bc37fd4ec 100644 --- a/packages/ui-backend-claude/tests/worker-state.test.ts +++ b/packages/ui-backend-claude/tests/worker-state.test.ts @@ -3,6 +3,7 @@ import * as fs from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { createHash } from "node:crypto"; +import { sessionReplay } from "@schlessera/brain-ui-sdk/internal"; import type { WorkerStateLimits } from "@schlessera/brain-ui-sdk/internal"; import { createClaudeWorkerState, rewriteAppendedAnswers, SessionTooLargeError, workerStateLimit } from "../src/worker-state.js"; import { createStateLedger, type StateLedger } from "../src/worker-state-budget.js"; @@ -188,8 +189,8 @@ test("ephemeral state copies nothing in and never persists, but still reserves", fs.writeFileSync(join(path, project, `${NEW}.jsonl`), "Athena"); snapshot.persist(path); expect(fs.existsSync(join(saved, project, `${NEW}.jsonl`))).toBe(false); }); -test("persist rewrites only the answer text this run appended to a session's own transcript (#1655)", () => { - const line = (type: string, text: string) => JSON.stringify({ type, message: { role: type, content: [{ type: "text", text }] } }); +test("persist preserves runtime answer bytes while promoting only this turn (#2049)", () => { + const line = (type: string, text: string) => " " + JSON.stringify({ type, uuid: text, message: { role: type, content: [{ type: "text", text }] } }) + " "; const before = `${line("assistant", "Earlier: [map](/tmp/map.png)")}\n`; put(ID, before); fs.mkdirSync(join(saved, project, ID)); fs.writeFileSync(join(saved, project, ID, "agent.jsonl"), `${line("assistant", "[x](/tmp/x)")}\n`); @@ -201,10 +202,15 @@ test("persist rewrites only the answer text this run appended to a session's own snapshot.persist(path, text => { seen.push(text); return text.replaceAll("/tmp/", ".brain/scratch/turns/t/"); }); expect(seen.sort()).toEqual(["New: [report](/tmp/report.pdf)", "Now: [report](/tmp/report.pdf) and [map](/tmp/map.png)"]); const out = fs.readFileSync(join(saved, project, `${ID}.jsonl`), "utf8"); + expect(out, "CLAUDE_NATIVE_TRANSCRIPT_BYTES_UNCHANGED").toBe(fs.readFileSync(join(target, `${ID}.jsonl`), "utf8")); + const display = sessionReplay(saved, brain, ID).texts(); + const original = "Now: [report](/tmp/report.pdf) and [map](/tmp/map.png)"; + expect(display(original, 0, original), "CLAUDE_DISPLAY_MAPPING_SURVIVES_RELOAD").toBe("Now: [report](.brain/scratch/turns/t/report.pdf) and [map](.brain/scratch/turns/t/map.png)"); + expect(display(original, 0, original + "changed"), "CLAUDE_MAPPING_BOUND_TO_NATIVE_BYTES").toBe(original + "changed"); expect(out.startsWith(before)).toBe(true); - expect(out.slice(before.length)).toBe(`${line("user", "[u](/tmp/u.txt)")}\n${line("assistant", "Now: [report](.brain/scratch/turns/t/report.pdf) and [map](.brain/scratch/turns/t/map.png)")}\n`); + expect(out.slice(before.length)).toBe(`${line("user", "[u](/tmp/u.txt)")}\n${line("assistant", "Now: [report](/tmp/report.pdf) and [map](/tmp/map.png)")}\n`); expect(fs.readFileSync(join(saved, project, ID, "agent.jsonl"), "utf8")).toBe(`${line("assistant", "[x](/tmp/x)")}\n${line("assistant", "[y](/tmp/y)")}\n`); - expect(fs.readFileSync(join(saved, project, `${NEW}.jsonl`), "utf8")).toBe(`${line("assistant", "New: [report](.brain/scratch/turns/t/report.pdf)")}\n`); + expect(fs.readFileSync(join(saved, project, `${NEW}.jsonl`), "utf8")).toBe(`${line("assistant", "New: [report](/tmp/report.pdf)")}\n`); }); test("a transcript whose saved prefix changed is not rewritten", () => { const record = (text: string) => JSON.stringify({ type: "assistant", message: { content: [{ type: "text", text }] } }); diff --git a/packages/ui-backend-claude/tests/worker.test.ts b/packages/ui-backend-claude/tests/worker.test.ts index 572066348..362134e6c 100644 --- a/packages/ui-backend-claude/tests/worker.test.ts +++ b/packages/ui-backend-claude/tests/worker.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { prepareWorkerSandbox, WORKER_STATE, workerHostBoundary } from "@schlessera/brain-ui-sdk/internal"; import { createClaudeBackend } from "../src/backend.js"; +import { createHistory } from "../src/history.js"; import { createWorkerSpawn } from "../src/worker-spawn.js"; import { createClaudeWorkerState } from "../src/worker-state.js"; import { createStateLedger } from "../src/worker-state-budget.js"; @@ -74,7 +75,7 @@ test("worker transcript export refuses symlinks and concurrent saved-state overw expect(existsSync(join(saved, project, "alias.jsonl"))).toBe(false); } finally { state.release(); rmSync(root, { recursive: true, force: true }); } }); -test.skipIf(process.platform !== "linux" || !Bun.which("bwrap"))("a real worker's linked turn-temp file is promoted and the saved answer links the copy (#1655)", async () => { +test.skipIf(process.platform !== "linux" || !Bun.which("bwrap"))("a real worker preserves native answer bytes while fresh UI history links the promoted copy (#2049)", async () => { const repo = resolve(import.meta.dir, "../../.."); mkdirSync(join(repo, "tmp"), { recursive: true }); const root = mkdtempSync(join(repo, "tmp/claude-promotion-")), brain = join(root, "brain"), saved = join(root, "config"), home = join(root, "home"); @@ -82,6 +83,11 @@ test.skipIf(process.platform !== "linux" || !Bun.which("bwrap"))("a real worker' const turnId = "9d8c7b6a-5f4e-4d3c-8b2a-1f0e9d8c7b6a", session = "33333333-3333-4333-8333-333333333333"; const project = `projects/${realpathSync(brain).replace(/[^a-zA-Z0-9]/g, "-")}`; const records: unknown[] = []; + const answer = "The report: [voyage](/tmp/brain-worker-scratch/report.pdf)."; + const entry = { type: "assistant", uuid: "44444444-4444-4444-8444-444444444444", sessionId: session, + message: { role: "assistant", content: [{ type: "text", text: answer }] } }; + // Keep noncanonical JSONL whitespace too: host promotion cannot reserialize it. + const nativeBytes = ` ${JSON.stringify(entry)} \n`; mkdirSync(join(brain, ".brain/scratch"), { recursive: true }); const view = createFixtureWorkerStartupView(brain); const outputSpy = spyOn(nativeOutput, "prepareNativeWorkerOutput").mockImplementation(prepareFixtureNativeWorkerOutput); @@ -93,8 +99,7 @@ test.skipIf(process.platform !== "linux" || !Bun.which("bwrap"))("a real worker' fs.writeFileSync(process.env.BRAIN_WORKER_SCRATCH + "/report.pdf", "Odysseus: the voyage report."); const dir = process.env.CLAUDE_CONFIG_DIR + "/${project}"; fs.mkdirSync(dir, { recursive: true }); - const text = "The report: [voyage](/tmp/brain-worker-scratch/report.pdf)."; - fs.writeFileSync(dir + "/${session}.jsonl", JSON.stringify({ type: "assistant", message: { role: "assistant", content: [{ type: "text", text }] } }) + "\\n");`; + fs.writeFileSync(dir + "/${session}.jsonl", ${JSON.stringify(nativeBytes)});`; const child = spawner.spawn({ command: process.execPath, args: ["-e", program], env: { CLAUDE_CONFIG_DIR: saved, HOME: home }, signal: new AbortController().signal } as never); child.stdin.end(); @@ -103,8 +108,19 @@ test.skipIf(process.platform !== "linux" || !Bun.which("bwrap"))("a real worker' expect(code).toBe(0); const destination = `.brain/scratch/turns/${turnId}/report.pdf`; expect(readFileSync(join(brain, destination), "utf8")).toBe("Odysseus: the voyage report."); - const transcript = JSON.parse(readFileSync(join(saved, project, `${session}.jsonl`), "utf8")); - expect(transcript.message.content[0].text).toBe(`The report: [voyage](${destination}).`); + const transcriptPath = join(saved, project, `${session}.jsonl`); + expect(readFileSync(transcriptPath, "utf8"), "REAL_WORKER_NATIVE_TRANSCRIPT_BYTES_UNCHANGED").toBe(nativeBytes); + const history = createHistory({ brainPath: brain, env: { CLAUDE_CONFIG_DIR: saved, HOME: home }, + // Reopen the actual saved file at the SDK-reader boundary. This is fresh + // production history normalization, not streamed or worker-owned objects. + getSessionMessagesFn: async id => [{ ...JSON.parse(readFileSync(join(saved, project, `${id}.jsonl`), "utf8")), + session_id: id, parent_tool_use_id: null, parent_agent_id: null }], + }); + const replay = await history.getHistory(session); + expect(replay, "REAL_WORKER_RELOADED_HISTORY_LINKS_PROMOTED_COPY").toEqual([ + { role: "assistant", content: `The report: [voyage](${destination}).`, thinking: undefined, toolCalls: [], + parts: [{ kind: "text", text: `The report: [voyage](${destination}).` }] }, + ]); expect(records).toEqual([{ kind: "promotion", source: "/tmp/brain-worker-scratch/report.pdf", destination, bytes: 28, originalLink: "/tmp/brain-worker-scratch/report.pdf", rewrittenLink: destination }]); } finally { await view.cleanup(); outputSpy.mockRestore(); rmSync(root, { recursive: true, force: true }); } diff --git a/packages/ui-backend-pi/README.md b/packages/ui-backend-pi/README.md index 3a1c35fe4..7fb356117 100644 --- a/packages/ui-backend-pi/README.md +++ b/packages/ui-backend-pi/README.md @@ -331,6 +331,19 @@ normalizer threads each result back onto its assistant tool call. `bashExecution field is additive). Session cost/turn aggregates are left to the host's `sessions` metadata table. +Promoted answer links and authoritative server receipts live in host-owned +`session-replay/` metadata below `sessionDir`, outside the provider transcript. +Replay applies each display mapping to its native message and exact original +text. Receipts overlay the matching call on the active branch; a direct server +application with no runtime call gets a display-only row at its original turn. +A direct effect committed before any native message retains a session-root +position and remains visible even when no runtime transcript file exists. +Neither path manufactures a model message in the session file. After a worker +exits, the host may append the missing result for an actual runtime-written +call whose committed server application has no runtime result; existing results +are never replaced. The installed Pi adapter otherwise repairs such orphaned +calls with its own error result. + ## Testing `bun test` — no live LLM calls. Covers permission gating, path containment, diff --git a/packages/ui-backend-pi/src/history.ts b/packages/ui-backend-pi/src/history.ts index 2c217fcf1..d6b30155f 100644 --- a/packages/ui-backend-pi/src/history.ts +++ b/packages/ui-backend-pi/src/history.ts @@ -1,4 +1,5 @@ -import { sessionPromptContext } from "@schlessera/brain-ui-sdk/internal"; +import { sessionPromptContext, sessionReplay } from "@schlessera/brain-ui-sdk/internal"; +import { dirname } from "node:path"; /** * Session listing + history normalization for the pi backend. * @@ -63,27 +64,51 @@ export async function getPiHistory( ): Promise { const infos = await SessionManager.list(cwd, sessionDir); const info = infos.find((i) => i.id === sessionId); - if (!info) return []; + if (!info && !sessionDir) return []; - const sm = SessionManager.open(info.path, sessionDir); + // A committed direct effect may be the session's only retained history. + // Reading host metadata must not manufacture a provider transcript to find it. // getBranch() returns the active path root → leaf (chronological order). - const branch = sm.getBranch(); + const branch = info ? SessionManager.open(info.path, sessionDir).getBranch() : []; const omitted = retriedAttempts(branch); const context = sessionDir ? sessionPromptContext(sessionDir, cwd, sessionId) : undefined; + const display = sessionReplay(sessionDir ?? dirname(info!.path), cwd, sessionId); + const displayText = display.texts(); const messages: AgentMessage[] = []; + const anchors = new Map(); + let visible = -1; for (const entry of branch) { if (entry.type === "message" && !omitted.has(entry.id)) { - if (context && entry.message.role === "user") { - const content = entry.message.content; + const message = structuredClone(entry.message); + if (context && message.role === "user") { + const content = message.content; const wire = typeof content === "string" ? content : content.filter(part => part.type === "text").map(part => (part as { text: string }).text).join(""); const text = context.userText(entry.id, wire); - messages.push({ ...entry.message, content: typeof content === "string" ? text : [ + message.content = typeof content === "string" ? text : [ { type: "text", text }, ...content.filter(part => part.type !== "text"), - ] }); - } else messages.push(entry.message); + ]; + } + if (message.role === "assistant") message.content.forEach((part, index) => { + if (part.type === "text") part.text = displayText(entry.id, index, part.text); + }); + messages.push(message); + if (message.role === "user" || message.role === "assistant") visible++; + anchors.set(entry.id, visible); } } - return normalizeMessages(messages); + const history = normalizeMessages(messages); + const extra: Array<{ after: number; receipt: ReturnType[number] }> = []; + for (const receipt of display.receipts()) { + const call = receipt.anchor ? history[anchors.get(receipt.anchor) ?? -1]?.toolCalls.find(call => call.id === receipt.id) : undefined; + if (call) { call.output = receipt.output; call.isError = receipt.isError; } + else if (receipt.anchor == null) extra.push({ after: -1, receipt }); + else if (anchors.has(receipt.anchor)) extra.push({ after: anchors.get(receipt.anchor)!, receipt }); + } + // Insert in reverse chronology so native positions keep their meaning. + for (const { after, receipt } of extra.sort((a, b) => a.after - b.after).reverse()) history.splice(after + 1, 0, + { role: "assistant", content: "", toolCalls: [{ id: receipt.id, name: receipt.name, input: receipt.input, + output: receipt.output, isError: receipt.isError }], parts: [{ kind: "tool", toolIndex: 0 }] }); + return history; } /** diff --git a/packages/ui-backend-pi/src/worker-session.ts b/packages/ui-backend-pi/src/worker-session.ts index 505d20bd9..92133a858 100644 --- a/packages/ui-backend-pi/src/worker-session.ts +++ b/packages/ui-backend-pi/src/worker-session.ts @@ -4,13 +4,13 @@ import { existsSync, readFileSync, realpathSync, statSync } from "node:fs"; import { join, relative, isAbsolute } from "node:path"; import { randomUUID } from "node:crypto"; import { stateRoomCopy, stateRoomWarning, type StateRoomObservation } from "@schlessera/brain-ui-sdk/protocol"; -import { getAgentDir, SessionManager, CURRENT_SESSION_VERSION, type AgentSessionEvent } from "@earendil-works/pi-coding-agent"; +import { getAgentDir, SessionManager, type AgentSessionEvent } from "@earendil-works/pi-coding-agent"; import type { ThinkingLevel } from "@earendil-works/pi-agent-core"; import type { Model } from "@earendil-works/pi-ai"; import { BackendRequestError, BRAIN_APPLICATION_TOOLS, brainApplicationInput, brainReadBaseInput, type BrainApplicationResult, compileConfirmPatterns, DEFAULT_CONFIRM_BASH_PATTERNS, isCompletedAutonomousToolCall } from "@schlessera/brain-ui-sdk/server"; import { bashLockKey, BRAIN_LOCK_KEY } from "@schlessera/brain-ui-sdk/server"; -import { NativeFileVersions, workerOutwardNote, workerCommandPolicyNote, createLimitTracker, createTurnTempPromoter, launchAgentWorker, launchLimits, prepareWorkerSandbox, processStateLedger, retainWorkerStartupView, prepareNativeWorkerOutput, type PreparedWorkerOutput, +import { sessionReplay, NativeFileVersions, workerOutwardNote, workerCommandPolicyNote, createLimitTracker, createTurnTempPromoter, launchAgentWorker, launchLimits, prepareWorkerSandbox, processStateLedger, retainWorkerStartupView, prepareNativeWorkerOutput, type PreparedWorkerOutput, outputStagingLimit, requireWorkerHost, resolveTurnTempLimits, restrictedReadPaths, turnTempLimit, workerNoteLimits, WorkerResourceLimitError, workerSandboxNote, workerSandboxWarnings, WORKER_SCRATCH, WORKER_XDG_DIRS, type Reservation, type WorkerSandbox, type WorkerStateHandle } from "@schlessera/brain-ui-sdk/internal"; @@ -63,14 +63,16 @@ export async function openPiWorkerSession(backend: CreatePiBackendOptions, sessi let ownedPrompt: { id: string; text: string; images: boolean } | undefined; const nativeVersions = nativeSessions.get(sessionId) ?? new NativeFileVersions(); if (!env.autonomous) nativeSessions.set(sessionId, nativeVersions); - /** Entries saved before this turn: their answers were promoted, if at all, by their own turn. */ + /** Entries saved before this turn: only their own turn owns their display mappings. */ const resumedIds = new Set((entries ?? []).map(entry => (entry as { id?: unknown }).id)); const turnContext = createTurnContext(); const allowedTools = new Set(env.autonomous?.allowedTools ?? backend.allowedTools ?? DEFAULT_PI_ALLOWED_TOOLS); const confirmPatterns = compileConfirmPatterns(backend.confirmBashPatterns ?? DEFAULT_CONFIRM_BASH_PATTERNS, () => {}); const bridgeTools = createPiBridgeTools({ brainPath: backend.brainPath, turn: turnContext, capabilities: env.caps }); const grants = new Map(); - const applications = new Map; result: BrainApplicationResult; output?: Awaited> }>(); + const replayTurnId = randomUUID(); + const display = sessionReplay(sessionDir, backend.brainPath, sessionId); + const applications = new Map; result: BrainApplicationResult; anchor?: string | null; output?: Awaited> }>(); let latestEntries: Record[] = []; const listeners = new Set<(event: AgentSessionEvent) => void>(); let active = false; @@ -260,57 +262,64 @@ export async function openPiWorkerSession(backend: CreatePiBackendOptions, sessi ready.reject(error); done.reject(error); child.kill("SIGKILL"); } - function persist(): void { - if (env.autonomous || !applications.size && !latestEntries.some(entry => entry.type === "message")) return; - const saved: Record[] = latestEntries.length ? structuredClone(latestEntries) : [{ type: "session", version: CURRENT_SESSION_VERSION, - id: sessionId, cwd: backend.brainPath, timestamp: new Date().toISOString() }]; - const append = (message: Record) => saved.push({ type: "message", id: randomUUID(), - parentId: saved.slice(1).at(-1)?.id ?? null, timestamp: new Date().toISOString(), message }); + function persist(recoverMissing = false): void { + if (env.autonomous) return; + // Receipt ownership survives even when no runtime call or snapshot exists. + // Never synthesize a model call or replace a runtime-written result. for (const [id, receipt] of applications) { - if (!saved.some(entry => entry.type === "message" && (entry.message as any)?.role === "assistant" - && (entry.message as any).content.some((part: any) => part.type === "toolCall" && part.id === id))) { - // A direct authorized proposal may have no runtime transcript call. - // Record the actual server effect instead of depending on worker history. - append({ role: "assistant", content: [{ type: "toolCall", id, name: receipt.name, arguments: receipt.input }], - api: model?.api ?? "server-application", provider: model?.provider ?? "server", model: model?.id ?? "server-application", - stopReason: "toolUse", timestamp: Date.now(), - usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } } }); - } - const result = { role: "toolResult", toolCallId: id, toolName: receipt.name, - content: receipt.output?.content ?? [{ type: "text", text: JSON.stringify(receipt.result) }], isError: !receipt.result.ok, timestamp: Date.now() }; - const prior = saved.find(entry => entry.type === "message" && (entry.message as any)?.role === "toolResult" && (entry.message as any).toolCallId === id); - if (prior) prior.message = result; - else { - // A committed server effect survives cancellation before the runtime - // can append its own tool result. Later worker snapshots cannot erase it. - append(result); + const called = latestEntries.find(entry => !resumedIds.has(entry.id) && entry.type === "message" + && (entry.message as any)?.role === "assistant" && (entry.message as any).content.some((part: any) => part.type === "toolCall" && part.id === id)); + if (typeof called?.id === "string") receipt.anchor = called.id; + else if (receipt.anchor === undefined) { + // Direct effects can precede every native message. Null is an owned + // session-root position, not an anchor to a later worker snapshot. + const prior = latestEntries.findLast(entry => entry.type === "message") + ?? entries?.findLast(entry => entry.type === "message"); + receipt.anchor = typeof prior?.id === "string" ? prior.id : null; } + display.retainReceipt({ id, name: receipt.name, input: receipt.input, anchor: receipt.anchor, turnId: replayTurnId, + output: receipt.output?.content.filter(part => part.type === "text").map(part => part.text).join("") ?? JSON.stringify(receipt.result), + isError: !receipt.result.ok }); + } + if (!latestEntries.some(entry => entry.type === "message")) return; + const saved = [...latestEntries]; + if (recoverMissing) for (const [id, receipt] of applications) { + const called = saved.some(entry => !resumedIds.has(entry.id) && entry.type === "message" && (entry.message as any)?.role === "assistant" + && (entry.message as any).content.some((part: any) => part.type === "toolCall" && part.id === id)); + const answered = saved.some(entry => !resumedIds.has(entry.id) && entry.type === "message" && (entry.message as any)?.role === "toolResult" + && (entry.message as any).toolCallId === id); + if (!called || answered) continue; + // The installed Pi adapter repairs orphaned calls with "No result provided". + // Preserve the committed effect instead, only after the runtime has exited. + saved.push({ type: "message", id: randomUUID(), parentId: saved.at(-1)?.id ?? null, + timestamp: new Date().toISOString(), message: { role: "toolResult", toolCallId: id, toolName: receipt.name, + content: receipt.output?.content ?? [{ type: "text", text: JSON.stringify(receipt.result) }], + isError: !receipt.result.ok, timestamp: Date.now() } }); } savePiSession(sessionDir, sessionId, saved); } + /** * Once the worker has exited, copies what this turn's answers link in turn - * temp into scratch, saves the answers linking the copies, and reports each + * temp into scratch, retains display mappings outside the transcript, and reports each * linked file to the host (#1655). Turn temp is removed either way. */ function promote(): void { - if (!turnTemp || env.autonomous) { closeTurnTemp(); return; } try { + persist(true); + if (!turnTemp || env.autonomous) return; const promoter = createTurnTempPromoter({ brain: sandbox.brain, tmp: turnTemp.root, turnId: turnContext.turnId, scratch: WORKER_SCRATCH, exclude: [PI_WORKER_STATE], scratchWritable: sandbox.areas.some(area => area.name === "scratch" && area.mount.writable) }); - let changed = false; for (const entry of latestEntries) { if (entry.type !== "message" || resumedIds.has(entry.id)) continue; const message = entry.message as { role?: unknown; content?: unknown } | undefined; if (message?.role !== "assistant" || !Array.isArray(message.content)) continue; - for (const part of message.content as Array<{ type?: unknown; text?: unknown }>) { + for (const [index, part] of (message.content as Array<{ type?: unknown; text?: unknown }>).entries()) { if (part?.type !== "text" || typeof part.text !== "string") continue; const next = promoter.rewrite(part.text); - if (next !== part.text) { part.text = next; changed = true; } + if (typeof entry.id === "string") display.retainText(entry.id, index, part.text, next); } } - if (changed) persist(); for (const record of promoter.records) { try { turnContext.bridge?.activity?.(record); } catch { /* Observability must not fail the observed turn. */ } } diff --git a/packages/ui-backend-pi/tests/direct-application-replay.test.ts b/packages/ui-backend-pi/tests/direct-application-replay.test.ts new file mode 100644 index 000000000..2da951aa5 --- /dev/null +++ b/packages/ui-backend-pi/tests/direct-application-replay.test.ts @@ -0,0 +1,72 @@ +import { expect, spyOn, test } from "bun:test"; +import { spawn } from "node:child_process"; +import { createHash } from "node:crypto"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import * as worker from "@schlessera/brain-ui-sdk/internal"; +import { savePiSession } from "../src/session-storage.js"; +import { createPiBackend } from "../src/backend.js"; +import type { ServerMessage } from "@schlessera/brain-ui-sdk/server"; + +for (const mode of ["none", "header", "late", "cancel", "multiple", "resume"]) test(`direct authorized application receipt: ${mode}`, async () => { + const root = mkdtempSync(join(tmpdir(), "odysseus-direct-receipt-")), brain = join(root, "ithaca"), sessionDir = join(root, "sessions"); + mkdirSync(join(brain, "notes"), { recursive: true }); mkdirSync(sessionDir); + const previous = process.env.PI_CODING_AGENT_DIR; process.env.PI_CODING_AGENT_DIR = join(root, "agent"); + const children: ReturnType[] = [], exits: Promise[] = []; + const probe = spyOn(worker.workerHostBoundary, "probe").mockReturnValue({ ok: true }); + const launch = spyOn(worker, "launchAgentWorker").mockImplementation(() => { + const child = spawn(process.execPath, [join(import.meta.dir, "fixtures/direct-application-worker.ts")], { stdio: ["pipe", "pipe", "pipe"], env: { PATH: process.env.PATH! } }); + children.push(child); + const exited = new Promise((resolve, reject) => { child.once("exit", (code, signal) => { expect(signal === "SIGKILL" || code === 0).toBe(true); resolve(code ?? 128); }); child.once("error", reject); }); + exits.push(exited); + return { stdin: child.stdin!, stdout: child.stdout!, stderr: child.stderr!, exited, + kill: (signal?: NodeJS.Signals) => child.kill(signal), limitReport: () => ({ hits: [], fatal: false }) } as unknown as ReturnType; + }); + try { + const resumeId = "11111111-1111-4111-8111-111111111111"; + if (mode === "resume") savePiSession(sessionDir, resumeId, [ + { type: "session", version: 3, id: resumeId, cwd: brain, timestamp: "2026-07-12T00:00:00.000Z" }, + { type: "message", id: "prior-user", parentId: null, timestamp: "2026-07-12T00:00:00.000Z", message: { role: "user", content: "Earlier voyage", timestamp: 0 } }, + ]); + const before = mode === "resume" ? readFileSync(join(sessionDir, `${resumeId}.jsonl`), "utf8") : undefined; + const backend = createPiBackend({ brainPath: brain, sessionDir }), frames: ServerMessage[] = []; + let applications = 0; + const controller = new AbortController(); + await backend.startTurn({ prompt: mode, ...(mode === "resume" ? { sessionId: resumeId } : {}), signal: controller.signal, bridge: { + emit: frame => frames.push(frame), requestPermission: async () => ({ behavior: "deny", message: "No generic grants" }), + applyBrain: async input => { + expect(input).toEqual({ operation: "write", path: "notes/raft.md", content: "Odysseus repairs the raft.\n", expectedBaseHash: null }); + applications++; writeFileSync(join(brain, "notes/raft.md"), "Odysseus repairs the raft.\n"); + if (mode === "cancel") controller.abort(); + return { ok: true, message: "Committed the direct raft plan", changes: [{ path: "notes/raft.md", contentHash: createHash("sha256").update("Odysseus repairs the raft.\n").digest("hex") }] }; + }, + } }); + expect(applications, "ACTUAL_PARENT_RPC_COMMITTED_EFFECT").toBe(mode === "multiple" ? 2 : 1); + expect(readFileSync(join(brain, "notes/raft.md"), "utf8")).toBe("Odysseus repairs the raft.\n"); + const result = frames.findLast(frame => frame.type === "result"); expect(result).toBeDefined(); + if (!result) throw new Error("Missing terminal result"); + expect(result.outcome).toBe(mode === "cancel" ? "cancelled" : "success"); + expect(existsSync(join(sessionDir, `${result.sessionId}.jsonl`)), "NO_MANUFACTURED_PROVIDER_TRANSCRIPT").toBe(mode === "late" || mode === "resume"); + if (before !== undefined) expect(readFileSync(join(sessionDir, `${resumeId}.jsonl`), "utf8"), "RESUMED_DIRECT_EFFECT_PRESERVES_ALL_NATIVE_BYTES").toBe(before); + if (mode === "late") { + const native = readFileSync(join(sessionDir, `${result.sessionId}.jsonl`), "utf8").trim().split("\n").map(line => JSON.parse(line)); + expect(native.map(entry => entry.type), "ONLY_RUNTIME_HEADER_AND_USER").toEqual(["session", "message"]); + expect(native[1].message).toEqual({ role: "user", content: "Next voyage", timestamp: 1 }); + } + const reloaded = createPiBackend({ brainPath: brain, sessionDir }); + const history = await reloaded.getHistory(result.sessionId); + if (mode === "late") expect(history.map(message => message.role), "DIRECT_EFFECT_KEEPS_PRE_SNAPSHOT_POSITION").toEqual(["assistant", "user"]); + if (mode === "resume") expect(history.map(message => message.role), "DIRECT_EFFECT_FOLLOWS_PRIOR_NATIVE_LEAF").toEqual(["user", "assistant"]); + const calls = history.flatMap(message => message.toolCalls); + expect(calls, "DIRECT_EFFECT_VISIBLE_WITHOUT_NATIVE_HISTORY").toHaveLength(mode === "multiple" ? 2 : 1); + if (mode === "multiple") expect(calls.map(call => call.id), "DIRECT_RECEIPTS_KEEP_COMMIT_ORDER").toEqual(["direct-raft", "direct-mast"]); + expect(calls[0]).toMatchObject({ id: "direct-raft", output: expect.stringContaining("Committed the direct raft plan"), isError: false }); + expect(await reloaded.getHistory("22222222-2222-4222-8222-222222222222"), "UNRELATED_SESSION_HAS_NO_RECEIPT").toEqual([]); + } finally { + children.forEach(child => child.kill("SIGKILL")); await Promise.allSettled(exits); + launch.mockRestore(); probe.mockRestore(); + if (previous === undefined) delete process.env.PI_CODING_AGENT_DIR; else process.env.PI_CODING_AGENT_DIR = previous; + rmSync(root, { recursive: true, force: true }); + } +}); diff --git a/packages/ui-backend-pi/tests/fixtures/direct-application-worker.ts b/packages/ui-backend-pi/tests/fixtures/direct-application-worker.ts new file mode 100644 index 000000000..338e243b6 --- /dev/null +++ b/packages/ui-backend-pi/tests/fixtures/direct-application-worker.ts @@ -0,0 +1,32 @@ +/** An untrusted worker may propose an authorized effect without supplying runtime history. */ +import { pipeMessages, sendPipe } from "../../src/worker-protocol.js"; +let init: any; +let mode = ""; +let answers = 0; +for await (const raw of pipeMessages(process.stdin)) { + const message = raw as any; + if (message.type === "init") { + init = message; + sendPipe(process.stdout, { type: "ready", sessionId: init.sessionId, levels: [], cost: 0 }); + } else if (message.type === "prompt") { + // The fixture command is its first line; owned host context may follow it. + mode = String(message.text).split("\n", 1)[0]!; + if (mode === "header") sendPipe(process.stdout, { type: "snapshot", entries: [{ type: "session", version: 3, + id: init.sessionId, cwd: init.backend.brainPath, timestamp: "2026-07-12T00:00:00.000Z" }] }); + sendPipe(process.stdout, { type: "rpc", id: 1, method: "apply", toolCallId: "direct-raft", toolName: "write_file", + input: { operation: "write", path: "notes/raft.md", content: "Odysseus repairs the raft.\n", expectedBaseHash: null } }); + } else if (message.type === "answer") { + if (message.error) throw new Error(message.error); + if (mode === "multiple" && answers++ === 0) { + sendPipe(process.stdout, { type: "rpc", id: 2, method: "apply", toolCallId: "direct-mast", toolName: "write_file", + input: { operation: "write", path: "notes/raft.md", content: "Odysseus repairs the raft.\n", expectedBaseHash: null } }); + continue; + } + if (mode === "late") sendPipe(process.stdout, { type: "snapshot", entries: [ + { type: "session", version: 3, id: init.sessionId, cwd: init.backend.brainPath, timestamp: "2026-07-12T00:00:00.000Z" }, + { type: "message", id: "later-user", parentId: null, timestamp: "2026-07-12T00:00:01.000Z", message: { role: "user", content: "Next voyage", timestamp: 1 } }, + ] }); + // The server committed the effect; the worker never supplied a message snapshot. + sendPipe(process.stdout, { type: "done", cost: 0 }); + } +} diff --git a/packages/ui-backend-pi/tests/fixtures/transcript-prefix-probe.ts b/packages/ui-backend-pi/tests/fixtures/transcript-prefix-probe.ts new file mode 100644 index 000000000..d1e188829 --- /dev/null +++ b/packages/ui-backend-pi/tests/fixtures/transcript-prefix-probe.ts @@ -0,0 +1,61 @@ +/** Native Pi workers and local OpenAI-compatible requests, confined by the caller. */ +import { cpSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { createPiBackend } from "../../src/backend.js"; +import { prepareFixtureWorkerSandbox } from "../../../ui-sdk/tests/helpers/worker-startup-view-fixture.js"; +import { startFixtureTurn } from "../../../../tests/fixtures/worker-input-fixture.js"; +import type { ServerMessage } from "@schlessera/brain-ui-sdk/server"; +const scratch = process.argv[2]!; +process.env.PI_CODING_AGENT_DIR = join(scratch, "agent"); +mkdirSync(process.env.PI_CODING_AGENT_DIR, { recursive: true }); +const brain = join(scratch, "ithaca"), sessionDir = join(scratch, "sessions"); +cpSync(resolve("packages/core/fixtures/corpus"), brain, { recursive: true }); +const config = join(brain, "brain.config.ts"); +writeFileSync(config, readFileSync(config, "utf8").replace('"@schlessera/brain"', JSON.stringify(resolve("packages/core/src/index.ts")))); +writeFileSync(join(brain, ".gitignore"), ".brain/\n"); +mkdirSync(join(brain, ".brain/scratch"), { recursive: true }); +mkdirSync(join(brain, "docs"), { recursive: true }); +writeFileSync(join(brain, "docs/ithaca-chart.md"), "Odysseus verifies the Ithaca chart.\n"); +const requests: any[] = []; +let step = 0; +const link = "[Ithaca chart](/tmp/ithaca-chart.pdf)"; +const server = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(request) { + if (new URL(request.url).pathname !== "/v1/chat/completions") throw new Error(`Unexpected endpoint ${request.url}`); + const body = await request.json() as any; requests.push(body); + const index = step++; + const tool = index === 0 ? { id: "call_chart", type: "function", function: { name: "bash", arguments: JSON.stringify({ command: "printf 'Odysseus charts Ithaca.' > /tmp/ithaca-chart.pdf" }) } } + : index === 1 ? { id: "call_read", type: "function", function: { name: "read_file", arguments: JSON.stringify({ path: "docs/ithaca-chart.md" }) } } : undefined; + const text = index === 1 ? `Before checking: ${link}` : index === 2 ? `After checking: ${link}` : index > 2 ? "Odysseus follows the chart." : ""; + const chunks = [ + { id: `chat_odysseus_${index}`, object: "chat.completion.chunk", model: body.model, + choices: [{ index: 0, delta: { role: "assistant", content: text, ...(tool ? { tool_calls: [{ index: 0, ...tool }] } : {}) }, finish_reason: null }] }, + { id: `chat_odysseus_${index}`, object: "chat.completion.chunk", model: body.model, + choices: [{ index: 0, delta: {}, finish_reason: tool ? "tool_calls" : "stop" }], usage: { prompt_tokens: 3, completion_tokens: 5, total_tokens: 8 } }, + ]; + return new Response(chunks.map(chunk => `data: ${JSON.stringify(chunk)}\n\n`).join("") + "data: [DONE]\n\n", { headers: { "content-type": "text/event-stream" } }); +} }); +writeFileSync(join(process.env.PI_CODING_AGENT_DIR, "models.json"), JSON.stringify({ providers: { openai: { + baseUrl: `${server.url.origin}/v1`, api: "openai-completions", apiKey: "offline-odysseus", models: [{ id: "gpt-4o-mini" }], +} } })); +writeFileSync(join(process.env.PI_CODING_AGENT_DIR, "settings.json"), JSON.stringify({ retry: { enabled: false }, compaction: { enabled: false }, cacheWarming: "off" })); +let sessionId: string | undefined; +try { + for (let index = 0; index < 2; index++) { + const backend = createPiBackend({ brainPath: brain, sessionDir, loadExtensions: false, confirmBashPatterns: [], + profiles: [{ id: "fixture", label: "Fixture", vendor: "openai", model: "gpt-4o-mini", thinkingLevel: "off" }] }); + const sandbox = prepareFixtureWorkerSandbox(brain, { profile: "strict" }); + const frames: ServerMessage[] = [], controller = new AbortController(); + const deadline = setTimeout(() => controller.abort(), 45000); + try { + await startFixtureTurn(backend, brain, { prompt: index ? "Use the chart" : "Prepare the chart", profileId: "fixture", sessionId, + signal: controller.signal, turnId: `odysseus-${index}`, workerSandbox: sandbox, + bridge: { emit(frame) { frames.push(frame); }, async requestPermission() { return { behavior: "allow" }; } } }); + const result = frames.findLast(frame => frame.type === "result"); + if (!result || result.outcome !== "success") throw new Error(JSON.stringify(frames)); + sessionId = result.sessionId; + } finally { clearTimeout(deadline); } + } + const history = await createPiBackend({ brainPath: brain, sessionDir }).getHistory(sessionId!); + const transcript = readFileSync(join(sessionDir, `${sessionId}.jsonl`), "utf8"); + writeFileSync(join(scratch, "requests.json"), JSON.stringify({ requests, history, transcript })); +} finally { server.stop(true); } diff --git a/packages/ui-backend-pi/tests/transcript-prefix.test.ts b/packages/ui-backend-pi/tests/transcript-prefix.test.ts new file mode 100644 index 000000000..e805d3943 --- /dev/null +++ b/packages/ui-backend-pi/tests/transcript-prefix.test.ts @@ -0,0 +1,26 @@ +/** Exact provider-history regression using the installed runtimes, no live provider. */ +import { expect, test } from "bun:test"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { tmpdir } from "node:os"; +const root = resolve(import.meta.dir, "../../.."); +test("PI promoted links preserve the previous request messages byte for byte", async () => { + const scratch = mkdtempSync(join(tmpdir(), "transcript-prefix-pi-")); + try { + const child = Bun.spawn(["bwrap", "--unshare-net", "--bind", "/", "/", "--proc", "/proc", "--dev", "/dev", process.execPath, + "--preload", join(root, "scripts/test-network-child-preload.ts"), join(import.meta.dir, "fixtures/transcript-prefix-probe.ts"), scratch], + { cwd: root, env: { PATH: process.env.PATH!, LANG: "C.UTF-8", TMPDIR: tmpdir() }, stdout: "pipe", stderr: "pipe" }); + const [out, err, code] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text(), child.exited]); + expect(code, err || out).toBe(0); + const { requests, history, transcript } = JSON.parse(readFileSync(join(scratch, "requests.json"), "utf8")); + expect(requests, "REAL_TOOL_ROUND_TRIP_AND_RESUMED_REQUEST").toHaveLength(4); + const previous = requests[2].messages, resumed = requests[3].messages; + expect(JSON.stringify(previous), "INTERMEDIATE_PROVIDER_LINK_BEFORE_TOOL").toContain("Before checking: [Ithaca chart](/tmp/ithaca-chart.pdf)"); + expect(JSON.stringify(resumed.slice(0, previous.length)), "PI_BYTE_IDENTICAL_MESSAGES_PREFIX").toBe(JSON.stringify(previous)); + expect(transcript, "SAVED_RUNTIME_ANSWER_UNCHANGED").toContain("After checking: [Ithaca chart](/tmp/ithaca-chart.pdf)"); + const displayed = JSON.stringify(history); + expect(displayed, "RELOADED_DISPLAY_HAS_BOTH_PROMOTED_LINKS").toContain("Before checking: [Ithaca chart](.brain/scratch/turns/odysseus-0/ithaca-chart.pdf)"); + expect(displayed).toContain("After checking: [Ithaca chart](.brain/scratch/turns/odysseus-0/ithaca-chart.pdf)"); + expect(displayed).not.toContain("[Ithaca chart](/tmp/"); + } finally { rmSync(scratch, { recursive: true, force: true }); } +}, 180_000); diff --git a/packages/ui-backend-pi/tests/worker-session.test.ts b/packages/ui-backend-pi/tests/worker-session.test.ts index 53f5e7e2b..7751568ad 100644 --- a/packages/ui-backend-pi/tests/worker-session.test.ts +++ b/packages/ui-backend-pi/tests/worker-session.test.ts @@ -8,6 +8,7 @@ import * as worker from "@schlessera/brain-ui-sdk/internal"; import type { ServerMessage } from "@schlessera/brain-ui-sdk/server"; import { createPiBackend } from "../src/backend.js"; import { openPiWorkerSession } from "../src/worker-session.js"; +import { getPiHistory } from "../src/history.js"; import { savePiSession } from "../src/session-storage.js"; import { PI_PIPE_MAX_BYTES, PI_WORKER_STATE, sendPipe, sessionReserveBytes, SessionTooLargeError } from "../src/worker-protocol.js"; @@ -136,7 +137,7 @@ test("snapshot overflow reaches the terminal TurnFailure", async () => { expect(result.stateRoom).toBeUndefined(); expect(launch).toHaveBeenCalledTimes(1); }); -test("a turn's linked turn-temp file is promoted into scratch and the saved answer links the copy (#1655)", async () => { +test("turn-temp promotion preserves native answer bytes and reloads promoted UI links (#2049)", async () => { const turnId = "0b7e3c1a-4f2d-4c6b-8a9e-1d2c3b4a5f60"; const answer = "The voyage report is [ready](/tmp/brain-worker-scratch/report.pdf). Not [this](/tmp/brain-worker-scratch/pi-state/auth.json)."; // Stands in for the worker's private /tmp, as the server holds it (#1703). @@ -189,8 +190,11 @@ test("a turn's linked turn-temp file is promoted into scratch and the saved answ const destination = `.brain/scratch/turns/${turnId}/report.pdf`; expect(readFileSync(join(brain, destination), "utf8")).toBe("Odysseus: the voyage report."); const saved = readFileSync(join(sessionDir, readdirSync(sessionDir).find(name => name.endsWith(".jsonl"))!), "utf8"); - expect(saved).toContain(`The voyage report is [ready](${destination}).`); - expect(saved).not.toContain("[ready](/tmp/"); + expect(saved, "PI_NATIVE_ASSISTANT_TEXT_UNCHANGED").toContain(answer); + const nativeId = JSON.parse(saved.split("\n")[0]!).id; + const replay = await getPiHistory(brain, nativeId, sessionDir); + expect(replay.find(message => message.role === "assistant")?.content, "PI_RELOADED_DISPLAY_LINK").toContain(`[ready](${destination})`); + expect(replay.find(message => message.role === "assistant")?.parts).toContainEqual({ kind: "text", text: answer.replace("/tmp/brain-worker-scratch/report.pdf", destination) }); // Runtime state in turn temp (the login) is never promoted. expect(saved).toContain("[this](/tmp/brain-worker-scratch/pi-state/auth.json)"); expect(events).toEqual([{ kind: "promotion", source: "/tmp/brain-worker-scratch/report.pdf", destination, bytes: 28, @@ -304,3 +308,70 @@ test.each(["id", "cwd"])("canonical transcript refuses a mismatched %s before wo .rejects.toThrow("Cannot resume unknown session"); expect(launch).not.toHaveBeenCalled(); }); + + +for (const mode of ["result", "orphan", "direct"] as const) test(`Pi authoritative ${mode} receipt is replay metadata without rewriting runtime entries`, async () => { + const input = { operation: "write", path: "notes/raft.md", content: "Odysseus repairs the raft.", expectedBaseHash: null }; + const receipt = { ok: true, message: "Server committed the raft plan.", changes: [{ path: input.path, contentHash: "a".repeat(64) }] }; + let runtime: any[] = []; + launch.mockImplementation(() => { + const stdout = new PassThrough(), stderr = new PassThrough(), exited = Promise.withResolvers(); + const stop = () => { stdout.end(); stderr.end(); exited.resolve(0); }; + children.push(stop); + let sessionId = ""; + const snapshot = () => sendPipe(stdout, { type: "snapshot", entries: runtime }); + const stdin = new Writable({ write(chunk, _encoding, callback) { + const message = JSON.parse(String(chunk)); + if (message.type === "init") { + sessionId = message.sessionId; + sendPipe(stdout, { type: "ready", sessionId, levels: [], cost: 0 }); + } else if (message.type === "prompt") { + runtime = [ + { type: "session", version: 3, id: sessionId, cwd: brain, timestamp: "2026-07-12T00:00:00.000Z" }, + { type: "message", id: "u1", parentId: null, timestamp: "2026-07-12T00:00:00.000Z", message: { role: "user", content: "Repair the raft", timestamp: 0 } }, + ]; + if (mode !== "direct") runtime.push({ type: "message", id: "a1", parentId: "u1", timestamp: "2026-07-12T00:00:01.000Z", + message: { role: "assistant", content: [{ type: "toolCall", id: "raft-call", name: "write_file", arguments: input }], + model: "odysseus-1", stopReason: "toolUse", timestamp: 1, + usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, totalTokens: 2, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } } } }); + snapshot(); + sendPipe(stdout, { type: "rpc", id: 1, method: "apply", toolCallId: "raft-call", toolName: "write_file", input }); + } else if (message.type === "answer") { + expect(message.error).toBeUndefined(); + if (mode === "result") runtime.push({ type: "message", id: "r1", parentId: "a1", timestamp: "2026-07-12T00:00:02.000Z", + message: { role: "toolResult", toolCallId: "raft-call", toolName: "write_file", content: [{ type: "text", text: "Runtime wrote these exact result bytes." }], isError: false, timestamp: 2 } }); + snapshot(); + sendPipe(stdout, { type: "done", cost: 0 }); + } + callback(); + } }); + return { stdin, stdout, stderr, exited: exited.promise, kill: stop, limitReport: () => ({ hits: [], fatal: false }) } as unknown as ReturnType; + }); + const frames: ServerMessage[] = []; + await createPiBackend({ brainPath: brain, sessionDir }).startTurn({ prompt: "Repair the raft", signal: new AbortController().signal, + bridge: { emit: frame => frames.push(frame), requestPermission: async () => ({ behavior: "allow" }), applyBrain: async () => receipt } }); + const result = terminal(frames); + expect(result).toMatchObject({ outcome: "success" }); + const bytes = readFileSync(join(sessionDir, result.sessionId + ".jsonl"), "utf8"); + const runtimeBytes = runtime.map(entry => JSON.stringify(entry)).join("\n") + "\n"; + if (mode === "orphan") { + expect(bytes, "PI_ORPHAN_ONLY_APPENDS_MISSING_RESULT").toStartWith(runtimeBytes); + const appended = bytes.slice(runtimeBytes.length).trim().split("\n").map(line => JSON.parse(line)); + expect(appended).toHaveLength(1); + expect(appended[0].message).toMatchObject({ role: "toolResult", toolCallId: "raft-call", content: [{ type: "text", text: JSON.stringify(receipt) }] }); + } else expect(bytes, "PI_RUNTIME_TRANSCRIPT_UNCHANGED").toBe(runtimeBytes); + const replay = await getPiHistory(brain, result.sessionId, sessionDir); + const calls = replay.flatMap(message => message.toolCalls); + expect(calls, "PI_SERVER_RECEIPT_REPLAYED_ONCE").toHaveLength(1); + expect(calls[0]).toMatchObject({ id: "raft-call", name: "write_file", output: JSON.stringify(receipt), isError: false }); + if (mode === "direct") { + const later = { type: "message", id: "u2", parentId: "u1", timestamp: "2026-07-12T00:00:03.000Z", message: { role: "user", content: "Next voyage", timestamp: 3 } }; + savePiSession(sessionDir, result.sessionId, [...runtime, later]); + const resumed = await getPiHistory(brain, result.sessionId, sessionDir); + expect(resumed.map(message => message.role), "DIRECT_RECEIPT_STAYS_WITH_ITS_TURN").toEqual(["user", "assistant", "user"]); + savePiSession(sessionDir, result.sessionId, [runtime[0], { ...later, parentId: null }]); + const branched = await getPiHistory(brain, result.sessionId, sessionDir); + expect(branched.flatMap(message => message.toolCalls), "UNRELATED_BRANCH_HAS_NO_RECEIPT").toEqual([]); + } + +}); diff --git a/packages/ui-sdk/src/internal.ts b/packages/ui-sdk/src/internal.ts index e2dea19fa..30bd18afa 100644 --- a/packages/ui-sdk/src/internal.ts +++ b/packages/ui-sdk/src/internal.ts @@ -80,3 +80,6 @@ export { sessionPromptContext } from "./server/session-prompt-context.js"; export type { SessionPromptFacts } from "./server/session-prompt-context.js"; export { describeClient } from "./server/system-prompt.js"; + +export { sessionReplay } from "./server/session-replay.js"; +export type { ReplayReceipt } from "./server/session-replay.js"; diff --git a/packages/ui-sdk/src/server/session-replay.ts b/packages/ui-sdk/src/server/session-replay.ts new file mode 100644 index 000000000..4edd05c05 --- /dev/null +++ b/packages/ui-sdk/src/server/session-replay.ts @@ -0,0 +1,73 @@ +/** Host-owned display metadata. These files are never runtime transcript input. */ +import { createHash, randomUUID } from "node:crypto"; +import { closeSync, constants, existsSync, fstatSync, mkdirSync, openSync, readFileSync, realpathSync, renameSync, rmSync, writeFileSync } from "node:fs"; +import { dirname, join, resolve } from "node:path"; + +interface TextMapping { digest: string; text: string } +export interface ReplayReceipt { id: string; name: string; input: Record; output: string; isError: boolean; anchor?: string | null; turnId?: string } +interface Snapshot { texts: Record>; receipts: Record } +const digest = (text: string) => createHash("sha256").update(text).digest("hex"); +const MAX_BYTES = 64 * 1024 * 1024; +function directory(path: string): void { + if (existsSync(path)) { + if (realpathSync(path) !== path) throw new Error("Session replay storage must not contain directory aliases"); + return; + } + directory(dirname(path)); mkdirSync(path, { mode: 0o700 }); +} +/** Concrete first-party storage, keyed by native session, message and text part. */ +export function sessionReplay(root: string, brainPath: string, sessionId: string) { + const brain = existsSync(brainPath) ? realpathSync(brainPath) : resolve(brainPath), dir = resolve(root, "session-replay", digest(brain)); + let ancestor = dir; + while (!existsSync(ancestor)) ancestor = dirname(ancestor); + const actual = resolve(realpathSync(ancestor), dir.slice(ancestor.length).replace(/^\//, "")); + const outside = actual !== brain && !actual.startsWith(brain + "/") && !brain.startsWith(actual + "/"); + const validate = () => { + if (!outside) throw new Error("Session replay storage must be outside the authoritative brain"); + if (existsSync(dir)) directory(dir); + }; + const path = join(dir, `${digest(sessionId)}.json`); + function read(): Snapshot { + let fd: number; + try { fd = openSync(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); } + catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return { texts: {}, receipts: {} }; throw error; } + try { + validate(); + const stat = fstatSync(fd); + if (!stat.isFile() || stat.nlink !== 1 || stat.size > MAX_BYTES) throw new Error("Invalid session replay storage file"); + const bytes = readFileSync(fd); + if (bytes.length > MAX_BYTES) throw new Error("Session replay storage exceeds its byte bound"); + return JSON.parse(bytes.toString("utf8")) as Snapshot; + } finally { closeSync(fd); } + } + function write(state: Snapshot): void { + validate(); + const bytes = JSON.stringify(state); + if (Buffer.byteLength(bytes) > MAX_BYTES) throw new Error("Session replay storage exceeds its byte bound"); + directory(dir); + const temp = `${path}.${randomUUID()}.tmp`; + try { writeFileSync(temp, bytes, { flag: "wx", mode: 0o600 }); renameSync(temp, path); } + finally { rmSync(temp, { force: true }); } + } + return { + /** Retain this message's promoted-link display text without editing the message. */ + retainText(messageId: string, part: number, original: string, text: string): void { + if (original === text) return; + const state = read(); + (state.texts[digest(messageId)] ??= {})[part] = { digest: digest(original), text }; + write(state); + }, + /** A changed/reused native identity never receives another message's mapping. */ + texts() { + const state = read(); + return (messageId: string, part: number, original: string): string => { + const mapping = state.texts[digest(messageId)]?.[part]; + return mapping?.digest === digest(original) ? mapping.text : original; + }; + }, + retainReceipt(receipt: ReplayReceipt): void { + const state = read(); state.receipts[digest(`${receipt.turnId ?? ""}:${receipt.id}`)] = receipt; write(state); + }, + receipts(): ReplayReceipt[] { return Object.values(read().receipts); }, + }; +} diff --git a/packages/ui-sdk/tests/session-replay.test.ts b/packages/ui-sdk/tests/session-replay.test.ts new file mode 100644 index 000000000..2be4c70d6 --- /dev/null +++ b/packages/ui-sdk/tests/session-replay.test.ts @@ -0,0 +1,55 @@ +import { afterEach, expect, test } from "bun:test"; +import { linkSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, symlinkSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { sessionReplay } from "../src/server/session-replay.js"; +const roots: string[] = []; +afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); +function fixture() { + const root = mkdtempSync(join(tmpdir(), "odysseus-replay-")); roots.push(root); + const brain = join(root, "ithaca"), storage = join(root, "state"); mkdirSync(brain); mkdirSync(storage); + return { root, brain, storage }; +} +test("display mappings persist by session, message, part and exact native text", () => { + const { brain, storage } = fixture(), original = "[Ithaca](/tmp/chart.pdf)", displayed = "[Ithaca](.brain/scratch/chart.pdf)"; + const store = sessionReplay(storage, brain, "odysseus"); + store.retainText("athena", 1, original, displayed); + const read = sessionReplay(storage, brain, "odysseus").texts(); + expect(read("athena", 1, original)).toBe(displayed); + expect(read("athena", 0, original)).toBe(original); + expect(read("penelope", 1, original)).toBe(original); + expect(read("athena", 1, original + "changed")).toBe(original + "changed"); + expect(sessionReplay(storage, brain, "telemachus").texts()("athena", 1, original)).toBe(original); + // Native ids cannot address a property or a path in the host store. + store.retainText("__proto__", 0, original, displayed); + expect(sessionReplay(storage, brain, "odysseus").texts()("__proto__", 0, original)).toBe(displayed); +}); +test("receipt identity is stable across snapshots without duplicate replay records", () => { + const { brain, storage } = fixture(), store = sessionReplay(storage, brain, "odysseus"); + const receipt = { id: "raft-call", name: "write_file", input: { path: "notes/raft.md" }, output: "committed", isError: false, anchor: "u1" }; + store.retainReceipt(receipt); store.retainReceipt({ ...receipt, output: "server committed" }); + expect(sessionReplay(storage, brain, "odysseus").receipts()).toEqual([{ ...receipt, output: "server committed" }]); +}); +test("replay storage is never created inside the authoritative brain", () => { + const { brain } = fixture(), store = sessionReplay(brain, brain, "odysseus"); + expect(() => store.retainText("athena", 0, "original", "promoted")).toThrow("outside"); + expect(readdirSync(brain)).toEqual([]); +}); +for (const alias of ["symlink", "hardlink"] as const) test(`replay storage refuses ${alias} file aliases`, () => { + const { root, brain, storage } = fixture(), store = sessionReplay(storage, brain, "odysseus"); + store.retainText("athena", 0, "original", "promoted"); + const parent = join(storage, "session-replay", readdirSync(join(storage, "session-replay"))[0]!); + const file = join(parent, readdirSync(parent)[0]!), outside = join(root, "outside"); + if (alias === "symlink") { linkSync(file, outside); rmSync(file); symlinkSync(outside, file); } + else linkSync(file, outside); + const original = readFileSync(outside, "utf8"); + expect(() => store.texts()).toThrow(); + expect(() => store.retainText("athena", 0, "original", "changed")).toThrow(); + expect(readFileSync(outside, "utf8")).toBe(original); +}); +test("replay storage refuses an ancestor redirected into the brain before writing", () => { + const { root, brain } = fixture(); symlinkSync(brain, join(root, "redirected")); + const store = sessionReplay(join(root, "redirected"), brain, "odysseus"); + expect(() => store.retainText("athena", 0, "original", "promoted")).toThrow("outside"); + expect(readdirSync(brain)).toEqual([]); +}); diff --git a/packages/ui-server/src/ws/message-sources.ts b/packages/ui-server/src/ws/message-sources.ts index cfbba904b..78ae24280 100644 --- a/packages/ui-server/src/ws/message-sources.ts +++ b/packages/ui-server/src/ws/message-sources.ts @@ -16,7 +16,7 @@ import type { SharedFileMeta } from "@schlessera/brain-ui-sdk/protocol"; * client-environment paths: Both backends replay the user's original text through host-owned message * provenance; current host facts are new tail content. Pi's history reader strips the * image notes pi appends to the stored text (`stripImageNotes`, - * `packages/ui-backend-pi/src/history.ts:229-231`). It is not for a pi `/skill:` or + * `packages/ui-backend-pi/src/history.ts:254-256`). It is not for a pi `/skill:` or * prompt-template command, which pi stores expanded: that message matches * no row and replays as `typed`. Its row is orphaned, and because every * message with the same text expands the same way, the orphan can only