From 8034623ce52c0dc9d030ab3a57f5c591f7ca4966 Mon Sep 17 00:00:00 2001 From: Duy Pham Date: Mon, 2 Nov 2026 15:35:58 +0700 Subject: [PATCH 1/2] fix(conversation): isolate Claude image disk reads --- changelog.d/claude-image-disk-isolation.md | 3 + server/claude-image-file.ts | 17 ++ server/claude-image-isolation.test.ts | 285 +++++++++++++++++++++ server/claude-image-pool.test.ts | 283 ++++++++++++++++++++ server/claude-image-pool.ts | 191 ++++++++++++++ server/claude-image-reader.fake.ts | 32 +++ server/claude-image-worker.fake.ts | 37 +++ server/claude-image-worker.ts | 31 +++ server/conversation.ts | 98 ++++++- server/index.ts | 3 + server/pane-routes.ts | 2 +- 11 files changed, 970 insertions(+), 12 deletions(-) create mode 100644 changelog.d/claude-image-disk-isolation.md create mode 100644 server/claude-image-file.ts create mode 100644 server/claude-image-isolation.test.ts create mode 100644 server/claude-image-pool.test.ts create mode 100644 server/claude-image-pool.ts create mode 100644 server/claude-image-reader.fake.ts create mode 100644 server/claude-image-worker.fake.ts create mode 100644 server/claude-image-worker.ts diff --git a/changelog.d/claude-image-disk-isolation.md b/changelog.d/claude-image-disk-isolation.md new file mode 100644 index 000000000..413984082 --- /dev/null +++ b/changelog.d/claude-image-disk-isolation.md @@ -0,0 +1,3 @@ +### Fixed +- Claude conversation images load on a separate, bounded worker so a slow transcript disk read + leaves terminal traffic, commands, task lists and other file requests responsive. diff --git a/server/claude-image-file.ts b/server/claude-image-file.ts new file mode 100644 index 000000000..ab7e53e2a --- /dev/null +++ b/server/claude-image-file.ts @@ -0,0 +1,17 @@ +import { fstatSync, statSync } from "node:fs"; + +/** A worker's file-generation fence; no transcript bytes or stat calls run on the bridge. */ +export function claudeImageFileGeneration(path: string | number): string | null { + try { + const stat = typeof path === "number" ? fstatSync(path, { bigint: true }) : statSync(path, { bigint: true }); + if (!stat.isFile()) return null; + return [stat.dev, stat.ino, stat.size, stat.mtimeNs, stat.ctimeNs].join(":"); + } catch { return null; } +} + +/** Only the worker supplies this context; the cooperative caller keeps its existing two arguments. */ +export interface ClaudeImageRead { + expected: string; + opened?: Int32Array; + observed?: string; +} diff --git a/server/claude-image-isolation.test.ts b/server/claude-image-isolation.test.ts new file mode 100644 index 000000000..2bccc308a --- /dev/null +++ b/server/claude-image-isolation.test.ts @@ -0,0 +1,285 @@ +import { afterEach, beforeEach, describe, expect, it, spyOn } from "bun:test"; +import * as fs from "node:fs"; +import { appendFileSync, closeSync, constants, fstatSync, mkdtempSync, openSync, renameSync, rmSync, truncateSync, writeFileSync, writeSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import type { HerdrPane, SessionSnapshot } from "../shared/protocol.ts"; +import * as store from "./claude-store.ts"; +import * as pool from "./claude-image-pool.ts"; +import { claudeImageFileGeneration } from "./claude-image-file.ts"; +import { forgetPaneTranscriptState, forgetTranscriptState } from "./conversation.ts"; +import * as herdr from "./herdr/client.ts"; +import { handlePaneRequest } from "./pane-routes.ts"; + +const id = (n: number) => `00000000-0000-4000-8000-${String(n).padStart(12, "0")}`; +const png = Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jRZkAAAAASUVORK5CYII=", "base64"); +const line = (entry: unknown): string => JSON.stringify(entry) + "\n"; +const picture = { type: "image", source: { type: "base64", media_type: "image/png", data: png.toString("base64") } }; +const ref = `${id(3)}:0:0`; +const fake = new URL("./claude-image-reader.fake.ts", import.meta.url); +const waitAsync = (Atomics as unknown as { waitAsync: (cells: Int32Array, index: number, value: number, timeout: number) => { async: boolean; value: Promise | string } }).waitAsync; +let root: string; +let path: string; +let nativePath: string; +let pane: HerdrPane; +const restores: (() => void)[] = []; +const holds: { release(): void; close(): void }[] = []; +function fixture(): string { + return line({ type: "user", uuid: id(1), parentUuid: null, message: { content: "read" } }) + + line({ type: "assistant", uuid: id(2), parentUuid: id(1), message: { content: [{ type: "tool_use", id: "read", name: "Read" }] } }) + + line({ type: "user", uuid: id(3), parentUuid: id(2), message: { content: [{ type: "tool_result", tool_use_id: "read", content: [picture, picture] }] } }); +} +function hold() { + const fifo = join(root, `gate-${holds.length}`); + expect(Bun.spawnSync(["mkfifo", fifo]).exitCode).toBe(0); + const cells = new Int32Array(new SharedArrayBuffer(12)); + cells[1] = -1; cells[2] = -1; + const gate = { + data: { cells: cells.buffer as SharedArrayBuffer, fifo }, + async ran(): Promise { + const wait = waitAsync(cells, 1, -1, 10_000); + if ((wait.async ? await wait.value : wait.value) === "timed-out") throw new Error("the real image reader never entered its held read"); + return cells[1]!; + }, + release() { + if (Atomics.exchange(cells, 0, 1) !== 0 || cells[1] === -1) return; + const fd = openSync(fifo, constants.O_WRONLY | constants.O_NONBLOCK); + try { writeSync(fd, "x"); } finally { closeSync(fd); } + }, + close() { const fd = Atomics.exchange(cells, 2, -1); if (fd >= 0) closeSync(fd); }, + }; + holds.push(gate); + return gate; +} +function heldRoute(gate: ReturnType) { + pool.setClaudeImagePoolOptions({ worker: fake }); + const original = pool.claudeImageTask; + const intercept = spyOn(pool, "claudeImageTask").mockImplementation((job, signal, deadline) => original({ ...job, readGate: gate.data } as typeof job, signal, deadline)); + restores.push(() => intercept.mockRestore()); +} +function request(signal?: AbortSignal): Promise { + const url = new URL(`http://127.0.0.1/api/pane/conversation/image?${new URLSearchParams({ pane_id: pane.pane_id, ref })}`); + return handlePaneRequest(new Request(url, { signal }), url, url.pathname, {}).then((response) => response!); +} +beforeEach(async () => { + pool.setClaudeImagePoolOptions(); + await pool.claudeImageThreadClosed(); + root = mkdtempSync(join(tmpdir(), "ovis-image-isolation-")); + path = join(root, "session.jsonl"); nativePath = path; + writeFileSync(path, fixture()); + pane = { pane_id: "isolated-image-pane", terminal_id: "isolated-terminal", cwd: root, foreground_cwd: root, agent: "claude", agent_session: { agent: "claude", kind: "id", value: id(9), source: "fixture" } } as HerdrPane; + const snapshot = spyOn(herdr, "sessionSnapshot").mockImplementation(async () => ({ panes: [{ ...pane }] }) as SessionSnapshot); + const rpc = spyOn(herdr, "herdrRpc").mockImplementation((async (method: string) => method === "agent.get" ? { agent: { agent_session: { value: pane.agent_session?.value } } } : { process_info: { foreground_processes: [] } }) as typeof herdr.herdrRpc); + const file = spyOn(store, "claudeTranscriptFile").mockImplementation(async () => nativePath); + restores.push(() => snapshot.mockRestore(), () => rpc.mockRestore(), () => file.mockRestore()); +}); +afterEach(async () => { + for (const gate of holds) gate.release(); + pool.setClaudeImagePoolOptions(); + await pool.claudeImageThreadClosed(); + for (const gate of holds.splice(0)) gate.close(); + for (const restore of restores.splice(0).reverse()) restore(); + forgetTranscriptState(); + rmSync(root, { recursive: true, force: true }); +}); + +describe("isolated Claude image reader and route", () => { + it("serves exact bytes through the real worker without bridge transcript reads, retaining image headers", async () => { + const reads = spyOn(fs, "readSync"); + try { + const responses = await Promise.all([request(), request()]); + for (const response of responses) { + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toBe("image/png"); + expect(response.headers.get("cache-control")).toBe("private, max-age=86400, immutable"); + expect(response.headers.get("x-content-type-options")).toBe("nosniff"); + expect(Buffer.from(await response.arrayBuffer()).equals(png)).toBe(true); + } + expect(reads).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); } + }); + + it.skipIf(process.platform === "win32")("imports and runs the worker without a herdr connection", async () => { + const socket = join(root, "no-worker-rpc.sock"); + let connections = 0; + const listener = Bun.listen({ unix: socket, socket: { open(peer) { connections += 1; peer.end(); }, data() {} } }); + const previous = process.env["HERDR_SOCKET"]; + const previousPath = process.env["HERDR_SOCKET_PATH"]; + process.env["HERDR_SOCKET"] = socket; + process.env["HERDR_SOCKET_PATH"] = socket; + try { + const generation = (await pool.claudeImageGeneration(path, "binding"))!; + const image = await pool.claudeImageTask({ path, ref, binding: "binding", generation }); + expect(Buffer.from(image!.bytes).equals(png)).toBe(true); + expect(connections).toBe(0); + } finally { + pool.setClaudeImagePoolOptions(); + await pool.claudeImageThreadClosed(); + listener.stop(true); + if (previous === undefined) delete process.env["HERDR_SOCKET"]; else process.env["HERDR_SOCKET"] = previous; + if (previousPath === undefined) delete process.env["HERDR_SOCKET_PATH"]; else process.env["HERDR_SOCKET_PATH"] = previousPath; + } + }); + + it("transfers an exact owned 8 MiB picture and refuses a picture one byte larger", async () => { + const bytes = Buffer.alloc(8 * 1024 * 1024); + png.copy(bytes); + const encoded = (data: Buffer) => ({ ...picture, source: { ...picture.source, data: data.toString("base64") } }); + writeFileSync(path, line({ type: "user", uuid: id(1), message: { content: [encoded(bytes), encoded(Buffer.concat([bytes, Buffer.from([0])]))] } })); + const generation = (await pool.claudeImageGeneration(path, "binding"))!; + const image = await pool.claudeImageTask({ path, ref: `${id(1)}:0`, binding: "binding", generation }); + expect(image!.bytes.byteOffset).toBe(0); + expect(image!.bytes.buffer.byteLength).toBe(bytes.length); + expect(Buffer.from(image!.bytes).equals(bytes)).toBe(true); + expect(await pool.claudeImageTask({ path, ref: `${id(1)}:1`, binding: "binding", generation })).toBeNull(); + }); + + for (const rejection of ["read", "check"] as const) it(`revokes a warm boundary after rejecting a stale ${rejection} generation`, async () => { + const drawn = line({ type: "user", uuid: id(1), message: { content: [picture] } }); + const clear = line({ type: "user", message: { role: "user", content: "/clear" } }); + const replaceable = line({ type: "progress", data: "x".repeat(clear.length - line({ type: "progress", data: "" }).length) }); + expect(replaceable.length).toBe(clear.length); + const padding = line({ type: "progress", data: "x".repeat(1024 * 1024) }).repeat(3); + writeFileSync(path, drawn + replaceable + padding); + const generation = (await pool.claudeImageGeneration(path, "binding"))!; + const job = { path, ref: `${id(1)}:0`, binding: "binding", generation }; + expect(await pool.claudeImageTask(job)).not.toBeNull(); + writeFileSync(path, drawn + clear + padding); + appendFileSync(path, line({ type: "progress", data: "append" })); + if (rejection === "read") expect(await pool.claudeImageTask(job)).toBeNull(); + else expect(await pool.claudeImageGenerationCurrent(job)).toBe(false); + const latest = (await pool.claudeImageGeneration(path, "binding"))!; + expect(latest).not.toBe(generation); + expect(await pool.claudeImageTask({ ...job, generation: latest })).toBeNull(); + }); + + it("refuses a queued read whose inspected file was rewritten before the scanner opened it", async () => { + const generation = await pool.claudeImageGeneration(path, "binding"); + writeFileSync(path, fixture().replace('"read"', '"gone"')); + expect(await pool.claudeImageTask({ path, ref, binding: "binding", generation: generation! })).toBeNull(); + }); + + for (const mutation of ["replace", "truncate", "rewrite", "clear", "rewind"] as const) it.skipIf(process.platform === "win32")(`rejects a pending worker image after ${mutation}`, async () => { + const gate = hold(); heldRoute(gate); + const reading = request(); + await gate.ran(); + if (mutation === "replace") { const next = join(root, "replacement"); writeFileSync(next, fixture()); renameSync(next, path); } + if (mutation === "truncate") truncateSync(path, 0); + if (mutation === "rewrite") writeFileSync(path, fixture().replaceAll('"read"', '"gone"')); + if (mutation === "clear") appendFileSync(path, line({ type: "system", subtype: "compact_boundary", logicalParentUuid: null }) + line({ type: "user", uuid: id(4), parentUuid: null, message: { content: "new" } })); + if (mutation === "rewind") appendFileSync(path, line({ type: "user", uuid: id(5), parentUuid: id(1), message: { content: "another branch" } })); + gate.release(); + expect((await reading).status).toBe(404); + }); + + it.skipIf(process.platform === "win32")("keeps harmless appended history readable and publishes the generation actually validated", async () => { + const gate = hold(); heldRoute(gate); + const before = claudeImageFileGeneration(path); + const reading = request(); + await gate.ran(); + appendFileSync(path, line({ type: "assistant", uuid: id(4), parentUuid: id(3), message: { content: [{ type: "text", text: "done" }] } })); + gate.release(); + expect(claudeImageFileGeneration(path)).not.toBe(before); + const response = await reading; + expect(response.status).toBe(200); + expect(Buffer.from(await response.arrayBuffer()).equals(png)).toBe(true); + }); + + for (const change of ["session", "source", "cwd", "path", "forget"] as const) it.skipIf(process.platform === "win32")(`revalidates the pane's ${change} binding after a held scan`, async () => { + const gate = hold(); heldRoute(gate); + const reading = request(); + await gate.ran(); + if (change === "session") pane = { ...pane, agent_session: { ...pane.agent_session!, value: id(10) } }; + if (change === "source") pane = { ...pane, agent: "pi" }; + if (change === "cwd") pane = { ...pane, cwd: join(root, "elsewhere") }; + if (change === "path") { nativePath = join(root, "second.jsonl"); writeFileSync(nativePath, fixture()); } + if (change === "forget") forgetPaneTranscriptState(pane.pane_id); + gate.release(); + const response = await reading; + expect(response.status).toBe(404); + expect(await response.json()).toMatchObject({ error: { code: "image_not_found" } }); + }); + + for (const stage of ["snapshot", "resolve"] as const) { + it(`bounds initial ${stage} authorization by deadline and cancellation`, async () => { + pool.setClaudeImagePoolOptions({ deadlineMs: 40 }); + const pending = new Promise(() => {}); + const blocked = stage === "snapshot" ? spyOn(herdr, "sessionSnapshot").mockReturnValue(pending) : spyOn(store, "claudeTranscriptFile").mockReturnValue(pending); + try { + const response = await request(); + expect(response.status).toBe(503); + const cancel = new AbortController(); + const reading = request(cancel.signal); + cancel.abort(); + expect((await reading).status).toBe(503); + expect(pool.claudeImagePoolSize().physical).toBe(0); + } finally { blocked.mockRestore(); } + }); + } + + for (const scope of ["pane", "all"] as const) it(`honors ${scope} transcript invalidation during initial authorization`, async () => { + let release: () => void = () => {}; + let entered: () => void = () => {}; + const waiting = new Promise((done) => { release = done; }); + const starting = new Promise((done) => { entered = done; }); + const file = spyOn(store, "claudeTranscriptFile").mockImplementation(async () => { entered(); await waiting; return path; }); + try { + const reading = request(); + await starting; + if (scope === "pane") forgetPaneTranscriptState(pane.pane_id); else forgetTranscriptState(); + release(); + expect((await reading).status).toBe(404); + expect(pool.claudeImagePoolSize().physical).toBe(0); + } finally { release(); file.mockRestore(); } + }); + + it("checks pane authority again after the final worker fence", async () => { + const original = pool.claudeImageGenerationCurrent; + const check = spyOn(pool, "claudeImageGenerationCurrent").mockImplementation(async (...args) => { + const answer = await original(...args); + pane = { ...pane, agent_session: { ...pane.agent_session!, value: id(10) } }; + return answer; + }); + try { expect((await request()).status).toBe(404); } finally { check.mockRestore(); } + }); + + it.skipIf(process.platform === "win32")("propagates request cancellation and returns the existing retryable busy policy", async () => { + const gate = hold(); heldRoute(gate); + const cancel = new AbortController(); + const reading = request(cancel.signal); + await gate.ran(); + cancel.abort(); + const response = await reading; + expect(response.status).toBe(503); + expect(response.headers.get("retry-after")).toBe("1"); + expect(await response.json()).toMatchObject({ error: { code: "file_busy" } }); + expect(pool.claudeImagePoolSize()).toMatchObject({ physical: 1, running: 1, waiters: 0 }); + gate.release(); + }); + + it.skipIf(process.platform === "win32")("answers a slow worker with retryable 503 inside its whole-request budget", async () => { + const gate = hold(); heldRoute(gate); + pool.setClaudeImagePoolOptions({ worker: fake, deadlineMs: 100 }); + const reading = request(); + await gate.ran(); + const response = await reading; + expect(response.status).toBe(503); + expect(response.headers.get("retry-after")).toBe("1"); + expect(await response.json()).toMatchObject({ error: { code: "file_busy" } }); + gate.release(); + }); + + it.skipIf(process.platform === "win32")("closes a real scanner descriptor only after its terminated read returns", async () => { + const gate = hold(); heldRoute(gate); + const reading = request(); + const fd = await gate.ran(); + expect(fstatSync(fd).isFile()).toBe(true); + pool.setClaudeImagePoolOptions(); + expect((await reading).status).toBe(503); + expect(fstatSync(fd).isFile()).toBe(true); + gate.release(); + await pool.claudeImageThreadClosed(); + expect(() => fstatSync(fd)).toThrow(); + }); +}); diff --git a/server/claude-image-pool.test.ts b/server/claude-image-pool.test.ts new file mode 100644 index 000000000..94b303d23 --- /dev/null +++ b/server/claude-image-pool.test.ts @@ -0,0 +1,283 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import { closeSync, constants, fstatSync, mkdtempSync, openSync, rmSync, writeFileSync, writeSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { claudeImageGeneration, claudeImagePoolSize, claudeImageTask, claudeImageThreadClosed, holdClaudeImagePool, setClaudeImagePoolOptions, type ClaudeImageJob } from "./claude-image-pool.ts"; +import { MAX_IMAGE_BYTES } from "./codex-images.ts"; +import { FileBusyError, fileTask, fileThreadsClosed, holdFilePool } from "./file-pool.ts"; + +const fake = new URL("./claude-image-worker.fake.ts", import.meta.url); +const waitAsync = (Atomics as unknown as { waitAsync: (cells: Int32Array, index: number, value: number, timeout: number) => { async: boolean; value: Promise | string } }).waitAsync; +const roots: string[] = []; +interface HeldRead { data: { cells: SharedArrayBuffer; fifo?: string }; ran(): Promise; release(): void; close(): void } +const gates: HeldRead[] = []; +const temporary = (): string => { const root = mkdtempSync(join(tmpdir(), "ovis-image-pool-")); roots.push(root); return root; }; +const job = (ref = "image", config: object = {}): ClaudeImageJob => ({ path: "/synthetic/history.jsonl", ref, binding: "pane-session", generation: JSON.stringify(config) }); +function gate(fifo?: string): HeldRead { + const cells = new Int32Array(new SharedArrayBuffer(12)); + cells[1] = -1; + cells[2] = -1; + const hold = { + data: { cells: cells.buffer as SharedArrayBuffer, ...(fifo === undefined ? {} : { fifo }) }, + async ran() { + const wait = waitAsync(cells, 1, -1, 10_000); + if ((wait.async ? await wait.value : wait.value) === "timed-out") throw new Error("the image thread never entered its held read"); + return cells[1]!; + }, + release() { + if (Atomics.exchange(cells, 0, 1) !== 0) return; + Atomics.notify(cells, 0); + if (fifo !== undefined && cells[1] !== -1) { + const fd = openSync(fifo, constants.O_WRONLY | constants.O_NONBLOCK); + try { writeSync(fd, "x"); } finally { closeSync(fd); } + } + }, + close() { const fd = Atomics.exchange(cells, 2, -1); if (fd >= 0) closeSync(fd); }, + }; + gates.push(hold); + return hold; +} +const held = (task: ClaudeImageJob, hold: ReturnType): ClaudeImageJob => ({ ...task, gate: hold.data } as ClaudeImageJob); +const configure = (deadlineMs = 10_000) => setClaudeImagePoolOptions({ worker: fake, deadlineMs }); + +afterEach(async () => { + for (const hold of gates) hold.release(); + setClaudeImagePoolOptions(); + await claudeImageThreadClosed(); + for (const hold of gates.splice(0)) hold.close(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +describe("Claude image disk isolation", () => { + it("bounds one physical thread and eight queued jobs, independently of the existing file pool", async () => { + configure(); + const hold = gate(); + const running = claudeImageTask(held(job("running"), hold)); + await hold.ran(); + const queued = Array.from({ length: 8 }, (_, at) => claudeImageTask(job(`queued-${at}`))); + expect(claudeImagePoolSize()).toEqual({ physical: 1, retiring: 0, running: 1, waiting: 8, waiters: 9 }); + await expect(claudeImageTask(job("overflow"))).rejects.toBeInstanceOf(FileBusyError); + const root = temporary(); + const file = join(root, "session.jsonl"); + writeFileSync(file, ""); + const releaseFiles = holdFilePool(); + try { + expect(await fileTask({ op: "info", path: file })).toMatchObject({ size: 0 }); + expect(await fileTask({ op: "commands", agent: "claude", cwd: root, home: root, roots: {}, version: null, serverEnv: {} })).toBeArray(); + expect(await fileTask({ op: "claude-tasks", path: file, live: true, since: null })).toEqual([]); + } finally { releaseFiles(); await fileThreadsClosed(); } + hold.release(); + for (const result of await Promise.all([running, ...queued])) expect([...result!.bytes]).toEqual([17, 17, 17, 17]); + }); + + it("uses worker metadata to identify a generation and refuses an already elapsed absolute deadline", async () => { + configure(); + expect(await claudeImageGeneration("/synthetic/history.jsonl", "pane-session")).toBe("test-file-generation"); + await expect(claudeImageTask(job(), undefined, performance.now() - 1)).rejects.toThrow("in time"); + expect(claudeImagePoolSize()).toMatchObject({ running: 0, waiting: 0, waiters: 0 }); + }); + + it("coalesces only matching bindings and generations, with bounded per-job and total waiters", async () => { + configure(); + const hold = gate(); + const task = held(job(), hold); + const shared = Array.from({ length: 8 }, () => claudeImageTask(task)); + await hold.ran(); + await expect(claudeImageTask(task)).rejects.toBeInstanceOf(FileBusyError); + const other = [ + { ...job(), binding: "another-pane-session" }, + { ...job(), generation: JSON.stringify({ size: 5 }) }, + job("another-ref"), + ].flatMap((different) => Array.from({ length: 8 }, () => claudeImageTask(different))); + expect(claudeImagePoolSize()).toMatchObject({ waiting: 3, waiters: 32 }); + await expect(claudeImageTask(job("total-overflow"))).rejects.toBeInstanceOf(FileBusyError); + hold.release(); + const same = await Promise.all(shared); + for (const result of same) expect(result).toBe(same[0]!); + await Promise.all(other); + expect(claudeImagePoolSize().waiters).toBe(0); + }); + + it("cancels one coalesced waiter without cancelling the others, and drops unclaimed queued jobs", async () => { + configure(); + const hold = gate(); + const cancel = new AbortController(); + const task = held(job(), hold); + const first = claudeImageTask(task, cancel.signal); + const second = claudeImageTask(task); + await hold.ran(); + cancel.abort(); + await expect(first).rejects.toThrow("cancelled"); + expect(claudeImagePoolSize()).toMatchObject({ physical: 1, running: 1, waiters: 1 }); + const queuedCancel = new AbortController(); + const queued = claudeImageTask(job("queued"), queuedCancel.signal); + queuedCancel.abort(); + await expect(queued).rejects.toThrow("cancelled"); + expect(claudeImagePoolSize().waiting).toBe(0); + hold.release(); + expect((await second)!.bytes.length).toBe(4); + await expect(claudeImageTask(job(), AbortSignal.abort())).rejects.toThrow("cancelled"); + }); + + it("includes queue time in deadlines and keeps expired running work attached until its answer", async () => { + configure(150); + const hold = gate(); + const running = claudeImageTask(held(job(), hold)); + const expired = expect(running).rejects.toThrow("in time"); + await hold.ran(); + const queued = expect(claudeImageTask(job("queued"))).rejects.toThrow("in time"); + await Promise.all([expired, queued]); + expect(claudeImagePoolSize()).toEqual({ physical: 1, retiring: 0, running: 1, waiting: 0, waiters: 0 }); + // Repeated retries cannot replace a syscall that still runs. + for (let at = 0; at < 2; at += 1) await expect(claudeImageTask(job(`retry-${at}`))).rejects.toThrow("in time"); + expect(claudeImagePoolSize().physical).toBe(1); + configure(); + const next = claudeImageTask(job("after-retirement")); + hold.release(); + expect((await next)!.bytes.length).toBe(4); + }); + + it.skipIf(process.platform === "win32")("retains its physical slot through a real blocked read, termination, final release and reacquisition", async () => { + configure(); + const root = temporary(); + const fifo = join(root, "held-disk-read"); + expect(Bun.spawnSync(["mkfifo", fifo]).exitCode).toBe(0); + const hold = gate(fifo); + const release = holdClaudeImagePool(); + const file = join(root, "transcript.jsonl"); + writeFileSync(file, "synthetic"); + const reading = claudeImageTask(held({ ...job("opened", { open: true }), path: file }, hold)); + const fd = await hold.ran(); + expect(fstatSync(fd).size).toBe(9); + release(); + release(); + await expect(reading).rejects.toThrow("stopped"); + expect(claudeImagePoolSize()).toMatchObject({ physical: 1, retiring: 1, running: 1 }); + expect(fstatSync(fd).size).toBe(9); + const later = holdClaudeImagePool(); + const next = claudeImageTask(job("new-server")); + expect(claudeImagePoolSize()).toMatchObject({ physical: 1, retiring: 1, waiting: 1 }); + // Reset must preserve the same bound too, even while a later server has work waiting. + configure(); + await expect(next).rejects.toThrow("replaced"); + const replacement = claudeImageTask(job("after-reset")); + expect(claudeImagePoolSize()).toMatchObject({ physical: 1, retiring: 1, waiting: 1 }); + hold.release(); + expect((await replacement)!.bytes.length).toBe(4); + expect(() => fstatSync(fd)).toThrow(); + later(); + await claudeImageThreadClosed(); + expect(claudeImagePoolSize().physical).toBe(0); + }); + + it.skipIf(process.platform === "win32")("lets the last server's process exit while its retired thread is still in a disk syscall", async () => { + const fifo = join(temporary(), "shutdown-read"); + expect(Bun.spawnSync(["mkfifo", fifo]).exitCode).toBe(0); + const source = ` + import { claudeImageTask, holdClaudeImagePool, setClaudeImagePoolOptions } from ${JSON.stringify(new URL("./claude-image-pool.ts", import.meta.url).pathname)}; + setClaudeImagePoolOptions({ worker: new URL(${JSON.stringify(fake.href)}) }); + const release = holdClaudeImagePool(); + const cells = new Int32Array(new SharedArrayBuffer(12)); + cells[1] = -1; cells[2] = -1; + const reading = claudeImageTask({ path: "/synthetic/history", ref: "image", binding: "pane", generation: "{}", gate: { cells: cells.buffer, fifo: ${JSON.stringify(fifo)} } }); + reading.catch(() => {}); + const wait = Atomics.waitAsync(cells, 1, -1, 10000); + if ((wait.async ? await wait.value : wait.value) === "timed-out") throw new Error("the child never reached its disk read"); + release(); + await reading.catch(() => {}); + console.log("last server released"); + `; + const child = Bun.spawn([process.execPath, "--eval", source], { stdout: "pipe", stderr: "pipe", env: { ...process.env, HERDR_TEST_MODE: "unit" } }); + let timer: ReturnType | undefined; + let code: number | null; + try { + code = await Promise.race([child.exited, new Promise((done) => { timer = setTimeout(() => done(null), 3_000); })]); + if (code === null) { + // Release a regressed child too; the test must never leave a blocked worker behind. + const fd = openSync(fifo, constants.O_WRONLY | constants.O_NONBLOCK); + try { writeSync(fd, "x"); } finally { closeSync(fd); } + await child.exited; + } + const stderr = await new Response(child.stderr).text(); + expect(stderr).toBe(""); + expect(await new Response(child.stdout).text()).toContain("last server released"); + expect(code).toBe(0); + } finally { clearTimeout(timer); child.kill(); await child.exited; } + }); + + it.skipIf(process.platform === "win32")("keeps timers, health HTTP and WebSocket messages responsive during a real held disk read", async () => { + configure(); + const fifo = join(temporary(), "held-read"); + expect(Bun.spawnSync(["mkfifo", fifo]).exitCode).toBe(0); + const hold = gate(fifo); + const reading = claudeImageTask(held(job(), hold)); + const server = Bun.serve({ + hostname: "127.0.0.1", port: 0, + fetch(request, server) { + if (new URL(request.url).pathname === "/ws" && server.upgrade(request)) return; + return new Response("healthy"); + }, + websocket: { message(ws, message) { ws.send(message); } }, + }); + let ws: WebSocket | undefined; + try { + await hold.ran(); + let timerRan = false; + await new Promise((done) => setImmediate(() => { timerRan = true; done(); })); + expect(timerRan).toBe(true); + expect(await (await fetch(`http://127.0.0.1:${server.port}/health`)).text()).toBe("healthy"); + const pong = new Promise((done, fail) => { + ws = new WebSocket(`ws://127.0.0.1:${server.port}/ws`); + ws.onopen = () => ws!.send("ping"); + ws.onmessage = (event) => done(String(event.data)); + ws.onerror = () => fail(new Error("the test websocket failed")); + }); + expect(await pong).toBe("ping"); + expect(claudeImagePoolSize().running).toBe(1); + } finally { ws?.close(); server.stop(true); hold.release(); await reading; } + }); + + it("accepts exact owned buffers through the size limit and rejects extra backing bytes", async () => { + configure(); + expect((await claudeImageTask(job("maximum", { size: MAX_IMAGE_BYTES })))!.bytes.buffer.byteLength).toBe(MAX_IMAGE_BYTES); + const oversized = await claudeImageTask(job("oversized", { size: MAX_IMAGE_BYTES + 1 })).catch((error: Error) => error); + expect(oversized).toBeInstanceOf(Error); + expect((oversized as Error).message).toContain("invalid answer"); + await claudeImageThreadClosed(); + const backing = await claudeImageTask(job("pooled-backing", { bad: true })).catch((error: Error) => error); + expect(backing).toBeInstanceOf(Error); + expect((backing as Error).message).toContain("invalid answer"); + await claudeImageThreadClosed(); + expect((await claudeImageTask(job("recovered")))!.bytes.length).toBe(4); + }); + + it("retires malformed image and metadata replies without throwing on the bridge", async () => { + configure(); + for (const name of ["number", "array", "undefined", "reply", "generation"]) { + const path = `/synthetic/malformed-${name}`; + for (const work of [() => claudeImageTask({ ...job(), path }), () => claudeImageGeneration(path, "pane-session")]) { + const error = await work().catch((error: Error) => error); + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain("invalid answer"); + await claudeImageThreadClosed(); + } + } + const malformedGeneration = await claudeImageTask(job("bad-image-generation", { invalidGeneration: true })).catch((error: Error) => error); + expect(malformedGeneration).toBeInstanceOf(Error); + expect((malformedGeneration as Error).message).toContain("invalid answer"); + await claudeImageThreadClosed(); + expect(await claudeImageTask({ ...job(), path: "/synthetic/missing-image" })).toBeNull(); + expect(await claudeImageGeneration("/synthetic/missing-image", "pane-session")).toBeNull(); + }); + + it("recovers from worker errors only after close, and rejects module startup failures", async () => { + configure(); + await expect(claudeImageTask(job("worker-error", { fail: true }))).rejects.toThrow("failed"); + await claudeImageThreadClosed(); + expect((await claudeImageTask(job("recovered")))!.bytes.length).toBe(4); + setClaudeImagePoolOptions({ worker: new URL("./missing-image-worker-for-test.ts", import.meta.url) }); + await expect(claudeImageTask(job("cannot-start"))).rejects.toThrow(); + await claudeImageThreadClosed(); + }); +}); diff --git a/server/claude-image-pool.ts b/server/claude-image-pool.ts new file mode 100644 index 000000000..b9aa5c3d8 --- /dev/null +++ b/server/claude-image-pool.ts @@ -0,0 +1,191 @@ +import { closeSync } from "node:fs"; + +import { MAX_IMAGE_BYTES } from "./codex-images.ts"; +import { FileBusyError } from "./file-pool.ts"; + +/** Only metadata crosses into the image thread; pane authorization stays with the caller. */ +export interface ClaudeImageJob { path: string; ref: string; binding: string; generation: string } +type ClaudeImageWork = ClaudeImageJob & { op?: "inspect" | "check" }; +export interface ClaudeImage { mediaType: string; bytes: Uint8Array; generation: string } +export type ClaudeImageResult = ClaudeImage | { generation: string }; +export interface ClaudeImageRequest { id: number; job: ClaudeImageWork; opened?: Int32Array } +export interface ClaudeImageReply { id: number; result?: ClaudeImageResult | null; error?: string } + +/** Queue time is part of an image request's deadline, just as for long file tasks. */ +export const CLAUDE_IMAGE_DEADLINE_MS = 15_000; +const WAITING = 8; +const JOB_WAITERS = 8; +const TOTAL_WAITERS = 32; +interface PoolOptions { worker: URL; deadlineMs: number } +const DEFAULTS: PoolOptions = { worker: new URL("./claude-image-worker.ts", import.meta.url), deadlineMs: CLAUDE_IMAGE_DEADLINE_MS }; +let options = DEFAULTS; +interface Waiter { done: (image: ClaudeImageResult | null) => void; fail: (error: Error) => void; timer: ReturnType; unlisten?: () => void } +interface Pending { key: string; request: ClaudeImageRequest; waiters: Set } +type BunWorker = Worker & { ref(): void; unref(): void }; +interface Thread { worker: BunWorker; running: Pending | null; retiring: boolean; opened: Int32Array; closed: Promise } +let thread: Thread | null = null; +const queue: Pending[] = []; +const jobs = new Map(); +let waiters = 0; +let sequence = 0; +let holders = 0; + +/** + * One physical image thread in the process, independent of commands, task lists and file routes. + * An expired request cannot stop a disk syscall. Even terminate/error/server shutdown leaves the + * physical slot taken until close; a server started meanwhile cannot grow another image thread. + */ +export function holdClaudeImagePool(): () => void { + holders += 1; + let released = false; + return () => { + if (released) return; + released = true; + holders -= 1; + if (holders === 0) stop("the server stopped"); + }; +} + +/** Test support: the previous physical thread must close before the replacement may start. */ +export function setClaudeImagePoolOptions(next: Partial = {}): void { + options = { ...DEFAULTS, ...next }; + stop("the image thread was replaced"); +} +export function claudeImagePoolSize(): { physical: number; retiring: number; running: number; waiting: number; waiters: number } { + return { physical: thread === null ? 0 : 1, retiring: thread?.retiring ? 1 : 0, running: thread?.running === null || thread === null ? 0 : 1, waiting: queue.length, waiters }; +} +/** The physical thread present now has closed, including one terminated inside a syscall. */ +export async function claudeImageThreadClosed(): Promise { await thread?.closed; } + +/** Coalesce only a caller's explicitly authorized binding and validated file generation. */ +export function claudeImageTask(job: ClaudeImageJob, signal?: AbortSignal, deadline?: number): Promise { + return task(job, signal, deadline) as Promise; +} +/** One absolute budget across authorization, metadata, scan and final binding checks. */ +export function claudeImageDeadline(began = performance.now()): number { return began + options.deadlineMs; } +/** File metadata stays on the same bounded thread too; unknown generations never coalesce. */ +export async function claudeImageGeneration(path: string, binding: string, signal?: AbortSignal, deadline?: number): Promise { + const result = await task({ op: "inspect", path, binding, ref: "", generation: `inspect-${++sequence}` }, signal, deadline); + return result !== null && "generation" in result ? result.generation ?? null : null; +} +/** A final metadata fence after pane reauthorization; a changed file is retried by a later request. */ +export async function claudeImageGenerationCurrent(job: ClaudeImageJob, signal?: AbortSignal, deadline?: number): Promise { + const result = await task({ ...job, op: "check", ref: "" }, signal, deadline); + return result !== null && "generation" in result && result.generation === job.generation; +} +function task(job: ClaudeImageWork, signal?: AbortSignal, deadline = performance.now() + options.deadlineMs): Promise { + if (signal?.aborted) return Promise.reject(new FileBusyError("the request was cancelled")); + if (deadline <= performance.now()) return Promise.reject(new FileBusyError("the image did not answer in time")); + const key = JSON.stringify([job.op ?? "read", job.binding, job.path, job.generation, job.ref]); + let pending = jobs.get(key); + if (waiters >= TOTAL_WAITERS || (pending !== undefined && pending.waiters.size >= JOB_WAITERS)) return Promise.reject(new FileBusyError("too many requests are waiting for an image")); + if (pending === undefined) { + if ((thread !== null && (thread.retiring || thread.running !== null)) && queue.length >= WAITING) return Promise.reject(new FileBusyError("the image thread is busy")); + pending = { key, request: { id: ++sequence, job: { ...job } }, waiters: new Set() }; + jobs.set(key, pending); + queue.push(pending); + } + const work = pending; + return new Promise((done, fail) => { + const waiter: Waiter = { done, fail, timer: setTimeout(() => leave(work, waiter, new FileBusyError("the image did not answer in time")), Math.max(0, deadline - performance.now())) }; + work.waiters.add(waiter); + waiters += 1; + if (signal !== undefined) { + const cancel = () => leave(work, waiter, new FileBusyError("the request was cancelled")); + signal.addEventListener("abort", cancel, { once: true }); + waiter.unlisten = () => signal.removeEventListener("abort", cancel); + } + pump(); + }); +} + +function leave(pending: Pending, waiter: Waiter, outcome: ClaudeImageResult | null | Error): void { + if (!pending.waiters.delete(waiter)) return; + waiters -= 1; + clearTimeout(waiter.timer); + waiter.unlisten?.(); + if (outcome instanceof Error) waiter.fail(outcome); + else waiter.done(outcome); + if (pending.waiters.size === 0) { + jobs.delete(pending.key); + const at = queue.indexOf(pending); + if (at >= 0) queue.splice(at, 1); + // A running job stays attached to its physical thread until its real answer/close. + } +} +function finish(pending: Pending, outcome: ClaudeImageResult | null | Error): void { + for (const waiter of [...pending.waiters]) leave(pending, waiter, outcome); +} +function stop(reason: string): void { + for (const pending of [...queue]) finish(pending, new FileBusyError(reason)); + if (thread !== null) retire(thread, new FileBusyError(reason)); +} +function retire(current: Thread, error: Error): void { + if (current.retiring) return; + current.retiring = true; + if (current.running !== null) finish(current.running, error); + // Do not clear thread/running here: termination can wait for a blocked disk read. + // Bun takes its shutdown keep-alive when termination begins: unref must happen BEFORE it, + // so the last server can exit even if the worker's syscall never returns. + current.worker.unref(); + void current.worker.terminate(); +} +function validReply(value: unknown, job: ClaudeImageWork): value is ClaudeImageReply { + if (value === null || typeof value !== "object" || Array.isArray(value)) return false; + const reply = value as ClaudeImageReply; + if (!Number.isSafeInteger(reply.id)) return false; + if (reply.error !== undefined) return typeof reply.error === "string"; + if (reply.result === null) return true; + const result = reply.result; + if (result === undefined || typeof result !== "object" || Array.isArray(result)) return false; + if (!("generation" in result) || typeof result.generation !== "string" || result.generation.length === 0) return false; + if (job.op !== undefined) return true; + return result !== undefined && "mediaType" in result && typeof result.mediaType === "string" && /^image\/(png|jpeg|gif|webp)$/.test(result.mediaType) && result.bytes instanceof Uint8Array && result.bytes.buffer instanceof ArrayBuffer && result.bytes.byteOffset === 0 && result.bytes.byteLength === result.bytes.buffer.byteLength && result.bytes.byteLength > 0 && result.bytes.byteLength <= MAX_IMAGE_BYTES; +} +function start(): Thread { + const worker = new Worker(options.worker) as BunWorker; + worker.unref(); + let closed = () => {}; + const opened = new Int32Array(new SharedArrayBuffer(4)); + opened[0] = -1; + const current: Thread = { worker, running: null, retiring: false, opened, closed: new Promise((done) => { closed = done; }) }; + thread = current; + worker.addEventListener("message", (event: MessageEvent) => { + if (current.retiring) return; + const pending = current.running; + const reply = event.data; + if (pending === null || !validReply(reply, pending.request.job) || reply.id !== pending.request.id) { + retire(current, new Error("the image thread returned an invalid answer")); + return; + } + current.running = null; + worker.unref(); + finish(pending, reply.error !== undefined ? new Error(reply.error) : reply.result ?? null); + pump(); + }); + worker.addEventListener("error", () => retire(current, new Error("the image thread failed"))); + worker.addEventListener("close", () => { + if (current.running !== null) finish(current.running, new Error("the image thread ended")); + const fd = Atomics.exchange(current.opened, 0, -1); + if (fd >= 0) { try { closeSync(fd); } catch { /* the thread closed it before stopping */ } } + if (thread === current) thread = null; + closed(); + pump(); + }); + return current; +} +function pump(): void { + if (queue.length === 0 || (thread !== null && (thread.retiring || thread.running !== null))) return; + const pending = queue.shift()!; + let current: Thread; + try { current = thread ?? start(); } + catch { + finish(pending, new Error("the image thread could not start")); + pump(); + return; + } + current.running = pending; + current.worker.ref(); + try { current.worker.postMessage({ ...pending.request, opened: current.opened }); } + catch { retire(current, new Error("the image task could not start")); } +} diff --git a/server/claude-image-reader.fake.ts b/server/claude-image-reader.fake.ts new file mode 100644 index 000000000..b25cbc36f --- /dev/null +++ b/server/claude-image-reader.fake.ts @@ -0,0 +1,32 @@ +/** Holds the real scanner inside its first filesystem read, never used by the server. */ +import { spyOn } from "bun:test"; +import * as fs from "node:fs"; +import type { ClaudeImageRequest } from "./claude-image-pool.ts"; +import { readClaudeImage } from "./claude-image-worker.ts"; + +declare const self: Worker; +interface Gate { cells: SharedArrayBuffer; fifo: string } +self.onmessage = async (event: MessageEvent) => { + const request = event.data; + const gate = (request.job as typeof request.job & { readGate?: Gate }).readGate; + const original = fs.readSync; + let held = false; + const spy = gate === undefined ? null : spyOn(fs, "readSync").mockImplementation(((fd: number, buffer: NodeJS.ArrayBufferView, offset: number, length: number, position: number | null) => { + if (!held) { + held = true; + const cells = new Int32Array(gate.cells); + const fifo = fs.openSync(gate.fifo, fs.constants.O_RDWR); + Atomics.store(cells, 2, fifo); + Atomics.store(cells, 1, fd); + Atomics.notify(cells, 1); + try { original(fifo, new Uint8Array(1), 0, 1, null); } + finally { if (Atomics.exchange(cells, 2, -1) >= 0) fs.closeSync(fifo); } + } + return original(fd, buffer, offset, length, position); + }) as typeof fs.readSync); + try { + const result = await readClaudeImage(request); + postMessage({ id: request.id, result }, result !== null && "bytes" in result ? [result.bytes.buffer] : []); + } catch (error) { postMessage({ id: request.id, error: error instanceof Error ? error.message : String(error) }); } + finally { spy?.mockRestore(); } +}; diff --git a/server/claude-image-worker.fake.ts b/server/claude-image-worker.fake.ts new file mode 100644 index 000000000..7396956c7 --- /dev/null +++ b/server/claude-image-worker.fake.ts @@ -0,0 +1,37 @@ +/** Real thread test harness: a held read, an Atomics gate, and deliberately invalid replies. */ +import { closeSync, constants, openSync, readSync } from "node:fs"; +import type { ClaudeImageRequest } from "./claude-image-pool.ts"; + +declare const self: Worker; +interface Gate { cells: SharedArrayBuffer; fifo?: string } +self.onmessage = (event: MessageEvent) => { + const request = event.data; + const malformed = request.job.path.split("/").at(-1); + if (malformed?.startsWith("malformed-")) { + if (malformed === "malformed-reply") postMessage(null); + else postMessage({ id: request.id, result: malformed === "malformed-number" ? 1 : malformed === "malformed-array" ? [] : malformed === "malformed-generation" ? { generation: 1 } : undefined }); + return; + } + if (malformed === "missing-image") { postMessage({ id: request.id, result: null }); return; } + if (request.job.op !== undefined) { postMessage({ id: request.id, result: { generation: "test-file-generation" } }); return; } + const task = JSON.parse(request.job.generation) as { bad?: boolean; fail?: boolean; size?: number; open?: boolean; invalidGeneration?: boolean }; + const opened = task.open ? openSync(request.job.path, "r") : -1; + if (request.opened !== undefined) Atomics.store(request.opened, 0, opened); + const held = (request.job as typeof request.job & { gate?: Gate }).gate; + if (held !== undefined) { + const cells = new Int32Array(held.cells); + const fd = held.fifo === undefined ? -1 : openSync(held.fifo, constants.O_RDWR); + Atomics.store(cells, 2, fd); + Atomics.store(cells, 1, opened >= 0 ? opened : 1); + Atomics.notify(cells, 1); + if (fd >= 0) { + try { readSync(fd, new Uint8Array(1), 0, 1, null); } + finally { if (Atomics.exchange(cells, 2, -1) >= 0) closeSync(fd); } + } else Atomics.wait(cells, 0, 0); + } + if (opened >= 0 && (request.opened === undefined || Atomics.compareExchange(request.opened, 0, opened, -1) === opened)) closeSync(opened); + if (task.fail) throw new Error("test failure"); + const bytes = new Uint8Array(task.size ?? 4).fill(17); + const result = { mediaType: "image/png", generation: task.invalidGeneration ? 1 : request.job.generation, bytes: task.bad ? bytes.subarray(1) : bytes }; + postMessage({ id: request.id, result }, [bytes.buffer]); +}; diff --git a/server/claude-image-worker.ts b/server/claude-image-worker.ts new file mode 100644 index 000000000..7497e382d --- /dev/null +++ b/server/claude-image-worker.ts @@ -0,0 +1,31 @@ +/** One image-only thread: no pane discovery, herdr calls or authentication runs here. */ +import { claudeImageFileGeneration, type ClaudeImageRead } from "./claude-image-file.ts"; +import type { ClaudeImageRequest, ClaudeImageResult } from "./claude-image-pool.ts"; +import { forgetTranscriptImage, transcriptImageAsync } from "./conversation.ts"; + +declare const self: Worker; + +/** The explicit-path reader is safe to import: its stores and RPC clients start only when called. */ +export async function readClaudeImage(request: ClaudeImageRequest): Promise { + const { job, opened } = request; + if (job.op !== undefined) { + const generation = claudeImageFileGeneration(job.path); + if (generation !== null && (job.op === "inspect" || generation === job.generation)) return { generation }; + if (job.op === "check") forgetTranscriptImage(job.path); + return null; + } + const read: ClaudeImageRead = { expected: job.generation, opened }; + // Keep the bounded boundary/revision caches in this thread: gallery images reuse their history. + const image = await transcriptImageAsync(job.path, job.ref, read); + return image !== null && read.observed !== undefined ? { ...image, generation: read.observed } : null; +} + +self.onmessage = async (event: MessageEvent) => { + const request = event.data; + try { + const result = await readClaudeImage(request); + postMessage({ id: request.id, result }, result !== null && "bytes" in result ? [result.bytes.buffer] : []); + } catch (error) { + postMessage({ id: request.id, error: error instanceof Error ? error.message : String(error) }); + } +}; diff --git a/server/conversation.ts b/server/conversation.ts index 7af305447..a4a89c9e1 100644 --- a/server/conversation.ts +++ b/server/conversation.ts @@ -34,6 +34,10 @@ * integration and lives in paneTranscript. */ +import { claudeImageFileGeneration, type ClaudeImageRead } from "./claude-image-file.ts"; +import { claudeImageDeadline, claudeImageGeneration, claudeImageGenerationCurrent, claudeImageTask } from "./claude-image-pool.ts"; +import { FileBusyError } from "./file-pool.ts"; + import { createHash, randomUUID } from "node:crypto"; import { closeSync, fstatSync, openSync, readFileSync, readSync, realpathSync, statSync } from "node:fs"; import { readStatuslineContext } from "./statusline.ts"; @@ -2153,16 +2157,23 @@ function transcriptPageRead(source: StreamSource, path: string, page: Conversati * picture a tool returned by its call (`codex-tool::`); omp names its pictures * as pi does. */ -export async function conversationImage(paneId: string, ref: string, codexHome?: string, opencodeDb?: string, grokHome?: string): Promise<{ mediaType: string; bytes: Uint8Array } | null> { +export async function conversationImage(paneId: string, ref: string, codexHome?: string, opencodeDb?: string, grokHome?: string, signal?: AbortSignal): Promise<{ mediaType: string; bytes: Uint8Array } | null> { + const deadline = claudeImageDeadline(); if (!CLAUDE_IMAGE_REF.test(ref) && !CODEX_IMAGE_REF.test(ref) && !CODEX_TOOL_IMAGE_REF.test(ref) && !PI_IMAGE_REF.test(ref) && !PI_PROMPT_IMAGE_REF.test(ref) && !OPENCODE_IMAGE_REF.test(ref) && !GROK_IMAGE_REF.test(ref) && !COPILOT_IMAGE_REF.test(ref)) return null; - const snapshot = await sessionSnapshot(); + // Claim the existing bounded pane state before authorization yields: a close/clear during + // discovery must invalidate this request too, rather than being recreated by its continuation. + const imageReads = CLAUDE_IMAGE_REF.test(ref) ? paneReads.get(paneId) ?? emptyPaneReads() : undefined; + if (imageReads !== undefined) remember(paneReads, paneId, imageReads, PANE_READS_MAX); + const authorize = (read: () => Promise): Promise => CLAUDE_IMAGE_REF.test(ref) ? imageBindingWait(read, signal, deadline) : read(); + const snapshot = await authorize(sessionSnapshot); const pane = snapshot.panes.find((candidate) => candidate.pane_id === paneId); if (pane === undefined || typeof pane.cwd !== "string" || pane.cwd.length === 0) return null; let resolved: ResolvedTranscript; - try { resolved = await resolveTranscript(pane, pane.cwd, codexHome, snapshot.panes, opencodeDb, undefined, grokHome); } + try { resolved = await authorize(() => resolveTranscript(pane, pane.cwd!, codexHome, snapshot.panes, opencodeDb, undefined, grokHome)); } catch (error) { if (error instanceof ConversationUnavailable) return null; throw error; } // OpenCode's store is read per request and keeps nothing for the pane to release if (resolved.source === "opencode-transcript") return opencodeImage(resolved.path, resolved.session, ref); + if (imageReads !== undefined && paneReads.get(paneId) !== imageReads) return null; rememberPaneRead(paneId, resolved.path); if (resolved.source === "codex-transcript") { if (CODEX_TOOL_IMAGE_REF.test(ref)) return codexToolImage(resolved.path, ref, resolved.codexHome ?? codexHome ?? defaultCodexHome()); @@ -2176,7 +2187,56 @@ export async function conversationImage(paneId: string, ref: string, codexHome?: // the store is held to the identity it had where the transcript was authorized (ompAuthorizedTranscript): // a path above it swapped since cannot move the read elsewhere if (resolved.source === "omp-transcript") return ompTranscriptImage(resolved.path, ref, resolved.ompBlobs ?? null); - return resolved.source === "claude-transcript" ? transcriptImageAsync(resolved.path, ref) : null; + if (resolved.source !== "claude-transcript" || !CLAUDE_IMAGE_REF.test(ref)) return null; + const reads = imageReads; + const paneBinding = (candidate: HerdrPane): string => JSON.stringify([candidate.pane_id, candidate.terminal_id, candidate.cwd, candidate.foreground_cwd, candidate.agent, candidate.agent_session]); + const binding = JSON.stringify([paneBinding(pane), resolved.source, resolved.path]); + const active = (): boolean => { + if (signal?.aborted) throw new FileBusyError("the request was cancelled"); + if (performance.now() >= deadline) throw new FileBusyError("the image did not answer in time"); + return reads !== undefined && paneReads.get(paneId) === reads; + }; + const current = async (): Promise => { + if (!active()) return false; + const latest = await imageBindingWait(sessionSnapshot, signal, deadline); + const candidate = latest.panes.find((entry) => entry.pane_id === paneId); + if (candidate === undefined || paneBinding(candidate) !== paneBinding(pane) || !active()) return false; + try { + const now = await imageBindingWait(() => resolveTranscript(candidate, candidate.cwd!, codexHome, latest.panes, opencodeDb, undefined, grokHome), signal, deadline); + return active() && now.source === "claude-transcript" && now.path === resolved.path; + } catch (error) { if (error instanceof ConversationUnavailable) return false; throw error; } + }; + if (!active()) return null; + const generation = await claudeImageGeneration(resolved.path, binding, signal, deadline); + if (generation === null || !await current()) return null; + const job = { path: resolved.path, ref, binding, generation }; + const image = await claudeImageTask(job, signal, deadline); + if (image === null || !await current()) return null; + if (!await claudeImageGenerationCurrent({ ...job, generation: image.generation }, signal, deadline) || !await current()) return null; + return { mediaType: image.mediaType, bytes: image.bytes }; +} + +/** Reauthorizing an image cannot outlive the same deadline or a client that has left. */ +function imageBindingWait(read: () => Promise, signal: AbortSignal | undefined, deadline: number): Promise { + return new Promise((done, fail) => { + let settled = false; + const finish = (result: T | Error, failed = false) => { + if (settled) return; + settled = true; + clearTimeout(timer); + signal?.removeEventListener("abort", cancel); + if (failed) fail(result); else done(result as T); + }; + const cancel = () => finish(new FileBusyError("the request was cancelled"), true); + const timer = setTimeout(() => finish(new FileBusyError("the image did not answer in time"), true), Math.max(0, deadline - performance.now())); + signal?.addEventListener("abort", cancel, { once: true }); + if (signal?.aborted) cancel(); + else if (performance.now() >= deadline) finish(new FileBusyError("the image did not answer in time"), true); + else { + try { read().then((result) => finish(result), (error: unknown) => finish(error instanceof Error ? error : new Error(String(error)), true)); } + catch (error) { finish(error instanceof Error ? error : new Error(String(error)), true); } + } + }); } /** How much of a transcript the image endpoint reads at a time while it looks for a ref's entry. */ @@ -2440,19 +2500,27 @@ export function transcriptImage(path: string, ref: string): TranscriptPicture | try { return finish(transcriptImageSteps(path, ref, false)); } catch { return null; } } +/** A worker rejected this file generation; its boundary must not seed a later image lookup. */ +export function forgetTranscriptImage(path: string): void { + clearScans.delete(path); + transcriptRevisions.delete(path); +} + /** Yield between windows and short records; individual synchronous disk reads can still block. */ -export async function transcriptImageAsync(path: string, ref: string): Promise { - try { return await finishAsync(transcriptImageSteps(path, ref, true)); } catch { return null; } +export async function transcriptImageAsync(path: string, ref: string, read?: ClaudeImageRead): Promise { + if (read !== undefined) delete read.observed; + try { return await finishAsync(transcriptImageSteps(path, ref, true, read)); } catch { return null; } } function sameImageFile(left: ReturnType, right: ReturnType): boolean { return left.dev === right.dev && left.ino === right.ino && left.size === right.size && left.mtimeMs === right.mtimeMs && left.ctimeMs === right.ctimeMs; } -function* transcriptImageSteps(path: string, ref: string, apart: boolean): Steps { +function* transcriptImageSteps(path: string, ref: string, apart: boolean, read?: ClaudeImageRead): Steps { const match = CLAUDE_IMAGE_REF.exec(ref); if (match === null) return null; const fd = openSync(path, "r"); + if (read?.opened !== undefined) Atomics.store(read.opened, 0, fd); const boundary = clearScans.get(path); const boundaryScanned = boundary?.scanned; const scans: TranscriptStream[] = []; @@ -2464,8 +2532,16 @@ function* transcriptImageSteps(path: string, ref: string, apart: boolean): Steps }; try { const stat = fstatSync(fd); + if (read !== undefined && claudeImageFileGeneration(fd) !== read.expected) return unpublish(...scans); const current = (snapshot: typeof stat): boolean => sameImageFile(snapshot, fstatSync(fd)) && sameImageFile(snapshot, statSync(path)); - if (!current(stat)) return null; + if (!current(stat)) return unpublish(...scans); + const accept = (picture: TranscriptPicture | null, snapshot: typeof stat): TranscriptPicture | null => { + if (picture === null || read === undefined) return picture; + const observed = claudeImageFileGeneration(fd); + if (observed === null || observed !== claudeImageFileGeneration(path) || !current(snapshot)) return unpublish(...scans); + read.observed = observed; + return picture; + }; const stream = transcriptStream("claude-transcript", path, stat, defaultCodexHome(), undefined, fd); scans.push(stream); stream.current = () => current(stat); @@ -2475,7 +2551,7 @@ function* transcriptImageSteps(path: string, ref: string, apart: boolean): Steps yield* applyHistoryBoundarySteps(path, stream, "claude-transcript", apart); const picture = yield* claudePictureSteps(stream, match); // the file held still for the whole read (its size, times and identity, by the pinned fd and by path) - if (!apart || current(stat)) return picture; + if (!apart || current(stat)) return accept(picture, stat); // a boundary kept from before the change is checked only by its tail, which a rewrite can keep if (picture === null) return unpublish(stream); // It changed under the read: only an append can keep the picture, and what was read may since be gone @@ -2499,9 +2575,9 @@ function* transcriptImageSteps(path: string, ref: string, apart: boolean): Steps yield* applyHistoryBoundarySteps(path, active, "claude-transcript", true, true); const again = yield* claudePictureSteps(active, match); // a further change during this read is retried by a later request - return again !== null && current(now) ? again : unpublish(stream, active); + return again !== null && current(now) ? accept(again, now) : unpublish(stream, active); } catch { return unpublish(...scans); } - finally { closeSync(fd); } + finally { if (read?.opened === undefined || Atomics.compareExchange(read.opened, 0, fd, -1) === fd) closeSync(fd); } } function* claudePictureSteps(stream: TranscriptStream, [, uuid, index, nested]: RegExpExecArray): Steps { diff --git a/server/index.ts b/server/index.ts index f7f65f7c5..d93ade1e0 100644 --- a/server/index.ts +++ b/server/index.ts @@ -42,6 +42,7 @@ import { PiSessionReader } from "./pi-session-metadata.ts"; import { HerdrSidebarConfig, herdrConfigPath, snapshotWithoutHiddenTokens } from "./herdr-sidebar.ts"; import { closeCodexDaemonSources } from "./codex-daemon.ts"; import { holdOpenCodeNative } from "./opencode-native.ts"; +import { holdClaudeImagePool } from "./claude-image-pool.ts"; import { holdFilePool } from "./file-pool.ts"; import { handleRequest, type RequestServices, type ServerContext } from "./dispatch.ts"; @@ -348,6 +349,7 @@ export function createServer( const releaseOpenCodeNative = holdOpenCodeNative(); // the file threads are shared by every server in this process too: the last one to stop ends them const releaseFilePool = holdFilePool(); + const releaseClaudeImagePool = holdClaudeImagePool(); core.start(); return { @@ -384,6 +386,7 @@ export function createServer( agentDetector.close(); server.stop(true); releaseFilePool(); + releaseClaudeImagePool(); requestLog.close(); }, }; diff --git a/server/pane-routes.ts b/server/pane-routes.ts index 4fb00d1cb..e08127518 100644 --- a/server/pane-routes.ts +++ b/server/pane-routes.ts @@ -253,7 +253,7 @@ export async function handlePaneRequest(request: Request, url: URL, pathname: st const ref = url.searchParams.get("ref"); if (!paneId || !ref) return badRequest("missing_parameter", "pane_id and ref query parameters are required"); try { - const image = await conversationImage(paneId, ref, options.codexHome, options.opencodeDb, options.grokHome); + const image = await conversationImage(paneId, ref, options.codexHome, options.opencodeDb, options.grokHome, request.signal); if (image === null) return jsonResponse({ error: { code: "image_not_found", message: "no such image in this pane's conversation" } }, 404); // Claude and pi embed immutable bytes, and omp, Grok and Copilot name a picture by its digest; Codex files and // OpenCode tool-image ordinals can change, and a Codex tool's picture (`codex-tool:`) loads with its row. From 645b38c2cbaad8d52275d3fa1ab51ff54a7b2b3e Mon Sep 17 00:00:00 2001 From: Duy Pham Date: Mon, 2 Nov 2026 15:55:49 +0700 Subject: [PATCH 2/2] test(conversation): check descriptor ownership across worker retirement --- server/claude-image-pool.test.ts | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/server/claude-image-pool.test.ts b/server/claude-image-pool.test.ts index 94b303d23..3e001856c 100644 --- a/server/claude-image-pool.test.ts +++ b/server/claude-image-pool.test.ts @@ -149,12 +149,17 @@ describe("Claude image disk isolation", () => { writeFileSync(file, "synthetic"); const reading = claudeImageTask(held({ ...job("opened", { open: true }), path: file }, hold)); const fd = await hold.ran(); - expect(fstatSync(fd).size).toBe(9); + const original = fstatSync(fd); + expect(original.size).toBe(9); + const ownsOriginal = (): boolean => { + try { const now = fstatSync(fd); return now.dev === original.dev && now.ino === original.ino; } + catch { return false; } + }; release(); release(); await expect(reading).rejects.toThrow("stopped"); expect(claudeImagePoolSize()).toMatchObject({ physical: 1, retiring: 1, running: 1 }); - expect(fstatSync(fd).size).toBe(9); + expect(fstatSync(fd)).toMatchObject({ dev: original.dev, ino: original.ino, size: 9 }); const later = holdClaudeImagePool(); const next = claudeImageTask(job("new-server")); expect(claudeImagePoolSize()).toMatchObject({ physical: 1, retiring: 1, waiting: 1 }); @@ -165,7 +170,8 @@ describe("Claude image disk isolation", () => { expect(claudeImagePoolSize()).toMatchObject({ physical: 1, retiring: 1, waiting: 1 }); hold.release(); expect((await replacement)!.bytes.length).toBe(4); - expect(() => fstatSync(fd)).toThrow(); + // The replacement thread may reuse the number after close; it must no longer own this file. + expect(ownsOriginal()).toBe(false); later(); await claudeImageThreadClosed(); expect(claudeImagePoolSize().physical).toBe(0);