fix: preserve plan fencing and journal ownership

.changeset/cold-journal-inspection.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Prevent cold session reads from replacing an event journal owned by another server instance.

.changeset/stale-write-fencing.md

@@ -2,4 +2,4 @@
 "@moonshot-ai/kimi-code": patch
 ---
 
-Each agent now tracks its own file-read baselines and refuses to overwrite files that changed on disk after its last read.
+Each agent now tracks its own file-read baselines and refuses to overwrite files, including active plan files, that changed on disk after its last read.

packages/agent-core-v2/src/agent/plan/planService.ts

@@ -14,12 +14,12 @@
  * the Model's replayed per-id `revisionCount`, starting at 1. Also carries
  * the plan-mode Harness constraints as an `onBeforeExecuteTool` veto
  * listener: while a plan is active, Write/Edit calls targeting only the
- * current plan file are allowed outright (`allow()`, ending all other
- * adjudication), any other Write/Edit and every TaskStop/CronCreate/
- * CronDelete call is vetoed with a `toolApproval.formatDenyMessage`-
- * formatted reason, and an `ExitPlanMode` call outside `auto` mode defers
- * to a cold `waitUntil` factory running the `exitPlanModeReview` user
- * review. Bound at Agent scope.
+ * current plan file are passed to later adjudication (`pass()`), so the
+ * file-fencing listener can still reject a stale write; any other Write/Edit
+ * and every TaskStop/CronCreate/CronDelete call is vetoed with a
+ * `toolApproval.formatDenyMessage`-formatted reason, and an `ExitPlanMode`
+ * call outside `auto` mode defers to a cold `waitUntil` factory running the
+ * `exitPlanModeReview` user review. Bound at Agent scope.
  */
 
 import { createHash, randomUUID } from 'node:crypto';
@@ -118,7 +118,9 @@ export class AgentPlanService extends Disposable implements IAgentPlanService {
 
     if (toolName === 'Write' || toolName === 'Edit') {
       if (writesOnlyPlanFile(event, plan.path)) {
-        event.allow();
+        // Plan mode authorizes this target, but must not short-circuit the
+        // file-fencing waitUntil that checks the read/write baseline.
+        event.pass();
         return;
       }
       event.veto(

packages/agent-core-v2/test/agent/plan/planGuard.test.ts

@@ -1,13 +1,13 @@
 /**
  * Scenario: plan-mode Harness constraints as an `onBeforeExecuteTool` veto
- * listener. Responsibilities: verify Write/Edit plan-file allow and vetoes,
+ * listener. Responsibilities: verify Write/Edit plan-file pass and vetoes,
  * TaskStop/Cron vetoes, abstention on unrelated tools, and every ExitPlanMode
  * review branch (approve with/without option, Reject and Exit, Revise,
  * dismiss, auto / no-plan / empty-plan / non-plan_review skips) with
  * telemetry.
  * Wiring: real wire and plan services against a fireable executor event
  * stub; a stand-in listener registered after the plan listener proves
- * whether the guard ended adjudication (veto/allow) or abstained;
+ * whether the guard vetoes, passes, or abstains;
  * `IAgentToolApprovalService` is a recording stub.
  * Run: `pnpm --filter @moonshot-ai/agent-core-v2 exec vitest run test/agent/plan/planGuard.test.ts`.
  */
@@ -231,7 +231,7 @@ describe('AgentPlanService plan-guard listener', () => {
 
   describe('guard', () => {
     it.each(['Write', 'Edit'] as const)(
-      'lets a %s that only targets the active plan file through without other adjudication',
+      'passes a %s that only targets the active plan file to later adjudication',
       async (toolName) => {
         await enterPlan();
         const decision = await run(
@@ -242,7 +242,7 @@ describe('AgentPlanService plan-guard listener', () => {
         );
 
         expect(decision).toBeUndefined();
-        expect(permissionRan).toBe(false);
+        expect(permissionRan).toBe(true);
       },
     );
 
@@ -259,7 +259,7 @@ describe('AgentPlanService plan-guard listener', () => {
       );
 
       expect(decision).toBeUndefined();
-      expect(permissionRan).toBe(false);
+      expect(permissionRan).toBe(true);
     });
 
     it.each(['Write', 'Edit'] as const)(
@@ -296,6 +296,29 @@ describe('AgentPlanService plan-guard listener', () => {
       expect(permissionRan).toBe(false);
     });
 
+    it('runs a later fencing adjudication for a plan-file write', async () => {
+      await enterPlan();
+      let fencingRan = false;
+      disposables.add(
+        executorEvents.executor.onBeforeExecuteTool((event) => {
+          fencingRan = true;
+          event.waitUntil(async () => ({
+            veto: { output: 'stale plan file', isError: true },
+          }));
+        }),
+      );
+
+      const decision = await executorEvents.fireBeforeExecute(
+        hookContext('Edit', {
+          args: { path: PLAN_PATH },
+          accesses: ToolAccesses.writeFile(PLAN_PATH),
+        }),
+      );
+
+      expect(fencingRan).toBe(true);
+      expect(decision?.veto).toEqual({ output: 'stale plan file', isError: true });
+    });
+
     it('blocks mixed plan-file and non-plan-file write accesses', async () => {
       await enterPlan();
       const decision = await run(

packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts

@@ -99,6 +99,7 @@ import { SubagentRosterTracker } from './subagentRosterTracker';
 import {
   type EventEnvelope,
   type JournalLogger,
+  type JournalWatermark,
   SessionEventJournal,
   sessionJournalPath,
 } from './sessionEventJournal';
@@ -719,24 +720,21 @@ export class SessionEventBroadcaster {
 
   /**
    * Watermark for a session that is not live in this process but exists on disk
-   * (carried over from a prior process, or created by v1). Opens the journal
-   * transiently — no agent/interaction listeners and not cached in
-   * `this.sessions` — so a later live activation still attaches subscriptions.
-   * The open is read-only (zero bytes written, no fabricated epoch).
+   * (carried over from a prior process, or created by v1). Inspects the journal
+   * transiently — no agent/interaction listeners, no repair/quarantine, and no
+   * cache entry in `this.sessions` — so a later live activation still attaches
+   * subscriptions without modifying a peer-owned file.
    * Returns `undefined` when the session is unknown to the index (truly absent).
    */
   private async readColdWatermark(
     sessionId: string,
-  ): Promise<{ seq: number; epoch: string | undefined } | undefined> {
+  ): Promise<JournalWatermark | undefined> {
     const summary = await this.opts.core.accessor.get(ISessionIndex).get(sessionId);
     if (summary === undefined) return undefined;
-    const journal = await SessionEventJournal.open(
+    return SessionEventJournal.inspect(
       sessionJournalPath(this.opts.eventsDir, sessionId),
       this.opts.logger,
     );
-    const watermark = { seq: journal.seq, epoch: journal.epoch };
-    await journal.close();
-    return watermark;
   }
 
   async close(): Promise<void> {

packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts

@@ -110,6 +110,11 @@ export interface JournalEntry {
   envelope: EventEnvelope;
 }
 
+export interface JournalWatermark {
+  readonly seq: number;
+  readonly epoch: string | undefined;
+}
+
 /** Minimal logger surface — keeps the journal decoupled from the server logger. */
 export interface JournalLogger {
   warn(obj: unknown, msg: string): void;
@@ -171,62 +176,26 @@ export class SessionEventJournal {
 
   /**
    * Open (or create-on-first-append) the journal for `filePath`. Scans an
-   * existing file to recover `{epoch, lastSeq}`. This open is READ-ONLY: a
-   * missing file or an unreadable/missing header yields a journal with
-   * `epoch: undefined` and seq 0, writes nothing, and defers the fresh epoch
-   * to the first real `append()`. With several headers the last one wins.
+   * existing file to recover `{epoch, lastSeq}`. This is the owner open: a
+   * corrupt journal may be quarantined and a torn tail may be repaired before
+   * the next append. A missing file or an unreadable/missing header yields a
+   * journal with `epoch: undefined` and seq 0, and defers the fresh epoch to
+   * the first real `append()`. With several headers the last one wins.
    */
   static async open(
     filePath: string,
     logger: JournalLogger = noopLogger,
   ): Promise<SessionEventJournal> {
-    let epoch: string | undefined;
-    let lastSeq = 0;
-    let sawAnyLine = false;
-    let corrupt = false;
-    let segmentSeq = 0;
-    let tailRepair: JournalTailRepair | undefined;
+    const scan = await scanJournal(filePath);
+    let { epoch, lastSeq } = scan;
+    const { sawAnyLine, corrupt } = scan;
+    let tailRepair = scan.tailRepair;
 
-    try {
-      for await (const line of readLines(filePath)) {
-        const raw = line.raw;
-        if (raw.trim().length === 0) {
-          if (!line.terminated) tailRepair = { kind: 'truncate', byteLength: line.startOffset };
-          continue;
-        }
-        sawAnyLine = true;
-        const parsed = parseJournalLine(raw);
-        if (parsed === undefined) {
-          if (!line.terminated) {
-            tailRepair = { kind: 'truncate', byteLength: line.startOffset };
-            continue;
-          }
-          corrupt = true;
-          break;
-        }
-        if (!line.terminated) tailRepair = { kind: 'terminate' };
-        if (parsed.kind === 'journal_header') {
-          epoch = parsed.epoch; // last header wins
-          segmentSeq = 0;
-          lastSeq = 0;
-          continue;
-        }
-        if (epoch === undefined || parsed.seq !== segmentSeq + 1) {
-          corrupt = true;
-          break;
-        }
-        segmentSeq = parsed.seq;
-        lastSeq = segmentSeq;
-      }
-    } catch (error) {
-      const code = (error as NodeJS.ErrnoException).code;
-      if (code !== 'ENOENT') {
-        corrupt = true;
-        logger.warn(
-          { filePath, err: String(error) },
-          'event journal unreadable; starting a fresh epoch on next append',
-        );
-      }
+    if (scan.readError !== undefined) {
+      logger.warn(
+        { filePath, err: String(scan.readError) },
+        'event journal unreadable; starting a fresh epoch on next append',
+      );
     }
 
     if (corrupt || (sawAnyLine && epoch === undefined)) {
@@ -248,6 +217,31 @@ export class SessionEventJournal {
     return new SessionEventJournal(filePath, logger, epoch, lastSeq, tailRepair);
   }
 
+  /**
+   * Inspect a journal watermark without taking ownership of the file.
+   *
+   * Unlike {@link open}, this method never quarantines corrupt data, repairs a
+   * torn tail, creates a directory, or writes a header. It is the only journal
+   * API that a cold peer should use while another process may own the session.
+   * A missing file and a corrupt/unusable file both return no baseline; the
+   * latter is logged so callers do not mistake it for a valid empty journal.
+   */
+  static async inspect(
+    filePath: string,
+    logger: JournalLogger = noopLogger,
+  ): Promise<JournalWatermark | undefined> {
+    const scan = await scanJournal(filePath);
+    if (scan.missing) return undefined;
+    if (scan.corrupt || (scan.sawAnyLine && scan.epoch === undefined)) {
+      logger.warn(
+        { filePath, err: scan.readError === undefined ? undefined : String(scan.readError) },
+        'event journal inspection found no usable watermark; leaving the file untouched',
+      );
+      return { seq: 0, epoch: undefined };
+    }
+    return { seq: scan.lastSeq, epoch: scan.epoch };
+  }
+
   /** Reserve the next durable seq. The caller must follow with `append()`. */
   nextSeq(): number {
     this.throwIfSticky();
@@ -513,6 +507,16 @@ interface RawJournalLine {
   startOffset: number;
 }
 
+interface JournalScan {
+  epoch: string | undefined;
+  lastSeq: number;
+  sawAnyLine: boolean;
+  corrupt: boolean;
+  missing: boolean;
+  readError: unknown | undefined;
+  tailRepair: JournalTailRepair | undefined;
+}
+
 async function* readLines(filePath: string): AsyncIterable<RawJournalLine> {
   let buffered = '';
   let offset = 0;
@@ -531,6 +535,60 @@ async function* readLines(filePath: string): AsyncIterable<RawJournalLine> {
   if (buffered.length > 0) yield { raw: buffered, terminated: false, startOffset: offset };
 }
 
+async function scanJournal(filePath: string): Promise<JournalScan> {
+  let epoch: string | undefined;
+  let lastSeq = 0;
+  let sawAnyLine = false;
+  let corrupt = false;
+  let missing = false;
+  let readError: unknown | undefined;
+  let segmentSeq = 0;
+  let tailRepair: JournalTailRepair | undefined;
+
+  try {
+    for await (const line of readLines(filePath)) {
+      const raw = line.raw;
+      if (raw.trim().length === 0) {
+        if (!line.terminated) tailRepair = { kind: 'truncate', byteLength: line.startOffset };
+        continue;
+      }
+      sawAnyLine = true;
+      const parsed = parseJournalLine(raw);
+      if (parsed === undefined) {
+        if (!line.terminated) {
+          tailRepair = { kind: 'truncate', byteLength: line.startOffset };
+          continue;
+        }
+        corrupt = true;
+        break;
+      }
+      if (!line.terminated) tailRepair = { kind: 'terminate' };
+      if (parsed.kind === 'journal_header') {
+        epoch = parsed.epoch; // last header wins
+        segmentSeq = 0;
+        lastSeq = 0;
+        continue;
+      }
+      if (epoch === undefined || parsed.seq !== segmentSeq + 1) {
+        corrupt = true;
+        break;
+      }
+      segmentSeq = parsed.seq;
+      lastSeq = segmentSeq;
+    }
+  } catch (error) {
+    const code = (error as NodeJS.ErrnoException).code;
+    if (code === 'ENOENT') {
+      missing = true;
+    } else {
+      corrupt = true;
+      readError = error;
+    }
+  }
+
+  return { epoch, lastSeq, sawAnyLine, corrupt, missing, readError, tailRepair };
+}
+
 async function quarantineCorruptJournal(filePath: string, logger: JournalLogger): Promise<void> {
   try {
     await rename(filePath, `${filePath}.corrupt.${ulid()}`);

packages/kap-server/test/sessionEventJournal.test.ts

@@ -116,6 +116,32 @@ describe('SessionEventJournal', () => {
     await j2.close();
   });
 
+  it('inspects a corrupt journal without quarantining or truncating it', async () => {
+    const corrupt = 'this is not json\n';
+    await writeFile(filePath, corrupt, 'utf8');
+
+    const watermark = await SessionEventJournal.inspect(filePath);
+
+    expect(watermark).toEqual({ seq: 0, epoch: undefined });
+    expect(await readFile(filePath, 'utf8')).toBe(corrupt);
+    await expect(stat(filePath)).resolves.toBeDefined();
+  });
+
+  it('inspects a torn tail without repairing the owner journal', async () => {
+    const owner = await SessionEventJournal.open(filePath);
+    owner.append(owner.nextSeq(), envelope(1));
+    await owner.close();
+    const durable = await readFile(filePath, 'utf8');
+    const torn = `${durable}{"kind":"event"}`;
+    await writeFile(filePath, torn, 'utf8');
+
+    const watermark = await SessionEventJournal.inspect(filePath);
+
+    expect(watermark).toEqual({ seq: 1, epoch: owner.epoch });
+    expect(await readFile(filePath, 'utf8')).toBe(torn);
+    await owner.close();
+  });
+
   it('repairs a torn trailing line before appending the next event', async () => {
     const j1 = await SessionEventJournal.open(filePath);
     j1.append(j1.nextSeq(), envelope(1));