fix(memory): reflection must not reject after a store closes

src/agent/agent-loop-reflection-fire-safety.test.ts

@@ -0,0 +1,302 @@
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { AgentLoop } from "./agent-loop.js";
+import type { MemoryContextProvider } from "./agent-loop.js";
+import { buildDefaultToolRegistry } from "../tools/index.js";
+import { SlotManager } from "../llm/slot-manager.js";
+import { createEmptySessionState } from "../session/session-state.js";
+import type { CompletionResult } from "../llm/llama-server-client.js";
+import type {
+  CapabilitiesSummary,
+  SkillCatalogEntry,
+  ToolDescriptor,
+} from "../prompt/stable-prefix.js";
+import type { ProfileFact } from "../memory/profile-store.js";
+import type { StructuredLogger } from "../tracing/structured-logger.js";
+import type {
+  ReflectionInput,
+  ReflectionRunner,
+} from "../memory/reflection/reflection-runner.js";
+
+/**
+ * `runTurn` fires reflection as a bare `void` — it is background
+ * bookkeeping the user is not waiting on. Two ways that used to hurt
+ * the turn:
+ *
+ *   - a decorator that reads a store closed by shutdown rejects, and
+ *     with nothing attached to the promise the process reports an
+ *     unhandled rejection (`error-reporting/error-reporter.ts`
+ *     forwards those to the crash reporter);
+ *   - `profileFactsProvider` is a raw `profileStore.list()` evaluated
+ *     synchronously to build the reflection allowlist, so a store
+ *     failure there failed the *turn*.
+ *
+ * Neither is recoverable by the loop and neither should be visible to
+ * the user, so both are pinned here.
+ */
+
+function makeCompletion(content: string): CompletionResult {
+  return {
+    content,
+    reasoningContent: "",
+    stop: true,
+    truncated: false,
+    timing: {
+      promptMs: 1,
+      predictedMs: 1,
+      promptTokens: 10,
+      predictedTokens: 5,
+    },
+    cacheHitTokens: 0,
+    slotId: 0,
+    modelId: "mock",
+  };
+}
+
+const TOOLS: ToolDescriptor[] = [
+  {
+    name: "finish",
+    summary: "Finish the session with a summary.",
+    argsSchema: '{"summary": string}',
+  },
+];
+
+const CAPS: CapabilitiesSummary = {
+  platform: "darwin",
+  arch: "arm64",
+  browserChannel: "chrome",
+  workingDir: "/work",
+  hasClipboard: true,
+  hasWmctrl: false,
+  hasNotifications: true,
+};
+
+const SKILLS: SkillCatalogEntry[] = [];
+
+const NOOP_PROVIDER: MemoryContextProvider = {
+  buildMemoryContext: () => ({ recalled: [], index: [] }),
+};
+
+function makeFact(id: number): ProfileFact {
+  return {
+    id,
+    key: "editor",
+    value: "vim",
+    validFrom: 1,
+    updatedAt: 1,
+    pinned: true,
+    keywords: [],
+    supersedes: null,
+    supersededBy: null,
+    voteScore: 0,
+  };
+}
+
+function makeLoop(deps: {
+  reflectionRunner: ReflectionRunner;
+  profileFactsProvider?: () => readonly ProfileFact[];
+  logger?: StructuredLogger;
+}): AgentLoop {
+  return new AgentLoop({
+    registry: buildDefaultToolRegistry(),
+    slotManager: new SlotManager(2),
+    grammar: 'root ::= "ok"',
+    llmComplete: async () =>
+      makeCompletion(JSON.stringify({ tool: "reply", args: { text: "ok" } })),
+    toolDescriptors: TOOLS,
+    capabilities: CAPS,
+    skillCatalog: SKILLS,
+    memoryContextProvider: NOOP_PROVIDER,
+    reflectionRunner: deps.reflectionRunner,
+    ...(deps.profileFactsProvider
+      ? { profileFactsProvider: deps.profileFactsProvider }
+      : {}),
+    ...(deps.logger ? { logger: deps.logger } : {}),
+  });
+}
+
+interface Warning {
+  message: string;
+  fields?: Record<string, unknown>;
+}
+
+function capturingLogger(into: Warning[]): StructuredLogger {
+  return {
+    debug() {
+      /* unused */
+    },
+    info() {
+      /* unused */
+    },
+    warn(message: string, fields?: Record<string, unknown>) {
+      into.push({ message, ...(fields ? { fields } : {}) });
+    },
+    error() {
+      /* unused */
+    },
+  } as unknown as StructuredLogger;
+}
+
+/** Collect unhandled rejections raised while `body` runs. */
+async function withUnhandledRejectionWatch(
+  body: () => Promise<void>,
+): Promise<unknown[]> {
+  const seen: unknown[] = [];
+  const onRejection = (reason: unknown) => seen.push(reason);
+  // Vitest installs its own handler; prepend so ours observes first
+  // and keep the runner's in place.
+  process.prependListener("unhandledRejection", onRejection);
+  try {
+    await body();
+    // An unhandled rejection is reported after the microtask queue
+    // drains — give the loop's `void` promise two macrotask ticks.
+    await new Promise((r) => setTimeout(r, 0));
+    await new Promise((r) => setTimeout(r, 0));
+  } finally {
+    process.removeListener("unhandledRejection", onRejection);
+  }
+  return seen;
+}
+
+describe("AgentLoop reflection is background work, never a turn hazard", () => {
+  let workingDir: string;
+
+  beforeEach(() => {
+    workingDir = mkdtempSync(join(tmpdir(), "atomic-reflect-loop-"));
+  });
+
+  afterEach(() => {
+    rmSync(workingDir, { recursive: true, force: true });
+  });
+
+  it("a rejecting reflectionRunner raises no unhandled rejection", async () => {
+    let called = false;
+    const loop = makeLoop({
+      reflectionRunner: {
+        async reflect(_input: ReflectionInput) {
+          called = true;
+          throw new TypeError("The database connection is not open");
+        },
+        abortPending() {
+          /* no-op */
+        },
+      },
+    });
+    const session = createEmptySessionState({ id: "s1", workingDir });
+
+    const seen = await withUnhandledRejectionWatch(async () => {
+      const result = await loop.runTurn(session, {
+        userMessage: "hello",
+        maxSteps: 2,
+        signal: new AbortController().signal,
+      });
+      expect(result.session.id).toBe("s1");
+      expect(result.reason).toBe("reply");
+    });
+
+    expect(called).toBe(true);
+    expect(seen).toEqual([]);
+  });
+
+  it("a throwing profileFactsProvider does not fail the turn", async () => {
+    const inputs: ReflectionInput[] = [];
+    const loop = makeLoop({
+      reflectionRunner: {
+        async reflect(input: ReflectionInput) {
+          inputs.push(input);
+        },
+        abortPending() {
+          /* no-op */
+        },
+      },
+      profileFactsProvider: () => {
+        throw new TypeError("The database connection is not open");
+      },
+    });
+    const session = createEmptySessionState({ id: "s2", workingDir });
+
+    const result = await loop.runTurn(session, {
+      userMessage: "hello",
+      maxSteps: 2,
+      signal: new AbortController().signal,
+    });
+
+    // The load-bearing assertion: on main this returns
+    // `reason: "failed"` / `status: "failed"` — the session id alone
+    // is the same either way, so it proves nothing on its own.
+    expect(result.reason).toBe("reply");
+    expect(result.session.status).not.toBe("failed");
+    expect(result.session.id).toBe("s2");
+    // Reflection still fires — just without profile candidates.
+    expect(inputs).toHaveLength(1);
+    expect(inputs[0]!.recalledProfileFactIds).toBeUndefined();
+  });
+
+  it("both profile-facts guards report rather than swallow", async () => {
+    const warnings: Warning[] = [];
+    const inputs: ReflectionInput[] = [];
+    const loop = makeLoop({
+      reflectionRunner: {
+        async reflect(input: ReflectionInput) {
+          inputs.push(input);
+        },
+        abortPending() {
+          /* no-op */
+        },
+      },
+      profileFactsProvider: () => {
+        throw new TypeError("The database connection is not open");
+      },
+      logger: capturingLogger(warnings),
+    });
+    const session = createEmptySessionState({ id: "s4", workingDir });
+
+    const result = await loop.runTurn(session, {
+      userMessage: "hello",
+      maxSteps: 2,
+      signal: new AbortController().signal,
+    });
+
+    expect(result.reason).toBe("reply");
+    expect(inputs).toHaveLength(1);
+    // The step guard fires per step; the reflection guard fires once
+    // at the end of the turn. Neither may be silent.
+    const messages = warnings.map((w) => w.message);
+    expect(messages).toContain("profile facts unavailable for this step");
+    expect(messages).toContain("profile facts unavailable for reflection");
+    for (const w of warnings) {
+      expect(w.fields).toMatchObject({ sessionId: "s4" });
+      expect(String(w.fields?.error)).toContain(
+        "database connection is not open",
+      );
+    }
+  });
+
+  it("a healthy profileFactsProvider still supplies the allowlist", async () => {
+    const inputs: ReflectionInput[] = [];
+    const loop = makeLoop({
+      reflectionRunner: {
+        async reflect(input: ReflectionInput) {
+          inputs.push(input);
+        },
+        abortPending() {
+          /* no-op */
+        },
+      },
+      profileFactsProvider: () => [makeFact(7)],
+    });
+    const session = createEmptySessionState({ id: "s3", workingDir });
+
+    await loop.runTurn(session, {
+      userMessage: "hello",
+      maxSteps: 2,
+      signal: new AbortController().signal,
+    });
+
+    expect(inputs).toHaveLength(1);
+    expect(inputs[0]!.recalledProfileFactIds).toEqual([7]);
+  });
+});

src/agent/agent-loop.ts

@@ -505,8 +505,11 @@ export class AgentLoop {
     // strictly larger memory set.
     //
     // Shutdown path still calls `abortPending()` with no sessionId
-    // to drain every in-flight reflection before the runtime tears
-    // down SQLite handles.
+    // before the runtime tears down SQLite handles. Note that it
+    // *signals* — nothing is awaited, so a reflection can still be
+    // resuming when the stores close. That is why the decorators and
+    // this call site guard their store reads rather than relying on
+    // the abort to have finished.
 
     if (options.userMessage !== undefined) {
       const text = options.userMessage;
@@ -651,7 +654,23 @@ export class AgentLoop {
         "This is the final allowed step. Do not call any non-terminal tool; " +
         "summarize the completed work with reply, or end the session with finish.";
       try {
-        const profileFacts = this.deps.profileFactsProvider?.();
+        // `profileFactsProvider` is a raw `profileStore.list()`.
+        // Dropping the facts is a real loss — `profile-renderer` emits
+        // pinned facts regardless of the contextual gate, so this step
+        // renders with no `### profile` section at all — but it is the
+        // lesser one: a throw here lands in the
+        // catch below, where a `TypeError` from a closed SQLite handle
+        // classifies `tool` and fails the turn outright.
+        let profileFacts: readonly ProfileFact[] | undefined;
+        try {
+          profileFacts = this.deps.profileFactsProvider?.();
+        } catch (err) {
+          this.deps.logger?.warn("profile facts unavailable for this step", {
+            sessionId: state.id,
+            stepIndex: i,
+            error: err instanceof Error ? err.message : String(err),
+          });
+        }
         const activeProfile =
           this.deps.profileManager?.getProfile() ??
           this.deps.profile ??
@@ -1144,11 +1163,31 @@ export class AgentLoop {
           // recalled across all steps of this turn) ∪ (profile
           // facts currently active). Profile facts are not gated
           // by recall — they're always candidates because the
-          // renderer already surfaces them whenever they pass the
-          // contextual-keyword gate. Sourcing them here keeps the
+          // renderer surfaces them whenever they are pinned or pass
+          // the contextual-keyword gate. Sourcing them here keeps the
           // decorator's hydration cheap.
-          const profileFacts =
-            this.deps.profileFactsProvider?.() ?? [];
+          // `profileFactsProvider` is a raw `profileStore.list()`.
+          // It is only ever an input to the fire-and-forget reflection
+          // below, so a store failure here must not fail the turn the
+          // user is waiting on — an empty allowlist just means the
+          // vote-runner sees no profile candidates this turn.
+          let profileFacts: readonly ProfileFact[] = [];
+          try {
+            profileFacts = this.deps.profileFactsProvider?.() ?? [];
+          } catch (err) {
+            // Usually the step guard above has already warned for this
+            // turn — same provider, same store. Not always: the store
+            // can close between the last step and this block.
+            this.deps.logger?.warn("profile facts unavailable for reflection", {
+              sessionId: state.id,
+              error: err instanceof Error ? err.message : String(err),
+            });
+          }
+          // `reflect()` is documented fire-safe, but it is composed at
+          // runtime from decorators that read SQLite stores. A bare
+          // `void` turns any escape into an unhandled rejection the
+          // loop can neither see nor recover from, so the trailing
+          // `.catch` pins the contract at the call site too.
           void this.deps.reflectionRunner.reflect({
             sessionId: state.id,
             userMessage,
@@ -1184,6 +1223,11 @@ export class AgentLoop {
             ...(segmentationActive && transcript.length > 0
               ? { transcript }
               : {}),
+          }).catch((err: unknown) => {
+            this.deps.logger?.warn("reflection failed after dispatch", {
+              sessionId: state.id,
+              error: err instanceof Error ? err.message : String(err),
+            });
           });
         }
       }