wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-08-05 06:24:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(codex): preserve transcript sync across mode switches

Browse Source
This commit is contained in:
weishu
2026-07-28 12:20:54 +08:00
parent f0c7d0b7e7
commit 92958890de
6 changed files with 118 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/codex/codexLocalLauncher.test.ts
+55 -4
View File
@@ -80,6 +80,7 @@ function createSessionStub(
let localLaunchFailure: { message: string; exitReason: 'switch' | 'exit' } | null = null;
let sessionId: string | null = null;
let transcriptPath: string | null = initialTranscriptPath;
let transcriptHistoryReplayPending = replayTranscriptHistoryOnStart;
let modelReasoningEffort: string | null = null;
const modelReasoningEffortUpdates: Array<string | null> = [];
const transcriptPathCallbacks: Array<(path: string) => void> = [];
@@ -96,7 +97,10 @@ function createSessionStub(
startedBy: 'terminal' as const,
startingMode: 'local' as const,
codexArgs,
replayTranscriptHistoryOnStart,
shouldReplayTranscriptHistory: () => transcriptHistoryReplayPending,
markTranscriptHistoryReplayConsumed: () => {
transcriptHistoryReplayPending = false;
},
client: {
isPending: () => pendingClient,
rpcHandlerManager: {
@@ -628,9 +632,9 @@ describe('codexLocalLauncher', () => {
}
});
it('replays existing transcript messages when importing a Codex thread into a new Hapi session', async () => {
it('replays imported transcript history only on the first local attachment', async () => {
const transcriptPath = join(tempDir, 'codex-import-transcript.jsonl');
const { session, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true);
const { session, userMessages, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true);
let releaseRunBarrier: (() => void) | undefined;
harness.runBarrier = new Promise((resolve) => {
releaseRunBarrier = resolve;
@@ -640,6 +644,7 @@ describe('codexLocalLauncher', () => {
transcriptPath,
[
JSON.stringify({ type: 'session_meta', payload: { id: 'codex-thread-import' } }),
JSON.stringify({ type: 'event_msg', payload: { type: 'user_message', message: 'old imported prompt' } }),
JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'old imported message' } })
].join('\n') + '\n'
);
@@ -652,14 +657,60 @@ describe('codexLocalLauncher', () => {
});
await wait(300);
await appendFile(
transcriptPath,
JSON.stringify({ type: 'event_msg', payload: { type: 'user_message', message: 'tail before switch' } }) + '\n'
);
if (releaseRunBarrier) {
releaseRunBarrier();
}
await launcherPromise;
expect(userMessages).toEqual(['old imported prompt', 'tail before switch']);
expect(agentMessages.filter((message) => (
message as { message?: string }
).message === 'old imported message')).toHaveLength(1);
let releaseSecondRunBarrier: (() => void) | undefined;
harness.runBarrier = new Promise((resolve) => {
releaseSecondRunBarrier = resolve;
});
const secondLauncherPromise = codexLocalLauncher(session as never);
await wait(50);
harness.sessionHookHandlers[1]?.('codex-thread-import', {
transcript_path: transcriptPath
});
await wait(300);
expect(userMessages).toEqual(['old imported prompt', 'tail before switch']);
expect(agentMessages.filter((message) => (
message as { message?: string }
).message === 'old imported message')).toHaveLength(1);
await appendFile(
transcriptPath,
[
JSON.stringify({ type: 'event_msg', payload: { type: 'user_message', message: 'new local prompt' } }),
JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'new local response' } })
].join('\n') + '\n'
);
await wait(700);
if (releaseSecondRunBarrier) {
releaseSecondRunBarrier();
}
await secondLauncherPromise;
expect(userMessages).toEqual(['old imported prompt', 'tail before switch', 'new local prompt']);
expect(agentMessages.filter((message) => (
message as { message?: string }
).message === 'old imported message')).toHaveLength(1);
expect(agentMessages).toContainEqual({
type: 'message',
message: 'old imported message',
message: 'new local response',
id: expect.any(String)
});
});
cli/src/codex/codexLocalLauncher.ts
+21 -3
View File
@@ -46,6 +46,7 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch
let pendingScannerSetup: Promise<void> | null = null;
let transcriptLocator: CodexTranscriptLocator | null = null;
let scannerTranscriptPath: string | null = null;
let scannerReplayedExistingHistory = false;
const pendingPlansByTurnId = new Map<string, ProposedPlanMessage>();
const pendingExecWrappers = new Map<string, PendingExecWrapper>();
const toolHookBridge = new CodexToolHookBridge();
@@ -76,6 +77,21 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch
});
};
const drainAndCleanupScanner = async (
activeScanner: CodexSessionScanner,
replayedExistingHistory: boolean
): Promise<void> => {
try {
// Codex can flush its final transcript records after the last watcher tick.
await activeScanner.flush();
if (replayedExistingHistory) {
session.markTranscriptHistoryReplayConsumed();
}
} finally {
await activeScanner.cleanup();
}
};
const handleSessionFound = (sessionId: string, allowSwitch = false): void => {
if (primarySessionId && primarySessionId !== sessionId && !allowSwitch) {
logger.debug(`[codex-local]: Ignoring non-primary Codex session id ${sessionId}; primary is ${primarySessionId}`);
@@ -174,10 +190,11 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch
scannerTranscriptPath = transcriptPath;
return;
}
const replayExistingHistory = session.shouldReplayTranscriptHistory();
const createdScanner = await createCodexSessionScanner({
transcriptPath,
// 中文注释：导入模式下允许 scanner 首次回放 transcript 全量内容，补齐 Codex 客户端里已有但 Hapi 还未看到的消息。
replayExistingHistory: session.replayTranscriptHistoryOnStart,
replayExistingHistory,
onSessionId: (sessionId) => {
if (!isPrimarySessionId(sessionId)) {
logger.debug(`[codex-local]: Ignoring transcript session id ${sessionId}; primary is ${primarySessionId}`);
@@ -237,11 +254,12 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch
}
});
if (shuttingDown) {
await createdScanner.cleanup();
await drainAndCleanupScanner(createdScanner, replayExistingHistory);
return;
}
scanner = createdScanner;
scannerTranscriptPath = transcriptPath;
scannerReplayedExistingHistory = replayExistingHistory;
};
const handleTranscriptPath = (transcriptPath: string): Promise<void> => {
@@ -376,7 +394,7 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch
}
const activeScanner = scanner as CodexSessionScanner | null;
if (activeScanner) {
await activeScanner.cleanup();
await drainAndCleanupScanner(activeScanner, scannerReplayedExistingHistory);
}
flushAllPendingExecWrappers();
for (const message of toolHookBridge.finish()) {
cli/src/codex/session.ts
+10 -2
View File
@@ -17,11 +17,11 @@ export class CodexSession extends AgentSessionBase<EnhancedMode> {
readonly codexCliOverrides?: CodexCliOverrides;
readonly startedBy: 'runner' | 'terminal';
readonly startingMode: 'local' | 'remote';
readonly replayTranscriptHistoryOnStart: boolean;
readonly sourceSessionId?: string;
localLaunchFailure: LocalLaunchFailure | null = null;
private transcriptPathCallbacks: Array<(path: string) => void> = [];
private transcriptHistoryReplayPending: boolean;
constructor(opts: {
api: ApiClient;
@@ -68,7 +68,7 @@ export class CodexSession extends AgentSessionBase<EnhancedMode> {
this.codexCliOverrides = opts.codexCliOverrides;
this.startedBy = opts.startedBy;
this.startingMode = opts.startingMode;
this.replayTranscriptHistoryOnStart = opts.replayTranscriptHistoryOnStart ?? false;
this.transcriptHistoryReplayPending = opts.replayTranscriptHistoryOnStart ?? false;
this.sourceSessionId = opts.sourceSessionId;
this.permissionMode = opts.permissionMode;
this.model = opts.model;
@@ -76,6 +76,14 @@ export class CodexSession extends AgentSessionBase<EnhancedMode> {
this.collaborationMode = opts.collaborationMode;
}
shouldReplayTranscriptHistory(): boolean {
return this.transcriptHistoryReplayPending;
}
markTranscriptHistoryReplayConsumed(): void {
this.transcriptHistoryReplayPending = false;
}
onTranscriptPathFound(path: string): void {
if (this.transcriptPath === path) {
return;
cli/src/codex/utils/codexSessionScanner.test.ts
+21
View File
@@ -59,6 +59,27 @@ describe('codexSessionScanner', () => {
expect(events[0]?.type).toBe('event_msg');
});
it('flushes an appended tail without waiting for the watcher', async () => {
await writeFile(
transcriptPath,
JSON.stringify({ type: 'session_meta', payload: { id: 'session-flush' } }) + '\n'
);
scanner = await createCodexSessionScanner({
transcriptPath,
onEvent: (event) => events.push(event)
});
await appendFile(
transcriptPath,
JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'final tail' } }) + '\n'
);
await scanner.flush();
expect(events).toHaveLength(1);
expect(events[0]?.payload).toEqual({ type: 'agent_message', message: 'final tail' });
});
it('reads exactly the requested transcript byte range', async () => {
const initial = 'existing transcript\n';
const appended = 'new event\n';
cli/src/codex/utils/codexSessionScanner.ts
+4
View File
@@ -11,6 +11,7 @@ interface CodexSessionScannerOptions {
}
export interface CodexSessionScanner {
flush: () => Promise<void>;
cleanup: () => Promise<void>;
setTranscriptPath: (transcriptPath: string) => Promise<void>;
}
@@ -20,6 +21,9 @@ export async function createCodexSessionScanner(opts: CodexSessionScannerOptions
await scanner.start();
return {
flush: async () => {
await scanner.flush();
},
cleanup: async () => {
await scanner.cleanup();
},
cli/src/modules/common/session/BaseSessionScanner.ts
+7
View File
@@ -108,6 +108,13 @@ export abstract class BaseSessionScanner<TEvent> {
this.intervalId = setInterval(() => this.sync.invalidate(), this.options.intervalMs);
}
public async flush(): Promise<void> {
if (this.stopped) {
return;
}
await this.sync.invalidateAndAwait();
}
public async cleanup(): Promise<void> {
this.stopped = true;
if (this.intervalId) {