mirror of
https://github.com/Canner/WrenAI.git
synced 2026-10-16 06:57:43 +08:00
feat(genbi): add gated Codex conversation lifecycle bridge
This commit is contained in:
Jax Liu
@@ -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}`);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<CodexEvent, { method: "command/exec/outputDelta" | "remoteControl/status/changed" | "thread/started" }>;
|
||||
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<void>;
|
||||
detach(): void;
|
||||
}
|
||||
type Driver = Pick<CodexSession, "startThread" | "runTurn" | "interruptTurn" | "close" | "onFailure">;
|
||||
type OpenInput = Omit<Parameters<CodexAppServerBackend["open"]>[1], "onEvent">;
|
||||
interface Backend { open(permit: CodexLaunchPermit, input: OpenInput & { onEvent(event: CodexEvent): void }): Promise<Driver> }
|
||||
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<void>;
|
||||
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<Driver>;
|
||||
private closing?: Promise<void>;
|
||||
private detachFailure?: () => void;
|
||||
private active: Promise<{ turnId: string; status: "completed" | "interrupted" | "failed" }> | undefined;
|
||||
private interrupting: Promise<void> | undefined;
|
||||
private turnStarted = false;
|
||||
private rejectStopped!: (error: CodexConversationError) => void;
|
||||
private readonly stopped = new Promise<never>((_, 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<void> {
|
||||
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<void> {
|
||||
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<void> { return this.finish(); }
|
||||
private finish(code?: ConversationFailure): Promise<void> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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<void> {
|
||||
this.closed = true;
|
||||
this.closePromise ??= this.rpc.close();
|
||||
|
||||
@@ -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<ReturnType<CodexSession["runTurn"]>>;
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void; let reject!: (error: unknown) => void;
|
||||
const promise = new Promise<T>((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<Turn>();
|
||||
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<typeof CodexConversation>[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<typeof f.driver>();
|
||||
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<void>();
|
||||
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" });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user