From 8e54bbeb53cd57d9b7f197f97968638694f4bb28 Mon Sep 17 00:00:00 2001 From: weishu Date: Fri, 24 Apr 2026 10:52:38 +0800 Subject: [PATCH] feat(codex): use hook to obtain the session id and transcript path --- cli/src/codex/codexLocal.ts | 14 +- cli/src/codex/codexLocalLauncher.test.ts | 244 +++++++- cli/src/codex/codexLocalLauncher.ts | 156 ++++- cli/src/codex/session.ts | 28 + cli/src/codex/utils/codexMcpConfig.test.ts | 17 +- cli/src/codex/utils/codexMcpConfig.ts | 34 +- .../codex/utils/codexSessionScanner.test.ts | 292 +++------ cli/src/codex/utils/codexSessionScanner.ts | 587 +++--------------- 8 files changed, 610 insertions(+), 762 deletions(-) diff --git a/cli/src/codex/codexLocal.ts b/cli/src/codex/codexLocal.ts index 5e84eb43..5f00b567 100644 --- a/cli/src/codex/codexLocal.ts +++ b/cli/src/codex/codexLocal.ts @@ -1,6 +1,10 @@ import { logger } from '@/ui/logger'; import { spawnWithTerminalGuard } from '@/utils/spawnWithTerminalGuard'; -import { buildMcpServerConfigArgs, buildDeveloperInstructionsArg } from './utils/codexMcpConfig'; +import { + buildMcpServerConfigArgs, + buildDeveloperInstructionsArg, + buildSessionStartHookConfigArgs +} from './utils/codexMcpConfig'; import { codexSystemPrompt } from './utils/systemPrompt'; import type { ReasoningEffort } from './appServerTypes'; @@ -33,6 +37,10 @@ export async function codexLocal(opts: { onSessionFound: (id: string) => void; codexArgs?: string[]; mcpServers?: Record; + sessionHook?: { + port: number; + token: string; + }; }): Promise { const args: string[] = []; @@ -58,6 +66,10 @@ export async function codexLocal(opts: { args.push(...buildMcpServerConfigArgs(opts.mcpServers)); } + if (opts.sessionHook) { + args.push(...buildSessionStartHookConfigArgs(opts.sessionHook.port, opts.sessionHook.token)); + } + // Add developer instructions (system prompt) args.push(...buildDeveloperInstructionsArg(codexSystemPrompt)); diff --git a/cli/src/codex/codexLocalLauncher.test.ts b/cli/src/codex/codexLocalLauncher.test.ts index 98741bab..00d8b3f1 100644 --- a/cli/src/codex/codexLocalLauncher.test.ts +++ b/cli/src/codex/codexLocalLauncher.test.ts @@ -1,10 +1,14 @@ -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { appendFile, mkdir, rm, writeFile } from 'node:fs/promises'; +import { existsSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; -const harness = vi.hoisted(() => ({ +const harness = { launches: [] as Array>, - sessionScannerCalls: [] as Array>, - scannerFailureMessage: 'No Codex session found within 120000ms for cwd c:\\workspace\\project; refusing fallback.' -})); + sessionHookHandlers: [] as Array<(sessionId: string, data: Record) => void>, + runBarrier: null as Promise | null +}; vi.mock('./codexLocal', () => ({ codexLocal: async (opts: Record) => { @@ -22,17 +26,13 @@ vi.mock('./utils/buildHapiMcpBridge', () => ({ }) })); -vi.mock('./utils/codexSessionScanner', () => ({ - createCodexSessionScanner: async (opts: { - onSessionMatchFailed?: (message: string) => void; - }) => { - harness.sessionScannerCalls.push(opts as Record); +vi.mock('@/claude/utils/startHookServer', () => ({ + startHookServer: async (opts: { onSessionHook: (sessionId: string, data: Record) => void }) => { + harness.sessionHookHandlers.push(opts.onSessionHook); return { - cleanup: async () => {}, - onNewSession: () => {}, - triggerFailure: () => { - opts.onSessionMatchFailed?.(harness.scannerFailureMessage); - } + port: 4242, + token: 'hook-token', + stop: () => {} }; } })); @@ -47,6 +47,9 @@ vi.mock('@/modules/common/launcher/BaseLocalLauncher', () => ({ async run(): Promise<'exit'> { await this.opts.launch(new AbortController().signal); + if (harness.runBarrier) { + await harness.runBarrier; + } return 'exit'; } } @@ -62,13 +65,27 @@ function createQueueStub() { }; } -function createSessionStub(permissionMode: 'default' | 'read-only' | 'safe-yolo' | 'yolo', codexArgs?: string[], path = '/tmp/worktree') { +function createSessionStub( + permissionMode: 'default' | 'read-only' | 'safe-yolo' | 'yolo', + codexArgs?: string[], + path = '/tmp/worktree', + initialTranscriptPath: string | null = null +) { const sessionEvents: Array<{ type: string; message?: string }> = []; + const agentMessages: unknown[] = []; let localLaunchFailure: { message: string; exitReason: 'switch' | 'exit' } | null = null; + let sessionId: string | null = null; + let transcriptPath: string | null = initialTranscriptPath; + const transcriptPathCallbacks: Array<(path: string) => void> = []; return { session: { - sessionId: null, + get sessionId() { + return sessionId; + }, + get transcriptPath() { + return transcriptPath; + }, path, startedBy: 'terminal' as const, startingMode: 'local' as const, @@ -80,7 +97,27 @@ function createSessionStub(permissionMode: 'default' | 'read-only' | 'safe-yolo' }, getPermissionMode: () => permissionMode, getModelReasoningEffort: () => null, - onSessionFound: () => {}, + onSessionFound: (value: string) => { + sessionId = value; + }, + onTranscriptPathFound: (pathValue: string) => { + transcriptPath = pathValue; + for (const callback of transcriptPathCallbacks) { + callback(pathValue); + } + }, + addTranscriptPathCallback: (callback: (path: string) => void) => { + transcriptPathCallbacks.push(callback); + }, + removeTranscriptPathCallback: (callback: (path: string) => void) => { + const index = transcriptPathCallbacks.indexOf(callback); + if (index !== -1) { + transcriptPathCallbacks.splice(index, 1); + } + }, + resetTranscriptPath: () => { + transcriptPath = null; + }, sendSessionEvent: (event: { type: string; message?: string }) => { sessionEvents.push(event); }, @@ -88,18 +125,38 @@ function createSessionStub(permissionMode: 'default' | 'read-only' | 'safe-yolo' localLaunchFailure = { message, exitReason }; }, sendUserMessage: () => {}, - sendAgentMessage: () => {}, + sendAgentMessage: (message: unknown) => { + agentMessages.push(message); + }, queue: createQueueStub() }, sessionEvents, + agentMessages, getLocalLaunchFailure: () => localLaunchFailure }; } describe('codexLocalLauncher', () => { + let tempDir = ''; + + const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + + beforeEach(async () => { + tempDir = join(tmpdir(), `codex-local-launcher-${Date.now()}`); + await mkdir(tempDir, { recursive: true }); + }); + afterEach(() => { + vi.useRealTimers(); harness.launches = []; - harness.sessionScannerCalls = []; + harness.sessionHookHandlers = []; + harness.runBarrier = null; + }); + + afterEach(async () => { + if (existsSync(tempDir)) { + await rm(tempDir, { recursive: true, force: true }); + } }); it('rebuilds approval and sandbox args from yolo mode', async () => { @@ -175,16 +232,151 @@ describe('codexLocalLauncher', () => { it('warns on session match failure without aborting local Codex launch', async () => { const { session, sessionEvents, getLocalLaunchFailure } = createSessionStub('default', undefined, 'c:\\workspace\\project'); - await codexLocalLauncher(session as never); + vi.useFakeTimers(); + const launcherPromise = codexLocalLauncher(session as never); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + vi.advanceTimersByTime(10_000); + await Promise.resolve(); + await launcherPromise; - const scannerCall = harness.sessionScannerCalls[0] as { onSessionMatchFailed?: (message: string) => void } | undefined; - scannerCall?.onSessionMatchFailed?.(harness.scannerFailureMessage); - - expect(harness.launches).toHaveLength(1); + expect(harness.launches.length).toBeGreaterThan(0); expect(getLocalLaunchFailure()).toBeNull(); expect(sessionEvents).toContainEqual({ type: 'message', - message: `${harness.scannerFailureMessage} Keeping local Codex running; remote transcript sync may be unavailable for this launch.` + message: 'No Codex SessionStart hook transcript path received within 10000ms. Keeping local Codex running; remote transcript sync may be unavailable for this launch.' }); }); + + it('does not reuse a stale transcript path from a previous launch', async () => { + const staleTranscriptPath = join(tempDir, 'stale-transcript.jsonl'); + const { session, sessionEvents } = createSessionStub('default', undefined, '/tmp/worktree', staleTranscriptPath); + + vi.useFakeTimers(); + const launcherPromise = codexLocalLauncher(session as never); + await Promise.resolve(); + await Promise.resolve(); + expect(session.transcriptPath).toBeNull(); + + vi.advanceTimersByTime(10_000); + await Promise.resolve(); + await launcherPromise; + + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'No Codex SessionStart hook transcript path received within 10000ms. Keeping local Codex running; remote transcript sync may be unavailable for this launch.' + }); + }); + + it('passes SessionStart hook config into local Codex launch', async () => { + const { session } = createSessionStub('default'); + + await codexLocalLauncher(session as never); + + expect(harness.launches).toHaveLength(1); + expect(harness.launches[0]?.sessionHook).toEqual({ + port: 4242, + token: 'hook-token' + }); + }); + + it('creates scanner only after transcript path arrives from SessionStart hook', async () => { + const transcriptPath = join(tempDir, 'codex-transcript.jsonl'); + const { session, agentMessages } = createSessionStub('default'); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + await writeFile( + transcriptPath, + JSON.stringify({ type: 'session_meta', payload: { id: 'codex-thread-1' } }) + '\n' + ); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + expect(session.transcriptPath).toBeNull(); + expect(agentMessages).toHaveLength(0); + + harness.sessionHookHandlers[0]?.('codex-thread-1', { + transcript_path: transcriptPath + }); + await wait(100); + + await appendFile( + transcriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'hello from transcript' } }) + '\n' + ); + + await wait(700); + if (releaseRunBarrier) { + releaseRunBarrier(); + } + await launcherPromise; + + expect(session.transcriptPath).toBe(transcriptPath); + expect(agentMessages).toContainEqual({ + type: 'message', + message: 'hello from transcript', + id: expect.any(String) + }); + }); + + it('does not leave transcript scanning alive after launcher teardown', async () => { + const transcriptPath = join(tempDir, 'teardown-race-transcript.jsonl'); + const { session, agentMessages } = createSessionStub('default'); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + const oldLines = Array.from({ length: 20_000 }, (_, index) => + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: `old-${index}` } }) + ).join('\n'); + await writeFile(transcriptPath, oldLines + '\n'); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[0]?.('codex-thread-race', { + transcript_path: transcriptPath + }); + if (releaseRunBarrier) { + releaseRunBarrier(); + } + await launcherPromise; + + await appendFile( + transcriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'post-teardown' } }) + '\n' + ); + await wait(2300); + + expect(agentMessages).toHaveLength(0); + }); + + it('ignores late SessionStart hooks after shutdown begins', async () => { + const staleTranscriptPath = join(tempDir, 'late-hook-transcript.jsonl'); + const { session } = createSessionStub('default'); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + if (releaseRunBarrier) { + releaseRunBarrier(); + } + await launcherPromise; + + harness.sessionHookHandlers[0]?.('late-local-thread', { + transcript_path: staleTranscriptPath + }); + + expect(session.sessionId).toBeNull(); + expect(session.transcriptPath).toBeNull(); + }); }); diff --git a/cli/src/codex/codexLocalLauncher.ts b/cli/src/codex/codexLocalLauncher.ts index e13ae75b..0f59bc5e 100644 --- a/cli/src/codex/codexLocalLauncher.ts +++ b/cli/src/codex/codexLocalLauncher.ts @@ -1,17 +1,23 @@ import { logger } from '@/ui/logger'; +import { startHookServer } from '@/claude/utils/startHookServer'; import { codexLocal } from './codexLocal'; import type { ReasoningEffort } from './appServerTypes'; import { CodexSession } from './session'; -import { createCodexSessionScanner } from './utils/codexSessionScanner'; +import { createCodexSessionScanner, type CodexSessionScanner } from './utils/codexSessionScanner'; import { convertCodexEvent } from './utils/codexEventConverter'; import { buildHapiMcpBridge } from './utils/buildHapiMcpBridge'; import { stripCodexCliOverrides } from './utils/codexCliOverrides'; import { buildCodexPermissionModeCliArgs } from './utils/permissionModeConfig'; import { BaseLocalLauncher } from '@/modules/common/launcher/BaseLocalLauncher'; +const SESSION_HOOK_TIMEOUT_MS = 10_000; + export async function codexLocalLauncher(session: CodexSession): Promise<'switch' | 'exit'> { const resumeSessionId = session.sessionId; - let scanner: Awaited> | null = null; + let scanner: CodexSessionScanner | null = null; + let hookReady = false; + let shuttingDown = false; + let pendingScannerSetup: Promise | null = null; const permissionMode = session.getPermissionMode(); const managedPermissionMode = permissionMode === 'read-only' || permissionMode === 'safe-yolo' || permissionMode === 'yolo' ? permissionMode @@ -27,9 +33,99 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch const { server: happyServer, mcpServers } = await buildHapiMcpBridge(session.client); logger.debug(`[codex-local]: Started hapi MCP bridge server at ${happyServer.url}`); + const hookServer = await startHookServer({ + onSessionHook: (sessionId, data) => { + if (shuttingDown) { + return; + } + session.onSessionFound(sessionId); + if (typeof data.transcript_path === 'string' && data.transcript_path.length > 0) { + hookReady = true; + session.onTranscriptPathFound(data.transcript_path); + } + } + }); + logger.debug(`[codex-local]: Started Codex SessionStart hook server on port ${hookServer.port}`); + + let hookTimeout: ReturnType | null = setTimeout(() => { + hookTimeout = null; + const message = `No Codex SessionStart hook transcript path received within ${SESSION_HOOK_TIMEOUT_MS}ms.`; + logger.warn(`[codex-local]: ${message}`); + session.sendSessionEvent({ + type: 'message', + message: `${message} Keeping local Codex running; remote transcript sync may be unavailable for this launch.` + }); + }, SESSION_HOOK_TIMEOUT_MS); + + const clearHookTimeout = () => { + if (hookTimeout) { + clearTimeout(hookTimeout); + hookTimeout = null; + } + }; + + const reportTranscriptSyncFailure = (transcriptPath: string, error: unknown): void => { + const detail = error instanceof Error ? error.message : String(error); + const message = `Codex transcript sync failed for ${transcriptPath}: ${detail}`; + logger.warn(`[codex-local]: ${message}`); + session.sendSessionEvent({ + type: 'message', + message: `${message} Keeping local Codex running; remote transcript sync is unavailable for this launch.` + }); + }; + const handleSessionFound = (sessionId: string) => { session.onSessionFound(sessionId); - scanner?.onNewSession(sessionId); + }; + + const processTranscriptPath = async (transcriptPath: string): Promise => { + hookReady = true; + clearHookTimeout(); + if (shuttingDown) { + return; + } + if (scanner) { + await scanner.setTranscriptPath(transcriptPath); + return; + } + const createdScanner = await createCodexSessionScanner({ + transcriptPath, + onSessionId: (sessionId) => { + session.onSessionFound(sessionId); + }, + onEvent: (event) => { + const converted = convertCodexEvent(event); + if (converted?.sessionId) { + session.onSessionFound(converted.sessionId); + } + if (converted?.userMessage) { + session.sendUserMessage(converted.userMessage); + } + if (converted?.message) { + session.sendAgentMessage(converted.message); + } + } + }); + if (shuttingDown) { + await createdScanner.cleanup(); + return; + } + scanner = createdScanner; + }; + + const handleTranscriptPath = (transcriptPath: string): Promise => { + const setupTask = (pendingScannerSetup ?? Promise.resolve()).then(() => processTranscriptPath(transcriptPath)); + const observedTask = setupTask.catch((error) => { + if (!shuttingDown) { + reportTranscriptSyncFailure(transcriptPath, error); + } + }); + pendingScannerSetup = observedTask.finally(() => { + if (pendingScannerSetup === observedTask) { + pendingScannerSetup = null; + } + }); + return pendingScannerSetup; }; const launcher = new BaseLocalLauncher({ @@ -47,7 +143,11 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch onSessionFound: handleSessionFound, abort: abortSignal, codexArgs, - mcpServers + mcpServers, + sessionHook: { + port: hookServer.port, + token: hookServer.token + } }); }, sendFailureMessage: (message) => { @@ -60,42 +160,30 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch switchLogMessage: 'doSwitch' }); - const handleSessionMatchFailed = (message: string) => { - logger.warn(`[codex-local]: ${message}`); - session.sendSessionEvent({ - type: 'message', - message: `${message} Keeping local Codex running; remote transcript sync may be unavailable for this launch.` - }); + session.resetTranscriptPath(); + const handleTranscriptPathCallback = (transcriptPath: string) => { + void handleTranscriptPath(transcriptPath); }; - - scanner = await createCodexSessionScanner({ - sessionId: resumeSessionId, - cwd: session.path, - startupTimestampMs: Date.now(), - onSessionMatchFailed: handleSessionMatchFailed, - onSessionFound: (sessionId) => { - session.onSessionFound(sessionId); - }, - onEvent: (event) => { - const converted = convertCodexEvent(event); - if (converted?.sessionId) { - session.onSessionFound(converted.sessionId); - scanner?.onNewSession(converted.sessionId); - } - if (converted?.userMessage) { - session.sendUserMessage(converted.userMessage); - } - if (converted?.message) { - session.sendAgentMessage(converted.message); - } - } - }); + session.addTranscriptPathCallback(handleTranscriptPathCallback); try { return await launcher.run(); } finally { - await scanner?.cleanup(); + shuttingDown = true; + clearHookTimeout(); + session.removeTranscriptPathCallback(handleTranscriptPathCallback); + hookServer.stop(); + if (pendingScannerSetup) { + await pendingScannerSetup; + } + const activeScanner = scanner as CodexSessionScanner | null; + if (activeScanner) { + await activeScanner.cleanup(); + } happyServer.stop(); + if (!hookReady) { + logger.debug('[codex-local]: SessionStart hook did not provide transcript path before shutdown'); + } logger.debug('[codex-local]: Stopped hapi MCP bridge server'); } } diff --git a/cli/src/codex/session.ts b/cli/src/codex/session.ts index c490d748..524d21d7 100644 --- a/cli/src/codex/session.ts +++ b/cli/src/codex/session.ts @@ -12,12 +12,15 @@ type LocalLaunchFailure = { }; export class CodexSession extends AgentSessionBase { + transcriptPath: string | null = null; readonly codexArgs?: string[]; readonly codexCliOverrides?: CodexCliOverrides; readonly startedBy: 'runner' | 'terminal'; readonly startingMode: 'local' | 'remote'; localLaunchFailure: LocalLaunchFailure | null = null; + private transcriptPathCallbacks: Array<(path: string) => void> = []; + constructor(opts: { api: ApiClient; client: ApiSessionClient; @@ -67,6 +70,31 @@ export class CodexSession extends AgentSessionBase { this.collaborationMode = opts.collaborationMode; } + onTranscriptPathFound(path: string): void { + if (this.transcriptPath === path) { + return; + } + this.transcriptPath = path; + for (const callback of this.transcriptPathCallbacks) { + callback(path); + } + } + + addTranscriptPathCallback(cb: (path: string) => void): void { + this.transcriptPathCallbacks.push(cb); + } + + removeTranscriptPathCallback(cb: (path: string) => void): void { + const index = this.transcriptPathCallbacks.indexOf(cb); + if (index !== -1) { + this.transcriptPathCallbacks.splice(index, 1); + } + } + + resetTranscriptPath(): void { + this.transcriptPath = null; + } + setPermissionMode = (mode: PermissionMode): void => { this.permissionMode = mode; }; diff --git a/cli/src/codex/utils/codexMcpConfig.test.ts b/cli/src/codex/utils/codexMcpConfig.test.ts index 286c739b..e91d5f2f 100644 --- a/cli/src/codex/utils/codexMcpConfig.test.ts +++ b/cli/src/codex/utils/codexMcpConfig.test.ts @@ -1,5 +1,9 @@ import { describe, it, expect } from 'vitest'; -import { buildMcpServerConfigArgs, buildDeveloperInstructionsArg } from './codexMcpConfig'; +import { + buildMcpServerConfigArgs, + buildDeveloperInstructionsArg, + buildSessionStartHookConfigArgs +} from './codexMcpConfig'; describe('codexMcpConfig', () => { describe('buildMcpServerConfigArgs', () => { @@ -90,4 +94,15 @@ describe('codexMcpConfig', () => { expect(args[1]).toContain('\\\\'); }); }); + + describe('buildSessionStartHookConfigArgs', () => { + it('builds a SessionStart hook config override', () => { + const args = buildSessionStartHookConfigArgs(4312, 'secret-token'); + + expect(args[0]).toBe('-c'); + expect(args[1]).toContain('hooks.SessionStart=['); + expect(args[1]).toContain('type = "command"'); + expect(args[1]).toContain('hook-forwarder --port 4312 --token secret-token'); + }); + }); }); diff --git a/cli/src/codex/utils/codexMcpConfig.ts b/cli/src/codex/utils/codexMcpConfig.ts index a4648670..5228703f 100644 --- a/cli/src/codex/utils/codexMcpConfig.ts +++ b/cli/src/codex/utils/codexMcpConfig.ts @@ -1,12 +1,14 @@ /** * Utilities for building Codex CLI config arguments (-c) for MCP servers - * and developer instructions. + * hooks, MCP servers, and developer instructions. * * Codex CLI accepts -c / --config flags with TOML-formatted key=value pairs. * This module generates the appropriate arguments for passing MCP server * configuration and developer instructions at runtime. */ +import { getHappyCliCommand } from '@/utils/spawnHappyCLI'; + /** * Escape a string value for use in a TOML string literal. * Handles double quotes, backslashes, and newlines. @@ -33,6 +35,36 @@ function buildTomlLiteralArray(values: string[]): string { return `[${items.join(',')}]`; } +function shellQuote(value: string): string { + if (value.length === 0) { + return '""'; + } + + if (/^[A-Za-z0-9_\/:=-]+$/.test(value)) { + return value; + } + + return '"' + value.replace(/(["\\$`])/g, '\\$1') + '"'; +} + +function shellJoin(parts: string[]): string { + return parts.map(shellQuote).join(' '); +} + +export function buildSessionStartHookConfigArgs(port: number, token: string): string[] { + const { command, args } = getHappyCliCommand([ + 'hook-forwarder', + '--port', + String(port), + '--token', + token + ]); + const hookCommand = shellJoin([command, ...args]); + const escapedHookCommand = escapeTomlString(hookCommand); + const hookConfig = `hooks.SessionStart=[{ hooks = [{ type = "command", command = "${escapedHookCommand}" }] }]`; + return ['-c', hookConfig]; +} + /** * Build -c arguments for MCP server configuration. * diff --git a/cli/src/codex/utils/codexSessionScanner.test.ts b/cli/src/codex/utils/codexSessionScanner.test.ts index cca21df3..166fee7c 100644 --- a/cli/src/codex/utils/codexSessionScanner.test.ts +++ b/cli/src/codex/utils/codexSessionScanner.test.ts @@ -1,8 +1,8 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { mkdir, writeFile, appendFile, rm } from 'node:fs/promises'; +import { appendFile, mkdir, rm, writeFile } from 'node:fs/promises'; +import { existsSync } from 'node:fs'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; -import { existsSync } from 'node:fs'; import { createCodexSessionScanner } from './codexSessionScanner'; import type { CodexSessionEvent } from './codexEventConverter'; @@ -10,20 +10,14 @@ const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); describe('codexSessionScanner', () => { let testDir: string; - let sessionsDir: string; - let sessionFile: string; - let originalCodexHome: string | undefined; + let transcriptPath: string; let scanner: Awaited> | null = null; let events: CodexSessionEvent[] = []; beforeEach(async () => { testDir = join(tmpdir(), `codex-scanner-${Date.now()}`); - sessionsDir = join(testDir, 'sessions', '2025', '12', '22'); - await mkdir(sessionsDir, { recursive: true }); - - originalCodexHome = process.env.CODEX_HOME; - process.env.CODEX_HOME = testDir; - + await mkdir(testDir, { recursive: true }); + transcriptPath = join(testDir, 'codex-session.jsonl'); events = []; }); @@ -33,247 +27,119 @@ describe('codexSessionScanner', () => { scanner = null; } - if (originalCodexHome === undefined) { - delete process.env.CODEX_HOME; - } else { - process.env.CODEX_HOME = originalCodexHome; - } - if (existsSync(testDir)) { await rm(testDir, { recursive: true, force: true }); } }); it('emits only new events after startup', async () => { - const sessionId = 'session-123'; - sessionFile = join(sessionsDir, `codex-${sessionId}.jsonl`); - - const initialLines = [ - JSON.stringify({ type: 'session_meta', payload: { id: sessionId } }), - JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'hello' } }) - ]; - - await writeFile(sessionFile, initialLines.join('\n') + '\n'); + await writeFile( + transcriptPath, + [ + JSON.stringify({ type: 'session_meta', payload: { id: 'session-123' } }), + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'old' } }) + ].join('\n') + '\n' + ); scanner = await createCodexSessionScanner({ - sessionId, + transcriptPath, onEvent: (event) => events.push(event) }); - await wait(150); + await wait(300); expect(events).toHaveLength(0); - const newLine = JSON.stringify({ - type: 'response_item', - payload: { type: 'function_call', name: 'Tool', call_id: 'call-1', arguments: '{}' } - }); - await appendFile(sessionFile, newLine + '\n'); + await appendFile( + transcriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'new' } }) + '\n' + ); - await wait(200); + await wait(700); expect(events).toHaveLength(1); - expect(events[0].type).toBe('response_item'); + expect(events[0]?.type).toBe('event_msg'); }); - it('limits session scan to dates within the start window', async () => { - const referenceTimestampMs = Date.parse('2025-12-22T00:00:00.000Z'); - const windowMs = 2 * 60 * 1000; - const matchingSessionId = 'session-222'; - const outsideSessionId = 'session-999'; - const outsideDir = join(testDir, 'sessions', '2025', '12', '20'); - const matchingFile = join(sessionsDir, `codex-${matchingSessionId}.jsonl`); - const outsideFile = join(outsideDir, `codex-${outsideSessionId}.jsonl`); - - await mkdir(outsideDir, { recursive: true }); - const baseLines = [ - JSON.stringify({ type: 'session_meta', payload: { id: matchingSessionId, cwd: '/data/github/happy/hapi', timestamp: '2025-12-22T00:00:30.000Z' } }), - JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'hello' } }) - ]; - await writeFile(matchingFile, baseLines.join('\n') + '\n'); + it('reports session id from the transcript metadata', async () => { await writeFile( - outsideFile, - JSON.stringify({ type: 'session_meta', payload: { id: outsideSessionId, cwd: '/data/github/happy/hapi', timestamp: '2025-12-20T00:00:00.000Z' } }) + '\n' + transcriptPath, + JSON.stringify({ type: 'session_meta', payload: { id: 'session-xyz' } }) + '\n' + ); + + let observedSessionId: string | null = null; + scanner = await createCodexSessionScanner({ + transcriptPath, + onEvent: (event) => events.push(event), + onSessionId: (sessionId) => { + observedSessionId = sessionId; + } + }); + + expect(observedSessionId).toBe('session-xyz'); + expect(events).toHaveLength(0); + }); + + it('switches to a newly supplied transcript path without replaying history', async () => { + const firstTranscriptPath = join(testDir, 'first.jsonl'); + const secondTranscriptPath = join(testDir, 'second.jsonl'); + + await writeFile( + firstTranscriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'first-old' } }) + '\n' + ); + await writeFile( + secondTranscriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'second-old' } }) + '\n' ); scanner = await createCodexSessionScanner({ - sessionId: null, - cwd: '/data/github/happy/hapi', - startupTimestampMs: referenceTimestampMs, - sessionStartWindowMs: windowMs, + transcriptPath: firstTranscriptPath, onEvent: (event) => events.push(event) }); - await wait(200); + await wait(300); expect(events).toHaveLength(0); - const newLine = JSON.stringify({ - type: 'response_item', - payload: { type: 'function_call', name: 'Tool', call_id: 'call-2', arguments: '{}' } - }); - await appendFile(matchingFile, newLine + '\n'); - - await wait(200); + await appendFile( + firstTranscriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'first-new' } }) + '\n' + ); + await wait(700); expect(events).toHaveLength(1); - expect(events[0].type).toBe('response_item'); - }); - it('fails fast when cwd is missing and no sessionId is provided', async () => { - const sessionId = 'session-missing-cwd'; - const matchFailedMessage = 'No cwd provided for Codex session matching; refusing to fallback.'; - sessionFile = join(sessionsDir, `codex-${sessionId}.jsonl`); - - await writeFile( - sessionFile, - JSON.stringify({ type: 'session_meta', payload: { id: sessionId } }) + '\n' - ); - - let failureMessage: string | null = null; - scanner = await createCodexSessionScanner({ - sessionId: null, - onEvent: (event) => events.push(event), - onSessionMatchFailed: (message) => { - failureMessage = message; - } - }); - - await wait(150); - expect(failureMessage).toBe(matchFailedMessage); - expect(events).toHaveLength(0); - - const newLine = JSON.stringify({ - type: 'response_item', - payload: { type: 'function_call', name: 'Tool', call_id: 'call-3', arguments: '{}' } - }); - await appendFile(sessionFile, newLine + '\n'); - - await wait(200); - expect(events).toHaveLength(0); - }); - - it('adopts a reused older session file when fresh matching activity appears after startup', async () => { - const reusedSessionId = 'session-reused-old-file'; - const targetCwd = '/data/github/happy/hapi'; - const startupTimestampMs = Date.now(); - const now = new Date(startupTimestampMs); - const currentSessionsDir = join( - testDir, - 'sessions', - String(now.getFullYear()), - String(now.getMonth() + 1).padStart(2, '0'), - String(now.getDate()).padStart(2, '0') - ); - await mkdir(currentSessionsDir, { recursive: true }); - sessionFile = join(currentSessionsDir, `codex-${reusedSessionId}.jsonl`); - - await writeFile( - sessionFile, - JSON.stringify({ - type: 'session_meta', - payload: { - id: reusedSessionId, - cwd: targetCwd, - timestamp: new Date(startupTimestampMs - 10 * 60 * 1000).toISOString() - } - }) + '\n' - ); - - let matchedSessionId: string | null = null; - scanner = await createCodexSessionScanner({ - sessionId: null, - cwd: targetCwd, - startupTimestampMs, - onEvent: (event) => events.push(event), - onSessionFound: (sessionId) => { - matchedSessionId = sessionId; - } - }); - - await wait(150); - expect(events).toHaveLength(0); - expect(matchedSessionId).toBeNull(); - - const newLine = JSON.stringify({ - type: 'response_item', - payload: { type: 'function_call', name: 'Tool', call_id: 'call-reused', arguments: '{}' } - }); - await appendFile(sessionFile, newLine + '\n'); - - await wait(2300); - expect(matchedSessionId).toBe(reusedSessionId); + await scanner.setTranscriptPath(secondTranscriptPath); + await wait(300); expect(events).toHaveLength(1); - expect(events[0].type).toBe('response_item'); + + await appendFile( + secondTranscriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'second-new' } }) + '\n' + ); + await wait(700); + expect(events).toHaveLength(2); + expect(events[1]?.payload).toEqual({ type: 'agent_message', message: 'second-new' }); }); - it('does not adopt a reused session when first fresh matching activity is ambiguous', async () => { - const targetCwd = '/data/github/happy/hapi'; - const startupTimestampMs = Date.now(); - const now = new Date(startupTimestampMs); - const currentSessionsDir = join( - testDir, - 'sessions', - String(now.getFullYear()), - String(now.getMonth() + 1).padStart(2, '0'), - String(now.getDate()).padStart(2, '0') - ); - await mkdir(currentSessionsDir, { recursive: true }); - - const firstSessionId = 'session-reused-a'; - const secondSessionId = 'session-reused-b'; - const firstFile = join(currentSessionsDir, `codex-${firstSessionId}.jsonl`); - const secondFile = join(currentSessionsDir, `codex-${secondSessionId}.jsonl`); - const oldTimestamp = new Date(startupTimestampMs - 10 * 60 * 1000).toISOString(); - + it('resets line cursor when the transcript file is truncated', async () => { await writeFile( - firstFile, - JSON.stringify({ - type: 'session_meta', - payload: { id: firstSessionId, cwd: targetCwd, timestamp: oldTimestamp } - }) + '\n' - ); - await writeFile( - secondFile, - JSON.stringify({ - type: 'session_meta', - payload: { id: secondSessionId, cwd: targetCwd, timestamp: oldTimestamp } - }) + '\n' + transcriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'before-truncate' } }) + '\n' ); - let matchedSessionId: string | null = null; scanner = await createCodexSessionScanner({ - sessionId: null, - cwd: targetCwd, - startupTimestampMs, - onEvent: (event) => events.push(event), - onSessionFound: (sessionId) => { - matchedSessionId = sessionId; - } + transcriptPath, + onEvent: (event) => events.push(event) }); - await wait(150); - expect(matchedSessionId).toBeNull(); - - const firstNewLine = JSON.stringify({ - type: 'response_item', - payload: { type: 'function_call', name: 'Tool', call_id: 'call-reused-a-1', arguments: '{}' } - }); - const secondNewLine = JSON.stringify({ - type: 'response_item', - payload: { type: 'function_call', name: 'Tool', call_id: 'call-reused-b-1', arguments: '{}' } - }); - await appendFile(firstFile, firstNewLine + '\n'); - await appendFile(secondFile, secondNewLine + '\n'); - - await wait(2300); - expect(matchedSessionId).toBeNull(); + await wait(300); expect(events).toHaveLength(0); - const laterUniqueLine = JSON.stringify({ - type: 'response_item', - payload: { type: 'function_call', name: 'Tool', call_id: 'call-reused-a-2', arguments: '{}' } - }); - await appendFile(firstFile, laterUniqueLine + '\n'); + await writeFile( + transcriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'after-truncate' } }) + '\n' + ); - await wait(2300); - expect(matchedSessionId).toBeNull(); - expect(events).toHaveLength(0); + await wait(700); + expect(events).toHaveLength(1); + expect(events[0]?.payload).toEqual({ type: 'agent_message', message: 'after-truncate' }); }); }); diff --git a/cli/src/codex/utils/codexSessionScanner.ts b/cli/src/codex/utils/codexSessionScanner.ts index fd8f45b0..2c09c5dc 100644 --- a/cli/src/codex/utils/codexSessionScanner.ts +++ b/cli/src/codex/utils/codexSessionScanner.ts @@ -1,305 +1,99 @@ -import { BaseSessionScanner, SessionFileScanEntry, SessionFileScanResult, SessionFileScanStats } from "@/modules/common/session/BaseSessionScanner"; -import { logger } from "@/ui/logger"; -import { join, relative, resolve, sep } from "node:path"; -import { homedir } from "node:os"; -import { readFile, readdir, stat } from "node:fs/promises"; -import type { CodexSessionEvent } from "./codexEventConverter"; +import { readFile } from 'node:fs/promises'; +import { BaseSessionScanner, SessionFileScanEntry, SessionFileScanResult, SessionFileScanStats } from '@/modules/common/session/BaseSessionScanner'; +import { logger } from '@/ui/logger'; +import type { CodexSessionEvent } from './codexEventConverter'; interface CodexSessionScannerOptions { - sessionId: string | null; + transcriptPath: string | null; onEvent: (event: CodexSessionEvent) => void; - onSessionFound?: (sessionId: string) => void; - onSessionMatchFailed?: (message: string) => void; - cwd?: string; - startupTimestampMs?: number; - sessionStartWindowMs?: number; + onSessionId?: (sessionId: string) => void; } -interface CodexSessionScanner { +export interface CodexSessionScanner { cleanup: () => Promise; - onNewSession: (sessionId: string) => void; + setTranscriptPath: (transcriptPath: string) => Promise; } -type PendingEvents = { - events: CodexSessionEvent[]; - fileSessionId: string | null; -}; - -type Candidate = { - sessionId: string; - score: number; -}; - -const DEFAULT_SESSION_START_WINDOW_MS = 2 * 60 * 1000; - export async function createCodexSessionScanner(opts: CodexSessionScannerOptions): Promise { - const targetCwd = opts.cwd && opts.cwd.trim().length > 0 ? normalizePath(opts.cwd) : null; - - if (!targetCwd && !opts.sessionId) { - const message = 'No cwd provided for Codex session matching; refusing to fallback.'; - logger.warn(`[CODEX_SESSION_SCANNER] ${message}`); - opts.onSessionMatchFailed?.(message); - return { - cleanup: async () => {}, - onNewSession: () => {} - }; - } - - const scanner = new CodexSessionScannerImpl(opts, targetCwd); + const scanner = new CodexSessionScannerImpl(opts); await scanner.start(); return { cleanup: async () => { await scanner.cleanup(); }, - onNewSession: (sessionId: string) => { - scanner.onNewSession(sessionId); + setTranscriptPath: async (transcriptPath: string) => { + await scanner.setTranscriptPath(transcriptPath); } }; } class CodexSessionScannerImpl extends BaseSessionScanner { - private readonly sessionsRoot: string; + private transcriptPath: string | null; private readonly onEvent: (event: CodexSessionEvent) => void; - private readonly onSessionFound?: (sessionId: string) => void; - private readonly onSessionMatchFailed?: (message: string) => void; - private readonly sessionIdByFile = new Map(); - private readonly sessionCwdByFile = new Map(); - private readonly sessionTimestampByFile = new Map(); - private readonly pendingEventsByFile = new Map(); - private readonly sessionMetaParsed = new Set(); + private readonly onSessionId?: (sessionId: string) => void; private readonly fileEpochByPath = new Map(); - private readonly targetCwd: string | null; - private readonly referenceTimestampMs: number; - private readonly sessionStartWindowMs: number; - private readonly matchDeadlineMs: number; - private readonly sessionDatePrefixes: Set | null; + private readonly fileSizeByPath = new Map(); + private observedSessionId: string | null = null; - private activeSessionId: string | null; - private reportedSessionId: string | null; - private matchFailed = false; - private bestWithinWindow: Candidate | null = null; - private readonly recentActivitySessionIds = new Set(); - private firstRecentActivityCandidateResolved = false; - private readonly firstRecentActivitySessionIds = new Set(); - private loggedAmbiguousRecentActivity = false; - - constructor(opts: CodexSessionScannerOptions, targetCwd: string | null) { + constructor(opts: CodexSessionScannerOptions) { super({ intervalMs: 2000 }); - const codexHomeDir = process.env.CODEX_HOME || join(homedir(), '.codex'); - this.sessionsRoot = join(codexHomeDir, 'sessions'); + this.transcriptPath = opts.transcriptPath; this.onEvent = opts.onEvent; - this.onSessionFound = opts.onSessionFound; - this.onSessionMatchFailed = opts.onSessionMatchFailed; - this.activeSessionId = opts.sessionId; - this.reportedSessionId = opts.sessionId; - this.targetCwd = targetCwd; - this.referenceTimestampMs = opts.startupTimestampMs ?? Date.now(); - this.sessionStartWindowMs = opts.sessionStartWindowMs ?? DEFAULT_SESSION_START_WINDOW_MS; - this.matchDeadlineMs = this.referenceTimestampMs + this.sessionStartWindowMs; - this.sessionDatePrefixes = this.targetCwd - ? getSessionDatePrefixes(this.referenceTimestampMs, this.sessionStartWindowMs) - : null; - - logger.debug(`[CODEX_SESSION_SCANNER] Init: targetCwd=${this.targetCwd ?? 'none'} startupTs=${new Date(this.referenceTimestampMs).toISOString()} windowMs=${this.sessionStartWindowMs}`); + this.onSessionId = opts.onSessionId; } - public onNewSession(sessionId: string): void { - if (this.activeSessionId === sessionId) { + async setTranscriptPath(transcriptPath: string): Promise { + if (this.transcriptPath === transcriptPath) { return; } - logger.debug(`[CODEX_SESSION_SCANNER] Switching to new session: ${sessionId}`); - this.setActiveSessionId(sessionId); + this.transcriptPath = transcriptPath; + await this.primeTranscript(transcriptPath); + this.pruneWatchers(this.transcriptPath ? [this.transcriptPath] : []); this.invalidate(); } - protected shouldScan(): boolean { - return !this.matchFailed; - } - - protected shouldWatchFile(filePath: string): boolean { - if (!this.activeSessionId) { - if (!this.targetCwd) { - return false; - } - return this.getCandidateForFile(filePath) !== null; - } - const fileSessionId = this.sessionIdByFile.get(filePath); - if (fileSessionId) { - return fileSessionId === this.activeSessionId; - } - return filePath.endsWith(`-${this.activeSessionId}.jsonl`); - } - protected async initialize(): Promise { - const files = await this.listSessionFiles(this.sessionsRoot); - for (const filePath of files) { - const { nextCursor } = await this.readSessionFile(filePath, 0); - this.setCursor(filePath, nextCursor); - if (this.shouldWatchFile(filePath)) { - this.ensureWatcher(filePath); - } + if (this.transcriptPath) { + await this.primeTranscript(this.transcriptPath); } } - protected async beforeScan(): Promise { - this.bestWithinWindow = null; - this.recentActivitySessionIds.clear(); - } - protected async findSessionFiles(): Promise { - const files = await this.listSessionFiles(this.sessionsRoot); - return sortFilesByMtime(files); + if (!this.transcriptPath) { + return []; + } + return [this.transcriptPath]; + } + + protected shouldWatchFile(filePath: string): boolean { + return Boolean(this.transcriptPath && filePath === this.transcriptPath); } protected async parseSessionFile(filePath: string, cursor: number): Promise> { - if (this.shouldSkipFile(filePath)) { - return { events: [], nextCursor: cursor }; - } return this.readSessionFile(filePath, cursor); } - protected generateEventKey(event: CodexSessionEvent, context: { filePath: string; lineIndex?: number }): string { + protected generateEventKey(_event: CodexSessionEvent, context: { filePath: string; lineIndex?: number }): string { const epoch = this.fileEpochByPath.get(context.filePath) ?? 0; - const lineIndex = context.lineIndex ?? -1; - return `${context.filePath}:${epoch}:${lineIndex}`; + return `${context.filePath}:${epoch}:${context.lineIndex ?? -1}`; } protected async handleFileScan(stats: SessionFileScanStats): Promise { - const filePath = stats.filePath; - const fileSessionId = this.sessionIdByFile.get(filePath) ?? null; - - if (!this.activeSessionId && this.targetCwd) { - this.appendPendingEvents(filePath, stats.events, fileSessionId); - const candidate = this.getCandidateForFile(filePath); - if (candidate) { - if (!this.bestWithinWindow || candidate.score < this.bestWithinWindow.score) { - this.bestWithinWindow = candidate; - } - } - const recentActivityCandidate = this.getRecentActivityCandidateForFile(filePath, stats.newCount); - if (recentActivityCandidate) { - this.recentActivitySessionIds.add(recentActivityCandidate.sessionId); - } - if (stats.newCount > 0) { - logger.debug(`[CODEX_SESSION_SCANNER] Buffered ${stats.newCount} pending events from ${filePath}`); - } - return; + for (const event of stats.events) { + this.onEvent(event); } - - const emittedForFile = this.emitEvents(stats.events, fileSessionId); - if (emittedForFile > 0) { - logger.debug(`[CODEX_SESSION_SCANNER] Emitted ${emittedForFile} new events from ${filePath}`); + if (stats.newCount > 0) { + logger.debug(`[codex-session-scanner] ${stats.newCount} new events from ${stats.filePath}`); } + this.pruneWatchers(this.transcriptPath ? [this.transcriptPath] : []); } - protected async afterScan(): Promise { - if (!this.activeSessionId && this.targetCwd) { - if (this.bestWithinWindow) { - logger.debug(`[CODEX_SESSION_SCANNER] Selected session ${this.bestWithinWindow.sessionId} within start window`); - this.setActiveSessionId(this.bestWithinWindow.sessionId); - } else { - this.captureFirstRecentActivityCandidate(); - - if (this.firstRecentActivitySessionIds.size === 1) { - const [sessionId] = this.firstRecentActivitySessionIds; - if (sessionId) { - logger.debug(`[CODEX_SESSION_SCANNER] Selected session ${sessionId} from first unique matching activity after startup`); - this.setActiveSessionId(sessionId); - } - } else if ( - !this.loggedAmbiguousRecentActivity - && this.firstRecentActivityCandidateResolved - && this.firstRecentActivitySessionIds.size > 1 - ) { - this.loggedAmbiguousRecentActivity = true; - logger.debug('[CODEX_SESSION_SCANNER] First matching activity after startup was ambiguous; refusing reused-session adoption'); - } - - if (!this.activeSessionId) { - if (Date.now() > this.matchDeadlineMs) { - this.matchFailed = true; - this.pendingEventsByFile.clear(); - const message = `No Codex session found within ${this.sessionStartWindowMs}ms for cwd ${this.targetCwd}; refusing fallback.`; - logger.warn(`[CODEX_SESSION_SCANNER] ${message}`); - this.onSessionMatchFailed?.(message); - } else if (this.pendingEventsByFile.size > 0) { - logger.debug('[CODEX_SESSION_SCANNER] No session candidate matched yet; pending events buffered'); - } - } - } - } - } - - private captureFirstRecentActivityCandidate(): void { - if (this.firstRecentActivityCandidateResolved || this.recentActivitySessionIds.size === 0) { - return; - } - - this.firstRecentActivityCandidateResolved = true; - for (const sessionId of this.recentActivitySessionIds) { - this.firstRecentActivitySessionIds.add(sessionId); - } - } - - private shouldSkipFile(filePath: string): boolean { - if (!this.activeSessionId) { - return false; - } - const fileSessionId = this.sessionIdByFile.get(filePath); - if (fileSessionId && fileSessionId !== this.activeSessionId) { - return true; - } - if (!fileSessionId && !filePath.endsWith(`-${this.activeSessionId}.jsonl`)) { - return true; - } - return false; - } - - private reportSessionId(sessionId: string): void { - if (this.reportedSessionId === sessionId) { - return; - } - this.reportedSessionId = sessionId; - this.onSessionFound?.(sessionId); - } - - private setActiveSessionId(sessionId: string): void { - this.activeSessionId = sessionId; - this.reportSessionId(sessionId); - const candidateFiles = this.getFilesForSession(sessionId); - for (const filePath of candidateFiles) { - if (this.shouldWatchFile(filePath)) { - this.ensureWatcher(filePath); - } - } - this.pruneWatchers(this.getWatchedFiles().filter((filePath) => this.shouldWatchFile(filePath))); - if (this.targetCwd) { - this.flushPendingEventsForSession(sessionId); - } else { - this.pendingEventsByFile.clear(); - } - } - - private async listSessionFiles(dir: string): Promise { - try { - const entries = await readdir(dir, { withFileTypes: true }); - const results: string[] = []; - for (const entry of entries) { - const full = join(dir, entry.name); - if (!shouldIncludeSessionPath(full, this.sessionsRoot, this.sessionDatePrefixes)) { - continue; - } - if (entry.isDirectory()) { - results.push(...await this.listSessionFiles(full)); - } else if (entry.isFile() && entry.name.endsWith('.jsonl')) { - results.push(full); - } - } - return results; - } catch (error) { - return []; - } + private async primeTranscript(filePath: string): Promise { + const { events, nextCursor } = await this.readSessionFile(filePath, 0); + const keys = events.map((entry) => this.generateEventKey(entry.event, { filePath, lineIndex: entry.lineIndex })); + this.seedProcessedKeys(keys); + this.setCursor(filePath, nextCursor); } private async readSessionFile(filePath: string, startLine: number): Promise> { @@ -307,271 +101,92 @@ class CodexSessionScannerImpl extends BaseSessionScanner { try { content = await readFile(filePath, 'utf-8'); } catch (error) { + logger.debug(`[codex-session-scanner] Failed to read transcript ${filePath}: ${error}`); return { events: [], nextCursor: startLine }; } - const events: SessionFileScanEntry[] = []; const lines = content.split('\n'); const hasTrailingEmpty = lines.length > 0 && lines[lines.length - 1] === ''; const totalLines = hasTrailingEmpty ? lines.length - 1 : lines.length; + const currentSize = Buffer.byteLength(content); + const previousSize = this.fileSizeByPath.get(filePath); let effectiveStartLine = startLine; - if (effectiveStartLine > totalLines) { + + if ((previousSize !== undefined && currentSize < previousSize) || effectiveStartLine > totalLines) { effectiveStartLine = 0; const nextEpoch = (this.fileEpochByPath.get(filePath) ?? 0) + 1; this.fileEpochByPath.set(filePath, nextEpoch); } + this.fileSizeByPath.set(filePath, currentSize); - const hasSessionMeta = this.sessionMetaParsed.has(filePath); - const parseFrom = hasSessionMeta ? effectiveStartLine : 0; - - for (let index = parseFrom; index < lines.length; index += 1) { - const trimmed = lines[index].trim(); - if (!trimmed) { + const events: SessionFileScanEntry[] = []; + for (let lineIndex = 0; lineIndex < totalLines; lineIndex += 1) { + const line = lines[lineIndex]; + if (!line || line.trim().length === 0) { continue; } + + let parsed: unknown; try { - const parsed = JSON.parse(trimmed) as CodexSessionEvent; - if (parsed?.type === 'session_meta') { - const payload = asRecord(parsed.payload); - const sessionId = payload ? asString(payload.id) : null; - if (sessionId) { - this.sessionIdByFile.set(filePath, sessionId); - } - const sessionCwd = payload ? asString(payload.cwd) : null; - const normalizedCwd = sessionCwd ? normalizePath(sessionCwd) : null; - if (normalizedCwd) { - this.sessionCwdByFile.set(filePath, normalizedCwd); - } - const rawTimestamp = payload ? payload.timestamp : null; - const sessionTimestamp = payload ? parseTimestamp(payload.timestamp) : null; - if (sessionTimestamp !== null) { - this.sessionTimestampByFile.set(filePath, sessionTimestamp); - } - logger.debug(`[CODEX_SESSION_SCANNER] Session meta: file=${filePath} cwd=${sessionCwd ?? 'none'} normalizedCwd=${normalizedCwd ?? 'none'} timestamp=${rawTimestamp ?? 'none'} parsedTs=${sessionTimestamp ?? 'none'}`); - this.sessionMetaParsed.add(filePath); - } - if (index >= effectiveStartLine) { - events.push({ event: parsed, lineIndex: index }); - } + parsed = JSON.parse(line); } catch (error) { - logger.debug(`[CODEX_SESSION_SCANNER] Failed to parse line: ${error}`); - } - } - - return { events, nextCursor: totalLines }; - } - - private getCandidateForFile(filePath: string): Candidate | null { - const sessionId = this.sessionIdByFile.get(filePath); - if (!sessionId) { - return null; - } - - const fileCwd = this.sessionCwdByFile.get(filePath); - if (this.targetCwd && fileCwd !== this.targetCwd) { - return null; - } - - const sessionTimestamp = this.sessionTimestampByFile.get(filePath); - if (sessionTimestamp === undefined) { - return null; - } - - if (sessionTimestamp < this.referenceTimestampMs) { - return null; - } - - const diff = sessionTimestamp - this.referenceTimestampMs; - if (diff > this.sessionStartWindowMs) { - return null; - } - - return { - sessionId, - score: diff - }; - } - - private getRecentActivityCandidateForFile(filePath: string, newCount: number): Candidate | null { - if (newCount <= 0) { - return null; - } - - const sessionId = this.sessionIdByFile.get(filePath); - if (!sessionId) { - return null; - } - - const fileCwd = this.sessionCwdByFile.get(filePath); - if (this.targetCwd && fileCwd !== this.targetCwd) { - return null; - } - - return { - sessionId, - score: 0 - }; - } - - private getFilesForSession(sessionId: string): string[] { - const matches: string[] = []; - for (const [filePath, storedSessionId] of this.sessionIdByFile.entries()) { - if (storedSessionId === sessionId) { - matches.push(filePath); - } - } - if (matches.length > 0) { - return matches; - } - const suffix = `-${sessionId}.jsonl`; - return this.getWatchedFiles().filter((filePath) => filePath.endsWith(suffix)); - } - - private appendPendingEvents(filePath: string, events: CodexSessionEvent[], fileSessionId: string | null): void { - if (events.length === 0) { - return; - } - const existing = this.pendingEventsByFile.get(filePath); - if (existing) { - existing.events.push(...events); - if (!existing.fileSessionId && fileSessionId) { - existing.fileSessionId = fileSessionId; - } - return; - } - this.pendingEventsByFile.set(filePath, { - events: [...events], - fileSessionId - }); - } - - private emitEvents(events: CodexSessionEvent[], fileSessionId: string | null): number { - let emittedForFile = 0; - for (const event of events) { - const payload = asRecord(event.payload); - const payloadSessionId = payload ? asString(payload.id) : null; - const eventSessionId = payloadSessionId ?? fileSessionId ?? null; - - if (this.activeSessionId && eventSessionId && eventSessionId !== this.activeSessionId) { + logger.debug(`[codex-session-scanner] Failed to parse transcript line ${filePath}:${lineIndex + 1}: ${error}`); continue; } - this.onEvent(event); - emittedForFile += 1; - } - return emittedForFile; - } - - private flushPendingEventsForSession(sessionId: string): void { - if (this.pendingEventsByFile.size === 0) { - return; - } - let emitted = 0; - for (const [filePath, pending] of this.pendingEventsByFile.entries()) { - const matches = (pending.fileSessionId && pending.fileSessionId === sessionId) - || filePath.endsWith(`-${sessionId}.jsonl`); - if (!matches) { + const event = parseCodexSessionEvent(parsed); + if (!event) { continue; } - emitted += this.emitEvents(pending.events, pending.fileSessionId); + + if (event.type === 'session_meta') { + const sessionId = extractSessionId(event); + if (sessionId) { + this.updateSessionId(sessionId); + } + } + + if (lineIndex < effectiveStartLine) { + continue; + } + + events.push({ event, lineIndex }); } - this.pendingEventsByFile.clear(); - if (emitted > 0) { - logger.debug(`[CODEX_SESSION_SCANNER] Emitted ${emitted} pending events for session ${sessionId}`); + + return { + events, + nextCursor: totalLines + }; + } + + private updateSessionId(sessionId: string): void { + if (this.observedSessionId === sessionId) { + return; } + this.observedSessionId = sessionId; + this.onSessionId?.(sessionId); } } -async function sortFilesByMtime(files: string[]): Promise { - const entries = await Promise.all(files.map(async (file) => { - try { - const stats = await stat(file); - return { file, mtimeMs: stats.mtimeMs }; - } catch { - return { file, mtimeMs: 0 }; - } - })); - - return entries - .sort((a, b) => b.mtimeMs - a.mtimeMs) - .map((entry) => entry.file); -} - -function asRecord(value: unknown): Record | null { +function parseCodexSessionEvent(value: unknown): CodexSessionEvent | null { if (!value || typeof value !== 'object') { return null; } - return value as Record; + const record = value as Record; + if (typeof record.type !== 'string' || record.type.length === 0) { + return null; + } + return { + timestamp: typeof record.timestamp === 'string' ? record.timestamp : undefined, + type: record.type, + payload: record.payload + }; } -function asString(value: unknown): string | null { - return typeof value === 'string' && value.length > 0 ? value : null; -} - -function parseTimestamp(value: unknown): number | null { - if (typeof value === 'number' && Number.isFinite(value)) { - return value; +function extractSessionId(event: CodexSessionEvent): string | null { + if (!event.payload || typeof event.payload !== 'object') { + return null; } - if (typeof value === 'string' && value.length > 0) { - const parsed = Date.parse(value); - return Number.isNaN(parsed) ? null : parsed; - } - return null; -} - -function normalizePath(value: string): string { - const resolved = resolve(value); - return process.platform === 'win32' ? resolved.toLowerCase() : resolved; -} - -function getSessionDatePrefixes(referenceTimestampMs: number, windowMs: number): Set { - const startDate = new Date(referenceTimestampMs - windowMs); - const endDate = new Date(referenceTimestampMs + windowMs); - const current = new Date(startDate.getFullYear(), startDate.getMonth(), startDate.getDate()); - const last = new Date(endDate.getFullYear(), endDate.getMonth(), endDate.getDate()); - const prefixes = new Set(); - - while (current <= last) { - const year = String(current.getFullYear()); - const month = String(current.getMonth() + 1).padStart(2, '0'); - const day = String(current.getDate()).padStart(2, '0'); - prefixes.add(`${year}/${month}/${day}`); - current.setDate(current.getDate() + 1); - } - - return prefixes; -} - -function shouldIncludeSessionPath( - fullPath: string, - sessionsRoot: string, - prefixes: Set | null -): boolean { - if (!prefixes) { - return true; - } - - const relativePath = relative(sessionsRoot, fullPath); - if (!relativePath || relativePath.startsWith('..')) { - return true; - } - - const normalized = relativePath.split(sep).filter(Boolean).join('/'); - if (!normalized) { - return true; - } - - for (const prefix of prefixes) { - if (normalized === prefix) { - return true; - } - if (normalized.startsWith(`${prefix}/`)) { - return true; - } - if (prefix.startsWith(`${normalized}/`)) { - return true; - } - } - - return false; + const payload = event.payload as Record; + return typeof payload.id === 'string' && payload.id.length > 0 ? payload.id : null; }