diff --git a/src/lost-response-retry.test.ts b/src/lost-response-retry.test.ts new file mode 100644 index 00000000..fd937637 --- /dev/null +++ b/src/lost-response-retry.test.ts @@ -0,0 +1,41 @@ +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { writeFileTool } from "./pi-tools.js"; + +function hash(content: string): string { + return `sha256:${createHash("sha256").update(content).digest("hex")}`; +} + +test("stale retry after a lost mutation response fails closed", async (t) => { + const root = await mkdtemp(join(tmpdir(), "devspace-lost-response-retry-")); + t.after(() => rm(root, { recursive: true, force: true })); + const path = join(root, "note.txt"); + await writeFile(path, "before\n"); + const expectedBeforeHash = hash("before\n"); + + // The local mutation succeeds, but the caller is assumed to lose this response. + await writeFileTool( + { path: "note.txt", content: "agent-change\n" }, + { cwd: root, root, expectedBeforeHash }, + ); + assert.equal(await readFile(path, "utf8"), "agent-change\n"); + + // Another actor changes the file before the caller retries the uncertain operation. + await writeFile(path, "newer-external-change\n"); + + const retry = await writeFileTool( + { path: "note.txt", content: "agent-change\n" }, + { cwd: root, root, expectedBeforeHash }, + ); + + assert.equal(retry.isError, true); + assert.match( + retry.content[0]?.type === "text" ? retry.content[0].text : "", + /File precondition failed/, + ); + assert.equal(await readFile(path, "utf8"), "newer-external-change\n"); +}); diff --git a/src/operation-receipts.test.ts b/src/operation-receipts.test.ts new file mode 100644 index 00000000..bdb63b4d --- /dev/null +++ b/src/operation-receipts.test.ts @@ -0,0 +1,141 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { OperationReceiptManager } from "./operation-receipts.js"; + +const op = (suffix: string) => `op-test-${suffix}`; + +test("replays a completed operation without executing twice", async () => { + const manager = new OperationReceiptManager(); + let executions = 0; + const input = { + workspaceId: "ws_1", + operationId: op("completed"), + tool: "write", + request: { path: "a.txt", content: "hello" }, + execute: async () => ++executions, + }; + assert.deepEqual(await manager.run(input), { value: 1, replayed: false }); + assert.deepEqual(await manager.run(input), { value: 1, replayed: true }); + assert.equal(executions, 1); +}); + +test("joins an in-flight duplicate", async () => { + const manager = new OperationReceiptManager(); + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + let executions = 0; + const input = { + workspaceId: "ws_1", + operationId: op("inflight"), + tool: "exec_command", + request: { cmd: "slow" }, + execute: async () => { executions++; await gate; return 42; }, + }; + const first = manager.run(input); + const second = manager.run(input); + await Promise.resolve(); + assert.equal(executions, 1); + release(); + assert.deepEqual(await first, { value: 42, replayed: false }); + assert.deepEqual(await second, { value: 42, replayed: true }); +}); + +test("replays the same failure without repeating its side effect", async () => { + const manager = new OperationReceiptManager(); + let executions = 0; + const input = { + workspaceId: "ws_1", + operationId: op("failure"), + tool: "bash", + request: { command: "danger" }, + execute: async () => { executions++; throw new Error("failed after side effect"); }, + }; + await assert.rejects(manager.run(input), /failed after side effect/); + await assert.rejects(manager.run(input), /failed after side effect/); + assert.equal(executions, 1); +}); + +test("rejects reusing an operation id for a changed request", async () => { + const manager = new OperationReceiptManager(); + await manager.run({ + workspaceId: "ws_1", + operationId: op("conflict"), + tool: "write", + request: { content: "one" }, + execute: async () => "ok", + }); + await assert.rejects(manager.run({ + workspaceId: "ws_1", + operationId: op("conflict"), + tool: "write", + request: { content: "two" }, + execute: async () => "wrong", + }), /different request/); +}); + +test("canonicalizes object key order", async () => { + const manager = new OperationReceiptManager(); + let executions = 0; + await manager.run({ + workspaceId: "ws_1", + operationId: op("canonical"), + tool: "edit", + request: { a: 1, nested: { x: true, y: "z" } }, + execute: async () => ++executions, + }); + const replay = await manager.run({ + workspaceId: "ws_1", + operationId: op("canonical"), + tool: "edit", + request: { nested: { y: "z", x: true }, a: 1 }, + execute: async () => ++executions, + }); + assert.equal(replay.replayed, true); + assert.equal(executions, 1); +}); + +test("compacts expired results to fail-closed tombstones", async () => { + let now = 0; + const manager = new OperationReceiptManager({ receiptTtlMs: 10, now: () => now }); + const input = { + workspaceId: "ws_1", + operationId: op("expired"), + tool: "write", + request: { content: "one" }, + execute: async () => "ok", + }; + await manager.run(input); + now = 11; + await assert.rejects(manager.run(input), /stored result has expired/); +}); + +test("fails closed instead of evicting live receipts", async () => { + const manager = new OperationReceiptManager({ maxReceipts: 1 }); + await manager.run({ + workspaceId: "ws_1", + operationId: op("capacity1"), + tool: "write", + request: { content: "one" }, + execute: async () => "ok", + }); + await assert.rejects(manager.run({ + workspaceId: "ws_1", + operationId: op("capacity2"), + tool: "write", + request: { content: "two" }, + execute: async () => "ok", + }), /capacity reached/); +}); + +test("rejects malformed operation ids before execution", async () => { + const manager = new OperationReceiptManager(); + let executions = 0; + await assert.rejects(manager.run({ + workspaceId: "ws_1", + operationId: "bad", + tool: "write", + request: {}, + execute: async () => ++executions, + }), /operationId must be/); + assert.equal(executions, 0); +}); diff --git a/src/operation-receipts.ts b/src/operation-receipts.ts new file mode 100644 index 00000000..422e8a27 --- /dev/null +++ b/src/operation-receipts.ts @@ -0,0 +1,199 @@ +import { createHash } from "node:crypto"; + +export const OPERATION_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$/; +export const OPERATION_ID_DESCRIPTION = + "Stable ID for this logical side-effecting operation. Reuse the same ID only when retrying the exact same request after an unknown or lost response; use a new ID for a new operation."; + +const DEFAULT_RECEIPT_TTL_MS = 30 * 60 * 1_000; +const DEFAULT_MAX_RECEIPTS = 1_000; +const DEFAULT_MAX_TOMBSTONES = 100_000; + +type StoredReceipt = { + fingerprint: string; + promise: Promise; + settledAt?: number; +}; + +type Tombstone = { + fingerprint: string; +}; + +export interface OperationReceiptManagerOptions { + receiptTtlMs?: number; + maxReceipts?: number; + maxTombstones?: number; + now?: () => number; +} + +export interface RunRecoverableOperationInput { + workspaceId: string; + operationId: string; + tool: string; + request: unknown; + execute: () => Promise; +} + +export interface RecoverableOperationResult { + value: T; + replayed: boolean; +} + +export class OperationReceiptManager { + private readonly receipts = new Map(); + private readonly tombstones = new Map(); + private readonly receiptTtlMs: number; + private readonly maxReceipts: number; + private readonly maxTombstones: number; + private readonly now: () => number; + + constructor(options: OperationReceiptManagerOptions = {}) { + this.receiptTtlMs = options.receiptTtlMs ?? DEFAULT_RECEIPT_TTL_MS; + this.maxReceipts = options.maxReceipts ?? DEFAULT_MAX_RECEIPTS; + this.maxTombstones = options.maxTombstones ?? DEFAULT_MAX_TOMBSTONES; + this.now = options.now ?? Date.now; + + if (!Number.isFinite(this.receiptTtlMs) || this.receiptTtlMs < 0) { + throw new Error("Operation receipt TTL must be a non-negative number."); + } + if (!Number.isInteger(this.maxReceipts) || this.maxReceipts < 1) { + throw new Error("Operation receipt capacity must be a positive integer."); + } + if (!Number.isInteger(this.maxTombstones) || this.maxTombstones < 1) { + throw new Error("Operation tombstone capacity must be a positive integer."); + } + } + + async run(input: RunRecoverableOperationInput): Promise> { + validateOperationId(input.operationId); + this.compactExpiredReceipts(); + + const key = receiptKey(input.workspaceId, input.operationId); + const fingerprint = requestFingerprint(input.tool, input.request); + const receipt = this.receipts.get(key); + if (receipt) { + assertFingerprintMatches(input.operationId, receipt.fingerprint, fingerprint); + return { + value: await receipt.promise as T, + replayed: true, + }; + } + + const tombstone = this.tombstones.get(key); + if (tombstone) { + assertFingerprintMatches(input.operationId, tombstone.fingerprint, fingerprint); + throw new Error( + `Operation ${input.operationId} was already executed, but its stored result has expired. Do not execute it again; inspect current state and use a new operationId for any new action.`, + ); + } + + if (this.receipts.size >= this.maxReceipts) { + throw new Error( + "Operation receipt capacity reached. Refusing a new side-effecting operation rather than evicting a receipt that may still be needed for safe retry.", + ); + } + + const stored: StoredReceipt = { + fingerprint, + promise: Promise.resolve().then(input.execute), + }; + this.receipts.set(key, stored); + void stored.promise.then( + () => { + stored.settledAt = this.now(); + }, + () => { + stored.settledAt = this.now(); + }, + ); + + return { + value: await stored.promise as T, + replayed: false, + }; + } + + private compactExpiredReceipts(): void { + const now = this.now(); + for (const [key, receipt] of this.receipts) { + if (receipt.settledAt === undefined || now - receipt.settledAt < this.receiptTtlMs) { + continue; + } + if (this.tombstones.size >= this.maxTombstones) { + throw new Error( + "Operation tombstone capacity reached. Refusing further side-effecting operations until DevSpace is restarted, so an old operation ID can never be silently reused.", + ); + } + this.receipts.delete(key); + this.tombstones.set(key, { fingerprint: receipt.fingerprint }); + } + } +} + +const defaultOperationReceiptManager = new OperationReceiptManager(); + +export async function runRecoverableOperation( + input: RunRecoverableOperationInput, +): Promise> { + return defaultOperationReceiptManager.run(input); +} + +export function recoverableStructuredContent>( + structuredContent: T, + operationId: string, + replayed: boolean, +): T & { operationId: string; operationReplayed: boolean } { + return { + ...structuredContent, + operationId, + operationReplayed: replayed, + }; +} + +function receiptKey(workspaceId: string, operationId: string): string { + return `${workspaceId}\u0000${operationId}`; +} + +function requestFingerprint(tool: string, request: unknown): string { + return createHash("sha256") + .update(tool) + .update("\u0000") + .update(canonicalJson(request)) + .digest("hex"); +} + +function canonicalJson(value: unknown): string { + if (value === null || typeof value !== "object") { + const encoded = JSON.stringify(value); + return encoded === undefined ? "null" : encoded; + } + if (Array.isArray(value)) { + return `[${value.map((item) => canonicalJson(item)).join(",")}]`; + } + + const record = value as Record; + const entries = Object.keys(record) + .filter((key) => record[key] !== undefined) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`); + return `{${entries.join(",")}}`; +} + +function validateOperationId(operationId: string): void { + if (!OPERATION_ID_PATTERN.test(operationId)) { + throw new Error( + "operationId must be 8-128 characters and contain only letters, digits, '.', '_', ':', or '-', starting with a letter or digit.", + ); + } +} + +function assertFingerprintMatches( + operationId: string, + expected: string, + actual: string, +): void { + if (expected !== actual) { + throw new Error( + `Operation ${operationId} was already used for a different request. Reuse an operationId only for an exact retry of the same tool call.`, + ); + } +} diff --git a/src/pi-tools-preconditions.test.ts b/src/pi-tools-preconditions.test.ts new file mode 100644 index 00000000..e6502fb5 --- /dev/null +++ b/src/pi-tools-preconditions.test.ts @@ -0,0 +1,129 @@ +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { editFileTool, writeFileTool } from "./pi-tools.js"; + +function hash(content: string): string { + return `sha256:${createHash("sha256").update(content).digest("hex")}`; +} + +test("writeFileTool honors matching content-hash preconditions", async (t) => { + const root = await mkdtemp(join(tmpdir(), "devspace-precondition-write-")); + t.after(() => rm(root, { recursive: true, force: true })); + const path = join(root, "note.txt"); + await writeFile(path, "before\n"); + + const response = await writeFileTool( + { path: "note.txt", content: "after\n" }, + { cwd: root, root, expectedBeforeHash: hash("before\n") }, + ); + + assert.equal(response.isError, undefined); + assert.equal(await readFile(path, "utf8"), "after\n"); +}); + +test("writeFileTool rejects stale and missing-file preconditions", async (t) => { + const root = await mkdtemp(join(tmpdir(), "devspace-precondition-stale-")); + t.after(() => rm(root, { recursive: true, force: true })); + await writeFile(join(root, "note.txt"), "current\n"); + + const stale = await writeFileTool( + { path: "note.txt", content: "overwrite\n" }, + { cwd: root, root, expectedBeforeHash: hash("old\n") }, + ); + assert.equal(stale.isError, true); + assert.match(stale.content[0]?.type === "text" ? stale.content[0].text : "", /File precondition failed/); + assert.equal(await readFile(join(root, "note.txt"), "utf8"), "current\n"); + + const create = await writeFileTool( + { path: "new.txt", content: "created\n" }, + { cwd: root, root, expectedBeforeHash: "missing" }, + ); + assert.equal(create.isError, undefined); + assert.equal(await readFile(join(root, "new.txt"), "utf8"), "created\n"); + + const overwriteExisting = await writeFileTool( + { path: "new.txt", content: "wrong\n" }, + { cwd: root, root, expectedBeforeHash: "missing" }, + ); + assert.equal(overwriteExisting.isError, true); + assert.equal(await readFile(join(root, "new.txt"), "utf8"), "created\n"); +}); + +test("editFileTool rejects an edit after the file diverges", async (t) => { + const root = await mkdtemp(join(tmpdir(), "devspace-precondition-edit-")); + t.after(() => rm(root, { recursive: true, force: true })); + const path = join(root, "note.txt"); + await writeFile(path, "alpha\n"); + const expectedBeforeHash = hash("alpha\n"); + await writeFile(path, "beta\n"); + + const response = await editFileTool( + { path: "note.txt", edits: [{ oldText: "beta", newText: "gamma" }] }, + { cwd: root, root, expectedBeforeHash }, + ); + + assert.equal(response.isError, true); + assert.equal(await readFile(path, "utf8"), "beta\n"); +}); + +test("writeFileTool rejects an explicitly empty precondition", async (t) => { + const root = await mkdtemp(join(tmpdir(), "devspace-precondition-empty-")); + t.after(() => rm(root, { recursive: true, force: true })); + const path = join(root, "note.txt"); + await writeFile(path, "current\n"); + + const response = await writeFileTool( + { path: "note.txt", content: "overwrite\n" }, + { cwd: root, root, expectedBeforeHash: "" }, + ); + + assert.equal(response.isError, true); + assert.equal(await readFile(path, "utf8"), "current\n"); +}); + +test("writeFileTool serializes validation with concurrent mutations", async (t) => { + const root = await mkdtemp(join(tmpdir(), "devspace-precondition-concurrent-")); + t.after(() => rm(root, { recursive: true, force: true })); + const path = join(root, "note.txt"); + await writeFile(path, "before\n"); + + let signalChecked!: () => void; + const checked = new Promise((resolve) => { + signalChecked = resolve; + }); + let releaseFirst!: () => void; + const holdFirst = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const first = writeFileTool( + { path: "note.txt", content: "first\n" }, + { + cwd: root, + root, + expectedBeforeHash: hash("before\n"), + afterPreconditionCheck: async () => { + signalChecked(); + await holdFirst; + }, + }, + ); + + await checked; + + const second = writeFileTool( + { path: "note.txt", content: "second\n" }, + { cwd: root, root, expectedBeforeHash: hash("before\n") }, + ); + + releaseFirst(); + + const [firstResponse, secondResponse] = await Promise.all([first, second]); + assert.equal(firstResponse.isError, undefined); + assert.equal(secondResponse.isError, true); + assert.equal(await readFile(path, "utf8"), "first\n"); +}); diff --git a/src/pi-tools.ts b/src/pi-tools.ts index 06f82197..8040f4a4 100644 --- a/src/pi-tools.ts +++ b/src/pi-tools.ts @@ -1,3 +1,5 @@ +import { createHash } from "node:crypto"; +import { readFile } from "node:fs/promises"; import { createBashTool, createEditTool, @@ -23,8 +25,12 @@ interface ToolContext { cwd: string; root: string; readRoots?: string[]; + expectedBeforeHash?: string; + afterPreconditionCheck?: () => Promise; } +const fileMutationQueues = new Map>(); + function toMcpContent(result: AgentToolResult): McpContent[] { return result.content.map((content) => { if (content.type === "text") { @@ -73,22 +79,69 @@ export async function readFileTool(input: ReadToolInput, context: ToolContext): export async function writeFileTool(input: WriteToolInput, context: ToolContext): Promise { const path = resolveAllowedPath(input.path, context.cwd, [context.root]); - const tool = createWriteTool(context.cwd); - return runTool((params) => tool.execute("write_file", params), { - path, - content: input.content, - }, context); + return withFileMutationLock(path, async () => { + const preconditionError = await checkExpectedBeforeHash(path, context.expectedBeforeHash); + if (preconditionError) return { content: formatToolError(preconditionError), isError: true }; + await context.afterPreconditionCheck?.(); + const tool = createWriteTool(context.cwd); + + return runTool((params) => tool.execute("write_file", params), { + path, + content: input.content, + }, context); + }); } export async function editFileTool(input: EditToolInput, context: ToolContext): Promise> { const path = resolveAllowedPath(input.path, context.cwd, [context.root]); - const tool = createEditTool(context.cwd); - return runTool((params) => tool.execute("edit_file", params), { - path, - edits: input.edits, - }, context); + return withFileMutationLock(path, async () => { + const preconditionError = await checkExpectedBeforeHash(path, context.expectedBeforeHash); + if (preconditionError) return { content: formatToolError(preconditionError), isError: true }; + await context.afterPreconditionCheck?.(); + const tool = createEditTool(context.cwd); + + return runTool((params) => tool.execute("edit_file", params), { + path, + edits: input.edits, + }, context); + }); +} + +async function withFileMutationLock(path: string, mutation: () => Promise): Promise { + const previous = fileMutationQueues.get(path) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise((resolve) => { + release = resolve; + }); + const tail = previous.then(() => current); + fileMutationQueues.set(path, tail); + + await previous; + try { + return await mutation(); + } finally { + release(); + if (fileMutationQueues.get(path) === tail) fileMutationQueues.delete(path); + } +} + +async function checkExpectedBeforeHash(path: string, expectedBeforeHash: string | undefined): Promise { + if (expectedBeforeHash === undefined) return undefined; + const actual = await fileContentHash(path); + if (actual === expectedBeforeHash) return undefined; + return new Error(`File precondition failed: expected ${expectedBeforeHash}, found ${actual}.`); +} + +async function fileContentHash(path: string): Promise { + try { + const content = await readFile(path); + return `sha256:${createHash("sha256").update(content).digest("hex")}`; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return "missing"; + throw error; + } } export async function runShellTool(input: BashToolInput, context: ToolContext): Promise { diff --git a/src/tool-surfaces/claude.ts b/src/tool-surfaces/claude.ts index 912c5951..01061d9b 100644 --- a/src/tool-surfaces/claude.ts +++ b/src/tool-surfaces/claude.ts @@ -4,12 +4,19 @@ import { runShellTool, writeFileTool, } from "../pi-tools.js"; +import { + OPERATION_ID_DESCRIPTION, + OPERATION_ID_PATTERN, + recoverableStructuredContent, + runRecoverableOperation, +} from "../operation-receipts.js"; import { EDIT_TOOL_ANNOTATIONS, SHELL_TOOL_ANNOTATIONS, WRITE_TOOL_ANNOTATIONS, toolNames, workspaceIdDescription, + type ToolContent, type ToolInstructionContext, type ToolRegistrationContext, } from "./types.js"; @@ -22,7 +29,7 @@ import { textBlock, } from "./shared.js"; -const CLAUDE_INSTRUCTIONS = `Use ${toolNames.read} for direct file reads, ${toolNames.edit} for targeted modifications, ${toolNames.write} only for new files or complete rewrites, and ${toolNames.shell} for inspection, tests, builds, and other commands. Shell commands run with the local user's authority and are not sandboxed; workspace validation only selects their initial working directory. Follow instructions returned by ${toolNames.openWorkspace}; read applicable instruction and skill files before working in their scope.`; +const CLAUDE_INSTRUCTIONS = `Use ${toolNames.read} for direct file reads, ${toolNames.edit} for targeted modifications, ${toolNames.write} only for new files or complete rewrites, and ${toolNames.shell} for inspection, tests, builds, and other commands. For each side-effecting call, choose a fresh operationId and reuse that same ID only for an exact retry after an unknown or lost response. Shell commands run with the local user's authority and are not sandboxed; workspace validation only selects their initial working directory. Follow instructions returned by ${toolNames.openWorkspace}; read applicable instruction and skill files before working in their scope.`; export function claudeInstructions({ agents, @@ -38,6 +45,46 @@ export function registerClaudeTools(context: ToolRegistrationContext): void { const CLAUDE_SHELL_DESCRIPTION = `Run a shell command with the local user's authority. Commands are not sandboxed; workspace validation only selects the initial working directory. Use this for file inspection, tests, builds, package scripts, and other commands.`; +interface RecoverableToolResponse { + content: ToolContent[]; + details?: unknown; + isError?: boolean; + structuredContent?: Record; +} + +const operationIdSchema = z + .string() + .regex(OPERATION_ID_PATTERN) + .describe(OPERATION_ID_DESCRIPTION); + +function recoverableOutputSchema(extra: z.ZodRawShape = {}): z.ZodRawShape { + return resultOutputSchema({ + operationId: z.string(), + operationReplayed: z + .boolean() + .describe( + "True when DevSpace returned the stored result of an earlier identical operation instead of repeating its local side effect.", + ), + ...extra, + }); +} + +function attachRecoveryMetadata( + response: RecoverableToolResponse, + operationId: string, + replayed: boolean, +): RecoverableToolResponse { + if (!response.structuredContent) return response; + return { + ...response, + structuredContent: recoverableStructuredContent( + response.structuredContent, + operationId, + replayed, + ), + }; +} + function registerClaudeMutationTools(context: ToolRegistrationContext): void { const { server, config, workspaces } = context; @@ -48,51 +95,76 @@ function registerClaudeMutationTools(context: ToolRegistrationContext): void { description: `Create or completely overwrite a file in a workspace. Prefer ${toolNames.edit} for targeted changes to existing files.`, inputSchema: { workspaceId: z.string().describe(workspaceIdDescription), + operationId: operationIdSchema, path: z .string() .describe("File path to write, relative to the workspace root."), content: z.string().describe("Complete new file content."), + expectedBeforeHash: z + .union([ + z.string().regex(/^sha256:[0-9a-f]{64}$/), + z.literal("missing"), + ]) + .optional() + .describe( + "Optional precondition: sha256:<64 lowercase hex> for the current file contents, or 'missing' if the file must not exist.", + ), }, - outputSchema: resultOutputSchema(), + outputSchema: recoverableOutputSchema(), annotations: WRITE_TOOL_ANNOTATIONS, }, - async ({ workspaceId, ...input }) => { - const startedAt = performance.now(); - const workspace = await workspaces.getWorkspace(workspaceId); - workspaces.resolvePath(workspace, input.path); - const response = await writeFileTool(input, { - cwd: workspace.root, - root: workspace.root, - }); + async ({ workspaceId, operationId, expectedBeforeHash, ...input }) => { + const recovered = await runRecoverableOperation({ + workspaceId, + operationId, + tool: toolNames.write, + request: { ...input, expectedBeforeHash }, + execute: async () => { + const startedAt = performance.now(); + const workspace = await workspaces.getWorkspace(workspaceId); + workspaces.resolvePath(workspace, input.path); + const response = await writeFileTool(input, { + cwd: workspace.root, + root: workspace.root, + expectedBeforeHash, + }); + + if (response.isError) { + logFailedToolResponse( + config, + { + tool: toolNames.write, + workspaceId, + path: input.path, + }, + response.content, + startedAt, + ); + return response; + } - if (response.isError) { - logFailedToolResponse( - config, - { + logToolCall(config, { tool: toolNames.write, workspaceId, path: input.path, - }, - response.content, - startedAt, - ); - return response; - } - - logToolCall(config, { - tool: toolNames.write, - workspaceId, - path: input.path, - success: true, - durationMs: Math.round(performance.now() - startedAt), - }); + success: true, + durationMs: Math.round(performance.now() - startedAt), + }); - return { - ...response, - structuredContent: { - result: contentText(response.content), + return { + ...response, + structuredContent: { + result: contentText(response.content), + }, + }; }, - }; + }); + + return attachRecoveryMetadata( + recovered.value, + operationId, + recovered.replayed, + ); }, ); @@ -103,6 +175,7 @@ function registerClaudeMutationTools(context: ToolRegistrationContext): void { description: `Edit one file in a workspace by replacing exact text blocks. Prefer this over ${toolNames.write} for targeted changes. Each oldText must match a unique, non-overlapping region of the original file; merge nearby changes into one edit and keep oldText as small as possible while still unique.`, inputSchema: { workspaceId: z.string().describe(workspaceIdDescription), + operationId: operationIdSchema, path: z .string() .describe("File path to edit, relative to the workspace root."), @@ -118,55 +191,77 @@ function registerClaudeMutationTools(context: ToolRegistrationContext): void { }), ) .min(1), + expectedBeforeHash: z + .string() + .regex(/^sha256:[0-9a-f]{64}$/) + .optional() + .describe( + "Optional precondition: sha256:<64 lowercase hex> for the current file contents.", + ), }, - outputSchema: resultOutputSchema({ + outputSchema: recoverableOutputSchema({ status: z.literal("applied"), }), annotations: EDIT_TOOL_ANNOTATIONS, }, - async ({ workspaceId, ...input }) => { - const startedAt = performance.now(); - const workspace = await workspaces.getWorkspace(workspaceId); - workspaces.resolvePath(workspace, input.path); - const response = await editFileTool(input, { - cwd: workspace.root, - root: workspace.root, - }); + async ({ workspaceId, operationId, expectedBeforeHash, ...input }) => { + const recovered = await runRecoverableOperation({ + workspaceId, + operationId, + tool: toolNames.edit, + request: { ...input, expectedBeforeHash }, + execute: async () => { + const startedAt = performance.now(); + const workspace = await workspaces.getWorkspace(workspaceId); + workspaces.resolvePath(workspace, input.path); + const response = await editFileTool(input, { + cwd: workspace.root, + root: workspace.root, + expectedBeforeHash, + }); + + if (response.isError) { + logFailedToolResponse( + config, + { + tool: toolNames.edit, + workspaceId, + path: input.path, + }, + response.content, + startedAt, + ); + return response; + } - if (response.isError) { - logFailedToolResponse( - config, - { + const stats = countDiffStats( + response.details?.patch ?? response.details?.diff, + ); + const editResultText = `Edited ${input.path} (+${stats.additions} -${stats.removals}).`; + const editContent = [textBlock(editResultText)]; + logToolCall(config, { tool: toolNames.edit, workspaceId, path: input.path, - }, - response.content, - startedAt, - ); - return response; - } - - const stats = countDiffStats( - response.details?.patch ?? response.details?.diff, - ); - const editResultText = `Edited ${input.path} (+${stats.additions} -${stats.removals}).`; - const editContent = [textBlock(editResultText)]; - logToolCall(config, { - tool: toolNames.edit, - workspaceId, - path: input.path, - success: true, - durationMs: Math.round(performance.now() - startedAt), - }); + success: true, + durationMs: Math.round(performance.now() - startedAt), + }); - return { - content: editContent, - structuredContent: { - status: "applied", - result: contentText(editContent), + return { + content: editContent, + structuredContent: { + status: "applied", + result: contentText(editContent), + }, + }; }, - }; + }); + + return attachRecoveryMetadata( + recovered.value, + operationId, + recovered.replayed, + ); }, ); } @@ -181,6 +276,7 @@ function registerShellTool(context: ToolRegistrationContext): void { description: CLAUDE_SHELL_DESCRIPTION, inputSchema: { workspaceId: z.string().describe(workspaceIdDescription), + operationId: operationIdSchema, command: z .string() .describe("Shell command to execute."), @@ -197,53 +293,67 @@ function registerShellTool(context: ToolRegistrationContext): void { .optional() .describe("Timeout in seconds. Defaults to 30, max 300."), }, - outputSchema: resultOutputSchema(), + outputSchema: recoverableOutputSchema(), annotations: SHELL_TOOL_ANNOTATIONS, }, - async ({ workspaceId, workingDirectory, ...input }) => { - const startedAt = performance.now(); - const workspace = await workspaces.getWorkspace(workspaceId); - const cwd = workspaces.resolveWorkingDirectory( - workspace, - workingDirectory, - ); - const response = await runShellTool(input, { - cwd, - root: workspace.root, - }); + async ({ workspaceId, operationId, workingDirectory, ...input }) => { + const recovered = await runRecoverableOperation({ + workspaceId, + operationId, + tool: toolNames.shell, + request: { ...input, workingDirectory }, + execute: async () => { + const startedAt = performance.now(); + const workspace = await workspaces.getWorkspace(workspaceId); + const cwd = workspaces.resolveWorkingDirectory( + workspace, + workingDirectory, + ); + const response = await runShellTool(input, { + cwd, + root: workspace.root, + }); + + if (response.isError) { + logFailedToolResponse( + config, + { + tool: toolNames.shell, + workspaceId, + workingDirectory: workingDirectory ?? ".", + command: input.command, + commandLength: input.command.length, + }, + response.content, + startedAt, + ); + return response; + } - if (response.isError) { - logFailedToolResponse( - config, - { + logToolCall(config, { tool: toolNames.shell, workspaceId, workingDirectory: workingDirectory ?? ".", command: input.command, commandLength: input.command.length, - }, - response.content, - startedAt, - ); - return response; - } - - logToolCall(config, { - tool: toolNames.shell, - workspaceId, - workingDirectory: workingDirectory ?? ".", - command: input.command, - commandLength: input.command.length, - success: true, - durationMs: Math.round(performance.now() - startedAt), - }); + success: true, + durationMs: Math.round(performance.now() - startedAt), + }); - return { - ...response, - structuredContent: { - result: contentText(response.content), + return { + ...response, + structuredContent: { + result: contentText(response.content), + }, + }; }, - }; + }); + + return attachRecoveryMetadata( + recovered.value, + operationId, + recovered.replayed, + ); }, ); } diff --git a/src/tool-surfaces/codex.ts b/src/tool-surfaces/codex.ts index e3495eff..1d4d3ba2 100644 --- a/src/tool-surfaces/codex.ts +++ b/src/tool-surfaces/codex.ts @@ -1,5 +1,11 @@ import * as z from "zod/v4"; import { applyPatch } from "../apply-patch.js"; +import { + OPERATION_ID_DESCRIPTION, + OPERATION_ID_PATTERN, + recoverableStructuredContent, + runRecoverableOperation, +} from "../operation-receipts.js"; import type { ProcessSnapshot } from "../process-sessions.js"; import { EDIT_TOOL_ANNOTATIONS, @@ -17,7 +23,12 @@ import { type CodexRegistration = (context: ToolRegistrationContext) => void; -const CODEX_INSTRUCTIONS = `Use ${toolNames.read} for direct file reads, apply_patch for all file modifications, exec_command for inspection, tests, builds, and other commands, and write_stdin to poll or interact with running processes. Commands run with the local user's authority and are not sandboxed; workspace validation only selects their initial working directory. Follow instructions returned by ${toolNames.openWorkspace}; read applicable instruction and skill files before working in their scope.`; +const CODEX_INSTRUCTIONS = `Use ${toolNames.read} for direct file reads, apply_patch for all file modifications, exec_command for inspection, tests, builds, and other commands, and write_stdin to poll or interact with running processes. For each side-effecting call, choose a fresh operationId and reuse that same ID only for an exact retry after an unknown or lost response. Commands run with the local user's authority and are not sandboxed; workspace validation only selects their initial working directory. Follow instructions returned by ${toolNames.openWorkspace}; read applicable instruction and skill files before working in their scope.`; + +const operationIdSchema = z + .string() + .regex(OPERATION_ID_PATTERN) + .describe(OPERATION_ID_DESCRIPTION); export function codexInstructions(): string { return CODEX_INSTRUCTIONS; @@ -47,6 +58,12 @@ function processResult(snapshot: ProcessSnapshot): string { function processOutputSchema(): z.ZodRawShape { return resultOutputSchema({ + operationId: z.string(), + operationReplayed: z + .boolean() + .describe( + "True when DevSpace returned the stored result of an earlier identical operation instead of repeating its local side effect.", + ), sessionId: z.number().optional(), running: z.boolean(), exitCode: z.number().int().optional(), @@ -56,20 +73,28 @@ function processOutputSchema(): z.ZodRawShape { }); } -function processToolResponse(snapshot: ProcessSnapshot) { +function processToolResponse( + snapshot: ProcessSnapshot, + operationId: string, + replayed: boolean, +) { const result = processResult(snapshot); const content = [textBlock(result)]; return { content, - structuredContent: { - result, - sessionId: snapshot.sessionId, - running: snapshot.running, - exitCode: snapshot.exitCode, - signal: snapshot.signal, - wallTimeMs: snapshot.wallTimeMs, - outputTruncated: snapshot.outputTruncated, - }, + structuredContent: recoverableStructuredContent( + { + result, + sessionId: snapshot.sessionId, + running: snapshot.running, + exitCode: snapshot.exitCode, + signal: snapshot.signal, + wallTimeMs: snapshot.wallTimeMs, + outputTruncated: snapshot.outputTruncated, + }, + operationId, + replayed, + ), }; } @@ -84,6 +109,7 @@ function registerApplyPatchTool(context: ToolRegistrationContext): void { "Apply one Codex-style patch in a workspace. Supports adding, overwriting, updating, deleting, and moving files. Use this for all file modifications. Paths must be relative to the workspace.", inputSchema: { workspaceId: z.string().describe(workspaceIdDescription), + operationId: operationIdSchema, patch: z .string() .describe( @@ -91,6 +117,12 @@ function registerApplyPatchTool(context: ToolRegistrationContext): void { ), }, outputSchema: resultOutputSchema({ + operationId: z.string(), + operationReplayed: z + .boolean() + .describe( + "True when DevSpace returned the stored result of an earlier identical operation instead of applying the patch again.", + ), additions: z.number(), removals: z.number(), files: z.array( @@ -103,29 +135,46 @@ function registerApplyPatchTool(context: ToolRegistrationContext): void { }), annotations: EDIT_TOOL_ANNOTATIONS, }, - async ({ workspaceId, patch }) => { - const startedAt = performance.now(); - const applied = await runLoggedToolOperation( - config, - { tool: "apply_patch", workspaceId }, - startedAt, - async () => { - const workspace = await workspaces.getWorkspace(workspaceId); - return applyPatch(workspace.root, patch); + async ({ workspaceId, operationId, patch }) => { + const recovered = await runRecoverableOperation({ + workspaceId, + operationId, + tool: "apply_patch", + request: { patch }, + execute: async () => { + const startedAt = performance.now(); + const applied = await runLoggedToolOperation( + config, + { tool: "apply_patch", workspaceId }, + startedAt, + async () => { + const workspace = await workspaces.getWorkspace(workspaceId); + return applyPatch(workspace.root, patch); + }, + ); + const paths = applied.files.map((file) => file.path).join(", "); + const result = `Applied patch to ${applied.files.length} file(s): ${paths}`; + const content = [textBlock(result)]; + + return { + content, + structuredContent: { + result, + additions: applied.additions, + removals: applied.removals, + files: applied.files, + }, + }; }, - ); - const paths = applied.files.map((file) => file.path).join(", "); - const result = `Applied patch to ${applied.files.length} file(s): ${paths}`; - const content = [textBlock(result)]; + }); return { - content, - structuredContent: { - result, - additions: applied.additions, - removals: applied.removals, - files: applied.files, - }, + ...recovered.value, + structuredContent: recoverableStructuredContent( + recovered.value.structuredContent, + operationId, + recovered.replayed, + ), }; }, ); @@ -142,6 +191,7 @@ function registerCodexProcessTools(context: ToolRegistrationContext): void { "Run a command with the local user's authority. Commands are not sandboxed; workspace validation only selects the initial working directory. Returns the result when it exits during the yield window, otherwise returns a sessionId for write_stdin. Use this for file inspection, tests, builds, package scripts, and long-running processes.", inputSchema: { workspaceId: z.string().describe(workspaceIdDescription), + operationId: operationIdSchema, cmd: z.string().min(1).describe("Shell command to execute."), tty: z .boolean() @@ -191,6 +241,7 @@ function registerCodexProcessTools(context: ToolRegistrationContext): void { }, async ({ workspaceId, + operationId, cmd, tty, columns, @@ -199,38 +250,58 @@ function registerCodexProcessTools(context: ToolRegistrationContext): void { yieldTimeMs, maxOutputTokens, }) => { - const startedAt = performance.now(); - const snapshot = await runLoggedToolOperation( - config, - { - tool: "exec_command", - workspaceId, - workingDirectory: workingDirectory ?? ".", - command: cmd, - commandLength: cmd.length, + const recovered = await runRecoverableOperation({ + workspaceId, + operationId, + tool: "exec_command", + request: { + cmd, + tty, + columns, + rows, + workingDirectory, + yieldTimeMs, + maxOutputTokens, }, - startedAt, - async () => { - const workspace = await workspaces.getWorkspace(workspaceId); - const cwd = workspaces.resolveWorkingDirectory( - workspace, - workingDirectory, + execute: async () => { + const startedAt = performance.now(); + return runLoggedToolOperation( + config, + { + tool: "exec_command", + workspaceId, + workingDirectory: workingDirectory ?? ".", + command: cmd, + commandLength: cmd.length, + }, + startedAt, + async () => { + const workspace = await workspaces.getWorkspace(workspaceId); + const cwd = workspaces.resolveWorkingDirectory( + workspace, + workingDirectory, + ); + return processSessions.start({ + workspaceId, + command: cmd, + cwd, + workspaceRoot: workspace.root, + tty, + columns, + rows, + yieldTimeMs, + maxOutputTokens, + }); + }, ); - return processSessions.start({ - workspaceId, - command: cmd, - cwd, - workspaceRoot: workspace.root, - tty, - columns, - rows, - yieldTimeMs, - maxOutputTokens, - }); }, - ); + }); - return processToolResponse(snapshot); + return processToolResponse( + recovered.value, + operationId, + recovered.replayed, + ); }, ); @@ -244,6 +315,7 @@ function registerCodexProcessTools(context: ToolRegistrationContext): void { workspaceId: z .string() .describe("Workspace identifier used to start the process."), + operationId: operationIdSchema, sessionId: z .number() .describe("Process session identifier returned by exec_command."), @@ -289,6 +361,7 @@ function registerCodexProcessTools(context: ToolRegistrationContext): void { }, async ({ workspaceId, + operationId, sessionId, chars, columns, @@ -296,26 +369,45 @@ function registerCodexProcessTools(context: ToolRegistrationContext): void { yieldTimeMs, maxOutputTokens, }) => { - const startedAt = performance.now(); - const snapshot = await runLoggedToolOperation( - config, - { tool: "write_stdin", workspaceId }, - startedAt, - async () => { - await workspaces.getWorkspace(workspaceId); - return processSessions.write({ - workspaceId, - sessionId, - chars, - columns, - rows, - yieldTimeMs, - maxOutputTokens, - }); + const recovered = await runRecoverableOperation({ + workspaceId, + operationId, + tool: "write_stdin", + request: { + sessionId, + chars, + columns, + rows, + yieldTimeMs, + maxOutputTokens, }, - ); + execute: async () => { + const startedAt = performance.now(); + return runLoggedToolOperation( + config, + { tool: "write_stdin", workspaceId }, + startedAt, + async () => { + await workspaces.getWorkspace(workspaceId); + return processSessions.write({ + workspaceId, + sessionId, + chars, + columns, + rows, + yieldTimeMs, + maxOutputTokens, + }); + }, + ); + }, + }); - return processToolResponse(snapshot); + return processToolResponse( + recovered.value, + operationId, + recovered.replayed, + ); }, ); }