From fab562b5abc6e384fca46cd75f3178f9cfd048d9 Mon Sep 17 00:00:00 2001 From: 7Sageer <7sageer@djwcb.cn> Date: Thu, 23 Jul 2026 11:28:00 +0800 Subject: [PATCH] fix: preserve plan fencing and journal ownership --- .changeset/cold-journal-inspection.md | 5 + .changeset/stale-write-fencing.md | 2 +- .../src/agent/plan/planService.ts | 16 +- .../test/agent/plan/planGuard.test.ts | 33 +++- .../ws/v1/sessionEventBroadcaster.ts | 16 +- .../transport/ws/v1/sessionEventJournal.ts | 158 ++++++++++++------ .../test/sessionEventJournal.test.ts | 26 +++ 7 files changed, 184 insertions(+), 72 deletions(-) create mode 100644 .changeset/cold-journal-inspection.md diff --git a/.changeset/cold-journal-inspection.md b/.changeset/cold-journal-inspection.md new file mode 100644 index 000000000..6b016981a --- /dev/null +++ b/.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. diff --git a/.changeset/stale-write-fencing.md b/.changeset/stale-write-fencing.md index 8b10966d1..458f2a222 100644 --- a/.changeset/stale-write-fencing.md +++ b/.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. diff --git a/packages/agent-core-v2/src/agent/plan/planService.ts b/packages/agent-core-v2/src/agent/plan/planService.ts index 0980c6b22..f1278adc9 100644 --- a/packages/agent-core-v2/src/agent/plan/planService.ts +++ b/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( diff --git a/packages/agent-core-v2/test/agent/plan/planGuard.test.ts b/packages/agent-core-v2/test/agent/plan/planGuard.test.ts index 64e10e6da..88f2a06f5 100644 --- a/packages/agent-core-v2/test/agent/plan/planGuard.test.ts +++ b/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( diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts index ba4e2ad3c..253fad16e 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/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 { 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 { diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts b/packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts index 8498c4fb1..4ddb890cc 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts +++ b/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 { - 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 { + 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 { let buffered = ''; let offset = 0; @@ -531,6 +535,60 @@ async function* readLines(filePath: string): AsyncIterable { if (buffered.length > 0) yield { raw: buffered, terminated: false, startOffset: offset }; } +async function scanJournal(filePath: string): Promise { + 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 { try { await rename(filePath, `${filePath}.corrupt.${ulid()}`); diff --git a/packages/kap-server/test/sessionEventJournal.test.ts b/packages/kap-server/test/sessionEventJournal.test.ts index 94378c2de..6ba70cfb5 100644 --- a/packages/kap-server/test/sessionEventJournal.test.ts +++ b/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));