fix: recover lost MCP operation responses

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");
+});

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<void>((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);
+});

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<unknown>;
+  settledAt?: number;
+};
+
+type Tombstone = {
+  fingerprint: string;
+};
+
+export interface OperationReceiptManagerOptions {
+  receiptTtlMs?: number;
+  maxReceipts?: number;
+  maxTombstones?: number;
+  now?: () => number;
+}
+
+export interface RunRecoverableOperationInput<T> {
+  workspaceId: string;
+  operationId: string;
+  tool: string;
+  request: unknown;
+  execute: () => Promise<T>;
+}
+
+export interface RecoverableOperationResult<T> {
+  value: T;
+  replayed: boolean;
+}
+
+export class OperationReceiptManager {
+  private readonly receipts = new Map<string, StoredReceipt>();
+  private readonly tombstones = new Map<string, Tombstone>();
+  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<T>(input: RunRecoverableOperationInput<T>): Promise<RecoverableOperationResult<T>> {
+    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<T>(
+  input: RunRecoverableOperationInput<T>,
+): Promise<RecoverableOperationResult<T>> {
+  return defaultOperationReceiptManager.run(input);
+}
+
+export function recoverableStructuredContent<T extends Record<string, unknown>>(
+  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<string, unknown>;
+  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.`,
+    );
+  }
+}