fix(KB-153): detect exhausted-retry errors from pi-coding-agent sessions

- Add checkSessionError helper that re-raises errors stored on session.state.error after prompt() resolves silently when retries are exhausted - Integrate checkSessionError in executor, triage, merger, and reviewer agents so existing catch blocks with isUsageLimitError can trigger UsageLimitPauser - Add tests for checkSessionError and for each agent's error propagation path - Add audit report documenting the error propagation gap - Add changeset for the fix
2026-03-28 03:22:53 -04:00
parent a2a12f94eb
commit 3dc741c56e
12 changed files with 410 additions and 4 deletions
--- a/.changeset/fix-rate-limit-auto-pause.md
+++ b/.changeset/fix-rate-limit-auto-pause.md
@@ -0,0 +1,5 @@
+---
+"@dustinbyrne/kb": patch
+---
+
+Fix auto-pause on rate limit when pi-coding-agent exhausts retries. After `session.prompt()` resolves with exhausted retries, all four agent types (executor, triage, merger, reviewer) now detect the error on `session.state.error` and trigger `UsageLimitPauser` to activate global pause. Previously, rate-limit errors that pi-coding-agent handled internally were silently swallowed, causing tasks to be promoted to wrong columns with incomplete work.
--- a/.kb/tasks/KB-153/AUDIT.md
+++ b/.kb/tasks/KB-153/AUDIT.md
@@ -0,0 +1,139 @@
+# KB-153 Audit: Auto-Pause on Rate Limit When pi-coding-agent Exhausts Retries
+
+**Date:** 2026-03-28
+**Author:** KB Engine (automated audit)
+
+## 1. pi-coding-agent Retry Flow
+
+### How Retries Work
+
+When `session.prompt()` is called, it ultimately calls `this.agent.prompt(messages)` followed by `await this.waitForRetry()`. The retry mechanism works as follows:
+
+1. **Error Detection (`_isRetryableError`):** On `agent_end`, the session checks the last assistant message. If `stopReason === "error"` and `errorMessage` matches retryable patterns (overloaded, rate limit, 429, 500, 502, 503, 504, etc.), it's considered retryable.
+
+2. **Retry Handling (`_handleRetryableError`):** Increments `_retryAttempt` counter. If `_retryAttempt > settings.maxRetries` (currently 3), it:
+   - Emits `auto_retry_end` with `{ success: false, finalError: message.errorMessage }`
+   - Resets `_retryAttempt = 0`
+   - Calls `_resolveRetry()` to resolve the pending promise
+   - Returns `false` (did not retry)
+
+3. **Resolution (`_resolveRetry`):** Resolves the `_retryPromise` that `waitForRetry()` is awaiting.
+
+4. **State Propagation:** At `turn_end`, the agent core sets `this._state.error = event.message.errorMessage` when the assistant message has an error. This persists on `session.state.error`.
+
+### Key Behavior: `prompt()` Resolves Without Throwing
+
+After retries are exhausted:
+```
+session.prompt(message)
+  → this.agent.prompt(messages)         // LLM call fails
+  → agent_end event fires
+  → _handleRetryableError() returns false (max retries exceeded)
+  → _resolveRetry() resolves the retry promise
+  → await this.waitForRetry() completes
+  → prompt() returns normally (no exception)
+```
+
+The error is available on `session.state.error` (e.g., `"rate_limit_error: Rate limit exceeded"`), but **no exception is thrown**.
+
+### `auto_retry_end` Event
+
+pi-coding-agent emits `auto_retry_end` with `{ success: false, attempt: N, finalError: "..." }` when retries are exhausted. **No kb engine component subscribes to this event.**
+
+## 2. Gap Analysis by Agent Type
+
+### 2.1 Executor (`packages/engine/src/executor.ts`)
+
+**Current flow:**
+```typescript
+// In agentWork closure:
+await session.prompt(agentPrompt);  // ← resolves normally after exhausted retries
+
+// These checks run, but session.state.error is set:
+if (this.depAborted.has(task.id)) { ... }  // false
+if (this.pausedAborted.has(task.id)) { ... }  // false
+if (taskDone) { ... }  // false (agent didn't call task_done)
+else { ... }  // moves to in-review with incomplete work!
+```
+
+**Consequence:** Task is moved to `in-review` with incomplete work, or with a log message "Agent finished without calling task_done — moved to in-review for inspection". The `catch` block containing `isUsageLimitError` never fires, so `UsageLimitPauser.onUsageLimitHit()` is never called.
+
+### 2.2 Triage (`packages/engine/src/triage.ts`)
+
+**Current flow:**
+```typescript
+// In specifyTask():
+await session.prompt(agentPrompt, ...);  // ← resolves normally after exhausted retries
+
+// Duplicate check runs on possibly empty/incomplete PROMPT.md
+const written = await readFile(join(this.rootDir, promptPath), "utf-8").catch(() => "");
+// ... proceeds to moveTask(task.id, "todo") with broken spec
+```
+
+**Consequence:** A partially-written or empty PROMPT.md is treated as a valid specification. The task is moved from `triage` to `todo` with an incomplete/broken spec. The `catch` block containing `isUsageLimitError` never fires.
+
+### 2.3 Merger (`packages/engine/src/merger.ts`)
+
+**Current flow:**
+```typescript
+// In aiMergeTask():
+await session.prompt(prompt);  // ← resolves normally after exhausted retries
+
+// Staged-changes check runs:
+const staged = execSync("git diff --cached --quiet 2>&1; echo $?", ...);
+if (staged !== "0") {
+  // Fallback commit with possibly dirty/incomplete state!
+  execSync(`git commit -m "feat(${taskId}): merge ${branch}" -m "${escapedLog}"`, ...);
+}
+result.merged = true;  // Reports success even though the merge was incomplete
+```
+
+**Consequence:** If the merge had conflicts and the agent hit a rate limit mid-resolution, the merge proceeds with a fallback commit that may contain conflict markers or incomplete resolution. The `catch` block containing `isUsageLimitError` never fires.
+
+### 2.4 Reviewer (`packages/engine/src/reviewer.ts`)
+
+**Current flow:**
+```typescript
+// In reviewStep():
+await session.prompt(request);  // ← resolves normally after exhausted retries
+
+// reviewText is empty (agent didn't produce output)
+const verdict = extractVerdict(reviewText);  // Returns "UNAVAILABLE"
+```
+
+**Consequence:** The reviewer returns an `UNAVAILABLE` verdict from empty review text. The executor's `createReviewStepTool` catch block would handle a thrown error, but since no error is thrown, the review silently fails. **Additionally, the reviewer has NO usage-limit handling at all** — even if an error were thrown, there's no `isUsageLimitError` check in `reviewStep()`.
+
+## 3. Systemic Impact
+
+When any agent hits a rate limit and retries are exhausted:
+
+1. **No global pause triggered** — `UsageLimitPauser.onUsageLimitHit()` is never called
+2. **Engine keeps launching new agents** — The scheduler continues dispatching tasks to agents that will also hit the same rate limit
+3. **API credits wasted** — Each new agent makes 1 + maxRetries = 4 API calls before silently failing
+4. **Tasks in wrong columns** — Executor moves incomplete tasks to `in-review`, triage moves broken specs to `todo`, merger may commit conflict markers
+
+## 4. Fix Approach
+
+### Strategy: Post-Prompt Error Check
+
+Add a utility function `checkSessionError(session)` that:
+1. Reads `session.state.error`
+2. If set and non-empty, throws `new Error(session.state.error)`
+
+Call this function immediately after every `await session.prompt(...)` in all four agent types. This re-raises the error that pi-coding-agent swallowed, routing it into the existing `catch` blocks where `isUsageLimitError` already triggers `UsageLimitPauser`.
+
+### Why This Approach
+
+- **Minimal change** — One new helper function, one line added per agent
+- **Leverages existing patterns** — All four agents already have catch blocks with `isUsageLimitError` checks (except reviewer, which needs one added)
+- **No pi-coding-agent modifications** — Works with the library's current behavior
+- **Forward-compatible** — If pi-coding-agent later changes to throw on exhausted retries, the check becomes a no-op (session.state.error would be undefined since the error was thrown)
+
+### Per-Agent Fix Details
+
+| Agent    | Location                                    | Fix                                                        |
+|----------|---------------------------------------------|------------------------------------------------------------|
+| Executor | `agentWork` closure after `session.prompt()` | `checkSessionError(session)` before dep/pause/done checks  |
+| Triage   | `specifyTask()` after `session.prompt()`     | `checkSessionError(session)` before duplicate check         |
+| Merger   | `aiMergeTask()` after `session.prompt()`     | `checkSessionError(session)` before staged-changes check    |
+| Reviewer | `reviewStep()` after `session.prompt()`      | `checkSessionError(session)` inside try block; caller handles |
--- a/packages/engine/src/executor.test.ts
+++ b/packages/engine/src/executor.test.ts
@@ -2576,6 +2576,51 @@ describe("TaskExecutor usage limit detection", () => {
    expect(onError).toHaveBeenCalled();
  });

+  it("triggers global pause when session.prompt() resolves with exhausted-retry error on state.error", async () => {
+    const store = createMockStore();
+    const pauser = new UsageLimitPauser(store);
+    const onUsageLimitHitSpy = vi.spyOn(pauser, "onUsageLimitHit");
+
+    // session.prompt() resolves normally, but session.state.error is set
+    // (this is what happens when pi-coding-agent exhausts retries)
+    const mockSession = {
+      prompt: vi.fn().mockResolvedValue(undefined),
+      dispose: vi.fn(),
+      state: { error: "rate_limit_error: Rate limit exceeded" },
+    };
+    mockedCreateHaiAgent.mockResolvedValue({ session: mockSession } as any);
+
+    const onError = vi.fn();
+    const executor = new TaskExecutor(store, "/tmp/test", {
+      onError,
+      usageLimitPauser: pauser,
+    });
+
+    await executor.execute({
+      id: "KB-001",
+      title: "Test",
+      description: "Test",
+      column: "in-progress",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    // UsageLimitPauser should be called
+    expect(onUsageLimitHitSpy).toHaveBeenCalledWith(
+      "executor",
+      "KB-001",
+      "rate_limit_error: Rate limit exceeded",
+    );
+    // Task should be marked as failed
+    expect(store.updateTask).toHaveBeenCalledWith("KB-001", { status: "failed" });
+    // onError callback should fire
+    expect(onError).toHaveBeenCalled();
+  });
+
  it("triggers global pause for overloaded error", async () => {
    const store = createMockStore();
    const pauser = new UsageLimitPauser(store);
--- a/packages/engine/src/executor.ts
+++ b/packages/engine/src/executor.ts
@@ -12,7 +12,7 @@ import { PRIORITY_EXECUTE, type AgentSemaphore } from "./concurrency.js";
 import type { WorktreePool } from "./worktree-pool.js";
 import { AgentLogger } from "./agent-logger.js";
 import { executorLog, reviewerLog } from "./logger.js";
-import { isUsageLimitError, type UsageLimitPauser } from "./usage-limit-detector.js";
+import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./usage-limit-detector.js";

 // Re-export for backward compatibility (tests import from executor.ts)
 export { summarizeToolArgs } from "./agent-logger.js";
@@ -430,6 +430,11 @@ export class TaskExecutor {
          const agentPrompt = buildExecutionPrompt(detail, this.rootDir, settings);
          await session.prompt(agentPrompt);

+          // Re-raise errors that pi-coding-agent swallowed after exhausting retries.
+          // session.prompt() resolves normally even when retries are exhausted —
+          // the error is stored on session.state.error instead of being thrown.
+          checkSessionError(session);
+
          // If dependency was added during execution, discard worktree and move to triage
          if (this.depAborted.has(task.id)) {
            this.depAborted.delete(task.id);
--- a/packages/engine/src/merger.test.ts
+++ b/packages/engine/src/merger.test.ts
@@ -466,6 +466,39 @@ describe("aiMergeTask — usage limit detection", () => {
    expect(store.updateSettings).toHaveBeenCalledWith({ globalPause: true });
  });

+  it("triggers global pause when session.prompt() resolves with exhausted-retry error on state.error", async () => {
+    const store = createMockStore(
+      { id: "KB-050", worktree: "/tmp/root/.worktrees/KB-050" },
+      [{ id: "KB-050", worktree: "/tmp/root/.worktrees/KB-050", column: "in-review" } as Task],
+    );
+    const pauser = new UsageLimitPauser(store);
+    const onUsageLimitHitSpy = vi.spyOn(pauser, "onUsageLimitHit");
+
+    // session.prompt() resolves normally, but session.state.error is set
+    const mockSession = {
+      prompt: vi.fn().mockResolvedValue(undefined),
+      dispose: vi.fn(),
+      state: { error: "429 Too Many Requests" },
+    };
+    mockedCreateHaiAgent.mockResolvedValue({ session: mockSession } as any);
+
+    await expect(
+      aiMergeTask(store, "/tmp/root", "KB-050", { usageLimitPauser: pauser }),
+    ).rejects.toThrow("AI merge failed");
+
+    // UsageLimitPauser should be called with "merger" agent type
+    expect(onUsageLimitHitSpy).toHaveBeenCalledWith(
+      "merger",
+      "KB-050",
+      "429 Too Many Requests",
+    );
+    // git reset --merge should be called to abort the merge
+    const resetCalls = mockedExecSync.mock.calls.filter(
+      (c) => String(c[0]).includes("reset --merge"),
+    );
+    expect(resetCalls.length).toBeGreaterThan(0);
+  });
+
  it("does NOT trigger global pause for non-usage-limit errors", async () => {
    const store = createMockStore(
      { id: "KB-050", worktree: "/tmp/root/.worktrees/KB-050" },
--- a/packages/engine/src/merger.ts
+++ b/packages/engine/src/merger.ts
@@ -5,7 +5,7 @@ import { createKbAgent } from "./pi.js";
 import type { WorktreePool } from "./worktree-pool.js";
 import { AgentLogger } from "./agent-logger.js";
 import { mergerLog } from "./logger.js";
-import { isUsageLimitError, type UsageLimitPauser } from "./usage-limit-detector.js";
+import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./usage-limit-detector.js";

 /**
 * Build the merge system prompt. When `includeTaskId` is true (default),
@@ -259,6 +259,9 @@ export async function aiMergeTask(
    const prompt = buildMergePrompt(taskId, branch, commitLog, diffStat, hasConflicts);
    await session.prompt(prompt);

+    // Re-raise errors that pi-coding-agent swallowed after exhausting retries.
+    checkSessionError(session);
+
    // 6. Verify the commit happened — if there are still staged changes, agent didn't commit
    const staged = execSync("git diff --cached --quiet 2>&1; echo $?", {
      cwd: rootDir,
--- a/packages/engine/src/reviewer.test.ts
+++ b/packages/engine/src/reviewer.test.ts
@@ -79,3 +79,54 @@ describe("reviewStep — model settings threading", () => {
    expect(result.verdict).toBe("APPROVE");
  });
 });
+
+describe("reviewStep — exhausted-retry error detection", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("throws when session.prompt() resolves with exhausted-retry error on state.error", async () => {
+    // session.prompt() resolves normally, but session.state.error is set
+    const mockSession = {
+      prompt: vi.fn().mockResolvedValue(undefined),
+      subscribe: vi.fn(),
+      dispose: vi.fn(),
+      state: { error: "rate_limit_error: Rate limit exceeded" },
+    };
+    mockedCreateHaiAgent.mockResolvedValue({ session: mockSession } as any);
+
+    await expect(
+      reviewStep("/tmp/worktree", "KB-100", 1, "Test Step", "code", "# prompt"),
+    ).rejects.toThrow("rate_limit_error: Rate limit exceeded");
+  });
+
+  it("disposes session in finally block despite the error", async () => {
+    const disposeFn = vi.fn();
+    const mockSession = {
+      prompt: vi.fn().mockResolvedValue(undefined),
+      subscribe: vi.fn(),
+      dispose: disposeFn,
+      state: { error: "rate_limit_error: Rate limit exceeded" },
+    };
+    mockedCreateHaiAgent.mockResolvedValue({ session: mockSession } as any);
+
+    await expect(
+      reviewStep("/tmp/worktree", "KB-100", 1, "Test Step", "code", "# prompt"),
+    ).rejects.toThrow();
+
+    // Session should be disposed in the finally block
+    expect(disposeFn).toHaveBeenCalled();
+  });
+
+  it("does not throw when session completes without error", async () => {
+    mockedCreateHaiAgent.mockResolvedValue(
+      createMockSession("### Verdict: APPROVE\n### Summary\nLooks good."),
+    );
+
+    const result = await reviewStep(
+      "/tmp/worktree", "KB-100", 1, "Test Step", "plan", "# prompt",
+    );
+
+    expect(result.verdict).toBe("APPROVE");
+  });
+});
--- a/packages/engine/src/reviewer.ts
+++ b/packages/engine/src/reviewer.ts
@@ -11,6 +11,7 @@
 import type { TaskStore } from "@kb/core";
 import { createKbAgent } from "./pi.js";
 import { AgentLogger } from "./agent-logger.js";
+import { checkSessionError } from "./usage-limit-detector.js";

 const REVIEWER_SYSTEM_PROMPT = `You are an independent code and plan reviewer.

@@ -180,6 +181,11 @@ export async function reviewStep(

  try {
    await session.prompt(request);
+
+    // Re-raise errors that pi-coding-agent swallowed after exhausting retries.
+    // The caller (executor's createReviewStepTool) catches errors and returns
+    // UNAVAILABLE, so the thrown error will be handled there.
+    checkSessionError(session);
  } finally {
    if (agentLogger) await agentLogger.flush();
    session.dispose();
--- a/packages/engine/src/triage.test.ts
+++ b/packages/engine/src/triage.test.ts
@@ -1406,6 +1406,50 @@ describe("TriageProcessor usage limit detection", () => {
    expect(onError).toHaveBeenCalled();
  });

+  it("triggers global pause when session.prompt() resolves with exhausted-retry error on state.error", async () => {
+    const store = createMockStore();
+    const pauser = new UsageLimitPauser(store);
+    const onUsageLimitHitSpy = vi.spyOn(pauser, "onUsageLimitHit");
+
+    // session.prompt() resolves normally, but session.state.error is set
+    const mockSession = {
+      prompt: vi.fn().mockResolvedValue(undefined),
+      dispose: vi.fn(),
+      state: { error: "overloaded_error: Overloaded" },
+    };
+    mockedCreateHaiAgent.mockResolvedValue({ session: mockSession } as any);
+
+    const onError = vi.fn();
+    const triage = new TriageProcessor(store, "/tmp/test", {
+      onSpecifyError: onError,
+      usageLimitPauser: pauser,
+    });
+
+    await triage.specifyTask({
+      id: "KB-001",
+      title: "Test",
+      description: "Test",
+      column: "triage",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    // UsageLimitPauser should be called with "triage" agent type
+    expect(onUsageLimitHitSpy).toHaveBeenCalledWith(
+      "triage",
+      "KB-001",
+      "overloaded_error: Overloaded",
+    );
+    // Task status should be cleared (not moved to todo with broken spec)
+    expect(store.updateTask).toHaveBeenCalledWith("KB-001", { status: null });
+    // onSpecifyError callback should fire
+    expect(onError).toHaveBeenCalled();
+  });
+
  it("does NOT trigger global pause for non-usage-limit errors", async () => {
    const store = createMockStore();
    const pauser = new UsageLimitPauser(store);
--- a/packages/engine/src/triage.ts
+++ b/packages/engine/src/triage.ts
@@ -6,7 +6,7 @@ import { createKbAgent } from "./pi.js";
 import { PRIORITY_SPECIFY, type AgentSemaphore } from "./concurrency.js";
 import { AgentLogger } from "./agent-logger.js";
 import { triageLog } from "./logger.js";
-import { isUsageLimitError, type UsageLimitPauser } from "./usage-limit-detector.js";
+import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./usage-limit-detector.js";

 const TRIAGE_SYSTEM_PROMPT = `You are a task specification agent for "kb", an AI-orchestrated task board.

@@ -366,6 +366,9 @@ export class TriageProcessor {
          const agentPrompt = buildSpecificationPrompt(detail, promptPath, settings, attachmentContents);
          await session.prompt(agentPrompt, imageContents.length > 0 ? { images: imageContents } : undefined);

+          // Re-raise errors that pi-coding-agent swallowed after exhausting retries.
+          checkSessionError(session);
+
          // Check if the agent flagged a duplicate
          const { readFile } = await import("node:fs/promises");
          const { join } = await import("node:path");
--- a/packages/engine/src/usage-limit-detector.test.ts
+++ b/packages/engine/src/usage-limit-detector.test.ts
@@ -1,5 +1,5 @@
 import { describe, it, expect, vi, beforeEach } from "vitest";
-import { isUsageLimitError, UsageLimitPauser } from "./usage-limit-detector.js";
+import { isUsageLimitError, UsageLimitPauser, checkSessionError } from "./usage-limit-detector.js";

 // ── isUsageLimitError classification tests ───────────────────────────

@@ -73,6 +73,59 @@ describe("isUsageLimitError", () => {
  });
 });

+// ── checkSessionError tests ──────────────────────────────────────────
+
+describe("checkSessionError", () => {
+  it("throws when session.state.error is set", () => {
+    const session = { state: { error: "rate_limit_error: Rate limit exceeded" } };
+    expect(() => checkSessionError(session)).toThrow("rate_limit_error: Rate limit exceeded");
+  });
+
+  it("does not throw when session.state.error is undefined", () => {
+    const session = { state: { error: undefined } };
+    expect(() => checkSessionError(session)).not.toThrow();
+  });
+
+  it("does not throw when session.state.error is empty string", () => {
+    const session = { state: { error: "" } };
+    expect(() => checkSessionError(session)).not.toThrow();
+  });
+
+  it("thrown error message matches session.state.error exactly", () => {
+    const errorMessage = "overloaded_error: Overloaded";
+    const session = { state: { error: errorMessage } };
+
+    let thrownMessage: string | undefined;
+    try {
+      checkSessionError(session);
+    } catch (err: any) {
+      thrownMessage = err.message;
+    }
+
+    expect(thrownMessage).toBe(errorMessage);
+    // Verify isUsageLimitError can classify it
+    expect(isUsageLimitError(thrownMessage!)).toBe(true);
+  });
+
+  it("thrown error message for rate limit is classifiable by isUsageLimitError", () => {
+    const session = { state: { error: "429 Too Many Requests" } };
+
+    let thrownMessage: string | undefined;
+    try {
+      checkSessionError(session);
+    } catch (err: any) {
+      thrownMessage = err.message;
+    }
+
+    expect(isUsageLimitError(thrownMessage!)).toBe(true);
+  });
+
+  it("does not throw when state has no error property", () => {
+    const session = { state: {} };
+    expect(() => checkSessionError(session as any)).not.toThrow();
+  });
+});
+
 // ── UsageLimitPauser tests ───────────────────────────────────────────

 function createMockStore(globalPause = false) {
--- a/packages/engine/src/usage-limit-detector.ts
+++ b/packages/engine/src/usage-limit-detector.ts
@@ -50,6 +50,25 @@ export function isUsageLimitError(errorMessage: string): boolean {
 * agents hitting limits only trigger one pause. The flag resets when `globalPause`
 * is externally set back to `false` (detected by reading settings before pausing).
 */
+/**
+ * Check if an agent session resolved with an error after exhausting retries.
+ *
+ * pi-coding-agent's `session.prompt()` does **not** throw when retries are
+ * exhausted — it resolves normally and stores the error on `session.state.error`.
+ * Call this immediately after every `await session.prompt(...)` to re-raise
+ * the swallowed error so existing `catch` blocks (with `isUsageLimitError`
+ * checks) can detect rate-limit conditions and trigger `UsageLimitPauser`.
+ *
+ * @param session — The agent session (or any object with `state.error?: string`)
+ * @throws {Error} If `session.state.error` is set and non-empty
+ */
+export function checkSessionError(session: { state: { error?: string } }): void {
+  const error = session.state?.error;
+  if (error) {
+    throw new Error(error);
+  }
+}
+
 export class UsageLimitPauser {
  private paused = false;