diff --git a/apps/genbi/scripts/installed-package-acceptance.mjs b/apps/genbi/scripts/installed-package-acceptance.mjs index fc0ce6c9f..147276ee2 100644 --- a/apps/genbi/scripts/installed-package-acceptance.mjs +++ b/apps/genbi/scripts/installed-package-acceptance.mjs @@ -466,6 +466,7 @@ async function verifyInstalledCodexBackend(input) { import assert from 'node:assert/strict'; import { existsSync } from 'node:fs'; import { CodexAppServerBackend } from ${JSON.stringify(moduleUrl("codex-app-server"))}; + import { CodexConversation } from ${JSON.stringify(moduleUrl("codex-conversation"))}; import { CODEX_CERTIFIED_ROWS } from ${JSON.stringify(moduleUrl("codex-compatibility"))}; import { CodexSession } from ${JSON.stringify(moduleUrl("codex-session"))}; assert.equal(CODEX_CERTIFIED_ROWS.length, 0); @@ -474,6 +475,29 @@ async function verifyInstalledCodexBackend(input) { assert.notEqual(readiness.state, 'ready'); assert.throws(() => backend.prepareLaunch()); assert.equal(existsSync(${JSON.stringify(runtimeRoot)}), false); + let bridgeClosed = 0; + const bridgeFrames = []; + const bridge = new CodexConversation({ open: async (_permit, input) => ({ + startThread: async () => 'fixture-thread', + runTurn: async (text) => { + assert.equal(text, 'fixture prompt'); + const turn = { id: 'fixture-turn', status: 'completed', items: [] }; + input.onEvent({ method: 'turn/completed', params: { threadId: 'fixture-thread', turn } }); + return turn; + }, + interruptTurn: async () => {}, + onFailure: () => () => {}, + close: async () => { bridgeClosed++; }, + }) }, { assertActive() {}, release() {}, runtime: {} }, { spec: {}, wrenHome: {}, assertScopeActive() {} }); + await bridge.ready; + const attachment = bridge.attach(bridge.capability, frame => bridgeFrames.push(frame)); + assert.deepEqual(await attachment.submit('fixture prompt'), { turnId: 'fixture-turn', status: 'completed' }); + assert.equal(bridgeFrames[0].type, 'replay'); + assert.equal(bridgeFrames.some(frame => frame.type === 'event'), true); + attachment.detach(); + assert.throws(() => attachment.submit('stale')); + await bridge.close(); + assert.equal(bridgeClosed, 1); const messages = []; let handlers; let closes = 0; const events = []; const transport = { listen(value) { handlers = value; }, @@ -514,7 +538,7 @@ function assertPublishedFiles(files) { const forbidden = /(^|\/)(scripts|test|tests|fixtures|examples|\.git)(\/|$)|(^|\/)node_modules(\/|$)/; const unexpected = files.filter((file) => forbidden.test(file)); if (unexpected.length > 0) throw new Error(`package tarball contains repository-only files: ${JSON.stringify(unexpected)}`); - for (const required of ["package/bin/genbi.mjs", "package/dist/index.html", "package/dist-server/server/bin.js", "package/managed-wren/manifest.json", "package/dist-server/server/runtime-host/codex-app-server.js", "package/dist-server/server/runtime-host/codex-session.js"]) { + for (const required of ["package/bin/genbi.mjs", "package/dist/index.html", "package/dist-server/server/bin.js", "package/managed-wren/manifest.json", "package/dist-server/server/runtime-host/codex-app-server.js", "package/dist-server/server/runtime-host/codex-session.js", "package/dist-server/server/runtime-host/codex-conversation.js"]) { if (!files.includes(required)) throw new Error(`package tarball is missing ${required}`); } } diff --git a/apps/genbi/server/runtime-host/CODEX.md b/apps/genbi/server/runtime-host/CODEX.md index 86555cd44..ab7ccd922 100644 --- a/apps/genbi/server/runtime-host/CODEX.md +++ b/apps/genbi/server/runtime-host/CODEX.md @@ -104,6 +104,45 @@ integration and browser event UI remain later composition work. ## Direct events versus command/PTY +`CodexConversation` is the process-local Sessions bridge above this driver. +The host passes the existing backend, a previously obtained single-use permit, +and its materialized scope/Wren-home inputs. Await `ready` before presenting an +interactive conversation; construction begins an asynchronous open and one +ephemeral thread, never a turn. The host must guard durable writes and scope +materialization with the permit before constructing the bridge. + +An attachment requires the host-generated capability and grants only prompt, +interrupt and detach operations. One attachment controls a conversation at a +time; detached handles lose authority even when another attachment reconnects. +Detach does not cancel an already accepted turn. The existing Sessions owner +must enforce its initial-attachment/detach lease and invoke/await `close()`. +Do not persist or log capabilities, backend inputs, or driver objects. + +Replay starts with explicit cursor/truncation/state metadata, followed by +strictly sequenced event/state frames. Retention is process-local, limited to +256 frames and 1 MiB of serialized UTF-8; it is not durable history or a complete +snapshot. Consumers must show truncation and must not assume retained deltas +contain the start of an item. Listener payloads are isolated copies. Listener +failure closes the owned backend; browser transport backpressure must be +bounded separately by the API/WebSocket owner. Vendor transcript content remains +untrusted user-facing data, never diagnostic text. + +Only one turn may run at once. Interrupt waits for turn completion, not merely +the interrupt acknowledgement. Before a turn ID exists, cancellation closes the +connection instead of guessing an ID. Close also settles the bridge's pending +turn if the underlying driver has not yet settled it; cleanup failure is still +reported and generation retention remains the backend's responsibility. +Idle transport failure is observed through `CodexSession.onFailure()`. +Command/PTY events are rejected by this bridge, not rendered as conversation +text. After close, a capability can read the retained tail only if the enclosing +Sessions authorization still permits it; revocation and durable lifecycle state +belong to that owner. + +This bridge is not registered in the current application composition. Scoped +producer/Wren/MCP materialization, durable Sessions/API/UI wiring, and structured +Ask/Setup injection still need their own acceptance before activation. In +particular, it does not enable MCP or bypass the project `.codex` rejection. + `startThread()` creates one ephemeral thread. `runTurn(text, { timeoutMs, signal })` resolves on `turn/completed`, **not** the start acknowledgement. Turn statuses are `completed`, `interrupted`, or `failed`; raw vendor error details are omitted. diff --git a/apps/genbi/server/runtime-host/codex-conversation.ts b/apps/genbi/server/runtime-host/codex-conversation.ts new file mode 100644 index 000000000..ae9a56019 --- /dev/null +++ b/apps/genbi/server/runtime-host/codex-conversation.ts @@ -0,0 +1,256 @@ +import { randomUUID, timingSafeEqual } from "node:crypto"; +import { CodexBackendError, type CodexAppServerBackend, type CodexLaunchPermit } from "./codex-app-server.js"; +import { parseCodexEvent, type CodexEvent } from "./codex-events.js"; +import { CodexRpcError } from "./codex-rpc.js"; +import type { CodexSession } from "./codex-session.js"; +import type { RuntimeBackendReasonCode } from "./types.js"; + +export type ConversationState = "opening" | "ready" | "running" | "closing" | "closed" | "failed"; +export type ConversationFailure = "unavailable" | "protocol" | "timeout" | "cancelled" | "cleanup" | "delivery"; +export class CodexConversationError extends Error { + constructor(readonly code: ConversationFailure | "attachment" | "busy" | "input" | "closed", + readonly backendReason?: RuntimeBackendReasonCode<"codex-app-server">) { + super("Codex conversation: " + code); + } +} +type ConversationEvent = Exclude; +export type ConversationFrame = + | { readonly type: "event"; readonly sequence: number; readonly event: ConversationEvent } + | { readonly type: "state"; readonly sequence: number; readonly state: ConversationState; readonly failure?: ConversationFailure; readonly backendReason?: RuntimeBackendReasonCode<"codex-app-server"> }; +export interface ConversationReplay { + readonly type: "replay"; + readonly afterSequence: number; + readonly throughSequence: number; + readonly truncated: boolean; + readonly retainedBytes: number; + readonly retentionLimitBytes: number; + readonly state: ConversationState; + readonly failure?: ConversationFailure; + readonly backendReason?: RuntimeBackendReasonCode<"codex-app-server">; +} +export interface CodexConversationAttachment { + submit(text: string): Promise<{ readonly turnId: string; readonly status: "completed" | "interrupted" | "failed" }>; + interrupt(): Promise; + detach(): void; +} +type Driver = Pick; +type OpenInput = Omit[1], "onEvent">; +interface Backend { open(permit: CodexLaunchPermit, input: OpenInput & { onEvent(event: CodexEvent): void }): Promise } +interface Attachment { + readonly listener: (frame: ConversationFrame | ConversationReplay) => void; + readonly queue: string[]; + queuedBytes: number; + draining: boolean; +} +const MAX_BYTES = 1_048_576; +const MAX_FRAMES = 256; +function failure(error: unknown): ConversationFailure { + if (error instanceof CodexConversationError && ["protocol", "timeout", "cancelled", "cleanup", "delivery"].includes(error.code)) return error.code as ConversationFailure; + if (error instanceof CodexBackendError && error.code === "codex_app_server_cleanup_failed") return "cleanup"; + if (error instanceof CodexRpcError) { + if (error.reason === "cleanup" || error.reason === "timeout" || error.reason === "cancelled") return error.reason; + return "protocol"; + } + return "unavailable"; +} + +/** + * Process-local Phase 5 bridge, not an application activation path. + * The host obtains the permit before durable writes and owns scope materialization. + * Browser code receives only attachment methods and projected frames, never a driver. + */ +export class CodexConversation { + readonly capability = randomUUID(); + readonly ready: Promise; + private state: ConversationState = "opening"; + private failed?: ConversationFailure; + private backendReason: RuntimeBackendReasonCode<"codex-app-server"> | undefined; + private sequence = 0; + private retainedBytes = 0; + private frames: { sequence: number; json: string; bytes: number }[] = []; + private owner: Attachment | undefined; + private driver?: Driver; + private readonly connection: Promise; + private closing?: Promise; + private detachFailure?: () => void; + private active: Promise<{ turnId: string; status: "completed" | "interrupted" | "failed" }> | undefined; + private interrupting: Promise | undefined; + private turnStarted = false; + private rejectStopped!: (error: CodexConversationError) => void; + private readonly stopped = new Promise((_, reject) => { this.rejectStopped = reject; }); + + constructor(backend: Backend, permit: CodexLaunchPermit, input: OpenInput) { + void this.stopped.catch(() => {}); + // Defer open until ownership of both promises exists; synchronous fake peers + // and early events must obey the same lifecycle as the production backend. + this.connection = Promise.resolve().then(() => { + if (this.state !== "opening") throw new CodexConversationError("closed"); + permit.assertActive(); + return backend.open(permit, { ...input, onEvent: (event) => this.receive(event) }); + }).then((driver) => { + this.driver = driver; + this.detachFailure = driver.onFailure((error) => { + if (this.state !== "closing" && this.state !== "closed" && this.state !== "failed") { + void this.finish(failure(error)).catch(() => {}); + } + }); + return driver; + }).catch((error: unknown) => { + if (error instanceof CodexBackendError) this.backendReason = error.code; + permit.release(); throw error; + }); + this.ready = this.initialize(); + void this.ready.catch(() => {}); + } + snapshot(): { state: ConversationState; failure?: ConversationFailure; backendReason?: RuntimeBackendReasonCode<"codex-app-server">; sequence: number } { + return { state: this.state, ...(this.failed ? { failure: this.failed } : {}), + ...(this.backendReason ? { backendReason: this.backendReason } : {}), sequence: this.sequence }; + } + private async initialize(): Promise { + try { + const driver = await this.connection; + if (this.state !== "opening") throw new CodexConversationError("closed"); + await driver.startThread(); + if (this.state !== "opening") throw new CodexConversationError("closed"); + this.transition("ready"); + if (this.snapshot().state !== "ready") throw new CodexConversationError("closed"); + } catch (error) { + await this.finish(failure(error)); + throw new CodexConversationError(this.failed ?? (this.closing ? "closed" : failure(error)), this.backendReason); + } + } + private receive(value: CodexEvent): void { + if (this.state === "closing" || this.state === "closed" || this.state === "failed") return; + try { + const event = parseCodexEvent(value.method, value.params); + if (event.method === "command/exec/outputDelta" || event.method === "remoteControl/status/changed" || event.method === "thread/started") { + throw new CodexConversationError("protocol"); + } + if (event.method === "turn/started") this.turnStarted = true; + this.record({ type: "event", sequence: ++this.sequence, event }); + } catch { + void this.finish("protocol").catch(() => {}); + throw new CodexConversationError("protocol"); + } + } + private transition(state: ConversationState, code?: ConversationFailure): void { + this.state = state; + if (code) this.failed = code; + this.record({ type: "state", sequence: ++this.sequence, state, ...(this.failed ? { failure: this.failed } : {}), + ...(this.backendReason ? { backendReason: this.backendReason } : {}) }); + } + private record(frame: ConversationFrame): void { + const json = JSON.stringify(frame); + const bytes = Buffer.byteLength(json); + if (bytes > MAX_BYTES) throw new CodexConversationError("protocol"); + this.frames.push({ sequence: frame.sequence, json, bytes }); this.retainedBytes += bytes; + while (this.frames.length > MAX_FRAMES || this.retainedBytes > MAX_BYTES) { + this.retainedBytes -= this.frames.shift()!.bytes; + } + if (this.owner) { + if (this.owner.queue.length >= MAX_FRAMES + 1 || this.owner.queuedBytes + bytes > MAX_BYTES + 4096) { this.deliveryFailed(); return; } + this.owner.queue.push(json); this.owner.queuedBytes += bytes; this.drain(this.owner); + } + } + private deliveryFailed(): void { + this.owner = undefined; + void this.finish("delivery").catch(() => {}); + } + private drain(owner: Attachment): void { + if (owner.draining) return; + owner.draining = true; + try { + // Reentrant live events append behind the captured replay, never inside it. + let delivered = 0; + while (this.owner === owner && owner.queue.length) { + if (++delivered > MAX_FRAMES * 2 + 1) { this.deliveryFailed(); break; } + const json = owner.queue.shift()!; + owner.queuedBytes -= Buffer.byteLength(json); + owner.listener(JSON.parse(json)); + } + } catch { this.deliveryFailed(); } + finally { owner.draining = false; } + } + attach(capability: string, listener: Attachment["listener"], afterSequence = 0): CodexConversationAttachment { + if (typeof capability !== "string" || capability.length !== this.capability.length) throw new CodexConversationError("attachment"); + const supplied = Buffer.from(capability); + const expected = Buffer.from(this.capability); + if (supplied.length !== expected.length || !timingSafeEqual(supplied, expected) || this.owner || typeof listener !== "function") throw new CodexConversationError("attachment"); + if (!Number.isSafeInteger(afterSequence) || afterSequence < 0 || afterSequence > this.sequence) throw new CodexConversationError("input"); + const replay: ConversationReplay = { type: "replay", afterSequence, throughSequence: this.sequence, + truncated: afterSequence < (this.frames[0]?.sequence ?? 1) - 1, + retainedBytes: this.retainedBytes, retentionLimitBytes: MAX_BYTES, state: this.state, + ...(this.failed ? { failure: this.failed } : {}), ...(this.backendReason ? { backendReason: this.backendReason } : {}) }; + const queue = [JSON.stringify(replay), ...this.frames.filter((frame) => frame.sequence > afterSequence).map((frame) => frame.json)]; + const owner: Attachment = { listener, queue, queuedBytes: queue.reduce((sum, json) => sum + Buffer.byteLength(json), 0), draining: false }; + this.owner = owner; + const assertOwner = () => { if (this.owner !== owner) throw new CodexConversationError("attachment"); }; + const handle = Object.freeze({ + submit: (text: string) => { assertOwner(); return this.submit(text); }, + interrupt: () => { assertOwner(); return this.interrupt(); }, + detach: () => { if (this.owner === owner) this.owner = undefined; }, + }); + this.drain(owner); + return handle; + } + private submit(text: string): Promise<{ turnId: string; status: "completed" | "interrupted" | "failed" }> { + if (this.active) throw new CodexConversationError("busy"); + if (this.state !== "ready" || !this.driver) throw new CodexConversationError("closed"); + if (typeof text !== "string" || !text.trim() || Buffer.byteLength(text) > 262_144) throw new CodexConversationError("input"); + const driver = this.driver; + this.turnStarted = false; + const active = Promise.resolve().then(async () => { + if (this.state !== "running") throw new CodexConversationError("closed"); + const result = await Promise.race([driver.runTurn(text), this.stopped]); + if (result.status === "inProgress") throw new CodexConversationError("protocol"); + return { turnId: result.id, status: result.status }; + }).catch(async (error: unknown) => { + await this.finish(failure(error)); + throw new CodexConversationError(this.failed ?? (error instanceof CodexConversationError ? error.code : failure(error))); + }).finally(() => { + this.active = undefined; this.interrupting = undefined; + if (this.state === "running") this.transition("ready"); + }); + this.active = active; + void active.catch(() => {}); + this.transition("running"); + return active; + } + private interrupt(): Promise { + if (!this.active || this.state !== "running" || !this.driver) throw new CodexConversationError("closed"); + // Before turn/start is acknowledged there is no interruptible vendor ID. + // Closing the owned connection cancels that early window without guessing it. + if (!this.turnStarted) return this.finish("cancelled"); + const active = this.active; + this.interrupting ??= this.driver.interruptTurn().then(async () => { await active; }).catch(async (error: unknown) => { + await this.finish(failure(error)); throw new CodexConversationError(this.failed ?? failure(error)); + }); + void this.interrupting.catch(() => {}); + return this.interrupting; + } + /** Detach retains the conversation; the Sessions lease owner decides when to close. */ + close(): Promise { return this.finish(); } + private finish(code?: ConversationFailure): Promise { + if (this.closing) return this.closing; + if (code) this.failed = code; + this.state = "closing"; + this.rejectStopped(new CodexConversationError(code ?? "closed")); + this.closing = Promise.resolve().then(async () => { + try { + let driver: Driver | undefined; + try { driver = await this.connection; } + catch (error) { if (failure(error) === "cleanup") throw error; } + await driver?.close(); + this.detachFailure?.(); + this.transition(this.failed ? "failed" : "closed"); + } catch { + this.backendReason = "codex_app_server_cleanup_failed"; + this.detachFailure?.(); + this.transition("failed", "cleanup"); + throw new CodexConversationError("cleanup", this.backendReason); + } finally { this.owner = undefined; } + }); + void this.closing.catch(() => {}); + return this.closing; + } +} diff --git a/apps/genbi/server/runtime-host/codex-session.ts b/apps/genbi/server/runtime-host/codex-session.ts index 864e8bdae..4d2e5a297 100644 --- a/apps/genbi/server/runtime-host/codex-session.ts +++ b/apps/genbi/server/runtime-host/codex-session.ts @@ -262,6 +262,10 @@ export class CodexSession { this.emit(event); } /** Disconnect and BFF shutdown use the same bounded, connection-owned close. */ + onFailure(listener: (error: CodexRpcError) => void): () => void { + return this.rpc.onFailure(listener); + } + close(): Promise { this.closed = true; this.closePromise ??= this.rpc.close(); diff --git a/apps/genbi/test/codex-conversation.test.ts b/apps/genbi/test/codex-conversation.test.ts new file mode 100644 index 000000000..7e7f734f0 --- /dev/null +++ b/apps/genbi/test/codex-conversation.test.ts @@ -0,0 +1,229 @@ +import { describe, expect, it, vi } from "vitest"; +import { CodexConversation, CodexConversationError, type ConversationFrame, type ConversationReplay } from "../server/runtime-host/codex-conversation.js"; +import { CodexBackendError, type CodexLaunchPermit } from "../server/runtime-host/codex-app-server.js"; +import { CodexRpcError } from "../server/runtime-host/codex-rpc.js"; +import type { CodexEvent } from "../server/runtime-host/codex-events.js"; +import type { CodexSession } from "../server/runtime-host/codex-session.js"; + +type Turn = Awaited>; +function deferred() { + let resolve!: (value: T) => void; let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} +function fixture() { + let emit!: (event: CodexEvent) => void; + let failed: ((error: CodexRpcError) => void) | undefined; + const turn = deferred(); + const permit = { runtime: {}, assertActive: vi.fn(), release: vi.fn() } as unknown as CodexLaunchPermit; + const driver = { + startThread: vi.fn(async () => "thread"), + runTurn: vi.fn(() => turn.promise), + interruptTurn: vi.fn(async () => {}), + close: vi.fn(async () => { turn.reject(new CodexRpcError("closed")); }), + onFailure: vi.fn((listener: (error: CodexRpcError) => void) => { failed = listener; return () => { failed = undefined; }; }), + }; + void turn.promise.catch(() => {}); + const backend = { open: vi.fn(async (_permit: CodexLaunchPermit, input: { onEvent(event: CodexEvent): void }) => { emit = input.onEvent; return driver; }) }; + const input = { spec: {}, wrenHome: {}, assertScopeActive: vi.fn() } as unknown as ConstructorParameters[2]; + const conversation = new CodexConversation(backend, permit, input); + return { conversation, driver, backend, permit, turn, input, emit: (event: CodexEvent) => emit(event), fail: (error: CodexRpcError) => failed?.(error) }; +} +const status = (): CodexEvent => ({ method: "thread/status/changed", params: { threadId: "thread", status: { type: "idle" } } }); +const started = (): CodexEvent => ({ method: "turn/started", params: { threadId: "thread", turn: { id: "turn", status: "inProgress", items: [] } } }); +type Frame = ConversationFrame | ConversationReplay; + +describe("Codex conversation lifecycle bridge", () => { + it("uses the supplied permit/open contract, captures early events and starts exactly one thread", async () => { + const f = fixture(); + f.backend.open.mockImplementationOnce(async (_permit, input) => { input.onEvent(status()); return f.driver; }); + await f.conversation.ready; + expect(f.permit.assertActive).toHaveBeenCalledOnce(); + expect(f.backend.open.mock.calls[0]![0]).toBe(f.permit); + expect(f.backend.open.mock.calls[0]![1]).toMatchObject(f.input); + expect(f.driver.startThread).toHaveBeenCalledOnce(); + const frames: Frame[] = []; + f.conversation.attach(f.conversation.capability, (frame) => frames.push(frame)); + expect(frames.map((frame) => frame.type)).toEqual(["replay", "event", "state"]); + await f.conversation.close(); + }); + it("rejects wrong/cross-session capabilities and concurrent owners; stale handles cannot control a reattachment", async () => { + const f = fixture(); const other = fixture(); await Promise.all([f.conversation.ready, other.conversation.ready]); + expect(() => f.conversation.attach(other.conversation.capability, () => {})).toThrow("attachment"); + const first = f.conversation.attach(f.conversation.capability, () => {}); + expect(() => f.conversation.attach(f.conversation.capability, () => {})).toThrow("attachment"); + first.detach(); + const second = f.conversation.attach(f.conversation.capability, () => {}); + first.detach(); + expect(() => first.submit("old")).toThrow("attachment"); + expect(() => first.interrupt()).toThrow("attachment"); + expect(() => second.submit("")).toThrow("input"); + expect(f.driver.runTurn).not.toHaveBeenCalled(); + await Promise.all([f.conversation.close(), other.conversation.close()]); + }); + it("serializes turns, streams events and does not mistake interrupt acknowledgement for completion", async () => { + const f = fixture(); await f.conversation.ready; + const frames: Frame[] = []; + const attachment = f.conversation.attach(f.conversation.capability, (frame) => frames.push(frame)); + const submitted = attachment.submit("hello"); + expect(() => attachment.submit("concurrent")).toThrow("busy"); + await Promise.resolve(); f.emit(started()); + const interrupted = attachment.interrupt(); + let complete = false; void interrupted.then(() => { complete = true; }); + await Promise.resolve(); await Promise.resolve(); + expect(complete).toBe(false); + f.turn.resolve({ id: "turn", status: "interrupted", items: [] }); + await expect(submitted).resolves.toEqual({ turnId: "turn", status: "interrupted" }); + await interrupted; + expect(f.conversation.snapshot().state).toBe("ready"); + expect(frames.some((frame) => frame.type === "event")).toBe(true); + await f.conversation.close(); + }); + it("cancels before the turn acknowledgement by closing, never by guessing an interrupt ID", async () => { + const f = fixture(); await f.conversation.ready; + const attachment = f.conversation.attach(f.conversation.capability, () => {}); + const submitted = attachment.submit("hello"); + await attachment.interrupt(); + await expect(submitted).rejects.toBeInstanceOf(CodexConversationError); + expect(f.driver.interruptTurn).not.toHaveBeenCalled(); + expect(f.driver.close).toHaveBeenCalledOnce(); + expect(f.conversation.snapshot()).toMatchObject({ state: "failed", failure: "cancelled" }); + }); + it("bounds replay by count and bytes, reports truncation and rejects future cursors", async () => { + const f = fixture(); await f.conversation.ready; + for (let i = 0; i < 300; i++) f.emit(status()); + let frames: Frame[] = []; + const first = f.conversation.attach(f.conversation.capability, (frame) => frames.push(frame)); + expect(frames).toHaveLength(257); + expect(frames[0]).toMatchObject({ type: "replay", truncated: true, throughSequence: 301 }); + expect(() => { first.detach(); f.conversation.attach(f.conversation.capability, () => {}, 9999); }).toThrow("input"); + const big: CodexEvent = { method: "item/agentMessage/delta", params: { threadId: "thread", turnId: "turn", itemId: "item", delta: "字".repeat(100_000) } }; + for (let i = 0; i < 6; i++) f.emit(big); + frames = []; + f.conversation.attach(f.conversation.capability, (frame) => frames.push(frame)); + expect(frames[0]).toMatchObject({ type: "replay", truncated: true }); + expect((frames[0] as ConversationReplay).retainedBytes).toBeLessThanOrEqual(1_048_576); + await f.conversation.close(); + }); + it("reentrant live delivery follows the captured replay and subscriber mutation cannot corrupt retained frames", async () => { + const f = fixture(); await f.conversation.ready; f.emit(status()); + const frames: Frame[] = []; + const first = f.conversation.attach(f.conversation.capability, (frame) => { + frames.push(frame); + if (frame.type === "replay") f.emit(status()); + if (frame.type === "event") frame.event.params.threadId = "mutated"; + }); + expect(frames.map((frame) => frame.type === "replay" ? 0 : frame.sequence)).toEqual([0, 1, 2, 3]); + first.detach(); + const replay: Frame[] = []; + f.conversation.attach(f.conversation.capability, (frame) => replay.push(frame), 1); + expect(replay[1]).toMatchObject({ event: { params: { threadId: "thread" } } }); + await f.conversation.close(); + }); + it("detach preserves the active turn; host close settles work and forbids later operations", async () => { + const f = fixture(); await f.conversation.ready; + const attachment = f.conversation.attach(f.conversation.capability, () => {}); + const submitted = attachment.submit("hello"); await Promise.resolve(); + attachment.detach(); + expect(f.driver.close).not.toHaveBeenCalled(); + await f.conversation.close(); + await expect(submitted).rejects.toBeInstanceOf(CodexConversationError); + expect(() => attachment.submit("late")).toThrow("attachment"); + }); + it("closes idle transport failures and redacts arbitrary errors", async () => { + const f = fixture(); await f.conversation.ready; + f.fail(new CodexRpcError("transport")); + await f.conversation.close(); + expect(f.conversation.snapshot()).toMatchObject({ state: "failed", failure: "protocol" }); + expect(f.driver.close).toHaveBeenCalledOnce(); + const g = fixture(); g.backend.open.mockRejectedValueOnce(new Error("private credential output")); + await expect(g.conversation.ready).rejects.toThrow("unavailable"); + expect(JSON.stringify(g.conversation.snapshot())).not.toContain("private"); + expect(g.permit.release).toHaveBeenCalledOnce(); + }); + it("reports cleanup failure and never retries it as successful termination", async () => { + const f = fixture(); await f.conversation.ready; + f.driver.close.mockRejectedValueOnce(new Error("private cleanup path")); + const close = f.conversation.close(); + expect(f.conversation.close()).toBe(close); + await expect(close).rejects.toThrow("cleanup"); + expect(f.conversation.snapshot()).toMatchObject({ state: "failed", failure: "cleanup" }); + expect(f.driver.close).toHaveBeenCalledOnce(); + const g = fixture(); g.backend.open.mockRejectedValueOnce(new CodexBackendError("codex_app_server_cleanup_failed")); + await expect(g.conversation.ready).rejects.toThrow("cleanup"); + await expect(g.conversation.close()).rejects.toThrow("cleanup"); + }); + it("callback failure closes the backend without forwarding the thrown value", async () => { + const f = fixture(); await f.conversation.ready; + f.conversation.attach(f.conversation.capability, () => { throw new Error("secret in listener"); }); + await f.conversation.close(); + expect(f.conversation.snapshot()).toMatchObject({ state: "failed", failure: "delivery" }); + expect(JSON.stringify(f.conversation.snapshot())).not.toContain("secret"); + }); + it("close before opening releases the permit without spawning or starting a thread", async () => { + const f = fixture(); await f.conversation.close(); + await expect(f.conversation.ready).rejects.toThrow("closed"); + expect(f.backend.open).not.toHaveBeenCalled(); + expect(f.permit.release).toHaveBeenCalledOnce(); + }); + it("close during backend initialization owns and closes the eventual connection", async () => { + const f = fixture(); const opening = deferred(); + f.backend.open.mockReturnValueOnce(opening.promise); + await Promise.resolve(); await Promise.resolve(); + const closing = f.conversation.close(); opening.resolve(f.driver); + await closing; await expect(f.conversation.ready).rejects.toThrow("closed"); + expect(f.driver.close).toHaveBeenCalledOnce(); + expect(f.driver.startThread).not.toHaveBeenCalled(); + }); + it("opening failure waits for cleanup already triggered by the transport failure hook", async () => { + const f = fixture(); const cleanup = deferred(); + f.driver.close.mockReturnValueOnce(cleanup.promise); + f.driver.startThread.mockImplementationOnce(async () => { + f.fail(new CodexRpcError("transport")); + throw new CodexRpcError("protocol"); + }); + let settled = false; + void f.conversation.ready.catch(() => { settled = true; }); + await vi.waitFor(() => expect(f.driver.close).toHaveBeenCalledOnce()); + expect(settled).toBe(false); + cleanup.resolve(); + await expect(f.conversation.ready).rejects.toThrow("protocol"); + }); + it("stale permit fails before backend open and is released", async () => { + const f = fixture(); vi.mocked(f.permit.assertActive).mockImplementationOnce(() => { throw new Error("stale"); }); + await expect(f.conversation.ready).rejects.toThrow("unavailable"); + expect(f.backend.open).not.toHaveBeenCalled(); + expect(f.permit.release).toHaveBeenCalledOnce(); + }); + it("preserves fixed per-backend reasons through opening failure and replay metadata", async () => { + const f = fixture(); f.backend.open.mockRejectedValueOnce(new CodexBackendError("codex_identity_uncertified")); + await expect(f.conversation.ready).rejects.toMatchObject({ code: "unavailable", backendReason: "codex_identity_uncertified" }); + expect(f.conversation.snapshot()).toMatchObject({ backendReason: "codex_identity_uncertified" }); + const frames: Frame[] = []; + f.conversation.attach(f.conversation.capability, (frame) => frames.push(frame)); + expect(frames[0]).toMatchObject({ backendReason: "codex_identity_uncertified" }); + }); + it("refuses terminal bytes and unknown protocol events instead of presenting them as conversations", async () => { + const f = fixture(); await f.conversation.ready; + expect(() => f.emit({ method: "command/exec/outputDelta", params: { processId: "p", stream: "stdout", deltaBase64: "", capReached: false } })).toThrow("protocol"); + await f.conversation.close(); + expect(f.conversation.snapshot()).toMatchObject({ state: "failed", failure: "protocol" }); + const g = fixture(); await g.conversation.ready; + expect(() => g.emit({ method: "unknown", params: {} } as unknown as CodexEvent)).toThrow("protocol"); + await g.conversation.close(); + }); + it("settles the facade turn even if the closing driver leaves its own completion pending", async () => { + const f = fixture(); await f.conversation.ready; + f.driver.close.mockImplementationOnce(async () => {}); + const handle = f.conversation.attach(f.conversation.capability, () => {}); + const submitted = handle.submit("hello"); await Promise.resolve(); + await f.conversation.close(); + await expect(submitted).rejects.toThrow("closed"); + }); + it("an oversized event fails closed instead of silently disappearing from replay", async () => { + const f = fixture(); await f.conversation.ready; + expect(() => f.emit({ method: "item/agentMessage/delta", params: { threadId: "thread", turnId: "turn", itemId: "item", delta: "字".repeat(400_000) } })).toThrow("protocol"); + await f.conversation.close(); + expect(f.conversation.snapshot()).toMatchObject({ state: "failed", failure: "protocol" }); + }); +});