fix(memory): cancel a sub-call's request when its timeout fires

AGENTS.md

@@ -1163,6 +1163,7 @@ The three channels share one SQLite file `<stateDir>/memory.sqlite` (separate fr
 ### Reflection (async end-of-turn memory formation)
 
 - **When.** Fired at the end of every `AgentLoop.runTurn` after `assistant_reply` is emitted. **Fire-and-forget**, never awaited. `abortPending({ sessionId: state.id })` runs at the start of the next `runTurn` so at most one reflection is in flight **per session**; reflections on other sessions are never aborted as a side effect (load-bearing for cross-session parallelism — see §"Concurrency contract").
+- **A timeout cancels the request, not just the wait.** Every memory sub-call wrapper in bootstrap (reflection, link generator, vote, query rewriter, distill) goes through `abortableSubcall` ([src/runtime/abortable-subcall.ts](src/runtime/abortable-subcall.ts)), which forwards the runner's abort signal into `llmComplete` — so a fired timeout or `abortPending` closes the HTTP request and frees the slot — and still rejects the moment the signal aborts in case a provider ignores it.
 - **What.** A micro-prompt with its own small stable prefix asks the model to extract durable facts from the last `USER`/`ASSISTANT` exchange. Output is GBNF-constrained to either `NONE` or a bounded list of two flavours:
   - `SET key=value` (pinned fact) or `SET key=value [pinned=false; keywords=a,b,c]` (contextual fact). Caps at `memory.reflection.maxFactsPerCall` (default `3`).
   - `NOTE freeform observation [tag1, tag2]` → into `MemoryStore` with implicit `reflection` tag. Master switch `memory.reflection.autoStoreNotes` (default `true`); cap at `memory.reflection.maxNotesPerCall` (default `2`, set to `0` to disable).
@@ -2314,7 +2315,7 @@ Every terminal failure the agent loop surfaces is normalised into a canonical `L
 
 ## Provider fallback chain
 
-A cross-provider circuit breaker layered **above** the single-provider reliability policy. Where the two retry layers above recover a request on the *same* provider, the fallback chain switches to a *different* configured provider when the active one is unavailable. It lives in [src/llm/fallback/](src/llm/fallback/). The `llmComplete` / `llmCompleteStream` seams that wrap it are built by [src/runtime/llm-fallback-seam.ts](src/runtime/llm-fallback-seam.ts) (`createFallbackCompleter` / `createFallbackStreamer`, injected with `{ fallbackChain, resolveSlice, recordUnaryUsage, recordStreamUsage }`) and wired in [src/runtime/bootstrap.ts](src/runtime/bootstrap.ts) — strictly **after** the per-provider retry budget (PR #90 `runOpenAiWithRetry`, `LlamaServerClient.completionRetries`) is spent, never inside it.
+A cross-provider circuit breaker layered **above** the single-provider reliability policy. Where the two retry layers above recover a request on the *same* provider, the fallback chain switches to a *different* configured provider when the active one is unavailable. It lives in [src/llm/fallback/](src/llm/fallback/). The `llmComplete` / `llmCompleteStream` seams that wrap it are built by [src/runtime/llm-fallback-seam.ts](src/runtime/llm-fallback-seam.ts) (`createFallbackCompleter` / `createFallbackStreamer`, injected with `{ fallbackChain, resolveSlice, recordUnaryUsage, recordStreamUsage }`) and wired in [src/runtime/bootstrap.ts](src/runtime/bootstrap.ts) — strictly **after** the per-provider retry budget (PR #90 `runOpenAiWithRetry`, `LlamaServerClient.completionRetries`) is spent, never inside it. The unary seam rethrows a request whose caller aborted as the signal's reason, so it classifies `cancelled` and never advances the chain — `LlamaServerClient` and `runOpenAiWithRetry` otherwise surface an abort as a `status: null` error that files as `transport`, which is how an abandoned memory sub-call could trip a breaker and flip the override.
 
 ### Chain unit and config
 

src/llm/provider/openai/openai-http.ts

@@ -224,11 +224,56 @@ export async function openAiPostJson(
       if (!res.ok) {
         throw await httpErrorFromResponse(deps, path, res);
       }
-      return (await res.json()) as Record<string, unknown>;
+      return readJsonBody(res, request.signal);
     }),
   );
 }
 
+/**
+ * Read a unary JSON body without going deaf to the caller's abort.
+ *
+ * `openAiFetch` unlinks the caller's signal the moment `fetch` resolves —
+ * it must, because it also opens streams, whose consumer owns the signal
+ * from then on. For a unary request that left the body read unabortable:
+ * a provider that sends headers first and the completion later kept the
+ * socket, the slot and the bill running after the caller had given up
+ * (a memory sub-call's timeout, a cancelled turn). Cancelling the reader
+ * tears the connection down; the caller gets `signal.reason`, which
+ * classifies `cancelled`.
+ *
+ * `res.json()` cannot be used for this: it locks the body, and cancelling
+ * a locked stream from outside is refused. Without a signal nothing can
+ * cancel the read, so that path keeps `res.json()` exactly as before.
+ */
+async function readJsonBody(
+  res: Response,
+  signal: AbortSignal | undefined,
+): Promise<Record<string, unknown>> {
+  if (!signal || !res.body) return (await res.json()) as Record<string, unknown>;
+  const reader = res.body.getReader();
+  const onAbort = (): void => {
+    reader.cancel(signal.reason).catch(() => undefined);
+  };
+  if (signal.aborted) onAbort();
+  else signal.addEventListener("abort", onAbort, { once: true });
+  try {
+    const chunks: Uint8Array[] = [];
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      chunks.push(value);
+    }
+    if (signal.aborted) {
+      throw signal.reason ?? new DOMException("aborted", "AbortError");
+    }
+    return JSON.parse(
+      new TextDecoder().decode(Buffer.concat(chunks)),
+    ) as Record<string, unknown>;
+  } finally {
+    signal.removeEventListener("abort", onAbort);
+  }
+}
+
 /**
  * Recover the one HTTP failure that is fixable by changing the request
  * rather than by waiting or switching providers: a 402 that names how

src/runtime/abortable-subcall.network.test.ts

@@ -0,0 +1,215 @@
+import { createServer, type Server } from "node:http";
+import type { AddressInfo } from "node:net";
+import { afterEach, describe, expect, it } from "vitest";
+
+import type { LlmStreamParams } from "../agent/step-executor.js";
+import { ProviderFallbackChain } from "../llm/fallback/index.js";
+import { DEFAULT_FALLBACK_TIMING } from "../llm/fallback/fallback-config.js";
+import { LlamaServerClient } from "../llm/llama-server-client.js";
+import { QWEN_THINK_PROFILE } from "../llm/model-profile.js";
+import {
+  fakeAnswer,
+  fakeProvider,
+} from "../llm/provider/fake-provider.fixture.js";
+import { LlamaServerProvider } from "../llm/provider/llama-server/llama-server-provider.js";
+import type { LlmProvider } from "../llm/provider/llm-provider.js";
+import { OpenAiProvider } from "../llm/provider/openai/openai-provider.js";
+import {
+  createLinkGeneratorRunner,
+  type LinkGeneratorInput,
+  type LinkGeneratorLlmComplete,
+  type LinkGeneratorRunnerDeps,
+} from "../memory/links/link-generator-runner.js";
+import { abortableSubcall } from "./abortable-subcall.js";
+import { createFallbackCompleter } from "./llm-fallback-seam.js";
+
+/**
+ * End to end over a real socket: a memory sub-call runner whose timeout
+ * fires must close the HTTP request it started — not merely stop waiting
+ * for it — and the abandoned request must not advance the fallback chain.
+ *
+ * Wiring mirrors bootstrap: link-generator runner → `abortableSubcall`
+ * (bootstrap's link-gen request shape) → `createFallbackCompleter` → a
+ * real provider pointed at a local server that never finishes answering.
+ */
+
+type Mode = "silent" | "headers-then-stall";
+type Kind = "openai" | "llama-server";
+
+const servers: Server[] = [];
+
+afterEach(async () => {
+  await Promise.all(
+    servers.splice(0).map(
+      (server) =>
+        new Promise<void>((resolve) => {
+          server.closeAllConnections();
+          server.close(() => resolve());
+        }),
+    ),
+  );
+});
+
+/** A server that accepts the request and never completes the response. */
+async function startStallingServer(mode: Mode) {
+  let requests = 0;
+  let closedSockets = 0;
+  const server = createServer((req, res) => {
+    requests += 1;
+    req.socket.once("close", () => {
+      closedSockets += 1;
+    });
+    req.resume();
+    if (mode === "headers-then-stall") {
+      // Headers plus a byte of body, then nothing: `fetch` has resolved,
+      // so only a cancelled body read can let go of this socket.
+      res.writeHead(200, { "content-type": "application/json" });
+      res.write(" ");
+    }
+  });
+  servers.push(server);
+  await new Promise<void>((resolve) =>
+    server.listen(0, "127.0.0.1", () => resolve()),
+  );
+  const { port } = server.address() as AddressInfo;
+  return {
+    url: `http://127.0.0.1:${port}`,
+    requests: () => requests,
+    closedSockets: () => closedSockets,
+  };
+}
+
+function providerFor(kind: Kind, url: string): LlmProvider {
+  if (kind === "openai") {
+    return new OpenAiProvider({
+      id: "primary",
+      baseUrl: url,
+      apiKey: "test-key",
+      defaultChatModel: "test-model",
+    });
+  }
+  return new LlamaServerProvider(
+    new LlamaServerClient({ baseUrl: url, completionRetries: 1 }),
+    {
+      id: "primary",
+      getProfile: () => QWEN_THINK_PROFILE,
+      visionEnabledByConfig: false,
+      visionAutoDetect: false,
+      maxImageBytes: 1,
+      maxImagesPerCall: 1,
+      baseUrlOverride: url,
+    },
+  );
+}
+
+/** Polls `predicate` until it holds or `ms` elapses. */
+async function eventually(predicate: () => boolean, ms: number) {
+  const deadline = Date.now() + ms;
+  while (!predicate()) {
+    if (Date.now() > deadline) return false;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  return true;
+}
+
+/** Settles with "hung" when `promise` has not settled within `ms`. */
+function within(promise: Promise<unknown>, ms: number) {
+  return Promise.race([
+    promise.then(() => "settled" as const),
+    new Promise<"hung">((resolve) => setTimeout(() => resolve("hung"), ms)),
+  ]);
+}
+
+const INPUT: LinkGeneratorInput = {
+  sessionId: "s1",
+  userMessage: "where did we put the deploy notes?",
+  assistantReply: "In the ops wiki, next to the runbook.",
+  candidates: [
+    { id: "1", body: "deploy notes live in the ops wiki" },
+    { id: "2", body: "the runbook covers rollbacks" },
+  ] as unknown as LinkGeneratorInput["candidates"],
+};
+
+const CASES: ReadonlyArray<{ kind: Kind; mode: Mode }> = [
+  { kind: "openai", mode: "silent" },
+  { kind: "openai", mode: "headers-then-stall" },
+  { kind: "llama-server", mode: "silent" },
+  { kind: "llama-server", mode: "headers-then-stall" },
+];
+
+describe("a timed-out memory sub-call", () => {
+  for (const { kind, mode } of CASES) {
+    it(`closes its ${kind} request (${mode}) and leaves the fallback chain alone`, async () => {
+      const server = await startStallingServer(mode);
+      const chain = new ProviderFallbackChain({
+        resolve: () => ({
+          chain: ["primary", "backup"],
+          timing: DEFAULT_FALLBACK_TIMING,
+        }),
+      });
+      let backupCalls = 0;
+      const providers = new Map<string, LlmProvider>([
+        ["primary", providerFor(kind, server.url)],
+        [
+          "backup",
+          fakeProvider("backup", "grammar", async () => {
+            backupCalls += 1;
+            return fakeAnswer("backup");
+          }),
+        ],
+      ]);
+      const complete = createFallbackCompleter({
+        fallbackChain: chain,
+        resolveSlice: (providerId) => {
+          const provider = providers.get(providerId)!;
+          return { provider, transport: provider.capabilities.toolTransport };
+        },
+        recordUnaryUsage: () => {},
+        recordStreamUsage: () => {},
+      });
+      // The whole chain run — including any fallover an orphaned failure
+      // would trigger — is this promise; the runner stops watching it.
+      let chainRun: Promise<unknown> | undefined;
+      const tracked = (params: LlmStreamParams) => {
+        const run = complete(params);
+        chainRun = run.catch(() => undefined);
+        return run;
+      };
+      const llmComplete: LinkGeneratorLlmComplete = abortableSubcall(
+        tracked,
+        (params: Parameters<LinkGeneratorLlmComplete>[0]) => ({
+          prompt: params.prompt,
+          grammar: params.grammar,
+          slotId: params.slotId,
+          sessionId: params.sessionId,
+          ...(params.responseFormat
+            ? { responseFormat: params.responseFormat }
+            : {}),
+        }),
+      );
+      const outcomes: string[] = [];
+      const runner = createLinkGeneratorRunner({
+        llmComplete,
+        linkStore: {} as LinkGeneratorRunnerDeps["linkStore"],
+        reflectionSlotId: -1,
+        timeoutMs: 150,
+        emitTrace: (event) => outcomes.push(event.outcome),
+      });
+
+      await expect(runner.generate(INPUT)).resolves.toBe(0);
+
+      expect(outcomes).toEqual(["timeout"]);
+      expect(server.requests()).toBe(1);
+      // The load-bearing assertion: the server sees the socket go away.
+      // Without the signal reaching the request it stays open until the
+      // provider's own request timeout, minutes later.
+      expect(await eventually(() => server.closedSockets() > 0, 3_000)).toBe(
+        true,
+      );
+      expect(chainRun).toBeDefined();
+      expect(await within(chainRun!, 3_000)).toBe("settled");
+      expect(backupCalls).toBe(0);
+      expect(chain.activeOverrideFor("link-gen:s1")).toBeNull();
+    });
+  }
+});