From 5ec903df5be39a4b581bf556eaaba206b7f6f6d6 Mon Sep 17 00:00:00 2001 From: snowingfox <1503401882@qq.com> Date: Wed, 12 Aug 2026 07:05:11 +0000 Subject: [PATCH 1/2] fix(tui): detect external session wire-file appends before accepting input Fixes #2835 --- .../tui/controllers/session-event-handler.ts | 5 + apps/kimi-code/src/tui/kimi-tui.ts | 75 ++++++++++++ .../kimi-code/src/tui/utils/wire-staleness.ts | 102 +++++++++++++++++ .../kimi-code/test/tui/wire-staleness.test.ts | 107 ++++++++++++++++++ 4 files changed, 289 insertions(+) create mode 100644 apps/kimi-code/src/tui/utils/wire-staleness.ts create mode 100644 apps/kimi-code/test/tui/wire-staleness.test.ts diff --git a/apps/kimi-code/src/tui/controllers/session-event-handler.ts b/apps/kimi-code/src/tui/controllers/session-event-handler.ts index 9cb1029bdc5..67c70d0aaac 100644 --- a/apps/kimi-code/src/tui/controllers/session-event-handler.ts +++ b/apps/kimi-code/src/tui/controllers/session-event-handler.ts @@ -117,6 +117,8 @@ export interface SessionEventHost { handleShellStarted(event: { commandId: string; taskId: string }): void; sendNormalUserInput(text: string): void; updateTerminalTitle(): void; + /** Re-read the active session's wire journal into the staleness baseline. */ + refreshWireTipFromDisk(): void; sendQueuedMessage(session: Session, item: QueuedMessage): void; shiftQueuedMessage(): QueuedMessage | undefined; readonly btwPanelController: BtwPanelController; @@ -398,6 +400,9 @@ export class SessionEventHandler { } this.pluginMcpToolsUsedInTurn.clear(); this.scheduleQueuedGoalPromotion(); + // The turn's journal records are now on disk; advance the staleness + // baseline so the next input is not mistaken for an external append. + this.host.refreshWireTipFromDisk(); } private handleStepBegin(event: TurnStepStartedEvent): void { diff --git a/apps/kimi-code/src/tui/kimi-tui.ts b/apps/kimi-code/src/tui/kimi-tui.ts index 488b426cdb7..9d0754a73a9 100644 --- a/apps/kimi-code/src/tui/kimi-tui.ts +++ b/apps/kimi-code/src/tui/kimi-tui.ts @@ -178,6 +178,10 @@ import { groupTurns, turnsToTrim, } from './utils/transcript-window'; +import { + readWireTurnBoundaryTime, + wireTailAheadOfTranscript, +} from './utils/wire-staleness'; export type { TUIState } from './tui-state'; export { createTUIState } from './tui-state'; @@ -360,6 +364,16 @@ export class KimiTUI { private currentLoadingTip: { kind: LoadingTipKind; tip: string | undefined } | undefined = undefined; private lastHistoryContent: string | undefined; + /** + * The newest user-turn timestamp this TUI has observed on the active + * session's wire journal (seeded at session switch, advanced after each turn + * ends). The send guard compares the live journal against this to detect + * turns appended by external clients. `undefined` until a session exposes a + * readable journal (fresh sessions start undefined and stay fail-open). + */ + private wireTipTime: number | undefined; + /** In-flight {@link refreshWireTipFromDisk} read, awaited by the send guard. */ + private wireTipRefresh: Promise | null = null; // Live `!` shell output entries, keyed by commandId so concurrent commands // each update their own card and stale events are dropped. Mutated in place // as `shell.output` events arrive; removed when the command completes. @@ -1275,6 +1289,56 @@ export class KimiTUI { this.updateQueueDisplay(); } + /** + * Re-read the active session's newest wire user-turn into {@link wireTipTime}. + * Failures keep the last observed tip so the guard stays fail-open. + */ + refreshWireTipFromDisk(): void { + const sessionDir = this.session?.summary?.sessionDir; + if (sessionDir === undefined) return; + const task = readWireTurnBoundaryTime(sessionDir, MAIN_AGENT_ID) + .then((time) => { + this.wireTipTime = time; + }) + .catch(() => { + // Keep the last observed tip; the guard fails open when unreadable. + }); + this.wireTipRefresh = task; + void task.finally(() => { + if (this.wireTipRefresh === task) this.wireTipRefresh = null; + }); + } + + /** + * Block a prompt when an external client has appended a turn to the session's + * wire journal after this TUI last rendered one. Sending anyway would fork + * the session into two divergent histories, so the user is told to resume + * fresh instead. Only meaningful while the session is idle — in-flight input + * is queued by the caller and must not be gated on the live journal. + */ + private async assertWireFresh(session: Session): Promise { + if (this.state.appState.streamingPhase !== 'idle') return false; + // A tip refresh kicked off by the last turn may still be in flight; await + // it so the comparison never sees a stale baseline and false-positives. + if (this.wireTipRefresh !== null) { + await this.wireTipRefresh.catch(() => undefined); + } + const sessionDir = session.summary?.sessionDir; + if (sessionDir === undefined) return false; + const wireTailTime = await readWireTurnBoundaryTime(sessionDir, MAIN_AGENT_ID).catch( + () => undefined, + ); + if (!wireTailAheadOfTranscript({ transcriptTipTime: this.wireTipTime, wireTailTime })) { + return false; + } + this.showStatus( + 'Session was modified outside this terminal; your transcript is out of date.\n' + + `Resume with: kimi -S ${quoteShellArg(session.id)} to reload the latest history before continuing.`, + 'warning', + ); + return true; + } + async sendNormalUserInput(text: string, preExtracted?: ExtractionResult): Promise { if (this.btwPanelController.sendUserInput(text)) return; if (this.state.appState.model.trim().length === 0) { @@ -1311,6 +1375,14 @@ export class KimiTUI { session = await this.ensureSession(); if (session === undefined) return; } + // An external client may have appended turns to the session's wire journal + // while this TUI was idle. Sending now would continue from a stale context + // and silently fork the session, so refuse until the user resumes fresh. + if (await this.assertWireFresh(session)) { + this.updateQueueDisplay(); + this.state.ui.requestRender(); + return; + } if (extraction.hasMedia) { this.sendMessage(session, text, { hasMedia: true, @@ -1882,6 +1954,9 @@ export class KimiTUI { this.harness.setTelemetryContext({ sessionId: session.id }); this.registerSessionHandlers(session); this.syncAdditionalDirs(session); + // Seed the wire staleness baseline for the newly active session (fresh + // sessions yield `undefined` and stay fail-open until their first turn). + this.refreshWireTipFromDisk(); } async syncRuntimeState(session: Session = this.requireSession()): Promise { diff --git a/apps/kimi-code/src/tui/utils/wire-staleness.ts b/apps/kimi-code/src/tui/utils/wire-staleness.ts new file mode 100644 index 00000000000..34b858e56ce --- /dev/null +++ b/apps/kimi-code/src/tui/utils/wire-staleness.ts @@ -0,0 +1,102 @@ +/** + * Staleness detection for the interactive TUI transcript vs the session's + * on-disk wire journal. + * + * The TUI renders a pure in-memory transcript while the engine persists every + * op to `/agents//wire.jsonl`. External clients (ACP + * attach, mobile) can append turns to that journal without the TUI ever seeing + * an event, so the user would keep typing against a stale context and silently + * fork the session into two divergent histories. These helpers let the TUI + * compare the newest user-turn it has rendered against the journal's newest + * user-turn before accepting the next input. + * + * The comparison deliberately keys on `turn.prompt` records rather than any + * timestamped record: compaction, plan-mode toggles and config writes advance + * the journal tail without opening a new conversational turn, so they must not + * count as external activity. `turn.prompt` is the wire's authoritative + * user-turn boundary (see `agent-core-v2/src/agent/loop/turnOps.ts`). + */ + +import { open } from 'node:fs/promises'; +import { join } from 'node:path'; + +/** Record types that open a new conversational turn in the wire journal. */ +const TURN_BOUNDARY_TYPES = new Set(['turn.prompt']); + +/** Path of an agent's persisted journal inside a session directory. */ +export function agentWirePath(sessionDir: string, agentId: string): string { + return join(sessionDir, 'agents', agentId, 'wire.jsonl'); +} + +/** How many trailing bytes of the journal we read to locate the newest turn. */ +const WIRE_TAIL_READ_BYTES = 64 * 1024; + +/** + * Extract the newest user-turn (`turn.prompt`) timestamp from a chunk read off + * the tail of a JSONL wire journal. + * + * A tail chunk may begin mid-record (the read boundary split a line); the scan + * runs from the last line upward, skipping fragments that fail to parse and + * non-boundary records. Returns `undefined` when the chunk holds no complete + * `turn.prompt` record with a numeric `time`. + */ +export function lastTurnBoundaryTimeInChunk(chunk: string): number | undefined { + const lines = chunk.split('\n'); + for (let i = lines.length - 1; i >= 0; i -= 1) { + const line = lines[i]!.trim(); + if (line.length === 0) continue; + let record: { type?: unknown; time?: unknown }; + try { + record = JSON.parse(line) as { type?: unknown; time?: unknown }; + } catch { + // Fragment from the tail-read boundary — keep scanning upward. + continue; + } + if (typeof record.time === 'number' && TURN_BOUNDARY_TYPES.has(record.type as string)) { + return record.time; + } + } + return undefined; +} + +/** + * True when the wire journal has a user-turn newer than the newest turn the + * TUI has rendered — i.e. an external client appended a turn the user never + * saw. Fails open (false) whenever either side is unknown. + */ +export function wireTailAheadOfTranscript(opts: { + readonly transcriptTipTime: number | undefined; + readonly wireTailTime: number | undefined; +}): boolean { + const { transcriptTipTime, wireTailTime } = opts; + if (transcriptTipTime === undefined || wireTailTime === undefined) return false; + return wireTailTime > transcriptTipTime; +} + +/** + * Read the newest user-turn timestamp of an agent's `wire.jsonl`. + * + * Failures (missing session dir, unreadable journal, a tail record larger than + * {@link WIRE_TAIL_READ_BYTES}) degrade to `undefined` so the staleness guard + * always fails open rather than blocking the user on a corrupt file. + */ +export async function readWireTurnBoundaryTime( + sessionDir: string, + agentId: string, +): Promise { + try { + const file = await open(agentWirePath(sessionDir, agentId), 'r'); + try { + const { size } = await file.stat(); + if (size <= 0) return undefined; + const length = Math.min(size, WIRE_TAIL_READ_BYTES); + const buffer = Buffer.alloc(length); + await file.read(buffer, 0, length, size - length); + return lastTurnBoundaryTimeInChunk(buffer.toString('utf8')); + } finally { + await file.close(); + } + } catch { + return undefined; + } +} diff --git a/apps/kimi-code/test/tui/wire-staleness.test.ts b/apps/kimi-code/test/tui/wire-staleness.test.ts new file mode 100644 index 00000000000..8832035cdc5 --- /dev/null +++ b/apps/kimi-code/test/tui/wire-staleness.test.ts @@ -0,0 +1,107 @@ +/** + * Scenario: the interactive TUI detects when an external client has appended + * turns to the session's wire journal so it can warn the user before their + * next input silently forks the conversation. + * + * Responsibilities: `lastTurnBoundaryTimeInChunk` recovers the newest user-turn + * (`turn.prompt`) timestamp from a wire-journal tail chunk, and + * `wireTailAheadOfTranscript` decides staleness by comparing that against the + * newest turn the in-memory transcript has rendered. + * + * Wiring: pure helpers only — no TUI or SDK imports, so the check runs without + * any terminal, session, or engine dependency. + * Run: pnpm -C apps/kimi-code exec vitest run test/tui/wire-staleness.test.ts + */ + +import { describe, expect, it } from 'vitest'; + +import { + lastTurnBoundaryTimeInChunk, + wireTailAheadOfTranscript, +} from '@/tui/utils/wire-staleness'; + +describe('wireTailAheadOfTranscript', () => { + it('detects a wire user-turn newer than the in-memory transcript tip', () => { + expect( + wireTailAheadOfTranscript({ + transcriptTipTime: 1_700_000_000_000, + wireTailTime: 1_700_000_500_000, + }), + ).toBe(true); + }); + + it('is false when the wire user-turn matches the transcript tip', () => { + expect( + wireTailAheadOfTranscript({ + transcriptTipTime: 1_700_000_500_000, + wireTailTime: 1_700_000_500_000, + }), + ).toBe(false); + }); + + it('is false when the wire user-turn predates the transcript tip', () => { + expect( + wireTailAheadOfTranscript({ + transcriptTipTime: 1_700_000_500_000, + wireTailTime: 1_700_000_000_000, + }), + ).toBe(false); + }); + + it('fails open when either timestamp is unknown', () => { + expect( + wireTailAheadOfTranscript({ transcriptTipTime: undefined, wireTailTime: 1_700_000_500_000 }), + ).toBe(false); + expect( + wireTailAheadOfTranscript({ transcriptTipTime: 1_700_000_000_000, wireTailTime: undefined }), + ).toBe(false); + expect( + wireTailAheadOfTranscript({ transcriptTipTime: undefined, wireTailTime: undefined }), + ).toBe(false); + }); +}); + +describe('lastTurnBoundaryTimeInChunk', () => { + it('returns the newest `turn.prompt` time, skipping non-boundary records', () => { + const chunk = [ + '{"type":"turn.prompt","time":100,"input":{"input":[],"origin":{}}}', + '{"type":"turn.ended","time":200,"turnId":1,"reason":"completed"}', + '{"type":"context.append_loop_event","time":300,"event":{"type":"content.part","part":{"type":"text","text":"hi"}}}', + '{"type":"usage.record","time":400,"usage":{}}', + ].join('\n'); + expect(lastTurnBoundaryTimeInChunk(chunk)).toBe(100); + }); + + it('returns the last of several `turn.prompt` records', () => { + const chunk = [ + '{"type":"turn.prompt","time":100,"input":{}}', + '{"type":"turn.ended","time":150,"turnId":1,"reason":"completed"}', + '{"type":"turn.prompt","time":200,"input":{}}', + '{"type":"turn.ended","time":250,"turnId":2,"reason":"completed"}', + ].join('\n'); + expect(lastTurnBoundaryTimeInChunk(chunk)).toBe(200); + }); + + it('returns undefined when the chunk holds no user-turn record', () => { + expect( + lastTurnBoundaryTimeInChunk( + '{"type":"metadata","protocol_version":"1","created_at":10}\n{"type":"usage.record","time":20,"usage":{}}\n', + ), + ).toBeUndefined(); + expect(lastTurnBoundaryTimeInChunk('')).toBeUndefined(); + }); + + it('scans past a partial line fragment from the read-tail boundary', () => { + // The read boundary split a long record; the fragment fails to parse and + // the scan keeps going until it finds the complete `turn.prompt`. + const chunk = + '{"type":"context.append_loop_event","time":1,"event":{"type":"content.part","part":{"type":"text","text":"' + + '\n{"type":"turn.prompt","time":300,"input":{}}'; + expect(lastTurnBoundaryTimeInChunk(chunk)).toBe(300); + }); + + it('ignores empty trailing lines and returns the newest complete record', () => { + const chunk = '{"type":"turn.prompt","time":100,"input":{}}\n\n'; + expect(lastTurnBoundaryTimeInChunk(chunk)).toBe(100); + }); +}); From 4f9f818d22d07343ce5cf69c6acee2a91f375695 Mon Sep 17 00:00:00 2001 From: snowingfox <1503401882@qq.com> Date: Wed, 12 Aug 2026 16:00:35 +0800 Subject: [PATCH 2/2] fix(tui): scan past 64 KiB of post-prompt records when detecting staleness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex P1: readWireTurnBoundaryTime read only the last 64 KiB of the wire journal, so an external client whose turn response/tool output exceeds 64 KiB hid its turn.prompt boundary and wireTailAheadOfTranscript failed open — leaving the exact silent-fork path unfixed for large turns. Walk backward from the journal tail in 64 KiB windows until a turn.prompt boundary is found or the file head is reached. A record split across a read boundary is reassembled by carrying the top fragment into the next older window and joining it to the head half before scanning. Adds regression coverage for a >64 KiB post-prompt record and a record spanning several windows, plus a changeset (sibling PRs carry one). Fixes #2835 Signed-off-by: snowingfox <1503401882@qq.com> --- .../tui-wire-staleness-backward-scan.md | 5 + .../kimi-code/src/tui/utils/wire-staleness.ts | 51 ++++++-- .../kimi-code/test/tui/wire-staleness.test.ts | 109 +++++++++++++++++- 3 files changed, 151 insertions(+), 14 deletions(-) create mode 100644 .changeset/tui-wire-staleness-backward-scan.md diff --git a/.changeset/tui-wire-staleness-backward-scan.md b/.changeset/tui-wire-staleness-backward-scan.md new file mode 100644 index 00000000000..3c73dd207e7 --- /dev/null +++ b/.changeset/tui-wire-staleness-backward-scan.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +TUI: detect external wire-journal appends even when a turn's response or tool output exceeds 64 KiB. `readWireTurnBoundaryTime` now scans backward past the tail read window to locate the newest `turn.prompt` boundary, so the staleness guard no longer fails open for large external appends. diff --git a/apps/kimi-code/src/tui/utils/wire-staleness.ts b/apps/kimi-code/src/tui/utils/wire-staleness.ts index 34b858e56ce..c67b48d23a6 100644 --- a/apps/kimi-code/src/tui/utils/wire-staleness.ts +++ b/apps/kimi-code/src/tui/utils/wire-staleness.ts @@ -28,7 +28,7 @@ export function agentWirePath(sessionDir: string, agentId: string): string { return join(sessionDir, 'agents', agentId, 'wire.jsonl'); } -/** How many trailing bytes of the journal we read to locate the newest turn. */ +/** Size of the backward scan windows used to locate the newest turn. */ const WIRE_TAIL_READ_BYTES = 64 * 1024; /** @@ -49,7 +49,7 @@ export function lastTurnBoundaryTimeInChunk(chunk: string): number | undefined { try { record = JSON.parse(line) as { type?: unknown; time?: unknown }; } catch { - // Fragment from the tail-read boundary — keep scanning upward. + // Fragment from the read boundary — keep scanning upward. continue; } if (typeof record.time === 'number' && TURN_BOUNDARY_TYPES.has(record.type as string)) { @@ -76,9 +76,17 @@ export function wireTailAheadOfTranscript(opts: { /** * Read the newest user-turn timestamp of an agent's `wire.jsonl`. * - * Failures (missing session dir, unreadable journal, a tail record larger than - * {@link WIRE_TAIL_READ_BYTES}) degrade to `undefined` so the staleness guard - * always fails open rather than blocking the user on a corrupt file. + * Walks backward from the journal tail in {@link WIRE_TAIL_READ_BYTES} windows + * until a `turn.prompt` boundary is found or the start of the file is reached, + * so a single external turn whose response/tool output spans more than one + * window cannot hide its prompt boundary. A record split across a read + * boundary is reassembled before scanning: the fragment at the top of each + * window is the tail half of a record whose head lives at the end of the next + * (older) window, and the two are joined back into one line. + * + * Failures (missing session dir, unreadable journal) degrade to `undefined` + * so the staleness guard always fails open rather than blocking the user on a + * corrupt file. */ export async function readWireTurnBoundaryTime( sessionDir: string, @@ -89,10 +97,35 @@ export async function readWireTurnBoundaryTime( try { const { size } = await file.stat(); if (size <= 0) return undefined; - const length = Math.min(size, WIRE_TAIL_READ_BYTES); - const buffer = Buffer.alloc(length); - await file.read(buffer, 0, length, size - length); - return lastTurnBoundaryTimeInChunk(buffer.toString('utf8')); + let offset = size; + // Tail half of the record split by the last read boundary; its head is + // the final line of the next (older) window. + let carry = ''; + while (offset > 0) { + const start = Math.max(0, offset - WIRE_TAIL_READ_BYTES); + const length = offset - start; + const buffer = Buffer.alloc(length); + await file.read(buffer, 0, length, start); + const chunk = buffer.toString('utf8'); + const firstNl = chunk.indexOf('\n'); + if (firstNl >= 0) { + // The first line may be the tail half of a split record; the rest + // are complete. Appending the carried tail to the window's own last + // line (the head half) reassembles the split record, which is the + // newest line this window contributes and is scanned first. + const rest = chunk.slice(firstNl + 1); + const time = lastTurnBoundaryTimeInChunk(rest + carry); + if (time !== undefined) return time; + carry = chunk.slice(0, firstNl); + } else { + // The whole window is one un-terminated record — accumulate it with + // the carried tail so the record is reassembled in an older window. + carry = chunk + carry; + } + offset = start; + } + // The oldest record reached the head of the file still split. + return carry.length > 0 ? lastTurnBoundaryTimeInChunk(carry) : undefined; } finally { await file.close(); } diff --git a/apps/kimi-code/test/tui/wire-staleness.test.ts b/apps/kimi-code/test/tui/wire-staleness.test.ts index 8832035cdc5..fab217b3077 100644 --- a/apps/kimi-code/test/tui/wire-staleness.test.ts +++ b/apps/kimi-code/test/tui/wire-staleness.test.ts @@ -4,19 +4,26 @@ * next input silently forks the conversation. * * Responsibilities: `lastTurnBoundaryTimeInChunk` recovers the newest user-turn - * (`turn.prompt`) timestamp from a wire-journal tail chunk, and - * `wireTailAheadOfTranscript` decides staleness by comparing that against the - * newest turn the in-memory transcript has rendered. + * (`turn.prompt`) timestamp from a wire-journal tail chunk, + * `readWireTurnBoundaryTime` locates that boundary in the on-disk journal even + * when it has been pushed past the tail read window, and + * `wireTailAheadOfTranscript` decides staleness by comparing the boundary + * against the newest turn the in-memory transcript has rendered. * - * Wiring: pure helpers only — no TUI or SDK imports, so the check runs without - * any terminal, session, or engine dependency. + * Wiring: pure helpers + a single file read only — no TUI or SDK imports, so + * the check runs without any terminal, session, or engine dependency. * Run: pnpm -C apps/kimi-code exec vitest run test/tui/wire-staleness.test.ts */ +import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { describe, expect, it } from 'vitest'; import { + agentWirePath, lastTurnBoundaryTimeInChunk, + readWireTurnBoundaryTime, wireTailAheadOfTranscript, } from '@/tui/utils/wire-staleness'; @@ -105,3 +112,95 @@ describe('lastTurnBoundaryTimeInChunk', () => { expect(lastTurnBoundaryTimeInChunk(chunk)).toBe(100); }); }); + +describe('readWireTurnBoundaryTime', () => { + /** Writes a journal and runs the assertion against it, cleaning up after. */ + async function withJournal( + lines: string[], + run: (dir: string, agentId: string) => Promise, + ): Promise { + const dir = await mkdtemp(join(tmpdir(), 'wire-staleness-')); + try { + const agentId = 'agent-1'; + await mkdir(join(dir, 'agents', agentId), { recursive: true }); + await writeFile(agentWirePath(dir, agentId), lines.join('\n')); + await run(dir, agentId); + } finally { + await rm(dir, { recursive: true, force: true }); + } + } + + it('reads the newest `turn.prompt` from a small journal', async () => { + await withJournal( + [ + '{"type":"metadata","protocol_version":"1","created_at":10}', + '{"type":"turn.prompt","time":100,"input":{}}', + '{"type":"turn.ended","time":200,"turnId":1,"reason":"completed"}', + '{"type":"turn.prompt","time":300,"input":{}}', + '{"type":"usage.record","time":400,"usage":{}}', + ], + async (dir, agentId) => { + await expect(readWireTurnBoundaryTime(dir, agentId)).resolves.toBe(300); + }, + ); + }); + + it('recovers a `turn.prompt` pushed past the 64 KiB tail window by a large post-prompt record', async () => { + await withJournal( + [ + '{"type":"metadata","protocol_version":"1","created_at":10}', + '{"type":"turn.prompt","time":100,"input":{}}', + // A single tool-output record larger than the 64 KiB tail read window + // sits between the prompt and the journal tail. A single tail read + // would only see the end of this record plus the newer records, miss + // the prompt boundary, and fail open — the exact silent-fork path this + // guard exists to catch. + `{"type":"context.append_loop_event","time":200,"event":{"type":"content.part","part":{"type":"text","text":"${'x'.repeat(80 * 1024)}"}}}`, + '{"type":"usage.record","time":300,"usage":{}}', + ], + async (dir, agentId) => { + await expect(readWireTurnBoundaryTime(dir, agentId)).resolves.toBe(100); + }, + ); + }); + + it('recovers a `turn.prompt` when a record spanning several windows sits after it', async () => { + await withJournal( + [ + '{"type":"turn.prompt","time":50,"input":{}}', + `{"type":"context.append_loop_event","time":60,"event":{"type":"content.part","part":{"type":"text","text":"${'y'.repeat(200 * 1024)}"}}}`, + '{"type":"usage.record","time":70,"usage":{}}', + ], + async (dir, agentId) => { + await expect(readWireTurnBoundaryTime(dir, agentId)).resolves.toBe(50); + }, + ); + }); + + it('fails open (undefined) when the journal holds no user-turn record', async () => { + await withJournal( + [ + '{"type":"metadata","protocol_version":"1","created_at":10}', + `{"type":"context.append_loop_event","time":20,"event":{"type":"content.part","part":{"type":"text","text":"${'z'.repeat(80 * 1024)}"}}}`, + '{"type":"usage.record","time":30,"usage":{}}', + ], + async (dir, agentId) => { + await expect(readWireTurnBoundaryTime(dir, agentId)).resolves.toBeUndefined(); + }, + ); + }); + + it('fails open (undefined) for an empty journal, missing session, or missing agent', async () => { + const dir = await mkdtemp(join(tmpdir(), 'wire-staleness-')); + try { + const agentId = 'agent-1'; + await mkdir(join(dir, 'agents', agentId), { recursive: true }); + await writeFile(agentWirePath(dir, agentId), ''); + await expect(readWireTurnBoundaryTime(dir, agentId)).resolves.toBeUndefined(); + await expect(readWireTurnBoundaryTime('/nonexistent-session', agentId)).resolves.toBeUndefined(); + await expect(readWireTurnBoundaryTime(dir, 'no-such-agent')).resolves.toBeUndefined(); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); +});