fix: avoid infinite recursion

.changeset/tidy-runs-converge.md

@@ -0,0 +1,5 @@
+---
+"@agent-native/core": patch
+---
+
+Fix a server hang caused by run reconciliation never reaching a fixed point. A run row already holding its reconciled terminal values still matched the repair UPDATE, and an unchanged rewrite counts as an affected row, so `reconcileTerminalRunFromEvents` reported a repair on every call. `getRunByThread` re-reconciles whenever a repair is reported, so a single settled `errored`/`stale_run` row made every lookup for that thread recurse without terminating, pinning the event loop and hanging all requests.

packages/core/src/agent/run-store.reconcile-convergence.spec.ts

@@ -0,0 +1,114 @@
+import Database from "better-sqlite3";
+import { describe, expect, it, vi } from "vitest";
+
+/**
+ * A run row that already carries its reconciled terminal values is converged,
+ * and `reconcileTerminalRunFromEvents` must say so by returning false.
+ *
+ * The repair UPDATE still matches such a row (its WHERE admits
+ * errored/stale_run so a real repair can re-run), and SQL counts an unchanged
+ * rewrite as an affected row. Returning `rowsAffected > 0` therefore reported
+ * "repaired" on every call for a row that never changed, and `getRunByThread`
+ * re-reads and re-reconciles whenever reconciliation claims a repair — so a
+ * single settled errored/stale_run row made every lookup for that thread
+ * recurse without a fixed point, pinning the event loop and hanging the server
+ * for every request, not just chat.
+ */
+
+const sqlite = new Database(":memory:");
+
+const client = {
+  execute: vi.fn(async (input: string | { sql: string; args?: unknown[] }) => {
+    if (typeof input === "string") {
+      sqlite.exec(input);
+      return { rows: [] as unknown[], rowsAffected: 0 };
+    }
+    const stmt = sqlite.prepare(input.sql);
+    const args = (input.args ?? []) as unknown[];
+    if (/^\s*select/i.test(input.sql)) {
+      return { rows: stmt.all(...args), rowsAffected: 0 };
+    }
+    const info = stmt.run(...args);
+    return { rows: [] as unknown[], rowsAffected: info.changes };
+  }),
+};
+
+vi.mock("../db/client.js", () => ({
+  getDbExec: () => client,
+  intType: () => "INTEGER",
+  isPostgres: () => false,
+  retryOnDdlRace: (fn: () => any) => fn(),
+}));
+
+const {
+  insertRun,
+  insertRunEvent,
+  reconcileTerminalRunFromEvents,
+  getRunByThread,
+  STALE_RUN_ERROR_EVENT,
+} = await import("./run-store.js");
+
+async function seedStaleRun(name: string) {
+  const runId = `run-${name}`;
+  const threadId = `thread-${name}`;
+  await insertRun(runId, threadId, `turn-${name}`);
+  await insertRunEvent(runId, 0, JSON.stringify(STALE_RUN_ERROR_EVENT));
+  return { runId, threadId };
+}
+
+function markReapedStale(runId: string) {
+  sqlite
+    .prepare(
+      `UPDATE agent_runs
+       SET status = 'errored', error_code = ?, error_detail = ?,
+           terminal_reason = 'stale_run', completed_at = ?
+       WHERE id = ?`,
+    )
+    .run(
+      STALE_RUN_ERROR_EVENT.errorCode,
+      STALE_RUN_ERROR_EVENT.details,
+      Date.now(),
+      runId,
+    );
+}
+
+describe("reconcileTerminalRunFromEvents convergence", () => {
+  it("repairs a still-running row whose terminal event already landed", async () => {
+    const { runId } = await seedStaleRun("needs-repair");
+
+    await expect(reconcileTerminalRunFromEvents(runId)).resolves.toBe(true);
+
+    const row = sqlite
+      .prepare(`SELECT status, error_code FROM agent_runs WHERE id = ?`)
+      .get(runId) as { status: string; error_code: string };
+    expect(row.status).toBe("errored");
+    expect(row.error_code).toBe(STALE_RUN_ERROR_EVENT.errorCode);
+  });
+
+  it("reaches a fixed point instead of reporting a repair forever", async () => {
+    const { runId } = await seedStaleRun("converged");
+    markReapedStale(runId);
+
+    // The reaper's own terminal_reason can differ from the one derived from
+    // the event, so the first pass may legitimately repair it. What must not
+    // happen is a repair being reported on a row that no longer changes.
+    await reconcileTerminalRunFromEvents(runId);
+
+    await expect(reconcileTerminalRunFromEvents(runId)).resolves.toBe(false);
+    await expect(reconcileTerminalRunFromEvents(runId)).resolves.toBe(false);
+  });
+
+  it("terminates getRunByThread for a settled errored/stale_run row", async () => {
+    const { runId, threadId } = await seedStaleRun("lookup");
+    markReapedStale(runId);
+
+    const before = client.execute.mock.calls.length;
+    const run = await getRunByThread(threadId, { includeTerminal: true });
+
+    expect(run?.id).toBe(runId);
+    expect(run?.status).toBe("errored");
+    // Bounded query count is the actual regression signal: the recursion had
+    // no fixed point, so an unfixed build never reaches this assertion.
+    expect(client.execute.mock.calls.length - before).toBeLessThan(12);
+  });
+});

packages/core/src/agent/run-store.ts

@@ -1313,6 +1313,33 @@ export async function reconcileTerminalRunFromEvents(
   const client = getDbExec();
   const errorCode = errorCodeForTerminalEvent(latest.event);
   const errorDetail = errorDetailForTerminalEvent(latest.event);
+  // A row already holding the reconciled values is converged, and saying
+  // otherwise is not cosmetic: the UPDATE below still matches such a row and
+  // SQL counts an unchanged rewrite as affected, so `rowsAffected` reports
+  // "repaired" on every call. `getRunByThread` re-reads and re-reconciles on
+  // that answer, so a settled errored/stale_run row recurses forever and pins
+  // the event loop instead of returning. Converged must read as no work done.
+  const { rows: currentRows } = await client.execute({
+    sql: `SELECT status, error_code, terminal_reason, completed_at FROM agent_runs WHERE id = ?`,
+    args: [runId],
+  });
+  const current = currentRows[0] as
+    | {
+        status?: string | null;
+        error_code?: string | null;
+        terminal_reason?: string | null;
+        completed_at?: number | string | null;
+      }
+    | undefined;
+  if (
+    current &&
+    current.status === status &&
+    (current.error_code ?? null) === (errorCode ?? null) &&
+    (current.terminal_reason ?? null) === terminalReason &&
+    current.completed_at != null
+  ) {
+    return false;
+  }
   const { rowsAffected } = await client.execute({
     sql: `UPDATE agent_runs
           SET status = ?,