fix(tui): detect external session wire-file appends before accepting input

.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.

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 {

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<void> | 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<boolean> {
+    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<void> {
     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<void> {

apps/kimi-code/src/tui/utils/wire-staleness.ts

@@ -0,0 +1,135 @@
+/**
+ * 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 `<sessionDir>/agents/<agentId>/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');
+}
+
+/** Size of the backward scan windows used 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 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`.
+ *
+ * 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,
+  agentId: string,
+): Promise<number | undefined> {
+  try {
+    const file = await open(agentWirePath(sessionDir, agentId), 'r');
+    try {
+      const { size } = await file.stat();
+      if (size <= 0) return undefined;
+      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();
+    }
+  } catch {
+    return undefined;
+  }
+}