/* - Date: 2026-06-29 Spec: plans/2026-06-29-agent-backend-replay-orchestrator-spec.md Relationship: First executable backend replay orchestrator slice. It validates real provider fixtures, manifests, provider selection, replay timing, and dropped-event diagnostics before a fixture is allowed to drive the real Commander-to-Phoenix backend path. - Date: 2026-06-30 Spec: plans/2026-06-30-gstack-agent-timeline-event-model-plan.md Relationship: Queries generic agent status/task timeline rows and includes them in the backend replay oracle output set. */ import { randomUUID } from 'node:crypto'; import { mkdir, writeFile } from 'node:fs/promises'; import { join, resolve } from 'node:path'; import { collectAgentReplayDiagnostics } from '../src/agentReplayDiagnostics'; import { featureBackendReplayOracleChecks } from '../src/agentReplayBackendOracles'; import { loadAgentReplayFixture } from '../src/agentReplayFixture'; import { evaluateAgentReplayFeatureReadiness, loadAgentReplayManifestForFixture, summarizeAgentReplayManifest, type AgentReplayFeatureReadiness, type AgentReplayManifestSummary, } from '../src/agentReplayManifest'; import { agentReplayProviderSelectionFromValue, createAgentReplayRuntime, parseAgentReplayClockMode, type AgentReplayRunnerProviderSelection, } from '../src/agentReplayRunner'; import { connectPhoenixClient } from '../src/phoenix'; import { type JsonObject } from '../src/protocol'; import { runLocalCodexRunner, type RunnerOptions } from '../src/runner'; import { linzumiCliVersion } from '../src/version'; const backendReplayOracleTimeoutMs = 30_000; const backendReplayOraclePollMs = 100; type BackendReplayOptions = { readonly fixturePath: string; readonly provider: AgentReplayRunnerProviderSelection; readonly speed: string | undefined; readonly dryRun: boolean; readonly probeBackend: boolean; readonly backendUrl: string | undefined; readonly workspace: string | undefined; readonly channel: string | undefined; readonly threadId: string | undefined; readonly token: string | undefined; readonly sessionToken: string | undefined; readonly runnerId: string; readonly cwd: string; readonly workDescription: string; readonly setupFixture: boolean; readonly setupLabel: string; readonly outDir: string; readonly summary: boolean; readonly browserAuthFile: string | undefined; readonly browserAuth: BackendReplayBrowserAuth | undefined; readonly requiredFeatureId: string | undefined; readonly oracleFeatureId: string | undefined; // Wire replay mode (spec/codex-wire-capture-replay.spec.md): serve the // fixture from the stub codex app-server websocket instead of the // in-process provider-boundary client, so the runner's REAL socket // machinery is exercised against the REAL backend. readonly wire: boolean; readonly wireStrictness: string | undefined; // Replay-to-a-point (spec §Replay-to-a-point): --wire-breakpoint specs // for the wire stub (`ordinal:`, `method:[:]`, // `first-delta`, `turn-started[:]`, `turn-completed[:]`), in hold // order. While the stub holds, the whole stack (runner -> backend -> // client projection) is frozen at the recorded prefix; an external // driver (Playwright DOM assertions, the film rig) observes and releases // the hold over the stub's HTTP control routes (GET /replay/breakpoint, // POST /replay/resume against the printed control URL). readonly wireBreakpoints: readonly string[]; }; type WireReplayServerHandle = { readonly url: string; readonly port: number; readonly report: () => JsonObject; readonly waitForCompletion: (timeoutMs: number) => Promise; readonly close: () => Promise; }; async function startWireReplayServerForFixture( options: BackendReplayOptions ): Promise { // The implementation is ReScript (zero-TS/zero-raw-JS ratchets); the // compiled module is loaded through a non-literal specifier so the type // checker does not require declarations. The CLI test/build pipeline // compiles the package (`rescript build ../linzumi-cli-rescript`). const wireModulePath = new URL( '../../linzumi-cli-rescript/src/harness/HarnessCodexWireReplay.res.mjs', import.meta.url ).href; const wireModule = (await import(wireModulePath)) as { startWireReplayServerJs: (serverOptions: { readonly capturePath?: string; readonly strictness?: string | undefined; readonly clock?: string | undefined; readonly breakpoints?: readonly string[]; }) => Promise; wireReplayClockForBackendSpeed: ( speed: string | undefined ) => string | undefined; }; // Translate the orchestrator's --speed grammar into the wire clock // grammar (realtime -> recorded, x passes through, ms refuses // loudly - it has no wire counterpart); property-pinned in // test/agentReplayFixture.test.ts. const wireClock = wireModule.wireReplayClockForBackendSpeed(options.speed); const server = await wireModule.startWireReplayServerJs({ capturePath: options.fixturePath, ...(options.wireStrictness === undefined ? {} : { strictness: options.wireStrictness }), ...(wireClock === undefined ? {} : { clock: wireClock }), ...(options.wireBreakpoints.length === 0 ? {} : { breakpoints: options.wireBreakpoints }), }); if (options.wireBreakpoints.length > 0) { // The external driver's handle on the frozen stack: poll GET // /replay/breakpoint until held, assert on the frozen DOM/projection, // then POST /replay/resume. console.error( `[wire-breakpoint] control http://127.0.0.1:${server.port} breakpoints=${options.wireBreakpoints.join( ',' )}` ); } return server; } type BackendReplayBrowserAuth = { readonly accessToken: string; readonly refreshToken: string | undefined; readonly userId: string | number | undefined; readonly username: string; readonly user: JsonObject; }; type BackendReplaySummaryArgs = { readonly options: BackendReplayOptions; readonly fixturePath: string; readonly recordCount: number; readonly provider: string; readonly clockMode: ReturnType; readonly manifest: AgentReplayManifestSummary | undefined; readonly featureReadiness: readonly AgentReplayFeatureReadiness[] | undefined; readonly diagnostics: ReturnType; readonly backendReplay: JsonObject; }; async function main(): Promise { const args = process.argv.slice(2); if (args.includes('--help')) { printHelp(); return; } let options = parseArgs(args); const clockMode = parseAgentReplayClockMode(options.speed); const loaded = await loadAgentReplayFixture(options.fixturePath); if (loaded.type === 'error') { throw new Error(JSON.stringify(loaded.error)); } const runtime = await createAgentReplayRuntime({ fixturePath: options.fixturePath, provider: options.provider, clockMode, }); if (runtime.type === 'error') { throw new Error(runtime.message); } const manifest = await manifestSummary(options.fixturePath); const featureReadiness = await manifestReadiness(options.fixturePath); const diagnostics = collectAgentReplayDiagnostics(loaded.fixture); const requiredFeatureGate = evaluateRequiredFeatureGate( options.requiredFeatureId, featureReadiness ); if (diagnostics.droppedEvents.length > 0) { const summary = backendReplaySummary({ options, fixturePath: loaded.fixture.path, recordCount: loaded.fixture.records.length, provider: runtime.runtime.provider, clockMode, manifest, featureReadiness, diagnostics, backendReplay: { status: 'blocked', reason: 'fixture diagnostics reported dropped provider events', }, }); await emitBackendReplaySummary(options, summary); process.exitCode = 1; return; } if (requiredFeatureGate.status === 'failed') { const summary = backendReplaySummary({ options, fixturePath: loaded.fixture.path, recordCount: loaded.fixture.records.length, provider: runtime.runtime.provider, clockMode, manifest, featureReadiness, diagnostics, backendReplay: { status: 'blocked', reason: 'required feature is not ready for backend replay', requiredFeature: requiredFeatureGate, }, }); await emitBackendReplaySummary(options, summary); process.exitCode = 1; return; } if (!options.dryRun || options.probeBackend) { options = await maybeSetupBackendReplayFixture(options); } await maybeWriteBackendReplayBrowserAuth(options); if (options.probeBackend) { const probe = await probeBackend(options); const summary = backendReplaySummary({ options, fixturePath: loaded.fixture.path, recordCount: loaded.fixture.records.length, provider: runtime.runtime.provider, clockMode, manifest, featureReadiness, diagnostics, backendReplay: { status: 'backend-probed', reason: 'joined the real Phoenix local_runner channel; replay/oracles are the next slice', probe, }, }); await emitBackendReplaySummary(options, summary); return; } if (options.dryRun) { const summary = backendReplaySummary({ options, fixturePath: loaded.fixture.path, recordCount: loaded.fixture.records.length, provider: runtime.runtime.provider, clockMode, manifest, featureReadiness, diagnostics, backendReplay: { status: 'not-run', reason: 'dry-run validates fixture readiness before backend connection', }, }); await emitBackendReplaySummary(options, summary); return; } const replay = await runBackendReplay(options, runtime.runtime.provider); const summary = backendReplaySummary({ options, fixturePath: loaded.fixture.path, recordCount: loaded.fixture.records.length, provider: runtime.runtime.provider, clockMode, manifest, featureReadiness, diagnostics, backendReplay: { status: 'backend-replay-started', reason: 'started the real local runner with provider replay, triggered the normal GraphQL start mutation, and ran backend GraphQL oracles', replay, }, }); await emitBackendReplaySummary(options, summary); if (backendReplayOracleStatus(replay) === 'failed') { process.exitCode = 1; } } async function emitBackendReplaySummary( options: BackendReplayOptions, summary: JsonObject ): Promise { if (options.summary) { await mkdir(options.outDir, { recursive: true }); await writeFile( join(options.outDir, 'summary.json'), `${JSON.stringify(summary, null, 2)}\n`, 'utf8' ); } process.stdout.write(`${JSON.stringify(summary, null, 2)}\n`); } function frontendReplayTarget( options: BackendReplayOptions, threadId: string | undefined ): JsonObject | undefined { if ( options.backendUrl === undefined || options.workspace === undefined || options.channel === undefined || threadId === undefined ) { return undefined; } const url = new URL(options.backendUrl); switch (url.protocol) { case 'http:': case 'https:': break; case 'ws:': url.protocol = 'http:'; break; case 'wss:': url.protocol = 'https:'; break; default: throw new Error( '--backend-url must use http:, https:, ws:, or wss: protocol' ); } const path = `/w/${encodeURIComponent(options.workspace)}/c/${encodeURIComponent( options.channel )}/thread/${encodeURIComponent(threadId)}`; url.pathname = path; url.search = ''; url.hash = ''; return { path, url: url.toString(), }; } function backendReplayOracleStatus(replay: JsonObject): string | undefined { const oracles = replay.oracles; if ( typeof oracles === 'object' && oracles !== null && !Array.isArray(oracles) && typeof oracles.status === 'string' ) { return oracles.status; } return undefined; } function parseArgs(args: readonly string[]): BackendReplayOptions { const fixture = argValue(args, '--fixture'); if (fixture === undefined) { throw new Error('provide --fixture '); } return { fixturePath: resolve(fixture), provider: agentReplayProviderSelectionFromValue( argValue(args, '--provider') ), speed: argValue(args, '--speed'), dryRun: args.includes('--dry-run'), probeBackend: args.includes('--probe-backend'), backendUrl: argValue(args, '--backend-url'), workspace: argValue(args, '--workspace'), channel: argValue(args, '--channel'), threadId: argValue(args, '--thread-id'), token: argValue(args, '--token') ?? process.env.LINZUMI_AGENT_BACKEND_REPLAY_TOKEN, sessionToken: argValue(args, '--session-token') ?? process.env.LINZUMI_AGENT_BACKEND_REPLAY_SESSION_TOKEN, runnerId: argValue(args, '--runner-id') ?? `agent-backend-replay-${randomUUID()}`, cwd: resolve(argValue(args, '--cwd') ?? process.cwd()), workDescription: argValue(args, '--work-description') ?? 'Replay recorded agent fixture through the local backend path.', setupFixture: args.includes('--setup-fixture'), setupLabel: argValue(args, '--setup-label') ?? 'agent-backend-replay', outDir: resolve( argValue(args, '--out-dir') ?? 'agent-backend-replay-output' ), summary: !args.includes('--no-summary'), browserAuthFile: argValue(args, '--browser-auth-file') === undefined ? undefined : resolve(argValue(args, '--browser-auth-file') ?? ''), browserAuth: undefined, requiredFeatureId: argValue(args, '--require-feature'), oracleFeatureId: argValue(args, '--oracle-feature'), wire: args.includes('--wire'), wireStrictness: argValue(args, '--wire-strictness'), wireBreakpoints: argValues(args, '--wire-breakpoint'), }; } function backendReplaySummary(args: BackendReplaySummaryArgs): JsonObject { return { mode: args.options.probeBackend ? 'backend-probe' : args.options.dryRun ? 'dry-run' : 'backend-replay', fixture: { path: args.fixturePath, records: args.recordCount, }, provider: args.provider, clockMode: args.clockMode, backendTarget: { backendUrl: args.options.backendUrl, workspace: args.options.workspace, channel: args.options.channel, threadId: args.options.threadId, runnerId: args.options.runnerId, hasToken: args.options.token !== undefined, hasSessionToken: args.options.sessionToken !== undefined, setupFixture: args.options.setupFixture, setupLabel: args.options.setupFixture ? args.options.setupLabel : undefined, }, artifacts: { summaryPath: args.options.summary ? join(args.options.outDir, 'summary.json') : undefined, browserAuthPath: args.options.browserAuthFile, }, requiredFeatureId: args.options.requiredFeatureId, oracleFeatureId: args.options.oracleFeatureId, frontendTarget: frontendReplayTarget(args.options, args.options.threadId), featureCoverage: args.manifest, featureReadiness: args.featureReadiness, diagnostics: args.diagnostics, backendReplay: args.backendReplay, }; } async function maybeSetupBackendReplayFixture( options: BackendReplayOptions ): Promise { if (!options.setupFixture) { return options; } if (options.backendUrl === undefined) { throw new Error('--setup-fixture requires --backend-url '); } const suffix = `${Date.now()}-${randomUUID().slice(0, 8)}`; const workspace = slugifyReplaySetup( `${options.setupLabel}-workspace-${suffix}` ); const channel = slugifyReplaySetup(`${options.setupLabel}-channel-${suffix}`); const session = await registerReplayFixtureUser( options.backendUrl, `${options.setupLabel}-${suffix}` ); await enableReplayFixtureFeatureFlag( options.backendUrl, session.accessToken, 'CLAUDE_CODE_SUPPORT_PREVIEW' ); await createReplayWorkspace(options.backendUrl, session.accessToken, { name: `${options.setupLabel} workspace ${suffix}`, slug: workspace, }); await createReplayChannel(options.backendUrl, session.accessToken, { workspace, name: `${options.setupLabel} channel ${suffix}`, slug: channel, }); return { ...options, workspace, channel, sessionToken: session.accessToken, browserAuth: session, }; } async function maybeWriteBackendReplayBrowserAuth( options: BackendReplayOptions ): Promise { if (options.browserAuthFile === undefined) { return; } const auth = options.browserAuth ?? (options.sessionToken === undefined ? undefined : { accessToken: options.sessionToken, refreshToken: process.env.LINZUMI_AGENT_BACKEND_REPLAY_REFRESH_TOKEN, userId: 'agent-backend-replay', username: 'agent-backend-replay', user: { id: 'agent-backend-replay', username: 'agent-backend-replay', }, }); if (auth === undefined) { throw new Error( '--browser-auth-file requires --setup-fixture, --session-token, or LINZUMI_AGENT_BACKEND_REPLAY_SESSION_TOKEN' ); } await mkdir(resolve(options.browserAuthFile, '..'), { recursive: true }); await writeFile( options.browserAuthFile, `${JSON.stringify(auth, null, 2)}\n`, 'utf8' ); } type ReplayFixtureSession = BackendReplayBrowserAuth; async function registerReplayFixtureUser( backendUrl: string, label: string ): Promise { const username = slugifyReplaySetup(label); const response = await backendFetch( backendHttpUrl(backendUrl, '/api/v2/auth/register'), { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ display_name: username, e2e_fixture: true, // updateViewerFeatureFlag requires a developer user (#2149 gate); the // fixture-registration path marks the throwaway user via this param // (auth_controller maybe_mark_e2e_fixture_developer), which is only // honored when KANDAN_E2E_FIXTURE_REGISTRATION enabled the stub-mode // dev backend. e2e_fixture_developer: true, email: `${username}@example.test`, password: `pw-${username}`, username, }), }, 'fixture user registration' ); const payload = await parseJsonObjectResponse( response, 'fixture user registration' ); const accessToken = (payload as { access_token?: unknown }).access_token; const refreshToken = (payload as { refresh_token?: unknown }).refresh_token; const user = (payload as { user?: unknown }).user; const payloadUsername = (payload as { username?: unknown }).username; const userId = (payload as { user_id?: unknown }).user_id; if (typeof accessToken !== 'string' || accessToken.trim() === '') { throw new Error( `fixture user registration did not return access_token: ${JSON.stringify(payload)}` ); } const userObject = typeof user === 'object' && user !== null && !Array.isArray(user) ? (user as JsonObject) : { id: typeof userId === 'string' || typeof userId === 'number' ? userId : typeof payloadUsername === 'string' ? payloadUsername : label, username: typeof payloadUsername === 'string' ? payloadUsername : label, }; return { accessToken, refreshToken: typeof refreshToken === 'string' ? refreshToken : undefined, userId: typeof userId === 'string' || typeof userId === 'number' ? userId : typeof userObject.id === 'string' || typeof userObject.id === 'number' ? userObject.id : label, username: typeof payloadUsername === 'string' && payloadUsername.trim() !== '' ? payloadUsername : label, user: userObject, }; } async function enableReplayFixtureFeatureFlag( backendUrl: string, accessToken: string, flag: 'CLAUDE_CODE_SUPPORT_PREVIEW' ): Promise { const response = await backendFetch( backendHttpUrl(backendUrl, '/api/v2/graphql'), { method: 'POST', headers: authenticatedJsonHeaders(accessToken), body: JSON.stringify({ query: ` mutation EnableReplayFixtureFeatureFlag( $flag: ViewerFeatureFlagKey! $enabled: Boolean! ) { updateViewerFeatureFlag(input: {flag: $flag, enabled: $enabled}) { viewer { featureFlags { claudeCodeSupportPreview } } } } `, variables: { enabled: true, flag }, }), }, 'fixture feature flag enable' ); const payload = await parseJsonObjectResponse( response, 'fixture feature flag enable' ); const errors = (payload as { errors?: unknown }).errors; if (errors !== undefined) { throw new Error( `fixture feature flag enable GraphQL errors: ${JSON.stringify(errors)}` ); } const data = objectField(payload, 'data'); const result = objectField(data, 'updateViewerFeatureFlag'); const viewer = objectField(result, 'viewer'); const featureFlags = objectField(viewer, 'featureFlags'); if ( (featureFlags as { readonly claudeCodeSupportPreview?: unknown }) .claudeCodeSupportPreview !== true ) { throw new Error( `fixture feature flag enable did not enable Claude Code support: ${JSON.stringify(payload)}` ); } } async function createReplayWorkspace( backendUrl: string, accessToken: string, input: { readonly name: string; readonly slug: string } ): Promise { const response = await backendFetch( backendHttpUrl(backendUrl, '/api/v2/workspaces'), { method: 'POST', headers: authenticatedJsonHeaders(accessToken), body: JSON.stringify(input), }, 'fixture workspace create' ); const payload = await parseJsonObjectResponse( response, 'fixture workspace create' ); if (typeof (payload as { workspace?: unknown }).workspace !== 'object') { throw new Error( `fixture workspace create did not return workspace: ${JSON.stringify(payload)}` ); } } async function createReplayChannel( backendUrl: string, accessToken: string, input: { readonly workspace: string; readonly name: string; readonly slug: string; } ): Promise { const response = await backendFetch( backendHttpUrl( backendUrl, `/api/v2/workspaces/${encodeURIComponent(input.workspace)}/channels` ), { method: 'POST', headers: authenticatedJsonHeaders(accessToken), body: JSON.stringify({ conversation_type: 'public_channel', name: input.name, slug: input.slug, topic: 'Agent backend replay fixture channel', }), }, 'fixture channel create' ); const payload = await parseJsonObjectResponse( response, 'fixture channel create' ); if (typeof (payload as { channel?: unknown }).channel !== 'object') { throw new Error( `fixture channel create did not return channel: ${JSON.stringify(payload)}` ); } } function authenticatedJsonHeaders(accessToken: string): Record { return { accept: 'application/json', authorization: `Bearer ${accessToken}`, 'content-type': 'application/json', }; } async function backendFetch( url: string, init: RequestInit, label: string ): Promise { try { return await fetch(url, init); } catch (error) { const message = error instanceof Error ? error.message : String(error); throw new Error(`${label} request failed for ${url}: ${message}`); } } async function parseJsonObjectResponse( response: Response, label: string ): Promise { const payload: unknown = await response.json().catch(() => undefined); if ( !response.ok || typeof payload !== 'object' || payload === null || Array.isArray(payload) || (payload as { ok?: unknown }).ok === false ) { throw new Error( `${label} failed: ${response.status}:${JSON.stringify(payload)}` ); } return payload; } function slugifyReplaySetup(value: string): string { const slug = value .toLowerCase() .replace(/[^a-z0-9]+/gu, '-') .replace(/^-+|-+$/gu, '') .slice(0, 58); if (slug.trim() === '') { throw new Error('fixture setup label produced an empty slug'); } return slug; } async function runBackendReplay( options: BackendReplayOptions, provider: string ): Promise { if (options.backendUrl === undefined) { throw new Error('backend replay requires --backend-url '); } if (options.workspace === undefined || options.workspace.trim() === '') { throw new Error('backend replay requires --workspace '); } if (options.channel === undefined || options.channel.trim() === '') { throw new Error('backend replay requires --channel '); } if ( options.sessionToken === undefined || options.sessionToken.trim() === '' ) { throw new Error( 'backend replay requires --session-token or LINZUMI_AGENT_BACKEND_REPLAY_SESSION_TOKEN so it can call the normal GraphQL start mutation' ); } const runnerToken = await resolveBackendReplayToken(options); const wireServer = options.wire ? await startWireReplayServerForFixture(options) : undefined; try { const runner = await runLocalCodexRunner( backendReplayRunnerOptions( options, provider, runnerToken, wireServer?.url ) ); try { const start = await startLocalCodexRunnerInstance(options, provider); const oracles = await runBackendReplayOracles(options, start, provider); return { runnerId: options.runnerId, instanceId: runner.instanceId, codexUrl: runner.codexUrl, frontendTarget: frontendReplayTarget(options, start.threadId), start, oracles, ...(wireServer === undefined ? {} : { wire: wireServer.report() }), }; } finally { await runner.close(); } } finally { await wireServer?.close(); } } type BackendReplayStartSuccess = { readonly typename: 'StartLocalCodexRunnerInstanceSuccess'; readonly runnerKey: string; readonly threadId: string; }; type BackendReplayOracleCheck = { readonly id: string; readonly status: 'passed' | 'failed'; readonly details: JsonObject; }; async function runBackendReplayOracles( options: BackendReplayOptions, start: BackendReplayStartSuccess, provider: string ): Promise { const startedAtMs = Date.now(); const deadlineMs = startedAtMs + backendReplayOracleTimeoutMs; let lastResult: JsonObject | undefined; let attempts = 0; while (Date.now() <= deadlineMs) { attempts += 1; const result = await evaluateBackendReplayOracles(options, start, provider); lastResult = result; if (result.status === 'passed') { return { ...result, attempts, elapsedMs: Date.now() - startedAtMs, timeoutMs: backendReplayOracleTimeoutMs, }; } await waitForBackendReplayOraclePoll(); } if (lastResult === undefined) { return { status: 'failed', checks: [], messageStates: [], threadOutput: [], attempts, elapsedMs: Date.now() - startedAtMs, timeoutMs: backendReplayOracleTimeoutMs, timeoutReason: 'backend replay oracle did not run before timeout', }; } return { ...lastResult, attempts, elapsedMs: Date.now() - startedAtMs, timeoutMs: backendReplayOracleTimeoutMs, timeoutReason: 'backend replay oracle conditions did not pass before timeout', }; } function waitForBackendReplayOraclePoll(): Promise { return new Promise((resolve) => { const timeout = setTimeout(resolve, backendReplayOraclePollMs); timeout.unref?.(); }); } async function evaluateBackendReplayOracles( options: BackendReplayOptions, start: BackendReplayStartSuccess, provider: string ): Promise { const statesResult = await queryBackendReplayMessageStates( options, start.threadId ); const checks: BackendReplayOracleCheck[] = []; checks.push({ id: 'message-states-present', status: statesResult.states.length > 0 ? 'passed' : 'failed', details: { count: statesResult.states.length }, }); const runnerStates = statesResult.states.filter( (state) => state.runnerKey === options.runnerId ); checks.push({ id: 'message-states-runner-key', status: runnerStates.length > 0 ? 'passed' : 'failed', details: { runnerId: options.runnerId, count: runnerStates.length }, }); const terminalStates = runnerStates.filter( (state) => state.status === 'processed' || state.status === 'failed' ); checks.push({ id: 'message-state-terminal', status: terminalStates.length > 0 ? 'passed' : 'failed', details: { terminalStatuses: terminalStates.map((state) => state.status), allStatuses: runnerStates.map((state) => state.status), }, }); const processedState = terminalStates.find( (state) => state.status === 'processed' ); checks.push({ id: 'message-state-processed', status: processedState === undefined ? 'failed' : 'passed', details: { processedSeq: processedState?.seq, failedReasons: terminalStates .filter((state) => state.status === 'failed') .map((state) => state.reason ?? 'unknown'), }, }); const output = processedState === undefined ? { sourceSeq: undefined, items: [] as BackendReplayThreadOutputItem[] } : await queryBackendReplayThreadOutputSettled( options, start.threadId, processedState.seq, provider ); checks.push({ id: 'thread-output-present', status: output.items.length > 0 ? 'passed' : 'failed', details: { sourceSeq: output.sourceSeq, count: output.items.length }, }); const supportedOutputTypenames = new Set([ 'ThreadFeedCodexAssistantMessage', 'ThreadFeedCodexCommandExecutionMessage', 'ThreadFeedCodexFileChangeMessage', 'ThreadFeedCodexReasoningMessage', 'ThreadFeedAgentStatusEvent', 'ThreadFeedAgentTaskUpdateEvent', ]); const unsupportedOutputItems = output.items.filter( (item) => !supportedOutputTypenames.has(item.typename) ); checks.push({ id: 'thread-output-supported-types', status: output.items.length > 0 && unsupportedOutputItems.length === 0 ? 'passed' : 'failed', details: { typenames: output.items.map((item) => item.typename), unsupportedTypenames: unsupportedOutputItems.map((item) => item.typename), }, }); const assistantItems = output.items.filter( (item) => item.typename === 'ThreadFeedCodexAssistantMessage' ); checks.push({ id: 'thread-output-assistant-present', status: assistantItems.length > 0 ? 'passed' : 'failed', details: { count: assistantItems.length }, }); const providerItems = output.items.filter((item) => { switch (provider) { case 'claude-code': return item.agentProvider === 'claude-code'; case 'codex': return ( item.agentProvider === 'codex' || item.agentProvider === undefined ); } }); checks.push({ id: 'thread-output-provider', status: output.items.length > 0 && providerItems.length === output.items.length ? 'passed' : 'failed', details: { expectedProvider: provider, acceptedProviders: provider === 'codex' ? ['codex', null] : ['claude-code'], providers: output.items.map((item) => item.agentProvider ?? null), }, }); checks.push( ...featureBackendReplayOracleChecks({ featureId: options.oracleFeatureId ?? options.requiredFeatureId, provider, outputItems: output.items, requireLane: options.oracleFeatureId !== undefined, }) ); const failedChecks = checks.filter((check) => check.status === 'failed'); return { status: failedChecks.length === 0 ? 'passed' : 'failed', checks, messageStates: statesResult.states, threadOutput: output.items, }; } type BackendReplayMessageState = { readonly seq: number; readonly status: string; readonly reason: string | undefined; readonly runnerKey: string | undefined; readonly updatedAt: string | undefined; }; type BackendReplayMessageStatesResult = { readonly states: readonly BackendReplayMessageState[]; }; async function queryBackendReplayMessageStates( options: BackendReplayOptions, threadId: string ): Promise { const data = await postGraphql(options, { operationName: 'AgentBackendReplayMessageStates', query: backendReplayMessageStatesQuery, variables: backendReplayGraphqlVariables(options, threadId), }); const connection = objectField(data, 'threadFeedConnection'); const rawStates = arrayField(connection, 'localCodexMessageStates'); return { states: rawStates.map(parseBackendReplayMessageState), }; } type BackendReplayThreadOutputItem = { readonly typename: string; readonly seq: number | undefined; readonly content: string | undefined; readonly sourceMessageSeq: number | undefined; readonly streamKey: string | undefined; readonly streamState: string | undefined; readonly agentProvider: string | undefined; readonly kind: string | undefined; readonly title: string | undefined; readonly body: string | undefined; readonly severity: string | undefined; readonly state: string | undefined; readonly progressLabel: string | undefined; }; async function queryBackendReplayThreadOutput( options: BackendReplayOptions, threadId: string, sourceSeq: number ): Promise<{ readonly sourceSeq: number; readonly items: readonly BackendReplayThreadOutputItem[]; }> { const data = await postGraphql(options, { operationName: 'AgentBackendReplayThreadOutput', query: backendReplayThreadOutputQuery, variables: { ...backendReplayGraphqlVariables(options, threadId), sourceSeq, }, }); const rawItems = arrayField(data, 'localCodexThreadOutput'); return { sourceSeq, items: rawItems.map(parseBackendReplayThreadOutputItem), }; } // The terminal message state can land before the async row projection // finishes, so a single post-terminal query races the projection (observed: // the same fixture produced different row subsets run to run). Poll until the // projected output is quiescent - and, when a feature oracle is requested, // until its checks pass - before snapshotting. async function queryBackendReplayThreadOutputSettled( options: BackendReplayOptions, threadId: string, sourceSeq: number, provider: 'codex' | 'claude-code' ): Promise<{ readonly sourceSeq: number; readonly items: readonly BackendReplayThreadOutputItem[]; }> { const timeoutMs = 30_000; const pollIntervalMs = 250; const startedAtMs = Date.now(); const featureId = options.oracleFeatureId ?? options.requiredFeatureId; let last = await queryBackendReplayThreadOutput(options, threadId, sourceSeq); // Content-keyed stability: a row can flip streaming -> completed (or grow // fields) without changing the count, so count-only comparison could // snapshot mid-stream rows as settled. let lastFingerprint = JSON.stringify(last.items); let stableRounds = 0; while (Date.now() - startedAtMs < timeoutMs) { const featureSatisfied = featureId === undefined || featureBackendReplayOracleChecks({ featureId, provider, outputItems: last.items, requireLane: options.oracleFeatureId !== undefined, }).every((check) => check.status === 'passed'); if (featureSatisfied && stableRounds >= 2) { return last; } await new Promise((resolvePoll) => setTimeout(resolvePoll, pollIntervalMs)); const next = await queryBackendReplayThreadOutput( options, threadId, sourceSeq ); const nextFingerprint = JSON.stringify(next.items); stableRounds = nextFingerprint === lastFingerprint ? stableRounds + 1 : 0; lastFingerprint = nextFingerprint; last = next; } return last; } function backendReplayGraphqlVariables( options: BackendReplayOptions, threadId: string ): JsonObject { if (options.workspace === undefined || options.channel === undefined) { throw new Error('backend oracle requires workspace and channel'); } return { workspaceSlug: options.workspace, channelSlug: options.channel, threadId, }; } function parseBackendReplayMessageState( value: unknown ): BackendReplayMessageState { const state = objectFromUnknown(value, 'localCodexMessageState'); const seq = numberField(state, 'seq'); const status = stringField(state, 'status'); return { seq, status, reason: optionalStringField(state, 'reason'), runnerKey: optionalStringField(state, 'runnerKey'), updatedAt: optionalStringField(state, 'updatedAt'), }; } function parseBackendReplayThreadOutputItem( value: unknown ): BackendReplayThreadOutputItem { const item = objectFromUnknown(value, 'localCodexThreadOutput item'); return { typename: stringField(item, '__typename'), seq: optionalNumberField(item, 'seq'), content: optionalStringField(item, 'content'), sourceMessageSeq: optionalNumberField(item, 'sourceMessageSeq'), streamKey: optionalStringField(item, 'streamKey'), streamState: optionalStringField(item, 'streamState'), agentProvider: optionalStringField(item, 'agentProvider'), kind: optionalStringField(item, 'kind'), title: optionalStringField(item, 'title'), body: optionalStringField(item, 'body'), severity: optionalStringField(item, 'severity'), state: optionalStringField(item, 'state'), progressLabel: optionalStringField(item, 'progressLabel'), }; } async function postGraphql( options: BackendReplayOptions, body: JsonObject ): Promise { if (options.backendUrl === undefined) { throw new Error('GraphQL request requires --backend-url '); } if ( options.sessionToken === undefined || options.sessionToken.trim() === '' ) { throw new Error('GraphQL request requires a non-empty session token'); } const response = await backendFetch( backendHttpUrl(options.backendUrl, '/api/v2/graphql'), { method: 'POST', headers: { accept: 'application/json', authorization: `Bearer ${options.sessionToken}`, 'content-type': 'application/json', }, body: JSON.stringify(body), }, 'GraphQL request' ); const payload = await parseJsonObjectResponse(response, 'GraphQL request'); const errors = (payload as { errors?: unknown }).errors; if (errors !== undefined) { throw new Error(`GraphQL errors: ${JSON.stringify(errors)}`); } return objectField(payload, 'data'); } function objectField(value: object, field: string): object { return objectFromUnknown((value as Record)[field], field); } function arrayField(value: object, field: string): readonly unknown[] { const raw = (value as Record)[field]; if (!Array.isArray(raw)) { throw new Error(`${field} must be an array`); } return raw; } function objectFromUnknown(value: unknown, label: string): object { if (typeof value !== 'object' || value === null || Array.isArray(value)) { throw new Error(`${label} must be an object`); } return value; } function stringField(value: object, field: string): string { const raw = (value as Record)[field]; if (typeof raw !== 'string') { throw new Error(`${field} must be a string`); } return raw; } function optionalStringField(value: object, field: string): string | undefined { const raw = (value as Record)[field]; if (raw === null || raw === undefined) { return undefined; } if (typeof raw !== 'string') { throw new Error(`${field} must be a string when present`); } return raw; } function numberField(value: object, field: string): number { const raw = (value as Record)[field]; if (!Number.isInteger(raw)) { throw new Error(`${field} must be an integer`); } return raw; } function optionalNumberField(value: object, field: string): number | undefined { const raw = (value as Record)[field]; if (raw === null || raw === undefined) { return undefined; } if (!Number.isInteger(raw)) { throw new Error(`${field} must be an integer when present`); } return raw; } const backendReplayMessageStatesQuery = ` query AgentBackendReplayMessageStates( $workspaceSlug: String! $channelSlug: String! $threadId: ID! ) { threadFeedConnection( workspaceSlug: $workspaceSlug channelSlug: $channelSlug threadId: $threadId last: 20 ) { localCodexMessageStates { seq status reason runnerKey updatedAt } } } `; const backendReplayThreadOutputQuery = ` query AgentBackendReplayThreadOutput( $workspaceSlug: String! $channelSlug: String! $threadId: ID! $sourceSeq: Int! ) { localCodexThreadOutput( workspaceSlug: $workspaceSlug channelSlug: $channelSlug threadId: $threadId sourceSeq: $sourceSeq ) { __typename ... on ThreadFeedEntry { seq } ... on ThreadFeedCodexReasoningMessage { sourceMessageSeq streamKey streamState agentProvider } ... on ThreadFeedCodexAssistantMessage { content sourceMessageSeq streamKey streamState agentProvider } ... on ThreadFeedCodexCommandExecutionMessage { sourceMessageSeq streamKey streamState agentProvider } ... on ThreadFeedCodexFileChangeMessage { sourceMessageSeq streamKey streamState agentProvider } ... on ThreadFeedAgentStatusEvent { sourceMessageSeq agentProvider kind title body severity } ... on ThreadFeedAgentTaskUpdateEvent { sourceMessageSeq agentProvider kind title body state progressLabel } } } `; function backendReplayRunnerOptions( options: BackendReplayOptions, provider: string, token: string, wireCodexUrl?: string | undefined ): RunnerOptions { return { kandanUrl: options.backendUrl ?? '', token, runnerId: options.runnerId, clientId: 'agent-backend-replay', workspaceSlug: options.workspace, cwd: options.cwd, codexBin: process.env.CODEX_BIN ?? 'codex', // Wire mode: the runner connects its REAL websocket client to the stub // codex app-server through the same --codex-url seam it uses for a real // pre-started codex, instead of substituting an in-process fake client. codexUrl: wireCodexUrl, launchTui: false, allowedCwds: [options.cwd], channelSession: undefined, agentReplay: wireCodexUrl !== undefined ? undefined : { fixturePath: options.fixturePath, provider: options.provider, clockMode: parseAgentReplayClockMode(options.speed), }, }; } async function startLocalCodexRunnerInstance( options: BackendReplayOptions, provider: string ): Promise { if (options.backendUrl === undefined) { throw new Error('start mutation requires --backend-url '); } if (options.workspace === undefined || options.channel === undefined) { throw new Error('start mutation requires --workspace and --channel'); } if ( options.sessionToken === undefined || options.sessionToken.trim() === '' ) { throw new Error('start mutation requires a non-empty session token'); } const response = await backendFetch( backendHttpUrl(options.backendUrl, '/api/v2/graphql'), { method: 'POST', headers: { accept: 'application/json', authorization: `Bearer ${options.sessionToken}`, 'content-type': 'application/json', }, body: JSON.stringify({ operationName: 'AgentBackendReplayStartLocalCodexRunnerInstance', query: startLocalCodexRunnerInstanceMutation, variables: { workspaceSlug: options.workspace, channelSlug: options.channel, runnerId: options.runnerId, cwd: options.cwd, workDescription: options.workDescription, agentProvider: provider, allowPortForwardingByDefault: true, }, }), }, 'startLocalCodexRunnerInstance GraphQL request' ); const payload: unknown = await response.json(); if (!response.ok) { throw new Error( `startLocalCodexRunnerInstance HTTP failed: ${response.status}:${JSON.stringify(payload)}` ); } if ( typeof payload !== 'object' || payload === null || Array.isArray(payload) ) { throw new Error( 'startLocalCodexRunnerInstance returned a non-object payload' ); } const errors = (payload as { errors?: unknown }).errors; if (errors !== undefined) { throw new Error( `startLocalCodexRunnerInstance GraphQL errors: ${JSON.stringify(errors)}` ); } const result = startMutationResult(payload); switch (result.__typename) { case 'StartLocalCodexRunnerInstanceSuccess': return { typename: result.__typename, runnerKey: result.runnerKey, threadId: result.threadId, }; case 'StartLocalCodexRunnerInstanceUnavailable': throw new Error( `startLocalCodexRunnerInstance unavailable: ${result.reason}` ); default: throw new Error( `startLocalCodexRunnerInstance returned unexpected result: ${JSON.stringify(result)}` ); } } function startMutationResult(payload: object): | { readonly __typename: 'StartLocalCodexRunnerInstanceSuccess'; readonly runnerKey: string; readonly threadId: string; } | { readonly __typename: 'StartLocalCodexRunnerInstanceUnavailable'; readonly reason: string; } | { readonly __typename: string } { const data = (payload as { data?: unknown }).data; if (typeof data !== 'object' || data === null || Array.isArray(data)) { throw new Error('startLocalCodexRunnerInstance missing data'); } const result = (data as { startLocalCodexRunnerInstance?: unknown }) .startLocalCodexRunnerInstance; if (typeof result !== 'object' || result === null || Array.isArray(result)) { throw new Error('startLocalCodexRunnerInstance missing result'); } const typename = (result as { __typename?: unknown }).__typename; if (typename === 'StartLocalCodexRunnerInstanceSuccess') { const runnerKey = (result as { runnerKey?: unknown }).runnerKey; const threadId = (result as { threadId?: unknown }).threadId; if (typeof runnerKey !== 'string' || typeof threadId !== 'string') { throw new Error( `startLocalCodexRunnerInstance success had invalid fields: ${JSON.stringify(result)}` ); } return { __typename: typename, runnerKey, threadId }; } if (typename === 'StartLocalCodexRunnerInstanceUnavailable') { const reason = (result as { reason?: unknown }).reason; if (typeof reason !== 'string') { throw new Error( `startLocalCodexRunnerInstance unavailable had invalid fields: ${JSON.stringify(result)}` ); } return { __typename: typename, reason }; } if (typeof typename === 'string') { return { __typename: typename }; } throw new Error( `startLocalCodexRunnerInstance result missing __typename: ${JSON.stringify(result)}` ); } const startLocalCodexRunnerInstanceMutation = ` mutation AgentBackendReplayStartLocalCodexRunnerInstance( $workspaceSlug: String! $channelSlug: String! $runnerId: String! $cwd: String! $workDescription: String! $agentProvider: String $allowPortForwardingByDefault: Boolean ) { startLocalCodexRunnerInstance( input: { workspaceSlug: $workspaceSlug channelSlug: $channelSlug runnerKey: $runnerId cwd: $cwd workDescription: $workDescription agentProvider: $agentProvider allowPortForwardingByDefault: $allowPortForwardingByDefault } ) { __typename ... on StartLocalCodexRunnerInstanceSuccess { runnerKey threadId } ... on StartLocalCodexRunnerInstanceUnavailable { reason } } } `; async function probeBackend( options: BackendReplayOptions ): Promise { if (options.backendUrl === undefined) { throw new Error('--probe-backend requires --backend-url '); } const token = await resolveBackendReplayToken(options); const client = await connectPhoenixClient(options.backendUrl, token); try { const topic = `local_runner:${options.runnerId}`; const reply = await client.join(topic, backendProbeJoinPayload(options), { registrationId: 'agent-backend-replay-probe', }); return { topic, ok: reply.ok === true, protocol: typeof reply.protocol === 'string' ? reply.protocol : undefined, runnerId: typeof reply.runner_id === 'string' ? reply.runner_id : undefined, serverCapabilities: typeof reply.server_capabilities === 'object' && reply.server_capabilities !== null && !Array.isArray(reply.server_capabilities) ? reply.server_capabilities : undefined, }; } finally { client.close(); } } async function resolveBackendReplayToken( options: BackendReplayOptions ): Promise { if (options.token !== undefined && options.token.trim() !== '') { return options.token; } if ( options.sessionToken !== undefined && options.sessionToken.trim() !== '' ) { return exchangeSessionTokenForLocalRunnerToken(options); } throw new Error( '--probe-backend requires --token , LINZUMI_AGENT_BACKEND_REPLAY_TOKEN, --session-token , or LINZUMI_AGENT_BACKEND_REPLAY_SESSION_TOKEN' ); } async function exchangeSessionTokenForLocalRunnerToken( options: BackendReplayOptions ): Promise { if (options.backendUrl === undefined) { throw new Error('--session-token exchange requires --backend-url '); } if (options.workspace === undefined || options.workspace.trim() === '') { throw new Error('--session-token exchange requires --workspace '); } if ( options.sessionToken === undefined || options.sessionToken.trim() === '' ) { throw new Error( '--session-token exchange requires a non-empty session token' ); } const endpoint = backendHttpUrl( options.backendUrl, '/api/v2/local-codex-runner/oauth/session-exchange' ); const body = options.channel === undefined || options.channel.trim() === '' ? { workspace: options.workspace } : { workspace: options.workspace, channel: options.channel }; const response = await backendFetch( endpoint, { method: 'POST', headers: { accept: 'application/json', authorization: `Bearer ${options.sessionToken}`, 'content-type': 'application/json', }, body: JSON.stringify(body), }, 'local-runner session token exchange' ); const payload: unknown = await response.json(); if ( !response.ok || typeof payload !== 'object' || payload === null || Array.isArray(payload) || (payload as { ok?: unknown }).ok !== true || typeof (payload as { access_token?: unknown }).access_token !== 'string' ) { const error = typeof payload === 'object' && payload !== null && !Array.isArray(payload) && typeof (payload as { error?: unknown }).error === 'string' ? (payload as { error: string }).error : 'unknown'; throw new Error( `local-runner session token exchange failed: ${response.status}:${error}` ); } return (payload as { access_token: string }).access_token; } function backendHttpUrl(backendUrl: string, pathname: string): string { const url = new URL(backendUrl); switch (url.protocol) { case 'http:': case 'https:': break; case 'ws:': url.protocol = 'http:'; break; case 'wss:': url.protocol = 'https:'; break; default: throw new Error( '--backend-url must use http:, https:, ws:, or wss: protocol' ); } url.pathname = pathname; url.search = ''; url.hash = ''; return url.toString(); } function backendProbeJoinPayload(options: BackendReplayOptions): JsonObject { return { clientName: 'kandan-local-codex-runner', clientId: 'agent-backend-replay-probe', runnerPid: process.pid, version: linzumiCliVersion, cwd: options.cwd, workspace: options.workspace ?? null, channel: options.channel ?? null, capabilities: { codexAppServer: true, codexRemoteTui: true, client_message_ids: true, commander_pipeline: 'pipeline', agentProviders: ['codex', 'claude-code'], agentReplay: true, }, }; } async function manifestSummary( fixturePath: string ): Promise { const loaded = await loadAgentReplayManifestForFixture(fixturePath); if (loaded === undefined) { return undefined; } if (loaded.type === 'error') { throw new Error(JSON.stringify(loaded.error)); } return summarizeAgentReplayManifest(loaded.manifest); } async function manifestReadiness( fixturePath: string ): Promise { const loaded = await loadAgentReplayManifestForFixture(fixturePath); if (loaded === undefined) { return undefined; } if (loaded.type === 'error') { throw new Error(JSON.stringify(loaded.error)); } return evaluateAgentReplayFeatureReadiness(loaded.manifest); } type RequiredFeatureGate = | { readonly status: 'passed'; readonly featureId: string; readonly readiness: AgentReplayFeatureReadiness; } | { readonly status: 'failed'; readonly featureId: string; readonly reason: string; readonly readiness?: AgentReplayFeatureReadiness | undefined; }; function evaluateRequiredFeatureGate( featureId: string | undefined, readiness: readonly AgentReplayFeatureReadiness[] | undefined ): RequiredFeatureGate | { readonly status: 'not-required' } { if (featureId === undefined) { return { status: 'not-required' }; } if (readiness === undefined) { return { status: 'failed', featureId, reason: '--require-feature needs a neighboring fixture manifest', }; } const feature = readiness.find((row) => row.featureId === featureId); if (feature === undefined) { return { status: 'failed', featureId, reason: `feature ${featureId} is not present in the fixture manifest`, }; } switch (feature.status) { case 'implemented': case 'ready-to-implement': return { status: 'passed', featureId, readiness: feature }; case 'blocked': case 'missing-proof': case 'not-started': return { status: 'failed', featureId, readiness: feature, reason: feature.reason, }; } } // Every value of a repeatable flag, in argv order (--wire-breakpoint may // name several ordered holds). function argValues(args: readonly string[], name: string): readonly string[] { const values: string[] = []; for (let index = 0; index < args.length; index += 1) { if (args[index] !== name) { continue; } const value = args[index + 1]; if (value === undefined) { throw new Error(`${name} requires a value`); } values.push(value); index += 1; } return values; } function argValue(args: readonly string[], name: string): string | undefined { const index = args.indexOf(name); if (index === -1) { return undefined; } const value = args[index + 1]; if (value === undefined) { throw new Error(`${name} requires a value`); } return value; } function printHelp(): void { process.stdout.write( `Usage: pnpm run replay:agent-backend -- --fixture [options]\n\nOptions:\n --dry-run Validate fixture/manifest/diagnostics without backend connection\n --probe-backend Also join the real Phoenix local_runner channel, then exit\n --provider auto|codex|claude-code Replay provider. Default: auto\n --speed immediate|realtime|x|ms\n Replay speed. Default: immediate\n --backend-url Local Phoenix backend URL; required by --probe-backend/replay\n --setup-fixture Create a throwaway fixture user/workspace/channel first; requires local fixture registration\n --setup-label