fix(advisor): make ledger commits atomic

.github/workflows/pr-review-advisor.yaml

@@ -159,6 +159,11 @@ jobs:
 
       - name: Install Pi SDK
         run: |
+          if ! command -v rg >/dev/null 2>&1; then
+            sudo apt-get update -qq
+            sudo apt-get install -y --no-install-recommends ripgrep
+          fi
+          rg --version
           PI_SDK_DIR="$RUNNER_TEMP/pi-sdk"
           npm install --prefix "$PI_SDK_DIR" --ignore-scripts --no-save --package-lock=false --before=2026-05-14T00:00:00.000Z "@earendil-works/pi-coding-agent@${PI_SDK_VERSION}"
           rm -rf "$ADVISOR_DIR/node_modules"

test/advisor-session-context-tools.test.ts

@@ -23,11 +23,12 @@ function contextTurn(name: string, content: string): AdvisorPromptTurn {
 }
 
 const ledgerToolName = "pr_review_update_ledger";
-const finalMutationTools = {
+const atomicMutationTools = {
   activeToolNames: [ledgerToolName],
   requiredToolNames: [ledgerToolName],
   requireToolsBeforeText: [],
-  requireTextBeforeToolNames: [ledgerToolName],
+  requireAssistantText: false,
+  atomicTerminalToolName: ledgerToolName,
 };
 const analysisEvent: AdvisorTurnFlowEvent = { type: "text", text: "analysis" };
 const ledgerStart: AdvisorTurnFlowEvent = { type: "tool_start", toolName: ledgerToolName };
@@ -38,23 +39,33 @@ const ledgerSuccess: AdvisorTurnFlowEvent = {
 };
 const ledgerFailure: AdvisorTurnFlowEvent = { ...ledgerSuccess, isError: true };
 const invalidFinalMutationFlows: Array<[string, AdvisorTurnFlowEvent[], string]> = [
-  ["an omitted call", [analysisEvent], "observed 0 starts"],
+  ["an omitted call", [], "observed 0 successful and 0 failed"],
+  ["an omitted completion", [ledgerStart], "observed 1 starts and 0 completions"],
   [
-    "duplicate starts",
-    [analysisEvent, ledgerStart, ledgerStart, ledgerSuccess],
-    "observed 2 starts",
+    "duplicate successful completions",
+    [ledgerStart, ledgerSuccess, ledgerStart, ledgerSuccess],
+    "observed 2 successful and 0 failed",
   ],
-  ["an omitted completion", [analysisEvent, ledgerStart], "0 successful of 0 total"],
+  ["a failed completion", [ledgerStart, ledgerFailure], "0 successful and 1 failed"],
   [
-    "duplicate successful completions",
-    [analysisEvent, ledgerStart, ledgerSuccess, ledgerSuccess],
-    "2 successful of 2 total",
+    "prose before a successful commit",
+    [analysisEvent, ledgerStart, ledgerSuccess],
+    "emitted prose during atomic",
+  ],
+  [
+    "a read before a successful commit",
+    [
+      { type: "tool_start", toolName: "read" },
+      { type: "tool_end", toolName: "read", isError: false },
+      ledgerStart,
+      ledgerSuccess,
+    ],
+    "called unexpected tool read during atomic commit",
   ],
-  ["a failed completion", [analysisEvent, ledgerStart, ledgerFailure], "0 successful of 1 total"],
   [
-    "a failed duplicate completion",
-    [analysisEvent, ledgerStart, ledgerSuccess, ledgerFailure],
-    "1 successful of 2 total",
+    "activity after a successful commit",
+    [ledgerStart, ledgerSuccess, analysisEvent],
+    "emitted activity after successful",
   ],
 ];
 
@@ -90,7 +101,6 @@ describe("advisor session context tool flow", () => {
       ...contextTurn("review", "{}"),
       activeToolNames: ["pr_review_update_ledger"],
       requiredToolNames: ["pr_review_update_ledger"],
-      requireTextBeforeToolNames: ["pr_review_update_ledger"],
     };
     const tools = resolveAdvisorTurnTools(
       turn,
@@ -126,18 +136,6 @@ describe("advisor session context tool flow", () => {
         tools,
       ).join("; "),
     ).toContain("text before pr_review_context completed");
-    expect(
-      advisorTurnFlowErrors(
-        "review",
-        [
-          { type: "tool_end", toolName: "pr_review_context", isError: false },
-          { type: "text", text: "analysis" },
-          { type: "tool_start", toolName: "pr_review_update_ledger" },
-          { type: "tool_start", toolName: "read" },
-        ],
-        tools,
-      ).join("; "),
-    ).toContain("called read after pr_review_update_ledger");
     expect(
       missingRequiredAdvisorToolNames(tools.requiredToolNames, new Set(["pr_review_context"])),
     ).toEqual(["pr_review_update_ledger"]);
@@ -149,11 +147,37 @@ describe("advisor session context tool flow", () => {
     ).toEqual([]);
   });
 
+  it("rejects an atomic commit configuration with context or extra tools (#6446)", () => {
+    const turn: AdvisorPromptTurn = {
+      ...contextTurn("invalid-atomic", "{}"),
+      activeToolNames: [ledgerToolName],
+      atomicTerminalToolName: ledgerToolName,
+    };
+
+    expect(() =>
+      resolveAdvisorTurnTools(
+        turn,
+        ["pr_review_context"],
+        new Set(["pr_review_context", ledgerToolName]),
+      ),
+    ).toThrow("atomic terminal tool must be the turn's only active and required tool");
+  });
+
   it.each(
     invalidFinalMutationFlows,
-  )("rejects %s for a final mutation tool (#6446)", (_case, events, expectedError) => {
-    expect(advisorTurnFlowErrors("review", events, finalMutationTools).join("; ")).toContain(
+  )("rejects %s for an atomic mutation tool (#6446)", (_case, events, expectedError) => {
+    expect(advisorTurnFlowErrors("review", events, atomicMutationTools).join("; ")).toContain(
       expectedError,
     );
   });
+
+  it("accepts failed atomic attempts before one successful commit (#6446)", () => {
+    const errors = advisorTurnFlowErrors(
+      "review",
+      [ledgerStart, ledgerFailure, ledgerStart, ledgerSuccess],
+      atomicMutationTools,
+    );
+
+    expect(errors).toEqual([]);
+  });
 });

test/advisor-session-runner.test.ts

@@ -10,6 +10,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
 
 const sdk = vi.hoisted(() => {
   type Listener = (event: unknown) => void;
+  type TerminalResponse = "omit" | "fail-once" | "fail-twice" | "fail-then-success" | "success";
   type MockTool = {
     name: string;
     execute: (
@@ -26,13 +27,40 @@ const sdk = vi.hoisted(() => {
     activeToolCalls: [] as string[][],
     contextContents: [] as string[],
     customTools: [] as MockTool[],
+    emitAnalysisError: false,
+    emitCommitProse: false,
+    emitRepairProse: false,
+    omitAnalysis: false,
+    prompts: [] as string[],
+    terminalResponses: [] as TerminalResponse[],
   };
 
   const reset = (): void => {
     state.omitContextTool = false;
     state.activeToolCalls = [];
     state.contextContents = [];
     state.customTools = [];
+    state.emitAnalysisError = false;
+    state.emitCommitProse = false;
+    state.emitRepairProse = false;
+    state.omitAnalysis = false;
+    state.prompts = [];
+    state.terminalResponses = [];
+  };
+
+  const executeTerminalTool = async (tool: MockTool, emit: Listener): Promise<void> => {
+    emit({ type: "tool_execution_start", toolName: tool.name });
+    try {
+      await tool.execute(`${tool.name}-call`, {}, undefined, undefined, undefined as never);
+      emit({ type: "tool_execution_end", toolName: tool.name, isError: false });
+    } catch {
+      emit({ type: "tool_execution_end", toolName: tool.name, isError: true });
+    }
+  };
+
+  const failTerminalTool = (tool: MockTool, emit: Listener): void => {
+    emit({ type: "tool_execution_start", toolName: tool.name });
+    emit({ type: "tool_execution_end", toolName: tool.name, isError: true });
   };
 
   const executeContextTool = async (contextTool: MockTool, emit: Listener): Promise<void> => {
@@ -69,16 +97,59 @@ const sdk = vi.hoisted(() => {
         state.activeToolCalls.push([...toolNames]);
       },
       async prompt(prompt: string) {
+        state.prompts.push(prompt);
         const contextTool = state.customTools.find(
           (tool) => activeToolNames.includes(tool.name) && tool.name.endsWith("_context"),
         );
+        const terminalTool = state.customTools.find(
+          (tool) => activeToolNames.includes(tool.name) && tool.name === "turn_action",
+        );
+        const terminalResponse = terminalTool
+          ? (state.terminalResponses.shift() ?? "omit")
+          : "omit";
         await (contextTool && !state.omitContextTool
           ? executeContextTool(contextTool, emit)
           : Promise.resolve());
-        emit({
-          type: "message_update",
-          assistantMessageEvent: { type: "text_delta", delta: `analysis for ${prompt}` },
-        });
+        if (
+          terminalTool &&
+          (terminalResponse === "fail-once" || terminalResponse === "fail-twice")
+        ) {
+          failTerminalTool(terminalTool, emit);
+        }
+        if (terminalTool && terminalResponse === "fail-twice") {
+          failTerminalTool(terminalTool, emit);
+        }
+        if (terminalTool && terminalResponse === "fail-then-success") {
+          failTerminalTool(terminalTool, emit);
+        }
+        const isRepairPrompt = prompt.includes("Call `turn_action` now");
+        if (
+          !state.omitAnalysis &&
+          (!prompt.includes("Emit no prose before or after") ||
+            (state.emitCommitProse && !isRepairPrompt) ||
+            (state.emitRepairProse && isRepairPrompt))
+        ) {
+          emit({
+            type: "message_update",
+            assistantMessageEvent: { type: "text_delta", delta: `analysis for ${prompt}` },
+          });
+        }
+        if (
+          terminalTool &&
+          (terminalResponse === "success" || terminalResponse === "fail-then-success")
+        ) {
+          await executeTerminalTool(terminalTool, emit);
+        }
+        if (state.emitAnalysisError && !terminalTool) {
+          emit({
+            type: "message_update",
+            assistantMessageEvent: {
+              type: "error",
+              error: { errorMessage: "analysis stream failed" },
+              reason: "error",
+            },
+          });
+        }
         emit({ type: "agent_end" });
       },
       abort: vi.fn(async () => {}),
@@ -133,6 +204,25 @@ function customTool(name: string): ToolDefinition {
   };
 }
 
+function analysisTurn(name: string): AdvisorPromptTurn {
+  return {
+    ...turn(name, '{"repair":true}'),
+    requireAssistantText: true,
+  };
+}
+
+function commitTurn(name: string): AdvisorPromptTurn {
+  return {
+    name,
+    prompt: "Commit the preceding analysis. Emit no prose before or after the tool call.",
+    activeToolNames: ["turn_action"],
+    requiredToolNames: ["turn_action"],
+    atomicTerminalToolName: "turn_action",
+    atomicTerminalRepairPrompt:
+      "Retry only the atomic turn action. Emit no prose before or after the tool call.",
+  };
+}
+
 async function run(promptTurns: AdvisorPromptTurn[]) {
   const dir = fs.mkdtempSync(path.join(os.tmpdir(), "advisor-session-runner-"));
   tempDirs.push(dir);
@@ -160,6 +250,109 @@ afterEach(() => {
 });
 
 describe("advisor session runner", () => {
+  it.each([
+    ["omitted", "omit"],
+    ["failed once", "fail-once"],
+    ["failed twice", "fail-twice"],
+  ] as const)("repairs a terminal tool that was %s (#6446)", async (_case, initialResponse) => {
+    sdk.state.terminalResponses = [initialResponse, "success"];
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toBeUndefined();
+    expect(result.turnErrors).toEqual([]);
+    expect(result.raw).toContain("atomic_terminal_repair_start only-commit turn_action");
+    expect(result.raw).toContain("atomic_terminal_repair_end only-commit turn_action ok");
+    expect(sdk.state.activeToolCalls).toEqual([
+      [...READ_ONLY_TOOLS, "review_context"],
+      READ_ONLY_TOOLS,
+      ["turn_action"],
+      ["turn_action"],
+      READ_ONLY_TOOLS,
+    ]);
+    expect(sdk.state.prompts).toHaveLength(3);
+    expect(sdk.state.prompts[2]).toContain("Call `turn_action` now");
+  });
+
+  it("accepts a failed atomic attempt followed by one same-turn success (#6446)", async () => {
+    sdk.state.terminalResponses = ["fail-then-success"];
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toBeUndefined();
+    expect(result.turnErrors).toEqual([]);
+    expect(result.raw).not.toContain("atomic_terminal_repair_start");
+    expect(sdk.state.prompts).toHaveLength(2);
+  });
+
+  it("rejects prose during the initial tool-only atomic commit (#6446)", async () => {
+    sdk.state.emitCommitProse = true;
+    sdk.state.terminalResponses = ["success"];
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toContain("emitted prose during atomic turn_action commit");
+    expect(result.turnErrors).toEqual([
+      expect.stringContaining("emitted prose during atomic turn_action commit"),
+    ]);
+    expect(sdk.state.prompts).toHaveLength(2);
+  });
+
+  it("does not repair a prose-only atomic commit by mutating the ledger (#6446)", async () => {
+    sdk.state.emitCommitProse = true;
+    sdk.state.terminalResponses = ["omit", "success"];
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toContain("emitted prose during atomic turn_action commit");
+    expect(result.raw).not.toContain("atomic_terminal_repair_start");
+    expect(sdk.state.prompts).toHaveLength(2);
+  });
+
+  it("fails closed after one unsuccessful atomic-terminal repair (#6446)", async () => {
+    sdk.state.terminalResponses = ["omit", "omit"];
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toContain(
+      "only-commit atomic-terminal repair must commit turn_action successfully exactly once",
+    );
+    expect(result.turnErrors).toEqual([
+      expect.stringContaining(
+        "only-commit atomic-terminal repair must commit turn_action successfully exactly once",
+      ),
+    ]);
+    expect(sdk.state.prompts).toHaveLength(3);
+  });
+
+  it("rejects prose during the tool-only atomic-terminal repair (#6446)", async () => {
+    sdk.state.emitRepairProse = true;
+    sdk.state.terminalResponses = ["omit", "success"];
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toContain(
+      "only-commit atomic-terminal repair emitted prose during atomic turn_action commit",
+    );
+    expect(result.turnErrors).toEqual([
+      expect.stringContaining("emitted prose during atomic turn_action commit"),
+    ]);
+  });
+
+  it("fails before the commit turn when required analysis is empty (#6446)", async () => {
+    sdk.state.omitAnalysis = true;
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toContain("only-analysis omitted required analysis");
+    expect(result.turnErrors).toEqual([
+      expect.stringContaining("only-analysis omitted required analysis"),
+    ]);
+    expect(sdk.state.prompts).toHaveLength(1);
+  });
+
+  it("stops before the commit turn when the SDK reports an analysis error (#6446)", async () => {
+    sdk.state.emitAnalysisError = true;
+    const result = await run([analysisTurn("only-analysis"), commitTurn("only-commit")]);
+
+    expect(result.fatalError).toBe("analysis stream failed");
+    expect(result.turnErrors).toEqual(["only-analysis: analysis stream failed"]);
+    expect(sdk.state.prompts).toHaveLength(1);
+  });
+
   it.each([
     ["omitted", false],
     ["failed", true],