diff --git a/packages/cli/src/chat.ts b/packages/cli/src/chat.ts index b6b1a30..dd56cfb 100644 --- a/packages/cli/src/chat.ts +++ b/packages/cli/src/chat.ts @@ -30,6 +30,7 @@ import { buildCompactedHistory, defaultEnvPath, forgetApiKey, + resumeBlock, runLoop, saveApiKey, summarizeTranscript, @@ -290,6 +291,8 @@ export class Chat { this.screen.append({ kind: "system", title: "continuing", lines: await this.resume(this.opts.continueFrom) }); } catch (err) { this.screen.append({ kind: "notice", level: "error", text: err instanceof Error ? err.message : String(err) }); + this.refresh(); + return; // Do not silently submit an initial task as a new run after continuation failed. } this.refresh(); } @@ -1490,6 +1493,13 @@ export class Chat { const state = loadResume(path); if (state.corruption !== undefined) throw new Error(`this journal is corrupt (${state.corruption})`); if (!state.checkpoint) throw new Error("no checkpoint was written in that session; there is nothing to continue"); + const blocker = resumeBlock(state.checkpoint, null); + if (blocker?.kind === "execution_uncertain") { + throw new Error( + `cannot resume: the run stopped while \`${blocker.tool}\` (${blocker.id}) was running, and whether it ` + + "took effect is unknowable from here. Inspect the working tree, then start a new run", + ); + } const lines: string[] = []; if (state.checkpoint.currentChannel !== this.settings.channel) { // The transcript is in that channel's format, so the session follows it. diff --git a/packages/cli/test/chat.test.ts b/packages/cli/test/chat.test.ts index baec3e4..0e991e1 100644 --- a/packages/cli/test/chat.test.ts +++ b/packages/cli/test/chat.test.ts @@ -9,7 +9,7 @@ */ import { execFileSync } from "node:child_process"; -import { existsSync, mkdirSync, mkdtempSync, readFileSync, statSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, statSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import { PassThrough } from "node:stream"; @@ -17,6 +17,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { AgentRegistry, BUILTIN_AGENTS } from "@motifcode/agents"; import { TransportError, type CompletionRequest, type CompletionResponse, type Transport } from "@motifcode/core"; import { DEFAULT_HOOKS } from "@motifcode/hooks"; +import type { JournalLine } from "@motifcode/journal"; import { doneBody, toolCallBody } from "@motifcode/replay"; import { BUILTIN_SKILLS, SkillRegistry } from "@motifcode/skills"; import { CORE_TOOLS, toolPrefix } from "@motifcode/tools"; @@ -410,6 +411,63 @@ describe("interactive session", () => { await vi.waitFor(() => expect(s.screen()).toContain("no session 9")); }); + it.each([ + ["slash", false], ["slash", true], ["continue", false], ["continue", true], + ] as const)("refuses %s of an uncertain write with mutating=%s before changing history", async (entry, mutating) => { + const t = new GateTransport([...reply("original reply"), ...reply("unexpected continuation")]); + const original = session(t); + open.push(original); + original.type("original task\r"); + await vi.waitFor(() => expect(original.chat.tasksCompleted).toBe(1)); + const previous = structuredClone(original.chat.transcript); + const folder = join(original.cwd, ".motif", "sessions"); + const file = join(folder, readdirSync(folder)[0]!); + const lines = readFileSync(file, "utf8").trimEnd().split("\n").map((line) => JSON.parse(line) as JournalLine); + for (const line of lines) { + if ("t" in line || line.record.t !== "checkpoint") continue; + line.record.state.inFlightTool = { name: "write", id: "root-c1", argumentsHash: "fixture", mutating }; + line.record.state.currentChannel = "object"; + line.record.state.messages = [{ role: "system", content: "old system" }, { role: "user", content: "UNSAFE_HISTORY" }]; + } + writeFileSync(file, lines.map((line) => JSON.stringify(line)).join("\n") + "\n"); + const target = entry === "slash" ? original : session(t, { continueFrom: file, initialTask: "must not run" }); + if (target !== original) open.push(target); + else target.type(`/resume ${file}\r`); + await vi.waitFor(() => expect(target.screen()).toContain("Inspect the working tree")); + expect(target.screen()).toContain("start a new run"); + expect(target.chat.transcript).toEqual(entry === "slash" ? previous : []); + expect(target.screen()).not.toContain("channel set to object"); + expect(t.seen).toHaveLength(1); + }); + + it.each([ + ["slash", "read"], ["slash", "complete"], ["continue", "read"], ["continue", "complete"], + ] as const)("still loads %s conversations with %s checkpoints", async (entry, state) => { + const t = new GateTransport([...reply("recorded reply")]); + const original = session(t); + open.push(original); + original.type("recorded task\r"); + await vi.waitFor(() => expect(original.chat.tasksCompleted).toBe(1)); + const folder = join(original.cwd, ".motif", "sessions"); + const file = join(folder, readdirSync(folder)[0]!); + const lines = readFileSync(file, "utf8").trimEnd().split("\n").map((line) => JSON.parse(line) as JournalLine); + for (const line of lines) { + if ("t" in line) line.header.model.id = "previous/model"; + else if (line.record.t === "checkpoint" && state === "read") { + line.record.state.inFlightTool = { name: "read", id: "root-c1", argumentsHash: "fixture", mutating: false }; + } + } + writeFileSync(file, lines.map((line) => JSON.stringify(line)).join("\n") + "\n"); + const target = session(t, entry === "continue" ? { continueFrom: file } : {}); + open.push(target); + if (entry === "slash") target.type(`/resume ${file}\r`); + await vi.waitFor(() => expect(target.screen()).toContain("continuing from")); + expect(target.chat.transcript).toEqual(original.chat.transcript); + expect(target.screen()).toContain("recorded against previous/model"); + expect(target.screen()).toContain("that session had finished"); + expect(t.seen).toHaveLength(1); + }); + it("shows tool output in full on ctrl-o and clips it again", async () => { const long = Array.from({ length: 30 }, (_, i) => `line ${i}`).join("\\n"); const t = new GateTransport([toolCallBody("bash", { command: `printf '${long}'` }), ...reply("ok")]); diff --git a/packages/cli/test/e2e.test.ts b/packages/cli/test/e2e.test.ts index 1596120..c836c82 100644 --- a/packages/cli/test/e2e.test.ts +++ b/packages/cli/test/e2e.test.ts @@ -18,6 +18,7 @@ import { tmpdir } from "node:os"; import { dirname, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import type { JournalEnvelopeV2 } from "@motifcode/journal"; const HERE = dirname(fileURLToPath(import.meta.url)); const REPO = resolve(HERE, "../../.."); @@ -289,6 +290,49 @@ describe("cli process end to end", () => { expect(server.bodies).toHaveLength(0); }, 30_000); + it("refuses uncertain writes before contacting the model, including old false flags", async () => { + server = new MockServer((turn) => turn === 1 + ? toolCall("write", { path: "write-marker.txt", content: "written once" }) + : "Continued after the completed write."); + await server.start(); + const initial = await runCli([ + "write the marker", "--max-turns", "1", "--permissions", "auto", "--endpoint", server.endpoint, "--no-hero", + ], dir); + expect(initial.code, `${initial.stdout}\n${initial.stderr}`).toBe(1); + expect(readFileSync(join(dir, "write-marker.txt"), "utf8")).toBe("written once"); + const sessions = join(dir, ".motif", "sessions"); + const journal = join(sessions, readdirSync(sessions)[0]!); + const lines = readFileSync(journal, "utf8").trimEnd().split("\n"); + const intentIndex = lines.findIndex((line) => { + const record = JSON.parse(line).record as JournalEnvelopeV2["record"] | undefined; + return record?.t === "checkpoint" && record.state.inFlightTool?.name === "write"; + }); + expect(intentIndex).toBeGreaterThan(0); + + for (const mutating of [false, true]) { + // Simulate a journal ending at intent, including the legacy misclassification. + const intent = JSON.parse(lines[intentIndex]!) as JournalEnvelopeV2; + if (intent.record.t !== "checkpoint") throw new Error("expected write intent"); + intent.record.state.inFlightTool!.mutating = mutating; + writeFileSync(journal, [...lines.slice(0, intentIndex), JSON.stringify(intent)].join("\n") + "\n"); + const resumed = await runCli(["resume", journal, "--print", "--endpoint", server.endpoint], dir); + expect(resumed.code, resumed.stderr).toBe(2); + expect(resumed.stdout).toBe(""); + expect(resumed.stderr).toContain("cannot resume: the run stopped while `write`"); + expect(resumed.stderr).toContain("Inspect the working tree, then start a new run"); + expect(server.bodies).toHaveLength(1); + expect(readFileSync(join(dir, "write-marker.txt"), "utf8")).toBe("written once"); + } + + // A recorded observation and completion checkpoint make ordinary continuation safe. + writeFileSync(journal, lines.filter((line) => JSON.parse(line).record?.t !== "scope_end").join("\n") + "\n"); + const completed = await runCli(["resume", journal, "--print", "--endpoint", server.endpoint], dir); + expect(completed.code, completed.stderr).toBe(0); + expect(completed.stdout).toBe("Continued after the completed write.\n"); + expect(server.bodies).toHaveLength(2); + expect(server.bodies[1]!.messages.at(-1)!.role).toBe("tool"); + }, 30_000); + it("refuses --interactive without a terminal, in one line", async () => { server = new MockServer(() => toolCall("done", { summary: "s" })); await server.start(); diff --git a/packages/core/src/checkpoint.ts b/packages/core/src/checkpoint.ts index d0f395a..f7454d8 100644 --- a/packages/core/src/checkpoint.ts +++ b/packages/core/src/checkpoint.ts @@ -81,7 +81,7 @@ export interface ResumeCompatibility { export const PROTOCOL_VERSION = 2; /** Tools whose effects outlive the call, and so must never be replayed blind. */ -const MUTATING = new Set(["bash", "apply_patch", "term", "task", "mcp"]); +const MUTATING = new Set(["bash", "write", "apply_patch", "term", "task", "mcp"]); export function isMutating(toolName: string): boolean { return MUTATING.has(toolName); @@ -130,7 +130,9 @@ export function resumeBlock( ): ResumeBlock | null { if (compat !== null) return { kind: "incompatible", detail: compat }; const inflight = checkpoint.inFlightTool; - if (inflight && inflight.mutating) { + // Older journals misclassified `write`. Keep a recorded true flag, but also + // apply today's classification so a legacy false flag cannot permit continuation. + if (inflight && (inflight.mutating || isMutating(inflight.name))) { return { kind: "execution_uncertain", tool: inflight.name, id: inflight.id }; } return null; diff --git a/packages/core/test/resume.test.ts b/packages/core/test/resume.test.ts index 7c9065f..6d8058b 100644 --- a/packages/core/test/resume.test.ts +++ b/packages/core/test/resume.test.ts @@ -166,7 +166,7 @@ describe("an uncertain command is not re-run", () => { }); it("classifies which tools can change the world", () => { - for (const t of ["bash", "apply_patch", "term", "task", "mcp"]) { + for (const t of ["bash", "write", "apply_patch", "term", "task", "mcp"]) { expect(isMutating(t), t).toBe(true); } for (const t of ["read", "skill", "done"]) { @@ -181,6 +181,20 @@ describe("an uncertain command is not re-run", () => { expect(resumeBlock(cp, null)).toMatchObject({ kind: "execution_uncertain", tool: "apply_patch" }); }); + it.each([ + ["write", false], + ["write", true], + ["apply_patch", false], + ["custom_mutation", true], + ] as const)("blocks in-flight %s with recorded mutating=%s", (name, mutating) => { + const cp = { + inFlightTool: { id: "root-c1", name, argumentsHash: "x", mutating }, + } as LoopCheckpoint; + expect(resumeBlock(cp, null)).toEqual({ kind: "execution_uncertain", tool: name, id: "root-c1" }); + // A classification correction must not override compatibility checks. + expect(resumeBlock(cp, "schema changed")).toEqual({ kind: "incompatible", detail: "schema changed" }); + }); + it("allows a resume whose in-flight tool only read", () => { const cp = { inFlightTool: { id: "root-c1", name: "read", argumentsHash: "x", mutating: false }, diff --git a/packages/journal/src/index.ts b/packages/journal/src/index.ts index 4c5d935..2dcbd84 100644 --- a/packages/journal/src/index.ts +++ b/packages/journal/src/index.ts @@ -33,7 +33,7 @@ import { appendFileSync, existsSync, mkdirSync, readFileSync, readdirSync, statSync } from "node:fs"; import { dirname, join } from "node:path"; -import type { LoopCheckpoint, LoopEvent, SessionEndReason } from "@motifcode/core"; +import { resumeBlock, type LoopCheckpoint, type LoopEvent, type SessionEndReason } from "@motifcode/core"; import type { ChannelId, Message } from "@motifcode/protocol"; export const JOURNAL_VERSION = 2; @@ -431,10 +431,10 @@ export function checkResumable( if (state.header.model.id !== current.model) { return `recorded against model ${state.header.model.id}, now ${current.model}`; } - const inflight = state.checkpoint.inFlightTool; - if (inflight?.mutating) { + const blocker = resumeBlock(state.checkpoint, null); + if (blocker?.kind === "execution_uncertain") { return ( - `the run stopped while \`${inflight.name}\` (${inflight.id}) was running, and whether it ` + + `the run stopped while \`${blocker.tool}\` (${blocker.id}) was running, and whether it ` + "took effect is unknowable from here. Inspect the working tree, then start a new run" ); } diff --git a/packages/journal/test/journal.test.ts b/packages/journal/test/journal.test.ts index d336c1a..f8864b8 100644 --- a/packages/journal/test/journal.test.ts +++ b/packages/journal/test/journal.test.ts @@ -11,7 +11,9 @@ import { mkdtempSync, readFileSync, writeFileSync, statSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import type { LoopCheckpoint } from "@motifcode/core"; +import { resumeBlock, runLoop, type LoopCheckpoint } from "@motifcode/core"; +import { CORE_TOOLS } from "@motifcode/tools"; +import { ScriptedTransport, toolCallBody } from "@motifcode/replay"; import { Journal, checkResumable, @@ -321,6 +323,73 @@ describe("resume", () => { expect(checkResumable(loadResume(j.path), current)).toBeNull(); }); + it.each([ + ["write", false, true], + ["write", true, true], + ["apply_patch", false, true], + ["custom_mutation", true, true], + ["read", false, false], + ] as const)("agrees with core for %s with recorded mutating=%s", (name, mutating, blocked) => { + const state = loadResume(interrupted(checkpoint({ + inFlightTool: { id: "root-c4", name, argumentsHash: "abc", mutating }, + })).path); + expect(resumeBlock(state.checkpoint!, null) !== null).toBe(blocked); + const reason = checkResumable(state, current); + if (blocked) { + expect(reason).toContain(`\`${name}\``); + expect(reason).toContain("Inspect the working tree, then start a new run"); + } else { + expect(reason).toBeNull(); + } + expect(checkResumable({ ...state, corruption: "broken record" }, current)).toContain("corrupt"); + expect(checkResumable(state, { ...current, model: "other" })).toContain("now other"); + }); + + it.each(["before", "intent", "effect", "observation", "truncated", "complete"] as const)( + "uses the last recorded checkpoint when a write stops at %s", + async (stopAt) => { + const j = newJournal(); + j.record(ROOT, { t: "scope_start", task: "write the marker", initialMessages: [] }); + let executed = false; + const stopped = new Error("simulated interruption"); + await expect(runLoop({ + tools: [...CORE_TOOLS], + system: () => "You are motifcode.", + userTask: "write the marker", + transport: new ScriptedTransport([toolCallBody("write", { path: "marker", content: "written" })]), + executor: { run: async () => { + // The fake executor can have an effect before the observation reaches the journal. + executed = true; + if (stopAt === "effect") throw stopped; + return { ok: true, output: "written" }; + } }, + emit: (event) => { + j.sinkFor(ROOT)(event); + if (stopAt === "observation" && event.type === "tool_end") throw stopped; + }, + onCheckpoint: (cp) => { + const completed = cp.messages.some((m) => m.role === "tool"); + if (stopAt === "truncated" && completed) { + // A partially appended completion must leave the preceding intent authoritative. + writeFileSync(j.path, readFileSync(j.path, "utf8") + '{"v":2,"record":{"t":"checkpoint"'); + throw stopped; + } + j.checkpointFor(ROOT)(cp); + if (stopAt === "before" || (stopAt === "intent" && cp.inFlightTool) || completed) throw stopped; + }, + })).rejects.toBe(stopped); + + const state = loadResume(j.path); + const uncertain = stopAt !== "before" && stopAt !== "complete"; + expect(executed).toBe(stopAt !== "before" && stopAt !== "intent"); + expect(state.truncatedTail).toBe(stopAt === "truncated"); + expect(state.checkpoint!.messages.some((m) => m.role === "tool")).toBe(stopAt === "complete"); + expect(resumeBlock(state.checkpoint!, null) !== null).toBe(uncertain); + expect(checkResumable(state, current) !== null).toBe(uncertain); + if (uncertain) expect(state.checkpoint!.inFlightTool).toMatchObject({ name: "write", mutating: true }); + }, + ); + it("refuses when no checkpoint was ever written", () => { const j = newJournal(); j.record(ROOT, { t: "scope_start", task: "t", initialMessages: [] });