mirror of
https://github.com/wu736139669/hapi.git
synced 2026-10-15 20:29:55 +00:00
refactor(codex): extract session management and launcher logic into separate modules
Reorganized runCodex.ts to improve maintainability by extracting: - CodexSession class for session lifecycle management - CodexLocalLauncher and CodexRemoteLauncher for mode-specific initialization - CodexEventConverter for MCP message handling and UI buffer updates - CodexSessionScanner for resume file discovery - emitReadyIfIdle utility for ready event emission Added codexSessionId field to metadata schema for session tracking. Updated UI components to work with refactored architecture.
This commit is contained in:
weishu
@@ -0,0 +1,85 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { convertCodexEvent } from './codexEventConverter';
|
||||
|
||||
describe('convertCodexEvent', () => {
|
||||
it('extracts session_meta id', () => {
|
||||
const result = convertCodexEvent({
|
||||
type: 'session_meta',
|
||||
payload: { id: 'session-123' }
|
||||
});
|
||||
|
||||
expect(result).toEqual({ sessionId: 'session-123' });
|
||||
});
|
||||
|
||||
it('converts agent_message events', () => {
|
||||
const result = convertCodexEvent({
|
||||
type: 'event_msg',
|
||||
payload: { type: 'agent_message', message: 'hello' }
|
||||
});
|
||||
|
||||
expect(result?.message).toMatchObject({
|
||||
type: 'message',
|
||||
message: 'hello'
|
||||
});
|
||||
});
|
||||
|
||||
it('converts reasoning events', () => {
|
||||
const result = convertCodexEvent({
|
||||
type: 'event_msg',
|
||||
payload: { type: 'agent_reasoning', text: 'thinking' }
|
||||
});
|
||||
|
||||
expect(result?.message).toMatchObject({
|
||||
type: 'reasoning',
|
||||
message: 'thinking'
|
||||
});
|
||||
});
|
||||
|
||||
it('converts reasoning delta events', () => {
|
||||
const result = convertCodexEvent({
|
||||
type: 'event_msg',
|
||||
payload: { type: 'agent_reasoning_delta', delta: 'step' }
|
||||
});
|
||||
|
||||
expect(result?.message).toEqual({
|
||||
type: 'reasoning-delta',
|
||||
delta: 'step'
|
||||
});
|
||||
});
|
||||
|
||||
it('converts function_call items', () => {
|
||||
const result = convertCodexEvent({
|
||||
type: 'response_item',
|
||||
payload: {
|
||||
type: 'function_call',
|
||||
name: 'ToolName',
|
||||
call_id: 'call-1',
|
||||
arguments: '{"foo":"bar"}'
|
||||
}
|
||||
});
|
||||
|
||||
expect(result?.message).toMatchObject({
|
||||
type: 'tool-call',
|
||||
name: 'ToolName',
|
||||
callId: 'call-1',
|
||||
input: { foo: 'bar' }
|
||||
});
|
||||
});
|
||||
|
||||
it('converts function_call_output items', () => {
|
||||
const result = convertCodexEvent({
|
||||
type: 'response_item',
|
||||
payload: {
|
||||
type: 'function_call_output',
|
||||
call_id: 'call-2',
|
||||
output: { ok: true }
|
||||
}
|
||||
});
|
||||
|
||||
expect(result?.message).toMatchObject({
|
||||
type: 'tool-call-result',
|
||||
callId: 'call-2',
|
||||
output: { ok: true }
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,224 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { z } from 'zod';
|
||||
import { logger } from '@/ui/logger';
|
||||
|
||||
const CodexSessionEventSchema = z.object({
|
||||
timestamp: z.string().optional(),
|
||||
type: z.string(),
|
||||
payload: z.unknown().optional()
|
||||
}).passthrough();
|
||||
|
||||
export type CodexSessionEvent = z.infer<typeof CodexSessionEventSchema>;
|
||||
|
||||
export type CodexMessage = {
|
||||
type: 'message';
|
||||
message: string;
|
||||
id: string;
|
||||
} | {
|
||||
type: 'reasoning';
|
||||
message: string;
|
||||
id: string;
|
||||
} | {
|
||||
type: 'reasoning-delta';
|
||||
delta: string;
|
||||
} | {
|
||||
type: 'token_count';
|
||||
info: Record<string, unknown>;
|
||||
id: string;
|
||||
} | {
|
||||
type: 'tool-call';
|
||||
name: string;
|
||||
callId: string;
|
||||
input: unknown;
|
||||
id: string;
|
||||
} | {
|
||||
type: 'tool-call-result';
|
||||
callId: string;
|
||||
output: unknown;
|
||||
id: string;
|
||||
};
|
||||
|
||||
export type CodexConversionResult = {
|
||||
sessionId?: string;
|
||||
message?: CodexMessage;
|
||||
};
|
||||
|
||||
function asRecord(value: unknown): Record<string, unknown> | null {
|
||||
if (!value || typeof value !== 'object') {
|
||||
return null;
|
||||
}
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function asString(value: unknown): string | null {
|
||||
return typeof value === 'string' && value.length > 0 ? value : null;
|
||||
}
|
||||
|
||||
function parseArguments(value: unknown): unknown {
|
||||
if (typeof value !== 'string') {
|
||||
return value;
|
||||
}
|
||||
|
||||
const trimmed = value.trim();
|
||||
if (trimmed.startsWith('{') || trimmed.startsWith('[')) {
|
||||
try {
|
||||
return JSON.parse(trimmed);
|
||||
} catch (error) {
|
||||
logger.debug('[codexEventConverter] Failed to parse function_call arguments as JSON:', error);
|
||||
}
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
function extractCallId(payload: Record<string, unknown>): string | null {
|
||||
const candidates = [
|
||||
'call_id',
|
||||
'callId',
|
||||
'tool_call_id',
|
||||
'toolCallId',
|
||||
'id'
|
||||
];
|
||||
|
||||
for (const key of candidates) {
|
||||
const value = payload[key];
|
||||
if (typeof value === 'string' && value.length > 0) {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
export function convertCodexEvent(rawEvent: unknown): CodexConversionResult | null {
|
||||
const parsed = CodexSessionEventSchema.safeParse(rawEvent);
|
||||
if (!parsed.success) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const { type, payload } = parsed.data;
|
||||
const payloadRecord = asRecord(payload);
|
||||
|
||||
if (type === 'session_meta') {
|
||||
const sessionId = payloadRecord ? asString(payloadRecord.id) : null;
|
||||
if (!sessionId) {
|
||||
return null;
|
||||
}
|
||||
return { sessionId };
|
||||
}
|
||||
|
||||
if (!payloadRecord) {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (type === 'event_msg') {
|
||||
const eventType = asString(payloadRecord.type);
|
||||
if (!eventType) {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (eventType === 'user_message') {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (eventType === 'agent_message') {
|
||||
const message = asString(payloadRecord.message);
|
||||
if (!message) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
message: {
|
||||
type: 'message',
|
||||
message,
|
||||
id: randomUUID()
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if (eventType === 'agent_reasoning') {
|
||||
const message = asString(payloadRecord.text) ?? asString(payloadRecord.message);
|
||||
if (!message) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
message: {
|
||||
type: 'reasoning',
|
||||
message,
|
||||
id: randomUUID()
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if (eventType === 'agent_reasoning_delta') {
|
||||
const delta = asString(payloadRecord.delta) ?? asString(payloadRecord.text) ?? asString(payloadRecord.message);
|
||||
if (!delta) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
message: {
|
||||
type: 'reasoning-delta',
|
||||
delta
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if (eventType === 'token_count') {
|
||||
const info = asRecord(payloadRecord.info);
|
||||
if (!info) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
message: {
|
||||
type: 'token_count',
|
||||
info,
|
||||
id: randomUUID()
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
if (type === 'response_item') {
|
||||
const itemType = asString(payloadRecord.type);
|
||||
if (!itemType) {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (itemType === 'function_call') {
|
||||
const name = asString(payloadRecord.name);
|
||||
const callId = extractCallId(payloadRecord);
|
||||
if (!name || !callId) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
message: {
|
||||
type: 'tool-call',
|
||||
name,
|
||||
callId,
|
||||
input: parseArguments(payloadRecord.arguments),
|
||||
id: randomUUID()
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if (itemType === 'function_call_output') {
|
||||
const callId = extractCallId(payloadRecord);
|
||||
if (!callId) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
message: {
|
||||
type: 'tool-call-result',
|
||||
callId,
|
||||
output: payloadRecord.output,
|
||||
id: randomUUID()
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
|
||||
import { mkdir, writeFile, appendFile, rm } from 'node:fs/promises';
|
||||
import { join } from 'node:path';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { existsSync } from 'node:fs';
|
||||
import { createCodexSessionScanner } from './codexSessionScanner';
|
||||
import type { CodexSessionEvent } from './codexEventConverter';
|
||||
|
||||
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 scanner: Awaited<ReturnType<typeof createCodexSessionScanner>> | 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;
|
||||
|
||||
events = [];
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
if (scanner) {
|
||||
await scanner.cleanup();
|
||||
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');
|
||||
|
||||
scanner = await createCodexSessionScanner({
|
||||
sessionId,
|
||||
onEvent: (event) => events.push(event)
|
||||
});
|
||||
|
||||
await wait(150);
|
||||
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 wait(200);
|
||||
expect(events).toHaveLength(1);
|
||||
expect(events[0].type).toBe('response_item');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,219 @@
|
||||
import { InvalidateSync } from '@/utils/sync';
|
||||
import { startFileWatcher } from '@/modules/watcher/startFileWatcher';
|
||||
import { logger } from '@/ui/logger';
|
||||
import { join } from 'node:path';
|
||||
import { homedir } from 'node:os';
|
||||
import { readFile, readdir, stat } from 'node:fs/promises';
|
||||
import type { CodexSessionEvent } from './codexEventConverter';
|
||||
|
||||
interface CodexSessionScannerOptions {
|
||||
sessionId: string | null;
|
||||
onEvent: (event: CodexSessionEvent) => void;
|
||||
onSessionFound?: (sessionId: string) => void;
|
||||
}
|
||||
|
||||
interface CodexSessionScanner {
|
||||
cleanup: () => Promise<void>;
|
||||
onNewSession: (sessionId: string) => void;
|
||||
}
|
||||
|
||||
export async function createCodexSessionScanner(opts: CodexSessionScannerOptions): Promise<CodexSessionScanner> {
|
||||
const codexHomeDir = process.env.CODEX_HOME || join(homedir(), '.codex');
|
||||
const sessionsRoot = join(codexHomeDir, 'sessions');
|
||||
|
||||
const processedLineCounts = new Map<string, number>();
|
||||
const watchers = new Map<string, () => void>();
|
||||
const sessionIdByFile = new Map<string, string>();
|
||||
|
||||
let activeSessionId: string | null = opts.sessionId;
|
||||
let reportedSessionId: string | null = opts.sessionId;
|
||||
let isClosing = false;
|
||||
|
||||
const reportSessionId = (sessionId: string) => {
|
||||
if (reportedSessionId === sessionId) {
|
||||
return;
|
||||
}
|
||||
reportedSessionId = sessionId;
|
||||
opts.onSessionFound?.(sessionId);
|
||||
};
|
||||
|
||||
const setActiveSessionId = (sessionId: string) => {
|
||||
activeSessionId = sessionId;
|
||||
reportSessionId(sessionId);
|
||||
};
|
||||
|
||||
async function listSessionFiles(dir: string): Promise<string[]> {
|
||||
try {
|
||||
const entries = await readdir(dir, { withFileTypes: true });
|
||||
const results: string[] = [];
|
||||
for (const entry of entries) {
|
||||
const full = join(dir, entry.name);
|
||||
if (entry.isDirectory()) {
|
||||
results.push(...await listSessionFiles(full));
|
||||
} else if (entry.isFile() && entry.name.endsWith('.jsonl')) {
|
||||
results.push(full);
|
||||
}
|
||||
}
|
||||
return results;
|
||||
} catch (error) {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
async function readSessionFile(filePath: string, startLine: number): Promise<{ events: CodexSessionEvent[]; totalLines: number }> {
|
||||
let content: string;
|
||||
try {
|
||||
content = await readFile(filePath, 'utf-8');
|
||||
} catch (error) {
|
||||
return { events: [], totalLines: startLine };
|
||||
}
|
||||
|
||||
const events: CodexSessionEvent[] = [];
|
||||
const lines = content.split('\n');
|
||||
const hasTrailingEmpty = lines.length > 0 && lines[lines.length - 1] === '';
|
||||
const totalLines = hasTrailingEmpty ? lines.length - 1 : lines.length;
|
||||
let effectiveStartLine = startLine;
|
||||
if (effectiveStartLine > totalLines) {
|
||||
effectiveStartLine = 0;
|
||||
}
|
||||
|
||||
const hasSessionId = sessionIdByFile.has(filePath);
|
||||
const parseFrom = hasSessionId ? effectiveStartLine : 0;
|
||||
|
||||
for (let index = parseFrom; index < lines.length; index += 1) {
|
||||
const trimmed = lines[index].trim();
|
||||
if (!trimmed) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed);
|
||||
if (parsed?.type === 'session_meta') {
|
||||
const payload = asRecord(parsed.payload);
|
||||
const sessionId = payload ? asString(payload.id) : null;
|
||||
if (sessionId) {
|
||||
sessionIdByFile.set(filePath, sessionId);
|
||||
}
|
||||
}
|
||||
if (index >= effectiveStartLine) {
|
||||
events.push(parsed);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Failed to parse line: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
return { events, totalLines };
|
||||
}
|
||||
|
||||
async function initializeProcessedMessages(): Promise<void> {
|
||||
const files = await listSessionFiles(sessionsRoot);
|
||||
for (const filePath of files) {
|
||||
const { totalLines } = await readSessionFile(filePath, 0);
|
||||
processedLineCounts.set(filePath, totalLines);
|
||||
if (!isClosing && !watchers.has(filePath)) {
|
||||
watchers.set(filePath, startFileWatcher(filePath, () => sync.invalidate()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const sync = new InvalidateSync(async () => {
|
||||
if (isClosing) {
|
||||
return;
|
||||
}
|
||||
const files = await listSessionFiles(sessionsRoot);
|
||||
const sortedFiles = await sortFilesByMtime(files);
|
||||
|
||||
for (const filePath of sortedFiles) {
|
||||
if (isClosing) {
|
||||
return;
|
||||
}
|
||||
if (!watchers.has(filePath)) {
|
||||
watchers.set(filePath, startFileWatcher(filePath, () => sync.invalidate()));
|
||||
}
|
||||
|
||||
const fileSessionId = sessionIdByFile.get(filePath);
|
||||
if (activeSessionId && fileSessionId && fileSessionId !== activeSessionId) {
|
||||
continue;
|
||||
}
|
||||
if (activeSessionId && !fileSessionId && !filePath.endsWith(`-${activeSessionId}.jsonl`)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const lastProcessedLine = processedLineCounts.get(filePath) ?? 0;
|
||||
const { events, totalLines } = await readSessionFile(filePath, lastProcessedLine);
|
||||
processedLineCounts.set(filePath, totalLines);
|
||||
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 (!activeSessionId && eventSessionId) {
|
||||
setActiveSessionId(eventSessionId);
|
||||
}
|
||||
|
||||
if (activeSessionId && eventSessionId && eventSessionId !== activeSessionId) {
|
||||
continue;
|
||||
}
|
||||
|
||||
opts.onEvent(event);
|
||||
emittedForFile += 1;
|
||||
}
|
||||
|
||||
if (emittedForFile > 0) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Emitted ${emittedForFile} new events from ${filePath}`);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
await initializeProcessedMessages();
|
||||
await sync.invalidateAndAwait();
|
||||
const intervalId = setInterval(() => sync.invalidate(), 2000);
|
||||
|
||||
return {
|
||||
cleanup: async () => {
|
||||
isClosing = true;
|
||||
clearInterval(intervalId);
|
||||
sync.stop();
|
||||
for (const stop of watchers.values()) {
|
||||
stop();
|
||||
}
|
||||
watchers.clear();
|
||||
},
|
||||
onNewSession: (sessionId: string) => {
|
||||
if (activeSessionId === sessionId) {
|
||||
return;
|
||||
}
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Switching to new session: ${sessionId}`);
|
||||
setActiveSessionId(sessionId);
|
||||
sync.invalidate();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
async function sortFilesByMtime(files: string[]): Promise<string[]> {
|
||||
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<string, unknown> | null {
|
||||
if (!value || typeof value !== 'object') {
|
||||
return null;
|
||||
}
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function asString(value: unknown): string | null {
|
||||
return typeof value === 'string' && value.length > 0 ? value : null;
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
export type ReadyEventOptions = {
|
||||
pending: unknown;
|
||||
queueSize: () => number;
|
||||
shouldExit: boolean;
|
||||
sendReady: () => void;
|
||||
notify?: () => void;
|
||||
};
|
||||
|
||||
/**
|
||||
* Notify connected clients when Codex finishes processing and the queue is idle.
|
||||
* Returns true when a ready event was emitted.
|
||||
*/
|
||||
export function emitReadyIfIdle({ pending, queueSize, shouldExit, sendReady, notify }: ReadyEventOptions): boolean {
|
||||
if (shouldExit) {
|
||||
return false;
|
||||
}
|
||||
if (pending) {
|
||||
return false;
|
||||
}
|
||||
if (queueSize() > 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
sendReady();
|
||||
notify?.();
|
||||
return true;
|
||||
}
|
||||
Reference in New Issue
Block a user