From c9c66384d3fec79345bafc9d7c8331df97d5be4a Mon Sep 17 00:00:00 2001 From: magonglei Date: Sat, 21 Mar 2026 03:29:25 +0800 Subject: [PATCH 1/5] Harden farfield thread state handling --- apps/server/package.json | 2 +- .../server/src/agents/adapters/codex-agent.ts | 1770 +++++++++++++++-- apps/server/src/debug-history.ts | 367 ++++ apps/server/src/index.ts | 80 +- apps/server/src/unified/adapter.ts | 219 +- .../test/codex-agent-live-state.test.ts | 827 ++++++++ apps/server/test/debug-history.test.ts | 165 ++ apps/server/test/unified-adapter.test.ts | 357 ++++ apps/web/index.html | 1 + apps/web/public/favicon.svg | 7 + apps/web/src/App.tsx | 750 ++++++- apps/web/src/components/ConversationItem.tsx | 157 +- apps/web/src/lib/api.ts | 19 +- apps/web/src/lib/server-target.ts | 16 +- apps/web/test/app.test.tsx | 815 +++++++- apps/web/test/conversation-item.test.tsx | 63 + packages/codex-api/src/app-server-client.ts | 45 +- .../codex-api/src/app-server-transport.ts | 97 +- .../codex-api/test/app-server-client.test.ts | 85 + .../scripts/generate-app-server-zod.mjs | 14 +- packages/codex-protocol/src/app-server.ts | 85 +- packages/codex-protocol/src/thread.ts | 52 +- packages/codex-protocol/test/protocol.test.ts | 75 +- packages/unified-surface/src/index.ts | 40 +- scripts/generate-opencode-manifest.mjs | 6 +- 25 files changed, 5751 insertions(+), 363 deletions(-) create mode 100644 apps/server/src/debug-history.ts create mode 100644 apps/server/test/codex-agent-live-state.test.ts create mode 100644 apps/server/test/debug-history.test.ts create mode 100644 apps/web/public/favicon.svg create mode 100644 apps/web/test/conversation-item.test.tsx diff --git a/apps/server/package.json b/apps/server/package.json index bf4088bc..63a601cb 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -13,7 +13,7 @@ }, "scripts": { "dev": "tsx watch src/index.ts", - "dev:remote": "HOST=0.0.0.0 PORT=4311 tsx watch src/index.ts", + "dev:remote": "HOST=0.0.0.0 PORT=4311 tsx src/index.ts", "build": "tsc -p tsconfig.json", "build:bundle": "rm -rf dist && esbuild src/index.ts --bundle --platform=node --format=esm --target=node20 --outfile=dist/index.js --banner:js='#!/usr/bin/env node' --external:pino --external:zod --external:@opencode-ai/sdk --external:@opencode-ai/sdk/*", "prepack": "bun run build:bundle", diff --git a/apps/server/src/agents/adapters/codex-agent.ts b/apps/server/src/agents/adapters/codex-agent.ts index 6874641f..e38b793c 100644 --- a/apps/server/src/agents/adapters/codex-agent.ts +++ b/apps/server/src/agents/adapters/codex-agent.ts @@ -1,3 +1,6 @@ +import { access, readFile, readdir } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import { AppServerClient, AppServerRpcError, @@ -71,6 +74,30 @@ export interface CodexAgentOptions { } const ANSI_ESCAPE_REGEX = /\u001B\[[0-?]*[ -/]*[@-~]/g; +const MAX_TRACKED_THREAD_RUNTIME_STATES = 80; +const MAX_STREAM_EVENTS_PER_THREAD = 120; +const ARCHIVED_TOOL_ARGUMENT_PREVIEW_MAX_CHARS = 600; +const ARCHIVED_TOOL_TEXT_PREVIEW_MAX_CHARS = 240; +const loggedArchivedUnknownPayloadTypes = new Set(); +const loggedArchivedMalformedJsonlLineKeys = new Set(); + +type ArchivedJsonValue = + | string + | number + | boolean + | null + | ArchivedJsonValue[] + | { [key: string]: ArchivedJsonValue }; + +type ArchivedThreadItem = ThreadConversationState["turns"][number]["items"][number]; +type ArchivedDynamicToolCallItem = Extract< + ArchivedThreadItem, + { type: "dynamicToolCall" } +>; +type ArchivedDynamicToolContentItem = NonNullable< + ArchivedDynamicToolCallItem["contentItems"] +>[number]; +type ArchivedTodoListItem = Extract; export class CodexAgentAdapter implements AgentAdapter { public readonly id = "codex"; @@ -102,9 +129,17 @@ export class CodexAgentAdapter implements AgentAdapter { StreamSnapshotOrigin >(); private readonly threadTitleById = new Map(); + private readonly trackedThreadIds = new Map(); private readonly ipcFrameListeners = new Set< (event: CodexIpcFrameEvent) => void >(); + private maxTrackedThreadCountSeen = 0; + private maxTotalBufferedStreamEventCountSeen = 0; + private maxBufferedStreamEventsPerThreadSeen = 0; + private archivedRestoreHitCount = 0; + private archivedRestoreFailureCount = 0; + private streamParseFailureCount = 0; + private streamReductionFailureCount = 0; private lastKnownOwnerClientId: string | null = null; private runtimeState: CodexAgentRuntimeState = { @@ -148,7 +183,7 @@ export class CodexAgentAdapter implements AgentAdapter { }); if (!state.connected) { - this.scheduleIpcReconnect(); + this.scheduleReconnect(); } else if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; @@ -184,10 +219,10 @@ export class CodexAgentAdapter implements AgentAdapter { if (frame.type === "broadcast" && threadId) { const current = this.streamEventsByThreadId.get(threadId) ?? []; current.push(frame); - if (current.length > 400) { - current.splice(0, current.length - 400); + if (current.length > MAX_STREAM_EVENTS_PER_THREAD) { + current.splice(0, current.length - MAX_STREAM_EVENTS_PER_THREAD); } - this.streamEventsByThreadId.set(threadId, current); + this.setTrackedThreadEvents(threadId, current); } if ( @@ -210,7 +245,7 @@ export class CodexAgentAdapter implements AgentAdapter { } if (sourceClientId) { - this.threadOwnerById.set(conversationId, sourceClientId); + this.setTrackedThreadOwner(conversationId, sourceClientId); } try { @@ -220,8 +255,7 @@ export class CodexAgentAdapter implements AgentAdapter { } const snapshot = parsedBroadcast.params.change.conversationState; - this.streamSnapshotByThreadId.set(conversationId, snapshot); - this.streamSnapshotOriginByThreadId.set(conversationId, "stream"); + this.setTrackedThreadSnapshot(conversationId, snapshot, "stream"); this.setThreadTitle(conversationId, snapshot.title); } catch (error) { logger.error( @@ -253,6 +287,43 @@ export class CodexAgentAdapter implements AgentAdapter { return this.threadOwnerById.size; } + public getTrackedThreadRuntimeCounts(): { + trackedThreadCount: number; + streamEventThreadCount: number; + totalBufferedStreamEventCount: number; + streamSnapshotThreadCount: number; + threadTitleCount: number; + maxTrackedThreadCountSeen: number; + maxTotalBufferedStreamEventCountSeen: number; + maxBufferedStreamEventsPerThreadSeen: number; + archivedRestoreHitCount: number; + archivedRestoreFailureCount: number; + streamParseFailureCount: number; + streamReductionFailureCount: number; + } { + let totalBufferedStreamEventCount = 0; + for (const events of this.streamEventsByThreadId.values()) { + totalBufferedStreamEventCount += events.length; + } + + return { + trackedThreadCount: this.trackedThreadIds.size, + streamEventThreadCount: this.streamEventsByThreadId.size, + totalBufferedStreamEventCount, + streamSnapshotThreadCount: this.streamSnapshotByThreadId.size, + threadTitleCount: this.threadTitleById.size, + maxTrackedThreadCountSeen: this.maxTrackedThreadCountSeen, + maxTotalBufferedStreamEventCountSeen: + this.maxTotalBufferedStreamEventCountSeen, + maxBufferedStreamEventsPerThreadSeen: + this.maxBufferedStreamEventsPerThreadSeen, + archivedRestoreHitCount: this.archivedRestoreHitCount, + archivedRestoreFailureCount: this.archivedRestoreFailureCount, + streamParseFailureCount: this.streamParseFailureCount, + streamReductionFailureCount: this.streamReductionFailureCount, + }; + } + public isEnabled(): boolean { return true; } @@ -415,53 +486,86 @@ export class CodexAgentAdapter implements AgentAdapter { let result: Awaited>; try { - result = await readThreadWithOption(input.includeTurns); - } catch (error) { - const typedError = error instanceof Error ? error : null; - const shouldTryResume = - isThreadNotLoadedAppServerRpcError(typedError) || - (input.includeTurns && - (isThreadNotMaterializedIncludeTurnsAppServerRpcError(typedError) || - isThreadNoRolloutIncludeTurnsAppServerRpcError(typedError))); - if (!shouldTryResume) { - throw error; - } - try { - await this.resumeThread(input.threadId); result = await readThreadWithOption(input.includeTurns); - } catch (resumeRetryError) { - const typedResumeRetryError = - resumeRetryError instanceof Error ? resumeRetryError : null; - const shouldRetryWithoutTurns = - input.includeTurns && - (isThreadNotMaterializedIncludeTurnsAppServerRpcError( - typedResumeRetryError, - ) || - isThreadNoRolloutIncludeTurnsAppServerRpcError( + } catch (error) { + const typedError = error instanceof Error ? error : null; + const shouldTryResume = + isThreadNotLoadedAppServerRpcError(typedError) || + (input.includeTurns && + (isThreadNotMaterializedIncludeTurnsAppServerRpcError(typedError) || + isThreadNoRolloutIncludeTurnsAppServerRpcError(typedError))); + if (!shouldTryResume) { + throw error; + } + + try { + await this.resumeThread(input.threadId); + result = await readThreadWithOption(input.includeTurns); + } catch (resumeRetryError) { + const typedResumeRetryError = + resumeRetryError instanceof Error ? resumeRetryError : null; + const shouldRetryWithoutTurns = + input.includeTurns && + (isThreadNotMaterializedIncludeTurnsAppServerRpcError( typedResumeRetryError, - )); - if (!shouldRetryWithoutTurns) { - throw resumeRetryError; + ) || + isThreadNoRolloutIncludeTurnsAppServerRpcError( + typedResumeRetryError, + )); + if (!shouldRetryWithoutTurns) { + throw resumeRetryError; + } + result = await readThreadWithOption(false); } - result = await readThreadWithOption(false); } + } catch (error) { + const restoredThread = await this.restoreArchivedThreadState( + input.threadId, + input.includeTurns, + ); + if (!restoredThread) { + throw error; + } + this.patchRuntimeState({ + lastError: null, + }); + result = { + thread: restoredThread, + }; + } + let parsedThread: ThreadConversationState; + try { + parsedThread = parseThreadConversationState(result.thread); + } catch (error) { + const restoredThread = await this.restoreArchivedThreadState( + input.threadId, + input.includeTurns, + ); + if (!restoredThread) { + throw error; + } + this.patchRuntimeState({ + lastError: null, + }); + parsedThread = restoredThread; } - const parsedThread = parseThreadConversationState(result.thread); const existingSnapshot = this.streamSnapshotByThreadId.get(input.threadId); const shouldStoreSnapshot = input.includeTurns || parsedThread.turns.length > 0 || existingSnapshot === undefined; if (shouldStoreSnapshot) { - this.streamSnapshotByThreadId.set(input.threadId, parsedThread); const snapshotOrigin: StreamSnapshotOrigin = input.includeTurns && parsedThread.turns.length > 0 ? "readThreadWithTurns" : "readThread"; - this.streamSnapshotOriginByThreadId.set(input.threadId, snapshotOrigin); + this.setTrackedThreadSnapshot(input.threadId, parsedThread, snapshotOrigin); } this.setThreadTitle(input.threadId, parsedThread.title); + this.patchRuntimeState({ + lastError: null, + }); return { thread: parsedThread, }; @@ -489,7 +593,7 @@ export class CodexAgentAdapter implements AgentAdapter { })(); if (ownerClientId && this.isIpcReady()) { - this.threadOwnerById.set(input.threadId, ownerClientId); + this.setTrackedThreadOwner(input.threadId, ownerClientId); try { await this.service.sendMessage({ threadId: input.threadId, @@ -642,18 +746,21 @@ export class CodexAgentAdapter implements AgentAdapter { () => this.appClient.readThread(input.threadId, true), ); const parsedThread = parseThreadConversationState(refreshedThread.thread); - this.streamSnapshotByThreadId.set(input.threadId, parsedThread); - this.streamSnapshotOriginByThreadId.set(input.threadId, "readThreadWithTurns"); this.setThreadTitle(input.threadId, parsedThread.title); const currentEvents = this.streamEventsByThreadId.get(input.threadId) ?? []; currentEvents.push( buildSyntheticSnapshotEvent(input.threadId, ownerClientIdForResult, parsedThread), ); - if (currentEvents.length > 400) { - currentEvents.splice(0, currentEvents.length - 400); + if (currentEvents.length > MAX_STREAM_EVENTS_PER_THREAD) { + currentEvents.splice(0, currentEvents.length - MAX_STREAM_EVENTS_PER_THREAD); } - this.streamEventsByThreadId.set(input.threadId, currentEvents); + this.setTrackedThreadSnapshot( + input.threadId, + parsedThread, + "readThreadWithTurns", + ); + this.setTrackedThreadEvents(input.threadId, currentEvents); return { ownerClientId: ownerClientIdForResult, @@ -668,7 +775,7 @@ export class CodexAgentAdapter implements AgentAdapter { input.ownerClientId, this.lastKnownOwnerClientId ?? undefined, ); - this.threadOwnerById.set(input.threadId, ownerClientId); + this.setTrackedThreadOwner(input.threadId, ownerClientId); const pendingIpcRequest = await this.resolvePendingIpcRequest( input.threadId, @@ -730,12 +837,20 @@ export class CodexAgentAdapter implements AgentAdapter { } public async readLiveState(threadId: string): Promise { - const snapshotState = this.streamSnapshotByThreadId.get(threadId) ?? null; - const snapshotOrigin = + this.touchTrackedThread(threadId); + let snapshotState = this.streamSnapshotByThreadId.get(threadId) ?? null; + let snapshotOrigin = this.streamSnapshotOriginByThreadId.get(threadId) ?? null; const ownerClientId = this.threadOwnerById.get(threadId) ?? this.lastKnownOwnerClientId ?? null; const rawEvents = this.streamEventsByThreadId.get(threadId) ?? []; + if (rawEvents.length === 0 && snapshotState === null) { + const restoredState = await this.restoreArchivedThreadState(threadId, true); + if (restoredState) { + snapshotState = restoredState; + snapshotOrigin = "readThreadWithTurns"; + } + } if (rawEvents.length === 0) { return { ownerClientId, @@ -744,37 +859,32 @@ export class CodexAgentAdapter implements AgentAdapter { }; } - const events: ReturnType[] = - []; - - for (let eventIndex = 0; eventIndex < rawEvents.length; eventIndex += 1) { - const event = rawEvents[eventIndex]; - try { - events.push(parseThreadStreamStateChangedBroadcast(event)); - } catch (error) { - logger.error( - { - threadId, - eventIndex, - error: toErrorMessage(error), - ...(error instanceof ProtocolValidationError - ? { issues: error.issues } - : {}), - }, - "thread-stream-event-parse-failed", - ); - return { - ownerClientId, - conversationState: snapshotState, - liveStateError: { - kind: "parseFailed", - message: toErrorMessage(error), - eventIndex, - patchIndex: null, - }, - }; - } + const collectedEvents = collectThreadStreamStateChangedEvents(rawEvents); + if (collectedEvents.parseError) { + this.streamParseFailureCount += 1; + logger.error( + { + threadId, + eventIndex: collectedEvents.parseError.eventIndex, + error: collectedEvents.parseError.message, + ...(collectedEvents.parseError.issues + ? { issues: collectedEvents.parseError.issues } + : {}), + }, + "thread-stream-event-parse-failed", + ); + return { + ownerClientId, + conversationState: snapshotState, + liveStateError: { + kind: "parseFailed", + message: collectedEvents.parseError.message, + eventIndex: collectedEvents.parseError.eventIndex, + patchIndex: null, + }, + }; } + const events = collectedEvents.events; if (events.length === 0) { return { @@ -806,7 +916,7 @@ export class CodexAgentAdapter implements AgentAdapter { buildSyntheticSnapshotEvent( threadId, ownerClientId ?? "farfield", - snapshotState, + snapshotState!, ), ...reductionEvents, ] @@ -820,6 +930,7 @@ export class CodexAgentAdapter implements AgentAdapter { liveStateError: null, }; } catch (error) { + this.streamReductionFailureCount += 1; const reductionErrorDetails = error instanceof ThreadStreamReductionError ? error.details : null; const eventIndex = reductionErrorDetails?.eventIndex ?? null; @@ -853,6 +964,7 @@ export class CodexAgentAdapter implements AgentAdapter { threadId: string, limit: number, ): Promise { + this.touchTrackedThread(threadId); return { ownerClientId: this.threadOwnerById.get(threadId) ?? @@ -970,7 +1082,7 @@ export class CodexAgentAdapter implements AgentAdapter { } } - private scheduleIpcReconnect(): void { + private scheduleReconnect(): void { if ( this.reconnectTimer || !this.runtimeState.codexAvailable || @@ -994,6 +1106,9 @@ export class CodexAgentAdapter implements AgentAdapter { }); return result; } catch (error) { + if (error instanceof AppServerTransportError) { + this.scheduleReconnect(); + } this.patchRuntimeState({ appReady: !(error instanceof AppServerTransportError), lastError: toErrorMessage(error), @@ -1027,6 +1142,8 @@ export class CodexAgentAdapter implements AgentAdapter { lastError: message, }); logger.warn({ error: message }, "codex-not-found"); + } else { + this.scheduleReconnect(); } } @@ -1053,7 +1170,7 @@ export class CodexAgentAdapter implements AgentAdapter { ipcConnected: this.ipcClient.isConnected(), lastError: toErrorMessage(error), }); - this.scheduleIpcReconnect(); + this.scheduleReconnect(); } finally { this.bootstrapInFlight = null; } @@ -1105,6 +1222,54 @@ export class CodexAgentAdapter implements AgentAdapter { ); } + private async restoreArchivedThreadState( + threadId: string, + includeTurns: boolean, + ): Promise { + const rolloutPath = await findArchivedRolloutPath(threadId); + if (!rolloutPath) { + return null; + } + + try { + const content = await readFile(rolloutPath, "utf8"); + const metadata = await readThreadIndexMetadata(threadId); + const restored = buildArchivedThreadConversationStateFromJsonl( + content, + threadId, + metadata, + ); + if (!restored) { + this.archivedRestoreFailureCount += 1; + return null; + } + + const normalized = includeTurns + ? restored + : { + ...restored, + turns: [], + }; + if (restored.turns.length > 0) { + this.setTrackedThreadSnapshot(threadId, restored, "readThreadWithTurns"); + } + this.setThreadTitle(threadId, normalized.title); + this.archivedRestoreHitCount += 1; + return normalized; + } catch (error) { + this.archivedRestoreFailureCount += 1; + logger.warn( + { + threadId, + rolloutPath, + error: toErrorMessage(error), + }, + "archived-thread-restore-failed", + ); + return null; + } + } + private async isThreadLoaded(threadId: string): Promise { let cursor: string | null = null; @@ -1207,10 +1372,59 @@ export class CodexAgentAdapter implements AgentAdapter { return snapshot.title; } + private touchTrackedThread(threadId: string): void { + const normalized = threadId.trim(); + if (normalized.length === 0) { + return; + } + if (this.trackedThreadIds.has(normalized)) { + this.trackedThreadIds.delete(normalized); + } + this.trackedThreadIds.set(normalized, true); + while (this.trackedThreadIds.size > MAX_TRACKED_THREAD_RUNTIME_STATES) { + const oldestThreadId = this.trackedThreadIds.keys().next().value; + if (typeof oldestThreadId !== "string") { + return; + } + this.trackedThreadIds.delete(oldestThreadId); + this.threadOwnerById.delete(oldestThreadId); + this.streamEventsByThreadId.delete(oldestThreadId); + this.streamSnapshotByThreadId.delete(oldestThreadId); + this.streamSnapshotOriginByThreadId.delete(oldestThreadId); + this.threadTitleById.delete(oldestThreadId); + } + this.updateRuntimeHighWaterMarks(); + } + + private setTrackedThreadOwner(threadId: string, ownerClientId: string): void { + this.touchTrackedThread(threadId); + this.threadOwnerById.set(threadId, ownerClientId); + } + + private setTrackedThreadEvents(threadId: string, events: IpcFrame[]): void { + this.touchTrackedThread(threadId); + this.streamEventsByThreadId.set( + threadId, + events.slice(-MAX_STREAM_EVENTS_PER_THREAD), + ); + this.updateRuntimeHighWaterMarks(); + } + + private setTrackedThreadSnapshot( + threadId: string, + snapshot: ThreadConversationState, + origin: StreamSnapshotOrigin, + ): void { + this.touchTrackedThread(threadId); + this.streamSnapshotByThreadId.set(threadId, snapshot); + this.streamSnapshotOriginByThreadId.set(threadId, origin); + } + private setThreadTitle( threadId: string, title: string | null | undefined, ): void { + this.touchTrackedThread(threadId); if (title === undefined) { this.threadTitleById.delete(threadId); return; @@ -1229,6 +1443,32 @@ export class CodexAgentAdapter implements AgentAdapter { this.threadTitleById.set(threadId, title); } + + private updateRuntimeHighWaterMarks(): void { + this.maxTrackedThreadCountSeen = Math.max( + this.maxTrackedThreadCountSeen, + this.trackedThreadIds.size, + ); + + let totalBufferedStreamEventCount = 0; + let maxBufferedStreamEventsPerThread = 0; + for (const events of this.streamEventsByThreadId.values()) { + totalBufferedStreamEventCount += events.length; + maxBufferedStreamEventsPerThread = Math.max( + maxBufferedStreamEventsPerThread, + events.length, + ); + } + + this.maxTotalBufferedStreamEventCountSeen = Math.max( + this.maxTotalBufferedStreamEventCountSeen, + totalBufferedStreamEventCount, + ); + this.maxBufferedStreamEventsPerThreadSeen = Math.max( + this.maxBufferedStreamEventsPerThreadSeen, + maxBufferedStreamEventsPerThread, + ); + } } function toErrorMessage(error: Error | string | unknown): string { @@ -1241,118 +1481,1182 @@ function toErrorMessage(error: Error | string | unknown): string { return String(error); } -const INVALID_REQUEST_ERROR_CODE = -32600; +export function collectThreadStreamStateChangedEvents(rawEvents: IpcFrame[]): { + events: ThreadStreamStateChangedBroadcast[]; + parseError: + | { + eventIndex: number; + message: string; + issues?: ProtocolValidationError["issues"]; + } + | null; +} { + const events: ThreadStreamStateChangedBroadcast[] = []; -export function isInvalidRequestAppServerRpcError( - error: Error | null, -): boolean { - if (!(error instanceof AppServerRpcError)) { - return false; + for (let eventIndex = 0; eventIndex < rawEvents.length; eventIndex += 1) { + const event = rawEvents[eventIndex]; + if (!event) { + continue; + } + if ( + event.type !== "broadcast" || + event.method !== "thread-stream-state-changed" + ) { + continue; + } + + try { + events.push(parseThreadStreamStateChangedBroadcast(event)); + } catch (error) { + return { + events, + parseError: { + eventIndex, + message: toErrorMessage(error), + ...(error instanceof ProtocolValidationError + ? { issues: error.issues } + : {}), + }, + }; + } } - return error.code === INVALID_REQUEST_ERROR_CODE; + + return { + events, + parseError: null, + }; } -export function isThreadNotMaterializedIncludeTurnsAppServerRpcError( - error: Error | null, -): boolean { - if (!isInvalidRequestAppServerRpcError(error)) { - return false; - } - if (!error) { - return false; - } - const normalized = error.message.trim().toLowerCase(); - return ( - normalized.includes("not materialized yet") && - normalized.includes("includeturns") - ); +interface ArchivedThreadIndexMetadata { + title: string | null; + updatedAt: number | null; } -export function isThreadNotLoadedAppServerRpcError( - error: Error | null, -): boolean { - if (!isInvalidRequestAppServerRpcError(error)) { - return false; - } - if (!error) { - return false; - } - const normalized = error.message.trim().toLowerCase(); - return normalized.includes("thread not loaded"); +interface ArchivedSessionEntry { + timestamp?: string; + type?: string; + payload?: Record; } -export function isThreadNoRolloutIncludeTurnsAppServerRpcError( - error: Error | null, -): boolean { - if (!isInvalidRequestAppServerRpcError(error)) { - return false; - } - if (!error) { - return false; +export function buildArchivedThreadConversationStateFromJsonl( + content: string, + threadId: string, + metadata?: ArchivedThreadIndexMetadata | null, +): ThreadConversationState | null { + const entries: ArchivedSessionEntry[] = []; + const lines = content.split(/\r?\n/); + for (let lineIndex = 0; lineIndex < lines.length; lineIndex += 1) { + const line = lines[lineIndex]; + const trimmed = line?.trim() ?? ""; + if (trimmed.length === 0) { + continue; + } + + try { + entries.push(JSON.parse(trimmed) as ArchivedSessionEntry); + } catch (error) { + logArchivedMalformedJsonlLine( + threadId, + lineIndex, + toErrorMessage(error), + lineIndex === lines.length - 1, + ); + } } - const normalized = error.message.trim().toLowerCase(); - return ( - normalized.includes("no rollout found for thread id") && - normalized.includes("app-server error -32600") - ); -} -export function isIpcNoClientFoundError(error: Error | null): boolean { - if (!(error instanceof DesktopIpcError)) { - return false; + if (entries.length === 0) { + return null; } - const normalized = error.message.trim().toLowerCase(); - return normalized.includes("no-client-found"); -} -function normalizeStderrLine(line: string): string { - return line.replace(ANSI_ESCAPE_REGEX, "").trim(); -} + let itemCounter = 0; + let createdAt = Number.POSITIVE_INFINITY; + let updatedAt = 0; + let cwd: string | undefined; + let source: string | undefined; + let title = metadata?.title ?? null; + let latestModel: string | null = null; + let latestReasoningEffort: string | null = null; + let latestTokenUsageInfo: ThreadConversationState["latestTokenUsageInfo"]; + let currentTurn: ThreadConversationState["turns"][number] | null = null; + const turns: ThreadConversationState["turns"] = []; + const archivedToolCallsById = new Map(); + + const nextItemId = () => `archived-item-${String(++itemCounter)}`; + + const ensureTurn = () => { + if (currentTurn) { + return currentTurn; + } -function isThreadStateGenerating(state: ThreadConversationState): boolean { - for (let index = state.turns.length - 1; index >= 0; index -= 1) { - const turn = state.turns[index]; - if (!turn) { - continue; + currentTurn = { + id: `archived-turn-${String(turns.length + 1)}`, + status: "completed", + items: [], + }; + turns.push(currentTurn); + return currentTurn; + }; + + const finalizeTurn = ( + turnId: string | null | undefined, + completedAtMs: number | null, + ) => { + if (!currentTurn) { + return; + } + if (turnId && !currentTurn.turnId) { + currentTurn.turnId = turnId; + } + currentTurn.status = "completed"; + if ( + completedAtMs !== null && + currentTurn.finalAssistantStartedAtMs === undefined + ) { + currentTurn.finalAssistantStartedAtMs = completedAtMs; } + currentTurn = null; + }; - const status = turn.status.trim().toLowerCase(); - const isTerminal = - status === "completed" || - status === "failed" || - status === "error" || - status === "cancelled" || - status === "canceled" || - status === "interrupted" || - status === "aborted"; - if (isTerminal) { - continue; + const ensureArchivedToolCall = ( + callId: string, + tool: unknown, + argumentsValue: unknown, + status: ArchivedDynamicToolCallItem["status"] = "inProgress", + ): ArchivedDynamicToolCallItem => { + const normalizedTool = normalizeArchivedToolName(tool); + const existing = archivedToolCallsById.get(callId); + if (existing) { + if (existing.tool === "legacyTool" && normalizedTool !== "legacyTool") { + existing.tool = normalizedTool; + } + if (argumentsValue !== undefined) { + existing.arguments = normalizeArchivedToolArguments(argumentsValue); + } + if (status !== "inProgress" || existing.status === "inProgress") { + existing.status = status; + } + return existing; } - return true; - } - return false; -} + const turn = ensureTurn(); + const item: ArchivedDynamicToolCallItem = { + id: nextItemId(), + type: "dynamicToolCall", + tool: normalizedTool, + arguments: normalizeArchivedToolArguments(argumentsValue), + status, + }; + turn.items.push(item); + archivedToolCallsById.set(callId, item); + return item; + }; -function deriveThreadWaitingState( - state: ThreadConversationState, -): { - waitingOnApproval: boolean; - waitingOnUserInput: boolean; -} { - let waitingOnApproval = false; - let waitingOnUserInput = false; + for (const entry of entries) { + const timestampSeconds = parseTimestampToUnixSeconds(entry.timestamp); + if (timestampSeconds !== null) { + createdAt = Math.min(createdAt, timestampSeconds); + updatedAt = Math.max(updatedAt, timestampSeconds); + } - for (const request of state.requests) { - if (request.completed === true) { + const payload = entry.payload; + const payloadType = payload?.["type"]; + if (!payload || typeof payloadType !== "string") { continue; } - switch (request.method) { - case "item/tool/requestUserInput": - waitingOnUserInput = true; + switch (payloadType) { + case "session_meta": { + const payloadCwd = payload["cwd"]; + if (typeof payloadCwd === "string" && payloadCwd.trim().length > 0) { + cwd = payloadCwd; + } + const payloadSource = payload["source"]; + if ( + typeof payloadSource === "string" && + payloadSource.trim().length > 0 + ) { + source = payloadSource; + } + title = readArchivedString(payload, [ + "thread_name", + "threadName", + "title", + "name", + ]) ?? title; + latestModel = + readArchivedString(payload, ["model", "latestModel", "model_name"]) ?? + latestModel; + latestReasoningEffort = + readArchivedString(payload, [ + "reasoning_effort", + "reasoningEffort", + "effort", + ]) ?? latestReasoningEffort; break; - case "item/commandExecution/requestApproval": + } + + case "task_started": { + if (currentTurn) { + finalizeTurn(null, null); + } + const payloadTurnId = payload["turn_id"]; + currentTurn = { + id: `archived-turn-${String(turns.length + 1)}`, + ...(typeof payloadTurnId === "string" && payloadTurnId.trim() + ? { turnId: payloadTurnId } + : {}), + status: "inProgress", + ...(timestampSeconds !== null + ? { turnStartedAtMs: timestampSeconds * 1000 } + : {}), + items: [], + }; + turns.push(currentTurn); + latestModel = + readArchivedString(payload, ["model", "latestModel", "model_name"]) ?? + latestModel; + latestReasoningEffort = + readArchivedString(payload, [ + "reasoning_effort", + "reasoningEffort", + "effort", + ]) ?? latestReasoningEffort; + break; + } + + case "user_message": { + const turn = ensureTurn(); + const contentItems: ThreadConversationState["turns"][number]["items"][number] & + { type: "userMessage" } = { + id: nextItemId(), + type: "userMessage", + content: [], + }; + + const payloadMessage = payload["message"]; + if (typeof payloadMessage === "string" && payloadMessage.length > 0) { + contentItems.content.push({ + type: "text", + text: payloadMessage, + }); + } + + const payloadImages = payload["images"]; + if (Array.isArray(payloadImages)) { + for (const image of payloadImages) { + if (typeof image === "string" && image.trim().length > 0) { + contentItems.content.push({ + type: "image", + url: image, + }); + } + } + } + + if (contentItems.content.length > 0) { + turn.items.push(contentItems); + } + break; + } + + case "message": { + const role = readArchivedString(payload, ["role"]); + if (!role) { + break; + } + + if (role === "assistant") { + const text = normalizeArchivedAssistantMessageText(payload["content"]); + if (!text) { + break; + } + const turn = ensureTurn(); + const payloadPhase = payload["phase"]; + turn.items.push({ + id: nextItemId(), + type: "agentMessage", + text, + ...(typeof payloadPhase === "string" ? { phase: payloadPhase } : {}), + }); + break; + } + + const content = normalizeArchivedMessageContent(payload["content"]); + if (content.length === 0) { + break; + } + + const turn = ensureTurn(); + if (role === "developer" || role === "system") { + turn.items.push({ + id: nextItemId(), + type: "steeringUserMessage", + content, + }); + break; + } + + turn.items.push({ + id: nextItemId(), + type: "userMessage", + content, + }); + break; + } + + case "agent_message": { + const payloadMessage = payload["message"]; + if (typeof payloadMessage !== "string" || payloadMessage.length === 0) { + break; + } + const turn = ensureTurn(); + const payloadPhase = payload["phase"]; + turn.items.push({ + id: nextItemId(), + type: "agentMessage", + text: payloadMessage, + ...(typeof payloadPhase === "string" ? { phase: payloadPhase } : {}), + }); + break; + } + + case "reasoning": { + const payloadSummary = payload["summary"]; + const summary = Array.isArray(payloadSummary) + ? payloadSummary.filter( + (entry): entry is string => typeof entry === "string", + ) + : []; + if (summary.length === 0) { + break; + } + const turn = ensureTurn(); + turn.items.push({ + id: nextItemId(), + type: "reasoning", + summary, + }); + break; + } + + case "agent_reasoning": { + const payloadText = readArchivedString(payload, ["text", "message"]); + const turn = ensureTurn(); + const payloadSummary = payload["summary"]; + const summary = Array.isArray(payloadSummary) + ? payloadSummary.filter( + (entry): entry is string => typeof entry === "string", + ) + : []; + if (summary.length === 0 && !payloadText) { + break; + } + turn.items.push({ + id: nextItemId(), + type: "reasoning", + ...(summary.length > 0 ? { summary } : {}), + ...(payloadText ? { text: payloadText } : {}), + }); + break; + } + + case "context_compacted": + case "compacted": { + const turn = ensureTurn(); + turn.items.push({ + id: nextItemId(), + type: "contextCompaction", + completed: true, + }); + break; + } + + case "token_count": { + const payloadInfo = payload["info"]; + if (payloadInfo !== undefined) { + latestTokenUsageInfo = + payloadInfo as ThreadConversationState["latestTokenUsageInfo"]; + } + break; + } + + case "web_search_call": { + const callId = + readArchivedString(payload, ["call_id", "callId"]) ?? + `archived-web-search-${entry.timestamp ?? String(itemCounter + 1)}`; + const action = payload["action"]; + const item = ensureArchivedToolCall( + callId, + "web_search", + action ?? payload, + normalizeArchivedToolStatus(payload["status"], null, "completed"), + ); + const contentItems = normalizeArchivedToolContentItems( + normalizeArchivedWebSearchContent(action), + ); + if (contentItems) { + item.contentItems = contentItems; + } + break; + } + + case "function_call": { + const callId = readArchivedString(payload, ["call_id", "callId"]); + if (!callId) { + break; + } + ensureArchivedToolCall(callId, payload["name"], payload["arguments"]); + break; + } + + case "function_call_output": { + const callId = readArchivedString(payload, ["call_id", "callId"]); + if (!callId) { + break; + } + const output = payload["output"]; + const success = normalizeArchivedToolSuccess(output, true); + const item = ensureArchivedToolCall( + callId, + payload["name"], + undefined, + normalizeArchivedToolStatus(undefined, success, "completed"), + ); + const contentItems = normalizeArchivedToolContentItems(output); + if (contentItems) { + item.contentItems = contentItems; + } + if (success !== null) { + item.success = success; + } + break; + } + + case "custom_tool_call": { + const callId = readArchivedString(payload, ["call_id", "callId"]); + if (!callId) { + break; + } + ensureArchivedToolCall( + callId, + payload["name"], + payload["input"], + normalizeArchivedToolStatus(payload["status"], null, "inProgress"), + ); + break; + } + + case "custom_tool_call_output": { + const callId = readArchivedString(payload, ["call_id", "callId"]); + if (!callId) { + break; + } + const output = payload["output"]; + const success = normalizeArchivedToolSuccess(output, true); + const item = ensureArchivedToolCall( + callId, + payload["name"], + undefined, + normalizeArchivedToolStatus(undefined, success, "completed"), + ); + const contentItems = normalizeArchivedToolContentItems(output); + if (contentItems) { + item.contentItems = contentItems; + } + if (success !== null) { + item.success = success; + } + break; + } + + case "dynamic_tool_call_request": { + const callId = readArchivedString(payload, ["call_id", "callId"]); + if (!callId) { + break; + } + ensureArchivedToolCall(callId, payload["tool"], payload["arguments"]); + break; + } + + case "dynamic_tool_call_response": { + const callId = readArchivedString(payload, ["call_id", "callId"]); + if (!callId) { + break; + } + const errorText = readArchivedString(payload, ["error"]); + const success = normalizeArchivedToolSuccess( + payload["success"], + errorText ? false : null, + ); + const item = ensureArchivedToolCall( + callId, + payload["tool"], + payload["arguments"], + normalizeArchivedToolStatus(payload["status"], success, "completed"), + ); + const contentItems = + normalizeArchivedToolContentItems(payload["content_items"]) ?? + (errorText + ? [ + { + type: "inputText", + text: truncateArchivedTextPreview( + errorText, + ARCHIVED_TOOL_TEXT_PREVIEW_MAX_CHARS, + ), + }, + ] + : null); + if (contentItems) { + item.contentItems = contentItems; + } + if (success !== null) { + item.success = success; + } + const durationMs = parseArchivedDurationToMs(payload["duration"]); + if (durationMs !== null) { + item.durationMs = durationMs; + } + break; + } + + case "todo_list": + case "todo-list": + case "todoList": + case "plan_update": { + const plan = normalizeArchivedTodoPlan(payload["plan"]); + if (plan.length === 0) { + break; + } + const explanation = payload["explanation"]; + const turn = ensureTurn(); + turn.items.push({ + id: nextItemId(), + type: "todoList", + ...(typeof explanation === "string" || explanation === null + ? { explanation } + : {}), + plan, + }); + break; + } + + case "thread_name_updated": { + title = + readArchivedString(payload, [ + "thread_name", + "threadName", + "title", + "name", + ]) ?? title; + break; + } + + case "task_complete": { + const payloadTurnId = payload["turn_id"]; + finalizeTurn( + typeof payloadTurnId === "string" ? payloadTurnId : null, + timestampSeconds !== null ? timestampSeconds * 1000 : null, + ); + break; + } + + default: + logArchivedUnknownPayloadType(threadId, payloadType); + break; + } + } + + if (currentTurn) { + currentTurn.status = "completed"; + currentTurn = null; + } + + if (turns.length === 0 && latestTokenUsageInfo === undefined) { + return null; + } + + const normalizedCreatedAt = Number.isFinite(createdAt) + ? createdAt + : metadata?.updatedAt ?? 0; + const normalizedUpdatedAt = + updatedAt > 0 ? updatedAt : metadata?.updatedAt ?? normalizedCreatedAt; + + return { + id: threadId, + turns, + requests: [], + ...(normalizedCreatedAt > 0 ? { createdAt: normalizedCreatedAt } : {}), + ...(normalizedUpdatedAt > 0 ? { updatedAt: normalizedUpdatedAt } : {}), + ...(title !== undefined ? { title } : {}), + latestModel, + latestReasoningEffort, + ...(cwd ? { cwd } : {}), + ...(source ? { source } : {}), + ...(latestTokenUsageInfo !== undefined ? { latestTokenUsageInfo } : {}), + }; +} + +function readArchivedString( + payload: Record, + keys: string[], +): string | null { + for (const key of keys) { + const value = payload[key]; + if (typeof value !== "string") { + continue; + } + const normalized = value.trim(); + if (normalized.length > 0) { + return normalized; + } + } + return null; +} + +function logArchivedUnknownPayloadType( + threadId: string, + payloadType: string, +): void { + if (loggedArchivedUnknownPayloadTypes.has(payloadType)) { + return; + } + loggedArchivedUnknownPayloadTypes.add(payloadType); + logger.warn( + { + threadId, + payloadType, + }, + "archived-thread-unknown-payload-type", + ); +} + +function logArchivedMalformedJsonlLine( + threadId: string, + lineIndex: number, + error: string, + truncatedTailCandidate: boolean, +): void { + const key = `${threadId}:${String(lineIndex)}`; + if (loggedArchivedMalformedJsonlLineKeys.has(key)) { + return; + } + loggedArchivedMalformedJsonlLineKeys.add(key); + logger.warn( + { + threadId, + lineIndex, + error, + truncatedTailCandidate, + }, + "archived-thread-jsonl-line-parse-failed", + ); +} + +function normalizeArchivedMessageContent( + value: unknown, +): Array<{ type: "text"; text: string } | { type: "image"; url: string }> { + if (!Array.isArray(value)) { + return []; + } + + return value.flatMap<{ type: "text"; text: string } | { type: "image"; url: string }>((entry) => { + if (!entry || typeof entry !== "object") { + return []; + } + + const record = entry as Record; + const rawType = readArchivedString(record, ["type"]); + if (!rawType) { + return []; + } + + switch (rawType) { + case "input_text": + case "output_text": { + const text = readArchivedString(record, ["text"]); + return text + ? [ + { + type: "text", + text, + }, + ] + : []; + } + + case "input_image": + case "output_image": { + const url = readArchivedString(record, ["image_url", "imageUrl", "url"]); + return url + ? [ + { + type: "image", + url, + }, + ] + : []; + } + + default: + return []; + } + }); +} + +function normalizeArchivedAssistantMessageText(value: unknown): string | null { + const textParts = normalizeArchivedMessageContent(value) + .filter((item): item is { type: "text"; text: string } => item.type === "text") + .map((item) => item.text); + if (textParts.length === 0) { + return null; + } + return textParts.join("\n\n"); +} + +function normalizeArchivedToolName(value: unknown): string { + if (typeof value !== "string") { + return "legacyTool"; + } + const normalized = value.trim(); + return normalized.length > 0 ? normalized : "legacyTool"; +} + +function normalizeArchivedToolArguments(value: unknown): ArchivedJsonValue { + if (typeof value === "string") { + const parsed = tryParseArchivedJson(value); + if (parsed !== undefined) { + return normalizeArchivedToolJsonValue(parsed); + } + } + return normalizeArchivedToolJsonValue(value); +} + +function normalizeArchivedToolJsonValue( + value: unknown, + maxLength = ARCHIVED_TOOL_ARGUMENT_PREVIEW_MAX_CHARS, +): ArchivedJsonValue { + const jsonValue = coerceArchivedJsonValue(value); + if (jsonValue === undefined) { + return truncateArchivedTextPreview(String(value), maxLength); + } + + const serialized = + typeof jsonValue === "string" ? jsonValue : JSON.stringify(jsonValue); + if (!serialized || serialized.length <= maxLength) { + return jsonValue; + } + + return `${serialized.slice(0, maxLength)}...`; +} + +function coerceArchivedJsonValue(value: unknown): ArchivedJsonValue | undefined { + if ( + value === null || + typeof value === "string" || + typeof value === "number" || + typeof value === "boolean" + ) { + return value; + } + + if (Array.isArray(value)) { + const result: ArchivedJsonValue[] = []; + for (const entry of value) { + const normalized = coerceArchivedJsonValue(entry); + if (normalized === undefined) { + continue; + } + result.push(normalized); + } + return result; + } + + if (typeof value !== "object") { + return undefined; + } + + const result: Record = {}; + for (const [key, nestedValue] of Object.entries( + value as Record, + )) { + const normalized = coerceArchivedJsonValue(nestedValue); + if (normalized === undefined) { + continue; + } + result[key] = normalized; + } + return result; +} + +function tryParseArchivedJson(value: string): unknown | undefined { + const trimmed = value.trim(); + if (trimmed.length === 0) { + return undefined; + } + + if (!/^[{\["0-9tfn-]/.test(trimmed)) { + return undefined; + } + + try { + return JSON.parse(trimmed); + } catch { + return undefined; + } +} + +function normalizeArchivedToolSuccess( + value: unknown, + fallback: boolean | null, +): boolean | null { + if (typeof value === "boolean") { + return value; + } + + if (value && typeof value === "object") { + const nestedSuccess = (value as Record)["success"]; + if (typeof nestedSuccess === "boolean") { + return nestedSuccess; + } + } + + return fallback; +} + +function normalizeArchivedToolStatus( + value: unknown, + success: boolean | null, + fallback: ArchivedDynamicToolCallItem["status"], +): ArchivedDynamicToolCallItem["status"] { + if (typeof value === "string") { + const normalized = value.trim().toLowerCase().replace(/[-_\s]+/g, ""); + if ( + normalized === "inprogress" || + normalized === "running" || + normalized === "pending" + ) { + return "inProgress"; + } + if ( + normalized === "completed" || + normalized === "complete" || + normalized === "success" || + normalized === "succeeded" + ) { + return "completed"; + } + if ( + normalized === "failed" || + normalized === "failure" || + normalized === "error" || + normalized === "errored" + ) { + return "failed"; + } + } + + if (success === true) { + return "completed"; + } + if (success === false) { + return "failed"; + } + return fallback; +} + +function normalizeArchivedToolContentItems( + value: unknown, +): ArchivedDynamicToolContentItem[] | null { + const body = + value && typeof value === "object" && !Array.isArray(value) + ? ((value as Record)["body"] ?? value) + : value; + + if (Array.isArray(body)) { + const items = body + .map(normalizeArchivedToolContentItem) + .filter( + (item): item is ArchivedDynamicToolContentItem => item !== null, + ); + return items.length > 0 ? items : null; + } + + if (body === null || body === undefined) { + return null; + } + + return [normalizeFallbackArchivedToolContentItem(body)]; +} + +function normalizeArchivedToolContentItem( + value: unknown, +): ArchivedDynamicToolContentItem | null { + if (!value || typeof value !== "object") { + return normalizeFallbackArchivedToolContentItem(value); + } + + const record = value as Record; + const rawType = readArchivedString(record, ["type"]); + if (!rawType) { + return normalizeFallbackArchivedToolContentItem(value); + } + + switch (rawType) { + case "inputText": + case "input_text": { + const text = readArchivedString(record, ["text"]); + if (!text) { + return null; + } + return { + type: "inputText", + text: truncateArchivedTextPreview( + text, + ARCHIVED_TOOL_TEXT_PREVIEW_MAX_CHARS, + ), + }; + } + + case "inputImage": + case "input_image": { + const imageUrl = readArchivedString(record, ["imageUrl", "image_url"]); + if (!imageUrl) { + return null; + } + return { + type: "inputImage", + imageUrl, + }; + } + + default: + return normalizeFallbackArchivedToolContentItem(value); + } +} + +function normalizeFallbackArchivedToolContentItem( + value: unknown, +): ArchivedDynamicToolContentItem { + return { + type: "inputText", + text: truncateArchivedTextPreview( + typeof value === "string" ? value : JSON.stringify(value), + ARCHIVED_TOOL_TEXT_PREVIEW_MAX_CHARS, + ), + }; +} + +function normalizeArchivedWebSearchContent( + value: unknown, +): Array<{ type: "inputText"; text: string }> | null { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return null; + } + + const record = value as Record; + const query = readArchivedString(record, ["query"]); + const queriesValue = record["queries"]; + const queries = Array.isArray(queriesValue) + ? queriesValue.filter( + (entry): entry is string => + typeof entry === "string" && entry.trim().length > 0, + ) + : []; + const lines = query ? [query, ...queries] : queries; + const uniqueLines = [...new Set(lines)]; + if (uniqueLines.length === 0) { + return null; + } + + return uniqueLines.map((line) => ({ + type: "inputText", + text: truncateArchivedTextPreview( + line, + ARCHIVED_TOOL_TEXT_PREVIEW_MAX_CHARS, + ), + })); +} + +function truncateArchivedTextPreview(value: string, maxLength: number): string { + if (value.length <= maxLength) { + return value; + } + return `${value.slice(0, maxLength)}...`; +} + +function normalizeArchivedTodoPlan( + value: unknown, +): ArchivedTodoListItem["plan"] { + if (!Array.isArray(value)) { + return []; + } + + return value.flatMap((entry) => { + if (!entry || typeof entry !== "object") { + return []; + } + const record = entry as Record; + const step = readArchivedString(record, ["step"]); + const status = readArchivedString(record, ["status"]); + if (!step || !status) { + return []; + } + return [{ step, status }]; + }); +} + +function parseArchivedDurationToMs(value: unknown): number | null { + if (typeof value === "number" && Number.isFinite(value) && value >= 0) { + return Math.round(value); + } + + if (typeof value !== "string") { + return null; + } + + const trimmed = value.trim(); + if (trimmed.length === 0) { + return null; + } + + const numeric = Number(trimmed); + if (Number.isFinite(numeric) && numeric >= 0) { + return Math.round(numeric); + } + + const match = /^PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?$/i.exec( + trimmed, + ); + if (!match) { + return null; + } + + const hours = Number(match[1] ?? 0); + const minutes = Number(match[2] ?? 0); + const seconds = Number(match[3] ?? 0); + if ( + !Number.isFinite(hours) || + !Number.isFinite(minutes) || + !Number.isFinite(seconds) + ) { + return null; + } + return Math.round(((hours * 60 * 60) + (minutes * 60) + seconds) * 1000); +} + +const INVALID_REQUEST_ERROR_CODE = -32600; + +export function isInvalidRequestAppServerRpcError( + error: Error | null, +): boolean { + if (!(error instanceof AppServerRpcError)) { + return false; + } + return error.code === INVALID_REQUEST_ERROR_CODE; +} + +export function isThreadNotMaterializedIncludeTurnsAppServerRpcError( + error: Error | null, +): boolean { + if (!isInvalidRequestAppServerRpcError(error)) { + return false; + } + if (!error) { + return false; + } + const normalized = error.message.trim().toLowerCase(); + return ( + normalized.includes("not materialized yet") && + normalized.includes("includeturns") + ); +} + +export function isThreadNotLoadedAppServerRpcError( + error: Error | null, +): boolean { + if (!isInvalidRequestAppServerRpcError(error)) { + return false; + } + if (!error) { + return false; + } + const normalized = error.message.trim().toLowerCase(); + return normalized.includes("thread not loaded"); +} + +export function isThreadNoRolloutIncludeTurnsAppServerRpcError( + error: Error | null, +): boolean { + if (!isInvalidRequestAppServerRpcError(error)) { + return false; + } + if (!error) { + return false; + } + const normalized = error.message.trim().toLowerCase(); + return ( + normalized.includes("no rollout found for thread id") && + normalized.includes("app-server error -32600") + ); +} + +export function isIpcNoClientFoundError(error: Error | null): boolean { + if (!(error instanceof DesktopIpcError)) { + return false; + } + const normalized = error.message.trim().toLowerCase(); + return normalized.includes("no-client-found"); +} + +function normalizeStderrLine(line: string): string { + return line.replace(ANSI_ESCAPE_REGEX, "").trim(); +} + +function isThreadStateGenerating(state: ThreadConversationState): boolean { + for (let index = state.turns.length - 1; index >= 0; index -= 1) { + const turn = state.turns[index]; + if (!turn) { + continue; + } + + const status = turn.status.trim().toLowerCase(); + const isTerminal = + status === "completed" || + status === "failed" || + status === "error" || + status === "cancelled" || + status === "canceled" || + status === "interrupted" || + status === "aborted"; + if (isTerminal) { + continue; + } + return true; + } + + return false; +} + +function deriveThreadWaitingState( + state: ThreadConversationState, +): { + waitingOnApproval: boolean; + waitingOnUserInput: boolean; +} { + let waitingOnApproval = false; + let waitingOnUserInput = false; + + for (const request of state.requests) { + if (request.completed === true) { + continue; + } + + switch (request.method) { + case "item/tool/requestUserInput": + waitingOnUserInput = true; + break; + case "item/commandExecution/requestApproval": case "item/fileChange/requestApproval": case "applyPatchApproval": case "execCommandApproval": @@ -1464,3 +2768,135 @@ function extractThreadId(frame: IpcFrame): string | null { return null; } + +async function readThreadIndexMetadata( + threadId: string, +): Promise { + const codexHomes = getCodexHomeCandidates(); + for (const codexHome of codexHomes) { + const indexPath = path.join(codexHome, "session_index.jsonl"); + try { + const content = await readFile(indexPath, "utf8"); + for (const line of content.split(/\r?\n/)) { + const trimmed = line.trim(); + if (!trimmed) { + continue; + } + const parsed = JSON.parse(trimmed) as Record; + if (parsed["id"] !== threadId) { + continue; + } + const parsedThreadName = parsed["thread_name"]; + const parsedUpdatedAt = parsed["updated_at"]; + return { + title: + typeof parsedThreadName === "string" ? parsedThreadName : null, + updatedAt: parseTimestampToUnixSeconds( + typeof parsedUpdatedAt === "string" ? parsedUpdatedAt : undefined, + ), + }; + } + } catch { + continue; + } + } + + return null; +} + +async function findArchivedRolloutPath(threadId: string): Promise { + for (const codexHome of getCodexHomeCandidates()) { + const archivedSessionsDir = path.join(codexHome, "archived_sessions"); + const archivedRollout = await findRolloutPathInDirectory( + archivedSessionsDir, + threadId, + 0, + ); + if (archivedRollout) { + return archivedRollout; + } + + const sessionsDir = path.join(codexHome, "sessions"); + const activeRollout = await findRolloutPathInDirectory(sessionsDir, threadId, 4); + if (activeRollout) { + return activeRollout; + } + } + + return null; +} + +async function findRolloutPathInDirectory( + directory: string, + threadId: string, + remainingDepth: number, +): Promise { + try { + await access(directory); + } catch { + return null; + } + + const entries = await readdir(directory, { withFileTypes: true }); + + for (const entry of entries) { + if (!entry.isFile()) { + continue; + } + if (!entry.name.endsWith(".jsonl") || !entry.name.includes(threadId)) { + continue; + } + return path.join(directory, entry.name); + } + + if (remainingDepth <= 0) { + return null; + } + + for (const entry of entries) { + if (!entry.isDirectory()) { + continue; + } + const nested = await findRolloutPathInDirectory( + path.join(directory, entry.name), + threadId, + remainingDepth - 1, + ); + if (nested) { + return nested; + } + } + + return null; +} + +function getCodexHomeCandidates(): string[] { + const candidates = [ + process.env["CODEX_HOME"], + process.env["USERPROFILE"] + ? path.join(process.env["USERPROFILE"], ".codex") + : null, + process.env["HOME"] ? path.join(process.env["HOME"], ".codex") : null, + path.join(os.homedir(), ".codex"), + ]; + + const unique = new Set(); + for (const candidate of candidates) { + if (!candidate || candidate.trim().length === 0) { + continue; + } + unique.add(candidate); + } + return [...unique]; +} + +function parseTimestampToUnixSeconds(value: string | undefined): number | null { + if (!value) { + return null; + } + const milliseconds = Date.parse(value); + if (!Number.isFinite(milliseconds)) { + return null; + } + return Math.floor(milliseconds / 1000); +} diff --git a/apps/server/src/debug-history.ts b/apps/server/src/debug-history.ts new file mode 100644 index 00000000..b93498eb --- /dev/null +++ b/apps/server/src/debug-history.ts @@ -0,0 +1,367 @@ +const MAX_HISTORY_STRING_LENGTH = 512; +const MAX_HISTORY_ARRAY_ITEMS = 24; +const MAX_HISTORY_OBJECT_KEYS = 24; +const MAX_HISTORY_TURN_PREVIEW_COUNT = 16; +const MAX_HISTORY_ITEM_PREVIEW_COUNT = 20; +const MAX_HISTORY_PATCH_PREVIEW_COUNT = 24; + +export interface StoredHistoryPayload { + json: string; + originalBytes: number; + storedBytes: number; + compacted: boolean; +} + +export function prepareHistoryPayloadForStorage( + payload: unknown, +): StoredHistoryPayload { + const originalJson = safeJsonStringify(payload); + let compactedPayload: unknown; + try { + compactedPayload = compactHistoryPayload(payload); + } catch { + compactedPayload = payload; + } + const compactedJson = safeJsonStringify(compactedPayload); + const originalBytes = Buffer.byteLength(originalJson, "utf8"); + const storedBytes = Buffer.byteLength(compactedJson, "utf8"); + + return { + json: compactedJson, + originalBytes, + storedBytes, + compacted: compactedJson !== originalJson, + }; +} + +function compactHistoryPayload(payload: unknown): unknown { + if (!isRecord(payload)) { + return compactJsonValue(payload, 0); + } + + if (payload["method"] === "thread-stream-state-changed") { + return compactThreadStreamStateChangedPayload(payload); + } + + return compactJsonValue(payload, 0); +} + +function compactThreadStreamStateChangedPayload( + payload: Record, +): Record { + const params = asRecord(payload["params"]); + const change = asRecord(params?.["change"]); + + if (!params || !change) { + return compactJsonValue(payload, 0) as Record; + } + + if (change["type"] === "snapshot") { + const conversationState = asRecord(change["conversationState"]); + if (!conversationState) { + return compactJsonValue(payload, 0) as Record; + } + + return { + ...payload, + params: { + ...params, + change: { + ...change, + conversationState: compactConversationState(conversationState), + }, + }, + }; + } + + if (change["type"] === "patches") { + const patches = Array.isArray(change["patches"]) ? change["patches"] : []; + return { + ...payload, + params: { + ...params, + change: { + ...change, + patchCount: patches.length, + patches: patches + .slice(0, MAX_HISTORY_PATCH_PREVIEW_COUNT) + .map((patch, index) => compactPatch(patch, index)), + ...(patches.length > MAX_HISTORY_PATCH_PREVIEW_COUNT + ? { + truncatedPatchCount: + patches.length - MAX_HISTORY_PATCH_PREVIEW_COUNT, + } + : {}), + }, + }, + }; + } + + return compactJsonValue(payload, 0) as Record; +} + +function compactConversationState( + conversationState: Record, +): Record { + const turns = Array.isArray(conversationState["turns"]) + ? conversationState["turns"] + : []; + const requests = Array.isArray(conversationState["requests"]) + ? conversationState["requests"] + : []; + const retainedTurns = turns.slice( + Math.max(0, turns.length - MAX_HISTORY_TURN_PREVIEW_COUNT), + ); + + return { + ...conversationState, + turnCount: turns.length, + requestCount: requests.length, + turns: retainedTurns.map((turn, index) => + compactTurn(turn, retainedTurns.length - 1 - index), + ), + requests: compactArray(requests, 1), + ...(turns.length > MAX_HISTORY_TURN_PREVIEW_COUNT + ? { + truncatedTurnCount: + turns.length - MAX_HISTORY_TURN_PREVIEW_COUNT, + } + : {}), + }; +} + +function compactTurn(turn: unknown, reverseIndex: number): unknown { + if (!isRecord(turn)) { + return compactJsonValue(turn, 0); + } + + const items = Array.isArray(turn["items"]) ? turn["items"] : []; + return { + ...(typeof turn["turnId"] === "string" ? { turnId: turn["turnId"] } : {}), + ...(typeof turn["status"] === "string" ? { status: turn["status"] } : {}), + ...(turn["error"] !== undefined + ? { error: compactJsonValue(turn["error"], 1) } + : {}), + itemCount: items.length, + itemTypes: items.map((item) => + isRecord(item) && typeof item["type"] === "string" + ? item["type"] + : "", + ), + items: items + .slice(Math.max(0, items.length - MAX_HISTORY_ITEM_PREVIEW_COUNT)) + .map((item) => compactTurnItem(item)), + ...(items.length > MAX_HISTORY_ITEM_PREVIEW_COUNT + ? { + truncatedItemCount: + items.length - MAX_HISTORY_ITEM_PREVIEW_COUNT, + } + : {}), + ...(turn["params"] !== undefined + ? { params: compactJsonValue(turn["params"], 1) } + : {}), + ...(turn["turnStartedAtMs"] !== undefined + ? { turnStartedAtMs: turn["turnStartedAtMs"] } + : {}), + ...(turn["finalAssistantStartedAtMs"] !== undefined + ? { finalAssistantStartedAtMs: turn["finalAssistantStartedAtMs"] } + : {}), + turnOffsetFromLatest: reverseIndex, + }; +} + +function compactTurnItem(item: unknown): unknown { + if (!isRecord(item)) { + return compactJsonValue(item, 0); + } + + const base = { + ...(typeof item["id"] === "string" ? { id: item["id"] } : {}), + ...(typeof item["type"] === "string" ? { type: item["type"] } : {}), + }; + + switch (item["type"]) { + case "userMessage": + case "steeringUserMessage": + return { + ...base, + ...(item["attachments"] !== undefined + ? { attachments: compactJsonValue(item["attachments"], 1) } + : {}), + content: compactJsonValue(item["content"], 1), + }; + case "agentMessage": + return { + ...base, + ...(typeof item["phase"] === "string" + ? { phase: item["phase"] } + : {}), + ...(typeof item["text"] === "string" + ? { text: compactString(item["text"]) } + : {}), + }; + case "reasoning": + return { + ...base, + ...(item["summary"] !== undefined + ? { summary: compactJsonValue(item["summary"], 1) } + : {}), + ...(typeof item["text"] === "string" + ? { text: compactString(item["text"]) } + : {}), + }; + case "error": + return { + ...base, + ...(typeof item["message"] === "string" + ? { message: compactString(item["message"]) } + : {}), + ...(item["errorInfo"] !== undefined + ? { errorInfo: compactJsonValue(item["errorInfo"], 1) } + : {}), + ...(item["additionalDetails"] !== undefined + ? { + additionalDetails: compactJsonValue( + item["additionalDetails"], + 1, + ), + } + : {}), + }; + default: + return compactJsonValue(item, 1); + } +} + +function compactPatch(patch: unknown, index: number): unknown { + if (!isRecord(patch)) { + return compactJsonValue(patch, 0); + } + + return { + index, + ...(typeof patch["op"] === "string" ? { op: patch["op"] } : {}), + ...(patch["path"] !== undefined + ? { path: compactJsonValue(patch["path"], 1) } + : {}), + ...(patch["value"] !== undefined + ? { value: compactJsonValue(patch["value"], 1) } + : {}), + ...(patch["from"] !== undefined + ? { from: compactJsonValue(patch["from"], 1) } + : {}), + }; +} + +function compactJsonValue(value: unknown, depth: number): unknown { + if (value === null || value === undefined) { + return value; + } + + if (typeof value === "string") { + return compactString(value); + } + + if ( + typeof value === "number" || + typeof value === "boolean" + ) { + return value; + } + + if (typeof value === "bigint") { + return value.toString(); + } + + if (Array.isArray(value)) { + return compactArray(value, depth + 1); + } + + if (!isRecord(value)) { + return value; + } + + const entries = Object.entries(value); + const limitedEntries = entries.slice(0, MAX_HISTORY_OBJECT_KEYS); + const result: Record = {}; + + for (const [key, nestedValue] of limitedEntries) { + result[key] = compactJsonValue(nestedValue, depth + 1); + } + + if (entries.length > MAX_HISTORY_OBJECT_KEYS) { + result["__truncatedKeyCount"] = + entries.length - MAX_HISTORY_OBJECT_KEYS; + } + + if (depth >= 6) { + result["__truncatedDepth"] = true; + } + + return result; +} + +function compactArray(values: unknown[], depth: number): unknown[] { + const limitedValues = values.slice(0, MAX_HISTORY_ARRAY_ITEMS); + const result = limitedValues.map((value) => compactJsonValue(value, depth + 1)); + + if (values.length > MAX_HISTORY_ARRAY_ITEMS) { + result.push({ + __truncatedArrayItems: values.length - MAX_HISTORY_ARRAY_ITEMS, + }); + } + + return result; +} + +function compactString(value: string): string { + if (value.length <= MAX_HISTORY_STRING_LENGTH) { + return value; + } + + const truncatedLength = value.length - MAX_HISTORY_STRING_LENGTH; + return `${value.slice(0, MAX_HISTORY_STRING_LENGTH)}... [${truncatedLength} chars truncated]`; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function asRecord(value: unknown): Record | null { + return isRecord(value) ? value : null; +} + +function safeJsonStringify(value: unknown): string { + const rootValue = normalizeJsonRootValue(value); + const seen = new WeakSet(); + const serialized = JSON.stringify( + rootValue, + (_key, nestedValue) => { + if (typeof nestedValue === "bigint") { + return nestedValue.toString(); + } + if (typeof nestedValue === "object" && nestedValue !== null) { + if (seen.has(nestedValue)) { + return "[Circular]"; + } + seen.add(nestedValue); + } + return nestedValue; + }, + 2, + ); + return serialized ?? "null"; +} + +function normalizeJsonRootValue(value: unknown): unknown { + if (value === undefined) { + return null; + } + if (typeof value === "bigint") { + return value.toString(); + } + if (typeof value === "function" || typeof value === "symbol") { + return String(value); + } + return value; +} diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 60c388fe..691432c6 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -16,6 +16,7 @@ import { TraceMarkBodySchema, TraceStartBodySchema, } from "./http-schemas.js"; +import { prepareHistoryPayloadForStorage } from "./debug-history.js"; import { logger } from "./logger.js"; import { parseServerCliOptions, @@ -34,7 +35,7 @@ import { const HOST = process.env["HOST"] ?? "127.0.0.1"; const PORT = Number(process.env["PORT"] ?? 4311); -const HISTORY_LIMIT = 2_000; +const HISTORY_LIMIT = 400; const USER_AGENT = "farfield/0.2.2"; const IPC_RECONNECT_DELAY_MS = 1_000; const SIDEBAR_PREVIEW_MAX_CHARS = 180; @@ -47,7 +48,6 @@ interface HistoryEntry { at: string; source: "ipc" | "app" | "system"; direction: "in" | "out" | "system"; - payload: unknown; meta: Record; } @@ -78,6 +78,31 @@ function resolveCodexExecutablePath(): string { return process.env["CODEX_CLI_PATH"]; } + if (process.platform === "win32") { + const windowsCandidates = [ + "codex.ps1", + "codex.cmd", + "codex", + "codex.exe", + ]; + + for (const candidate of windowsCandidates) { + try { + const resolved = execFileSync("where.exe", [candidate], { + encoding: "utf8", + }) + .split(/\r?\n/) + .map((entry) => entry.trim()) + .find((entry) => entry.length > 0); + if (resolved) { + return resolved; + } + } catch { + // Try the next candidate. + } + } + } + const desktopPath = "/Applications/Codex.app/Contents/Resources/codex"; if (fs.existsSync(desktopPath)) { return desktopPath; @@ -238,7 +263,7 @@ const ipcSocketPath = resolveIpcSocketPath(); const gitCommit = resolveGitCommitHash(); const history: HistoryEntry[] = []; -const historyById = new Map(); +const historyById = new Map(); const unifiedSseClients = new Set(); const activeSockets = new Set(); const SSE_KEEPALIVE_INTERVAL_MS = 15_000; @@ -263,17 +288,22 @@ function pushHistory( payload: unknown, meta: Record = {}, ): HistoryEntry { + const storedPayload = prepareHistoryPayloadForStorage(payload); const entry: HistoryEntry = { id: randomUUID(), at: new Date().toISOString(), source, direction, - payload, - meta, + meta: { + ...meta, + payloadBytes: storedPayload.originalBytes, + storedPayloadBytes: storedPayload.storedBytes, + payloadCompacted: storedPayload.compacted, + }, }; history.push(entry); - historyById.set(entry.id, payload); + historyById.set(entry.id, storedPayload.json); if (history.length > HISTORY_LIMIT) { const removed = history.shift(); @@ -282,7 +312,11 @@ function pushHistory( } } - recordTraceEvent({ type: "history", ...entry }); + recordTraceEvent({ + type: "history", + ...entry, + payloadJson: storedPayload.json, + }); return entry; } @@ -336,6 +370,7 @@ const unifiedAdapters = createUnifiedProviderAdapters({ function getRuntimeStateSnapshot(): Record { const codexRuntimeState = codexAdapter?.getRuntimeState(); + const codexRuntimeCounts = codexAdapter?.getTrackedThreadRuntimeCounts(); return { appExecutable: codexExecutable, @@ -348,6 +383,25 @@ function getRuntimeStateSnapshot(): Record { lastError: runtimeLastError ?? codexRuntimeState?.lastError ?? null, historyCount: history.length, threadOwnerCount: codexAdapter?.getThreadOwnerCount() ?? 0, + trackedThreadCount: codexRuntimeCounts?.trackedThreadCount ?? 0, + streamEventThreadCount: codexRuntimeCounts?.streamEventThreadCount ?? 0, + totalBufferedStreamEventCount: + codexRuntimeCounts?.totalBufferedStreamEventCount ?? 0, + streamSnapshotThreadCount: + codexRuntimeCounts?.streamSnapshotThreadCount ?? 0, + threadTitleCount: codexRuntimeCounts?.threadTitleCount ?? 0, + maxTrackedThreadCountSeen: + codexRuntimeCounts?.maxTrackedThreadCountSeen ?? 0, + maxTotalBufferedStreamEventCountSeen: + codexRuntimeCounts?.maxTotalBufferedStreamEventCountSeen ?? 0, + maxBufferedStreamEventsPerThreadSeen: + codexRuntimeCounts?.maxBufferedStreamEventsPerThreadSeen ?? 0, + archivedRestoreHitCount: codexRuntimeCounts?.archivedRestoreHitCount ?? 0, + archivedRestoreFailureCount: + codexRuntimeCounts?.archivedRestoreFailureCount ?? 0, + streamParseFailureCount: codexRuntimeCounts?.streamParseFailureCount ?? 0, + streamReductionFailureCount: + codexRuntimeCounts?.streamReductionFailureCount ?? 0, activeTrace: activeTrace?.summary ?? null, }; } @@ -818,6 +872,10 @@ const server = http.createServer(async (req, res) => { url.searchParams.get("includeTurns"), true, ); + const maxRenderableItems = parseInteger( + url.searchParams.get("maxRenderableItems"), + 0, + ); const knownProviders = threadIndex.providers(threadId); const resolvedProvider = threadIndex.resolve(threadId); const provider = providerFromQuery ?? resolvedProvider; @@ -872,6 +930,9 @@ const server = http.createServer(async (req, res) => { provider, threadId, includeTurns, + ...(maxRenderableItems > 0 + ? { maxRenderableItems } + : {}), }); threadIndex.register(result.thread.id, result.thread.provider); @@ -890,6 +951,9 @@ const server = http.createServer(async (req, res) => { provider, threadId, includeTurns, + ...(maxRenderableItems > 0 + ? { maxRenderableItems } + : {}), }, }, }); @@ -912,7 +976,7 @@ const server = http.createServer(async (req, res) => { jsonResponse(res, 200, { ok: true, entry: toDebugHistoryListEntry(entry), - fullPayloadJson: JSON.stringify(historyById.get(entryId) ?? null, null, 2), + fullPayloadJson: historyById.get(entryId) ?? "null", }); return; } diff --git a/apps/server/src/unified/adapter.ts b/apps/server/src/unified/adapter.ts index 563422a5..099e1a96 100644 --- a/apps/server/src/unified/adapter.ts +++ b/apps/server/src/unified/adapter.ts @@ -59,6 +59,10 @@ export const FEATURE_ID_BY_COMMAND_KIND: Record< listProjectDirectories: "listProjectDirectories", }; +const DYNAMIC_TOOL_ARGUMENT_PREVIEW_MAX_CHARS = 600; +const MCP_TOOL_ARGUMENT_PREVIEW_MAX_CHARS = 600; +const MCP_TOOL_STRUCTURED_CONTENT_PREVIEW_MAX_CHARS = 600; + const PROVIDER_FEATURE_SUPPORT: Record< UnifiedProviderId, Record @@ -93,6 +97,105 @@ const PROVIDER_FEATURE_SUPPORT: Record< }, }; +function humanizeCollaborationModeLabel(value: string): string { + const trimmed = value.trim(); + if (trimmed.length === 0) { + return "Default"; + } + + return trimmed + .replace(/[-_]+/g, " ") + .replace(/([a-z0-9])([A-Z])/g, "$1 $2") + .replace(/\s+/g, " ") + .trim() + .replace(/\b\w/g, (match) => match.toUpperCase()); +} + +function hasRenderableUnifiedUserMessageContent( + item: Extract, +): boolean { + return item.content.some((part) => { + if (part.type === "text") { + return part.text.length > 0; + } + return part.url.length > 0; + }); +} + +function shouldCountRenderableUnifiedItem(item: UnifiedItem): boolean { + switch (item.type) { + case "userMessage": + case "steeringUserMessage": + return hasRenderableUnifiedUserMessageContent(item); + case "agentMessage": + return item.text.length > 0; + case "reasoning": + return (item.summary?.length ?? 0) > 0 || Boolean(item.text); + case "userInputResponse": + return Object.values(item.answers).some((answers) => answers.length > 0); + default: + return true; + } +} + +function trimUnifiedThreadForRenderableItems( + thread: UnifiedThread, + maxRenderableItems: number, +): UnifiedThread { + if (maxRenderableItems <= 0 || thread.turns.length === 0) { + return thread; + } + + const renderableBudget = maxRenderableItems + 1; + let renderableCount = 0; + let firstTurnIndex = -1; + let firstItemIndex = -1; + + for ( + let turnIndex = thread.turns.length - 1; + turnIndex >= 0; + turnIndex -= 1 + ) { + const turn = thread.turns[turnIndex]; + if (!turn) { + continue; + } + const items = turn.items ?? []; + for (let itemIndex = items.length - 1; itemIndex >= 0; itemIndex -= 1) { + const item = items[itemIndex]; + if (!item || !shouldCountRenderableUnifiedItem(item)) { + continue; + } + renderableCount += 1; + if (renderableCount === renderableBudget) { + firstTurnIndex = turnIndex; + firstItemIndex = itemIndex; + } + } + } + + if ( + renderableCount <= renderableBudget || + firstTurnIndex === -1 || + firstItemIndex === -1 + ) { + return thread; + } + + return { + ...thread, + turns: thread.turns.slice(firstTurnIndex).map((turn, index) => { + if (index !== 0) { + return turn; + } + return { + ...turn, + items: (turn.items ?? []).slice(firstItemIndex), + }; + }), + }; +} + export class UnifiedBackendFeatureError extends Error { public readonly provider: UnifiedProviderId; public readonly featureId: UnifiedFeatureId; @@ -265,10 +368,17 @@ function createHandlerTable( threadId: command.threadId, includeTurns: command.includeTurns, }); + const mappedThread = mapThread(provider, result.thread); return { kind: "readThread", - thread: mapThread(provider, result.thread), + thread: + typeof command.maxRenderableItems === "number" + ? trimUnifiedThreadForRenderableItems( + mappedThread, + command.maxRenderableItems, + ) + : mappedThread, }; }, @@ -317,8 +427,8 @@ function createHandlerTable( kind: "listModels", data: result.data.map((model) => ({ id: model.id, - displayName: model.displayName, - description: model.description, + displayName: model.displayName ?? model.id, + description: model.description ?? "", defaultReasoningEffort: model.defaultReasoningEffort ?? null, supportedReasoningEfforts: model.supportedReasoningEfforts.map( (entry) => entry.reasoningEffort, @@ -342,12 +452,28 @@ function createHandlerTable( return { kind: "listCollaborationModes", data: result.data.map((mode) => ({ - name: mode.name, + name: + mode.name ?? + humanizeCollaborationModeLabel(mode.mode ?? "default"), mode: mode.mode ?? "default", - ...(mode.model !== undefined ? { model: mode.model } : {}), + ...(mode.model !== undefined + ? { model: mode.model } + : mode.settings?.model !== undefined + ? { model: mode.settings.model } + : {}), ...(mode.reasoning_effort !== undefined ? { reasoningEffort: mode.reasoning_effort } - : {}), + : mode.settings?.reasoning_effort !== undefined + ? { reasoningEffort: mode.settings.reasoning_effort } + : {}), + ...(mode.developer_instructions !== undefined + ? { developerInstructions: mode.developer_instructions } + : mode.settings?.developer_instructions !== undefined + ? { + developerInstructions: + mode.settings.developer_instructions, + } + : {}), })), }; }, @@ -430,13 +556,21 @@ function createHandlerTable( } const liveState = await adapter.readLiveState(command.threadId); + const mappedConversationState = liveState.conversationState + ? mapThread(provider, liveState.conversationState) + : null; return { kind: "readLiveState", threadId: command.threadId, ownerClientId: liveState.ownerClientId, - conversationState: liveState.conversationState - ? mapThread(provider, liveState.conversationState) - : null, + conversationState: + mappedConversationState && + typeof command.maxRenderableItems === "number" + ? trimUnifiedThreadForRenderableItems( + mappedConversationState, + command.maxRenderableItems, + ) + : mappedConversationState, ...(liveState.liveStateError ? { liveStateError: liveState.liveStateError } : {}), @@ -960,10 +1094,11 @@ function mapTurnItem( }; case "todo-list": + case "todoList": return { id: item.id, type: "todoList", - ...(item.explanation !== undefined + ...(item.explanation != null ? { explanation: item.explanation } : {}), plan: item.plan.map((entry) => ({ @@ -1081,21 +1216,23 @@ function mapTurnItem( server: item.server, tool: item.tool, status: item.status, - arguments: jsonValueFromString(JSON.stringify(item.arguments)), + arguments: buildJsonPreviewString( + item.arguments, + MCP_TOOL_ARGUMENT_PREVIEW_MAX_CHARS, + ), ...(item.result !== undefined ? { result: item.result ? { - content: item.result.content.map((entry) => - jsonValueFromString(JSON.stringify(entry)), - ), + content: item.result.content.map(() => null), ...(item.result.structuredContent !== undefined ? { structuredContent: item.result.structuredContent === null ? null - : jsonValueFromString( - JSON.stringify(item.result.structuredContent), + : buildJsonPreviewString( + item.result.structuredContent, + MCP_TOOL_STRUCTURED_CONTENT_PREVIEW_MAX_CHARS, ), } : {}), @@ -1111,6 +1248,33 @@ function mapTurnItem( : {}), }; + case "dynamicToolCall": + return { + id: item.id, + type: "dynamicToolCall", + tool: item.tool, + status: item.status, + arguments: buildJsonPreviewString( + item.arguments, + DYNAMIC_TOOL_ARGUMENT_PREVIEW_MAX_CHARS, + ), + ...(item.contentItems !== undefined + ? { + contentItems: item.contentItems + ? item.contentItems.map((contentItem) => + contentItem.type === "inputText" + ? { type: "inputText", text: contentItem.text } + : { type: "inputImage", imageUrl: contentItem.imageUrl }, + ) + : null, + } + : {}), + ...(item.success !== undefined ? { success: item.success } : {}), + ...(item.durationMs !== undefined + ? { durationMs: item.durationMs } + : {}), + }; + case "collabAgentToolCall": return { id: item.id, @@ -1178,6 +1342,29 @@ function jsonValueFromString(serialized: string): JsonValue { return JsonValueSchema.parse(JSON.parse(serialized)); } +function buildJsonPreviewString(value: unknown, maxLength: number): string { + if (typeof value === "string") { + return truncatePreviewText(value, maxLength); + } + + try { + const serialized = JSON.stringify(value); + if (!serialized) { + return ""; + } + return truncatePreviewText(serialized, maxLength); + } catch { + return truncatePreviewText(String(value), maxLength); + } +} + +function truncatePreviewText(value: string, maxLength: number): string { + if (value.length <= maxLength) { + return value; + } + return `${value.slice(0, maxLength)}...`; +} + function jsonArrayFromString(serialized: string): JsonValue[] { return z.array(JsonValueSchema).parse(JSON.parse(serialized)); } diff --git a/apps/server/test/codex-agent-live-state.test.ts b/apps/server/test/codex-agent-live-state.test.ts new file mode 100644 index 00000000..bfc175b9 --- /dev/null +++ b/apps/server/test/codex-agent-live-state.test.ts @@ -0,0 +1,827 @@ +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import type { + IpcFrame, + ThreadConversationState, + ThreadStreamStateChangedBroadcast, +} from "@farfield/protocol"; +import { + CodexAgentAdapter, + buildArchivedThreadConversationStateFromJsonl, + collectThreadStreamStateChangedEvents, +} from "../src/agents/adapters/codex-agent.js"; + +const SAMPLE_THREAD_STATE: ThreadConversationState = { + id: "thread-1", + turns: [], + requests: [], + createdAt: 1700000000, + updatedAt: 1700000100, + title: "Thread", + latestModel: null, + latestReasoningEffort: null, +}; + +function buildSnapshotEvent(): ThreadStreamStateChangedBroadcast { + return { + type: "broadcast", + method: "thread-stream-state-changed", + sourceClientId: "owner-1", + version: 1, + params: { + conversationId: "thread-1", + change: { + type: "snapshot", + conversationState: SAMPLE_THREAD_STATE, + }, + version: 1, + type: "thread-stream-state-changed", + }, + }; +} + +describe("collectThreadStreamStateChangedEvents", () => { + it("ignores unrelated thread broadcast frames", () => { + const archivedEvent: IpcFrame = { + type: "broadcast", + method: "thread-archived", + sourceClientId: "owner-1", + version: 2, + params: { + conversationId: "thread-1", + hostId: "local", + cwd: "C:\\repo", + }, + }; + + const result = collectThreadStreamStateChangedEvents([ + buildSnapshotEvent(), + archivedEvent, + ]); + + expect(result.parseError).toBeNull(); + expect(result.events).toHaveLength(1); + expect(result.events[0]?.params.conversationId).toBe("thread-1"); + }); + + it("still reports malformed thread-stream-state-changed frames", () => { + const malformedEvent: IpcFrame = { + type: "broadcast", + method: "thread-stream-state-changed", + sourceClientId: "owner-1", + version: 1, + params: { + conversationId: "thread-1", + }, + }; + + const result = collectThreadStreamStateChangedEvents([malformedEvent]); + + expect(result.events).toHaveLength(0); + expect(result.parseError?.eventIndex).toBe(0); + expect(result.parseError?.message).toContain("params.change"); + }); +}); + +describe("buildArchivedThreadConversationStateFromJsonl", () => { + it("rebuilds a minimal thread state from archived rollout events", () => { + const jsonl = [ + JSON.stringify({ + timestamp: "2026-03-19T10:41:10.772Z", + type: "session_meta", + payload: { + type: "session_meta", + id: "thread-archive-1", + cwd: "C:\\repo", + source: "vscode", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:22.961Z", + type: "event_msg", + payload: { + type: "task_started", + turn_id: "turn-1", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:22.962Z", + type: "event_msg", + payload: { + type: "user_message", + message: "hello", + images: ["https://example.com/image.png"], + local_images: [], + text_elements: [], + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.000Z", + type: "event_msg", + payload: { + type: "agent_message", + message: "working on it", + phase: "commentary", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.100Z", + type: "event_msg", + payload: { + type: "reasoning", + summary: ["check logs"], + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.200Z", + type: "event_msg", + payload: { + type: "context_compacted", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.300Z", + type: "event_msg", + payload: { + type: "token_count", + info: { + total_token_usage: { + input_tokens: 1, + output_tokens: 2, + total_tokens: 3, + }, + }, + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.400Z", + type: "event_msg", + payload: { + type: "task_complete", + turn_id: "turn-1", + }, + }), + ].join("\n"); + + const result = buildArchivedThreadConversationStateFromJsonl( + jsonl, + "thread-archive-1", + { + title: "Archived thread", + updatedAt: 1710844883, + }, + ); + + expect(result).not.toBeNull(); + expect(result?.title).toBe("Archived thread"); + expect(result?.cwd).toBe("C:\\repo"); + expect(result?.source).toBe("vscode"); + expect(result?.turns).toHaveLength(1); + expect(result?.turns[0]?.turnId).toBe("turn-1"); + expect(result?.turns[0]?.status).toBe("completed"); + expect(result?.turns[0]?.items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ type: "userMessage" }), + expect.objectContaining({ type: "agentMessage", text: "working on it" }), + expect.objectContaining({ type: "reasoning", summary: ["check logs"] }), + expect.objectContaining({ type: "contextCompaction" }), + ]), + ); + expect(result?.latestTokenUsageInfo).toEqual({ + total_token_usage: { + input_tokens: 1, + output_tokens: 2, + total_tokens: 3, + }, + }); + }); + + it("rebuilds legacy tool payloads into bounded dynamic tool items", () => { + const longOutput = "x".repeat(400); + const jsonl = [ + JSON.stringify({ + timestamp: "2026-03-20T14:40:00.000Z", + type: "session_meta", + payload: { + type: "session_meta", + id: "thread-archive-tools", + cwd: "C:\\repo", + source: "vscode", + model: "gpt-5.3-codex", + reasoning_effort: "high", + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:01.000Z", + type: "event_msg", + payload: { + type: "task_started", + turn_id: "turn-tools-1", + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:01.100Z", + type: "response_item", + payload: { + type: "message", + role: "developer", + content: [ + { + type: "input_text", + text: "Follow the runtime hardening checklist.", + }, + ], + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:01.200Z", + type: "response_item", + payload: { + type: "message", + role: "assistant", + phase: "commentary", + content: [ + { + type: "output_text", + text: "Inspecting archived replay coverage now.", + }, + ], + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:02.000Z", + type: "response_item", + payload: { + type: "function_call", + name: "shell_command", + arguments: JSON.stringify({ + command: "Get-ChildItem", + workdir: "C:\\repo", + }), + call_id: "call-function-1", + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:03.000Z", + type: "response_item", + payload: { + type: "function_call_output", + call_id: "call-function-1", + output: { + body: longOutput, + success: true, + }, + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:04.000Z", + type: "response_item", + payload: { + type: "custom_tool_call", + call_id: "call-custom-1", + name: "ask_user", + input: "Need confirmation", + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:05.000Z", + type: "response_item", + payload: { + type: "custom_tool_call_output", + call_id: "call-custom-1", + output: { + body: [ + { + type: "input_text", + text: "Confirmed by user", + }, + ], + success: false, + }, + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:06.000Z", + type: "event_msg", + payload: { + type: "dynamic_tool_call_request", + callId: "call-dynamic-1", + turnId: "turn-tools-1", + tool: "generate_image", + arguments: { + prompt: "render an architecture diagram", + }, + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:07.000Z", + type: "event_msg", + payload: { + type: "dynamic_tool_call_response", + call_id: "call-dynamic-1", + turn_id: "turn-tools-1", + tool: "generate_image", + arguments: { + prompt: "render an architecture diagram", + }, + content_items: [ + { + type: "input_image", + image_url: "https://example.com/diagram.png", + }, + ], + success: true, + error: null, + duration: "PT1.25S", + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:08.000Z", + type: "event_msg", + payload: { + type: "todo_list", + explanation: null, + plan: [ + { + step: "Verify archived replay", + status: "completed", + }, + ], + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:08.250Z", + type: "response_item", + payload: { + type: "web_search_call", + status: "completed", + action: { + type: "search", + query: "time: {\"utc_offset\":\"+08:00\"}", + queries: ['time: {"utc_offset":"+08:00"}'], + }, + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:08.500Z", + type: "event_msg", + payload: { + type: "unknown_future_payload", + extra: true, + }, + }), + JSON.stringify({ + timestamp: "2026-03-20T14:40:09.000Z", + type: "event_msg", + payload: { + type: "task_complete", + turn_id: "turn-tools-1", + }, + }), + ].join("\n"); + + const result = buildArchivedThreadConversationStateFromJsonl( + jsonl, + "thread-archive-tools", + { + title: "Archived tools", + updatedAt: 1710844883, + }, + ); + + expect(result).not.toBeNull(); + expect(result?.latestModel).toBe("gpt-5.3-codex"); + expect(result?.latestReasoningEffort).toBe("high"); + + const toolItems = + result?.turns[0]?.items.filter( + (item): item is Extract => + item.type === "dynamicToolCall", + ) ?? []; + expect(toolItems).toHaveLength(4); + expect(toolItems[0]).toEqual( + expect.objectContaining({ + type: "dynamicToolCall", + tool: "shell_command", + status: "completed", + success: true, + }), + ); + expect( + toolItems[0]?.contentItems?.[0]?.type === "inputText" + ? toolItems[0].contentItems[0].text.length + : 0, + ).toBeLessThanOrEqual(243); + expect(toolItems[1]).toEqual( + expect.objectContaining({ + tool: "ask_user", + status: "failed", + success: false, + }), + ); + expect(toolItems[2]).toEqual( + expect.objectContaining({ + tool: "generate_image", + status: "completed", + success: true, + durationMs: 1250, + }), + ); + expect(toolItems[2]?.contentItems?.[0]).toEqual({ + type: "inputImage", + imageUrl: "https://example.com/diagram.png", + }); + expect(toolItems[3]).toEqual( + expect.objectContaining({ + tool: "web_search", + status: "completed", + }), + ); + expect(toolItems[3]?.contentItems).toEqual([ + { + type: "inputText", + text: 'time: {"utc_offset":"+08:00"}', + }, + ]); + + expect(result?.turns[0]?.items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "steeringUserMessage", + }), + expect.objectContaining({ + type: "agentMessage", + text: "Inspecting archived replay coverage now.", + phase: "commentary", + }), + ]), + ); + + const todoItem = result?.turns[0]?.items.find( + (item) => item.type === "todoList", + ); + expect(todoItem).toEqual( + expect.objectContaining({ + type: "todoList", + plan: [ + { + step: "Verify archived replay", + status: "completed", + }, + ], + }), + ); + }); + + it("skips malformed trailing jsonl lines and restores earlier archived entries", () => { + const jsonl = [ + JSON.stringify({ + timestamp: "2026-03-19T10:41:10.772Z", + type: "session_meta", + payload: { + type: "session_meta", + id: "thread-archive-tail", + cwd: "C:\\repo", + source: "vscode", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:22.961Z", + type: "event_msg", + payload: { + type: "task_started", + turn_id: "turn-tail-1", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:22.962Z", + type: "event_msg", + payload: { + type: "agent_message", + message: "still restorable", + }, + }), + '{"timestamp":"2026-03-19T10:41:23.000Z","payload":', + ].join("\n"); + + const result = buildArchivedThreadConversationStateFromJsonl( + jsonl, + "thread-archive-tail", + null, + ); + + expect(result).not.toBeNull(); + expect(result?.turns).toHaveLength(1); + expect(result?.turns[0]?.items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "agentMessage", + text: "still restorable", + }), + ]), + ); + }); +}); + +describe("CodexAgentAdapter runtime retention", () => { + function createAdapterForTest(): CodexAgentAdapter { + return new CodexAgentAdapter({ + appExecutable: "codex", + socketPath: "\\\\.\\pipe\\farfield-test", + workspaceDir: process.cwd(), + userAgent: "farfield-test", + reconnectDelayMs: 1, + }); + } + + it("evicts the oldest tracked thread state after the runtime cap", () => { + const adapter = createAdapterForTest() as unknown as { + setTrackedThreadOwner: (threadId: string, ownerClientId: string) => void; + setTrackedThreadSnapshot: ( + threadId: string, + snapshot: ThreadConversationState, + origin: "stream" | "readThreadWithTurns" | "readThread", + ) => void; + setTrackedThreadEvents: (threadId: string, events: IpcFrame[]) => void; + setThreadTitle: (threadId: string, title: string | null) => void; + getTrackedThreadRuntimeCounts: CodexAgentAdapter["getTrackedThreadRuntimeCounts"]; + readStreamEvents: CodexAgentAdapter["readStreamEvents"]; + }; + + for (let index = 0; index < 82; index += 1) { + const threadId = `thread-${String(index)}`; + adapter.setTrackedThreadOwner(threadId, `owner-${String(index)}`); + adapter.setTrackedThreadSnapshot( + threadId, + { + ...SAMPLE_THREAD_STATE, + id: threadId, + title: `Thread ${String(index)}`, + }, + "stream", + ); + adapter.setTrackedThreadEvents(threadId, [buildSnapshotEvent()]); + adapter.setThreadTitle(threadId, `Thread ${String(index)}`); + } + + const counts = adapter.getTrackedThreadRuntimeCounts(); + expect(counts.trackedThreadCount).toBe(80); + expect(counts.maxTrackedThreadCountSeen).toBe(80); + expect(counts.streamEventThreadCount).toBe(80); + }); + + it("truncates buffered stream events per thread to the configured cap", async () => { + const adapter = createAdapterForTest() as unknown as { + setTrackedThreadEvents: (threadId: string, events: IpcFrame[]) => void; + getTrackedThreadRuntimeCounts: CodexAgentAdapter["getTrackedThreadRuntimeCounts"]; + readStreamEvents: CodexAgentAdapter["readStreamEvents"]; + }; + const events: IpcFrame[] = Array.from({ length: 160 }, (_, index) => ({ + type: "broadcast", + method: "thread-stream-state-changed", + sourceClientId: "owner-1", + version: 1, + params: { + conversationId: "thread-cap", + change: { + type: "snapshot", + conversationState: { + ...SAMPLE_THREAD_STATE, + id: `thread-cap-${String(index)}`, + }, + }, + version: 1, + type: "thread-stream-state-changed", + }, + })); + + adapter.setTrackedThreadEvents("thread-cap", events); + + const stream = await adapter.readStreamEvents("thread-cap", 500); + expect(stream.events).toHaveLength(120); + expect(adapter.getTrackedThreadRuntimeCounts().maxBufferedStreamEventsPerThreadSeen).toBe( + 120, + ); + }); + + it("increments the parse failure counter only for malformed stream-state frames", async () => { + const adapter = createAdapterForTest() as unknown as { + setTrackedThreadEvents: (threadId: string, events: IpcFrame[]) => void; + getTrackedThreadRuntimeCounts: CodexAgentAdapter["getTrackedThreadRuntimeCounts"]; + readLiveState: CodexAgentAdapter["readLiveState"]; + }; + + adapter.setTrackedThreadEvents("thread-parse", [ + { + type: "broadcast", + method: "thread-archived", + sourceClientId: "owner-1", + version: 1, + params: { + conversationId: "thread-parse", + }, + }, + ]); + + await adapter.readLiveState("thread-parse"); + expect(adapter.getTrackedThreadRuntimeCounts().streamParseFailureCount).toBe( + 0, + ); + + adapter.setTrackedThreadEvents("thread-parse", [ + { + type: "broadcast", + method: "thread-stream-state-changed", + sourceClientId: "owner-1", + version: 1, + params: { + conversationId: "thread-parse", + }, + }, + ]); + + const result = await adapter.readLiveState("thread-parse"); + expect(result.liveStateError?.kind).toBe("parseFailed"); + expect(adapter.getTrackedThreadRuntimeCounts().streamParseFailureCount).toBe( + 1, + ); + }); + + it("restores full archived live state after a metadata-only archived read", async () => { + const adapter = createAdapterForTest() as unknown as { + restoreArchivedThreadState: ( + threadId: string, + includeTurns: boolean, + ) => Promise; + readLiveState: CodexAgentAdapter["readLiveState"]; + }; + const threadId = "thread-archived-live-state"; + const previousCodexHome = process.env["CODEX_HOME"]; + const temporaryCodexHome = await mkdtemp( + path.join(os.tmpdir(), "farfield-codex-home-"), + ); + const archivedSessionsDir = path.join(temporaryCodexHome, "archived_sessions"); + const archivedJsonl = [ + JSON.stringify({ + timestamp: "2026-03-19T10:41:10.772Z", + type: "session_meta", + payload: { + type: "session_meta", + id: threadId, + cwd: "C:\\repo", + source: "vscode", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:22.961Z", + type: "event_msg", + payload: { + type: "task_started", + turn_id: "turn-archive-live-1", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.000Z", + type: "event_msg", + payload: { + type: "agent_message", + message: "restored from archive", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.400Z", + type: "event_msg", + payload: { + type: "task_complete", + turn_id: "turn-archive-live-1", + }, + }), + ].join("\n"); + + try { + process.env["CODEX_HOME"] = temporaryCodexHome; + await mkdir(archivedSessionsDir, { recursive: true }); + await writeFile( + path.join(archivedSessionsDir, `${threadId}.jsonl`), + archivedJsonl, + "utf8", + ); + + const metadataOnly = await adapter.restoreArchivedThreadState(threadId, false); + expect(metadataOnly?.turns).toHaveLength(0); + + const liveState = await adapter.readLiveState(threadId); + expect(liveState.liveStateError).toBeNull(); + expect(liveState.conversationState?.turns).toHaveLength(1); + expect(liveState.conversationState?.turns[0]?.items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "agentMessage", + text: "restored from archive", + }), + ]), + ); + } finally { + if (previousCodexHome === undefined) { + delete process.env["CODEX_HOME"]; + } else { + process.env["CODEX_HOME"] = previousCodexHome; + } + await rm(temporaryCodexHome, { recursive: true, force: true }); + } + }); + + it("clears runtime lastError after archived readThread fallback succeeds", async () => { + const adapter = createAdapterForTest() as unknown as { + appClient: { + readThread: (threadId: string, includeTurns: boolean) => Promise; + resumeThread: (threadId: string) => Promise; + }; + patchRuntimeState: (next: Partial<{ lastError: string | null }>) => void; + getRuntimeState: CodexAgentAdapter["getRuntimeState"]; + readThread: CodexAgentAdapter["readThread"]; + }; + const threadId = "thread-archived-read-thread"; + const previousCodexHome = process.env["CODEX_HOME"]; + const temporaryCodexHome = await mkdtemp( + path.join(os.tmpdir(), "farfield-codex-home-"), + ); + const archivedSessionsDir = path.join(temporaryCodexHome, "archived_sessions"); + const archivedJsonl = [ + JSON.stringify({ + timestamp: "2026-03-19T10:41:10.772Z", + type: "session_meta", + payload: { + type: "session_meta", + id: threadId, + cwd: "C:\\repo", + source: "vscode", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:22.961Z", + type: "event_msg", + payload: { + type: "task_started", + turn_id: "turn-archive-read-1", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.000Z", + type: "event_msg", + payload: { + type: "agent_message", + message: "fallback restored thread", + }, + }), + JSON.stringify({ + timestamp: "2026-03-19T10:41:23.400Z", + type: "event_msg", + payload: { + type: "task_complete", + turn_id: "turn-archive-read-1", + }, + }), + ].join("\n"); + const noRolloutError = new Error( + `app-server error -32600: no rollout found for thread id ${threadId}`, + ); + + try { + process.env["CODEX_HOME"] = temporaryCodexHome; + await mkdir(archivedSessionsDir, { recursive: true }); + await writeFile( + path.join(archivedSessionsDir, `${threadId}.jsonl`), + archivedJsonl, + "utf8", + ); + + adapter.patchRuntimeState({ + lastError: noRolloutError.message, + }); + adapter.appClient.readThread = async () => { + throw noRolloutError; + }; + adapter.appClient.resumeThread = async () => { + throw noRolloutError; + }; + + const result = await adapter.readThread({ + threadId, + includeTurns: true, + }); + + expect(result.thread.turns).toHaveLength(1); + expect(adapter.getRuntimeState().lastError).toBeNull(); + } finally { + if (previousCodexHome === undefined) { + delete process.env["CODEX_HOME"]; + } else { + process.env["CODEX_HOME"] = previousCodexHome; + } + await rm(temporaryCodexHome, { recursive: true, force: true }); + } + }); +}); diff --git a/apps/server/test/debug-history.test.ts b/apps/server/test/debug-history.test.ts new file mode 100644 index 00000000..8c8c3aae --- /dev/null +++ b/apps/server/test/debug-history.test.ts @@ -0,0 +1,165 @@ +import { describe, expect, it } from "vitest"; +import { prepareHistoryPayloadForStorage } from "../src/debug-history.js"; + +describe("prepareHistoryPayloadForStorage", () => { + it("preserves small payloads without compaction", () => { + const result = prepareHistoryPayloadForStorage({ + type: "broadcast", + method: "ping", + params: { ok: true }, + }); + + expect(result.compacted).toBe(false); + expect(JSON.parse(result.json)).toEqual({ + type: "broadcast", + method: "ping", + params: { ok: true }, + }); + }); + + it("compacts thread-stream snapshots while keeping debug-critical fields", () => { + const largeText = "x".repeat(2_000); + const result = prepareHistoryPayloadForStorage({ + type: "broadcast", + method: "thread-stream-state-changed", + params: { + conversationId: "thread-123", + change: { + type: "snapshot", + conversationState: { + id: "thread-123", + turns: [ + { + turnId: "turn-1", + status: "completed", + items: [ + { + id: "item-1", + type: "userMessage", + content: [{ type: "text", text: largeText }], + }, + { + id: "item-2", + type: "agentMessage", + phase: "commentary", + text: largeText, + }, + { + id: "item-3", + type: "contextCompaction", + }, + ], + }, + ], + requests: [], + }, + }, + }, + }); + + const stored = JSON.parse(result.json); + const turn = stored.params.change.conversationState.turns[0]; + + expect(result.compacted).toBe(true); + expect(turn.turnId).toBe("turn-1"); + expect(turn.itemCount).toBe(3); + expect(turn.itemTypes).toEqual([ + "userMessage", + "agentMessage", + "contextCompaction", + ]); + expect(turn.items[1]).toMatchObject({ + id: "item-2", + type: "agentMessage", + phase: "commentary", + }); + expect(turn.items[1].text).toContain("... ["); + }); + + it("compacts thread-stream patches into a bounded preview", () => { + const result = prepareHistoryPayloadForStorage({ + type: "broadcast", + method: "thread-stream-state-changed", + params: { + conversationId: "thread-456", + change: { + type: "patches", + patches: Array.from({ length: 40 }, (_, index) => ({ + op: "add", + path: ["turns", 0, "items", index], + value: { + id: `item-${index}`, + type: "agentMessage", + text: "y".repeat(1_000), + }, + })), + }, + }, + }); + + const stored = JSON.parse(result.json); + expect(result.compacted).toBe(true); + expect(stored.params.change.patchCount).toBe(40); + expect(stored.params.change.patches).toHaveLength(24); + expect(stored.params.change.truncatedPatchCount).toBe(16); + }); + + it("handles BigInt and undefined roots without throwing", () => { + const bigintResult = prepareHistoryPayloadForStorage({ + type: "system", + value: 42n, + }); + const undefinedResult = prepareHistoryPayloadForStorage(undefined); + + expect(JSON.parse(bigintResult.json)).toEqual({ + type: "system", + value: "42", + }); + expect(undefinedResult.json).toBe("null"); + }); + + it("falls back safely when payload contains circular references", () => { + const payload: { type: string; self?: unknown } = { + type: "system", + }; + payload.self = payload; + + const result = prepareHistoryPayloadForStorage(payload); + + expect(JSON.parse(result.json)).toEqual({ + type: "system", + self: "[Circular]", + }); + }); + + it("reports turnOffsetFromLatest relative to the latest retained turn", () => { + const result = prepareHistoryPayloadForStorage({ + type: "broadcast", + method: "thread-stream-state-changed", + params: { + conversationId: "thread-offsets", + change: { + type: "snapshot", + conversationState: { + id: "thread-offsets", + turns: Array.from({ length: 20 }, (_, index) => ({ + turnId: `turn-${index}`, + status: "completed", + items: [], + })), + requests: [], + }, + }, + }, + }); + + const stored = JSON.parse(result.json); + const retainedTurns = stored.params.change.conversationState.turns; + + expect(retainedTurns).toHaveLength(16); + expect(retainedTurns[0]?.turnId).toBe("turn-4"); + expect(retainedTurns[0]?.turnOffsetFromLatest).toBe(15); + expect(retainedTurns[15]?.turnId).toBe("turn-19"); + expect(retainedTurns[15]?.turnOffsetFromLatest).toBe(0); + }); +}); diff --git a/apps/server/test/unified-adapter.test.ts b/apps/server/test/unified-adapter.test.ts index d955e80e..29dc1102 100644 --- a/apps/server/test/unified-adapter.test.ts +++ b/apps/server/test/unified-adapter.test.ts @@ -615,4 +615,361 @@ describe("unified provider adapters", () => { : null, ).toBe("task-123"); }); + + it("maps dynamicToolCall turn items into unified items", async () => { + const adapter = createCodexAdapter(); + const longQuery = "x".repeat(800); + adapter.readThread = async () => ({ + thread: { + ...SAMPLE_THREAD, + turns: [ + { + id: "turn-1", + status: "completed", + items: [ + { + id: "item-dynamic", + type: "dynamicToolCall", + tool: "web_search", + status: "completed", + arguments: { + query: longQuery, + }, + contentItems: [ + { + type: "inputText", + text: "Search complete", + }, + { + type: "inputImage", + imageUrl: "https://example.com/result.png", + }, + ], + success: true, + durationMs: 64, + }, + ], + }, + ], + }, + }); + const unified = new AgentUnifiedProviderAdapter("codex", adapter); + + const result = await unified.execute( + UnifiedCommandSchema.parse({ + kind: "readThread", + provider: "codex", + threadId: SAMPLE_THREAD.id, + includeTurns: true, + }), + ); + + expect(result.kind).toBe("readThread"); + if (result.kind !== "readThread") { + return; + } + + const dynamicItem = result.thread.turns[0]?.items[0]; + expect(dynamicItem?.type).toBe("dynamicToolCall"); + expect( + dynamicItem && dynamicItem.type === "dynamicToolCall" + ? dynamicItem.tool + : null, + ).toBe("web_search"); + expect( + dynamicItem && dynamicItem.type === "dynamicToolCall" + ? dynamicItem.contentItems?.length + : null, + ).toBe(2); + expect( + dynamicItem && dynamicItem.type === "dynamicToolCall" + ? dynamicItem.success + : null, + ).toBe(true); + expect( + dynamicItem && dynamicItem.type === "dynamicToolCall" + ? typeof dynamicItem.arguments + : null, + ).toBe("string"); + expect( + dynamicItem && dynamicItem.type === "dynamicToolCall" + ? String(dynamicItem.arguments).length + : 0, + ).toBeLessThanOrEqual(603); + }); + + it("maps mcpToolCall items into lightweight unified previews", async () => { + const adapter = createCodexAdapter(); + const largeText = "y".repeat(800); + adapter.readThread = async () => ({ + thread: { + ...SAMPLE_THREAD, + turns: [ + { + id: "turn-1", + status: "completed", + items: [ + { + id: "item-mcp", + type: "mcpToolCall", + server: "filesystem", + tool: "read_file", + status: "completed", + arguments: { + path: "/tmp/huge.txt", + extra: largeText, + }, + result: { + content: [ + { + type: "text", + text: largeText, + }, + { + type: "text", + text: "tail", + }, + ], + structuredContent: { + body: largeText, + }, + }, + durationMs: 41, + }, + ], + }, + ], + }, + }); + const unified = new AgentUnifiedProviderAdapter("codex", adapter); + + const result = await unified.execute( + UnifiedCommandSchema.parse({ + kind: "readThread", + provider: "codex", + threadId: SAMPLE_THREAD.id, + includeTurns: true, + }), + ); + + expect(result.kind).toBe("readThread"); + if (result.kind !== "readThread") { + return; + } + + const mcpItem = result.thread.turns[0]?.items[0]; + expect(mcpItem?.type).toBe("mcpToolCall"); + expect( + mcpItem && mcpItem.type === "mcpToolCall" ? typeof mcpItem.arguments : null, + ).toBe("string"); + expect( + mcpItem && mcpItem.type === "mcpToolCall" + ? String(mcpItem.arguments).length + : 0, + ).toBeLessThanOrEqual(603); + expect( + mcpItem && mcpItem.type === "mcpToolCall" ? mcpItem.result?.content : null, + ).toEqual([null, null]); + expect( + mcpItem && mcpItem.type === "mcpToolCall" + ? typeof mcpItem.result?.structuredContent + : null, + ).toBe("string"); + expect( + mcpItem && + mcpItem.type === "mcpToolCall" && + typeof mcpItem.result?.structuredContent === "string" + ? mcpItem.result.structuredContent.length + : 0, + ).toBeLessThanOrEqual(603); + }); + + it("trims readThread results to the requested render window", async () => { + const adapter = createCodexAdapter(); + adapter.readThread = async () => ({ + thread: { + ...SAMPLE_THREAD, + turns: [ + { + id: "turn-1", + status: "completed", + items: Array.from({ length: 95 }, (_, index) => ({ + id: `item-agent-${index}`, + type: "agentMessage", + text: `agent-message-${index}`, + })), + }, + ], + }, + }); + const unified = new AgentUnifiedProviderAdapter("codex", adapter); + + const result = await unified.execute( + UnifiedCommandSchema.parse({ + kind: "readThread", + provider: "codex", + threadId: SAMPLE_THREAD.id, + includeTurns: true, + maxRenderableItems: 90, + }), + ); + + expect(result.kind).toBe("readThread"); + if (result.kind !== "readThread") { + return; + } + + expect(result.thread.turns[0]?.items.length).toBe(91); + const firstItem = result.thread.turns[0]?.items[0]; + expect(firstItem?.type).toBe("agentMessage"); + expect( + firstItem && firstItem.type === "agentMessage" ? firstItem.text : null, + ).toBe("agent-message-4"); + }); + + it("counts image-only user messages when trimming readThread results", async () => { + const adapter = createCodexAdapter(); + adapter.readThread = async () => ({ + thread: { + ...SAMPLE_THREAD, + turns: [ + { + id: "turn-images", + status: "completed", + items: Array.from({ length: 95 }, (_, index) => ({ + id: `item-image-${index}`, + type: "userMessage", + content: [ + { + type: "image", + url: `https://example.com/image-${index}.png`, + }, + ], + })), + }, + ], + }, + }); + const unified = new AgentUnifiedProviderAdapter("codex", adapter); + + const result = await unified.execute( + UnifiedCommandSchema.parse({ + kind: "readThread", + provider: "codex", + threadId: SAMPLE_THREAD.id, + includeTurns: true, + maxRenderableItems: 90, + }), + ); + + expect(result.kind).toBe("readThread"); + if (result.kind !== "readThread") { + return; + } + + expect(result.thread.turns[0]?.items.length).toBe(91); + const firstItem = result.thread.turns[0]?.items[0]; + expect(firstItem?.type).toBe("userMessage"); + expect( + firstItem && firstItem.type === "userMessage" + ? firstItem.content[0]?.type + : null, + ).toBe("image"); + }); + + it("trims readLiveState results to the requested render window", async () => { + const adapter = createCodexAdapter(); + adapter.readLiveState = async () => ({ + ownerClientId: "owner-1", + conversationState: { + ...SAMPLE_THREAD, + turns: [ + { + id: "turn-1", + status: "completed", + items: Array.from({ length: 95 }, (_, index) => ({ + id: `item-live-${index}`, + type: "agentMessage", + text: `live-message-${index}`, + })), + }, + ], + }, + liveStateError: null, + }); + const unified = new AgentUnifiedProviderAdapter("codex", adapter); + + const result = await unified.execute( + UnifiedCommandSchema.parse({ + kind: "readLiveState", + provider: "codex", + threadId: SAMPLE_THREAD.id, + maxRenderableItems: 90, + }), + ); + + expect(result.kind).toBe("readLiveState"); + if (result.kind !== "readLiveState") { + return; + } + + expect(result.conversationState?.turns[0]?.items.length).toBe(91); + const firstItem = result.conversationState?.turns[0]?.items[0]; + expect(firstItem?.type).toBe("agentMessage"); + expect( + firstItem && firstItem.type === "agentMessage" ? firstItem.text : null, + ).toBe("live-message-4"); + }); + + it("normalizes todoList live-state items with null explanation", async () => { + const adapter = createCodexAdapter(); + adapter.readLiveState = async () => ({ + ownerClientId: "owner-1", + conversationState: { + ...SAMPLE_THREAD, + turns: [ + { + id: "turn-1", + status: "inProgress", + items: [ + { + id: "item-todo", + type: "todoList", + explanation: null, + plan: [ + { + step: "Verify compatibility", + status: "completed", + }, + ], + }, + ], + }, + ], + }, + liveStateError: null, + }); + const unified = new AgentUnifiedProviderAdapter("codex", adapter); + + const result = await unified.execute( + UnifiedCommandSchema.parse({ + kind: "readLiveState", + provider: "codex", + threadId: SAMPLE_THREAD.id, + }), + ); + + expect(result.kind).toBe("readLiveState"); + if (result.kind !== "readLiveState") { + return; + } + + const todoItem = result.conversationState?.turns[0]?.items[0]; + expect(todoItem?.type).toBe("todoList"); + expect( + todoItem && todoItem.type === "todoList" + ? Object.prototype.hasOwnProperty.call(todoItem, "explanation") + : null, + ).toBe(false); + }); }); diff --git a/apps/web/index.html b/apps/web/index.html index bf66c64d..cfa7d318 100644 --- a/apps/web/index.html +++ b/apps/web/index.html @@ -16,6 +16,7 @@ content="#09090b" media="(prefers-color-scheme: dark)" /> +