fix(agents): honor abort signal during compaction summary retry backoff (#110114)

* fix(agents): honor abort signal during compaction summary retry backoff

* test(agents): cover compaction abort over real HTTP

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/agents/compaction.abort-http.test.ts

@@ -0,0 +1,108 @@
+import { createServer } from "node:http";
+import { performance } from "node:perf_hooks";
+import { describe, expect, it } from "vitest";
+import { summarizeInStages } from "./compaction.js";
+import type { AgentMessage } from "./runtime/index.js";
+import type { ExtensionContext } from "./sessions/index.js";
+
+describe("compaction retry backoff over real HTTP", () => {
+  it("aborts after one provider response without changing the source history", async () => {
+    const controller = new AbortController();
+    const requests: Array<{ method: string; path: string }> = [];
+    let responseCompletedAt: number | undefined;
+    let abortedAt: number | undefined;
+    let abortTimer: ReturnType<typeof setTimeout> | undefined;
+
+    const server = createServer((request, response) => {
+      request.resume();
+      requests.push({
+        method: request.method ?? "",
+        path: request.url ?? "",
+      });
+      response.writeHead(400, { "content-type": "application/json" });
+      response.end(
+        JSON.stringify({
+          error: {
+            message: "Transient loopback provider rejection",
+            type: "invalid_request_error",
+            code: "loopback_compaction_retry",
+          },
+        }),
+        () => {
+          // Start cancellation only after the real provider response can put
+          // production compaction into its minimum 500 ms retry backoff.
+          if (requests.length === 1) {
+            responseCompletedAt = performance.now();
+            abortTimer = setTimeout(() => {
+              abortedAt = performance.now();
+              controller.abort();
+            }, 200);
+          }
+        },
+      );
+    });
+
+    await new Promise<void>((resolve, reject) => {
+      server.once("error", reject);
+      server.listen(0, "127.0.0.1", () => {
+        server.removeListener("error", reject);
+        resolve();
+      });
+    });
+
+    try {
+      const address = server.address();
+      if (!address || typeof address === "string") {
+        throw new Error("Compaction loopback server did not expose a TCP port");
+      }
+      const model = {
+        id: "loopback-compaction-model",
+        name: "Loopback compaction model",
+        api: "openai-completions",
+        provider: "openai",
+        baseUrl: `http://127.0.0.1:${address.port}/v1`,
+        reasoning: false,
+        input: ["text"],
+        cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+        contextWindow: 200_000,
+        maxTokens: 8_192,
+      } satisfies NonNullable<ExtensionContext["model"]>;
+
+      const messages: AgentMessage[] = [
+        {
+          role: "user",
+          content: "Keep this original conversation and opaque ID 5cf86ba9 unchanged.",
+          timestamp: 1,
+        },
+      ];
+      const originalHistory = Buffer.from(JSON.stringify(messages));
+
+      const summary = summarizeInStages({
+        messages,
+        model,
+        apiKey: "loopback-test-key", // pragma: allowlist secret
+        signal: controller.signal,
+        reserveTokens: 1_000,
+        maxChunkTokens: 50_000,
+        contextWindow: model.contextWindow,
+        parts: 1,
+      });
+
+      await expect(summary).rejects.toThrow(/abort/i);
+      if (responseCompletedAt === undefined || abortedAt === undefined) {
+        throw new Error("Compaction did not abort after the real loopback response");
+      }
+      expect(performance.now() - abortedAt).toBeLessThan(400);
+      expect(performance.now() - responseCompletedAt).toBeLessThan(400);
+      expect(requests).toEqual([{ method: "POST", path: "/v1/chat/completions" }]);
+      expect(Buffer.from(JSON.stringify(messages)).equals(originalHistory)).toBe(true);
+    } finally {
+      if (abortTimer !== undefined) {
+        clearTimeout(abortTimer);
+      }
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    }
+  });
+});

src/agents/compaction.summarize-fallback.test.ts

@@ -138,6 +138,39 @@ describe("summarizeWithFallback", () => {
     expect(agentSessionMocks.generateSummary).toHaveBeenCalledTimes(1);
   });
 
+  it("stops retry backoff promptly when the caller aborts mid-sleep", async () => {
+    // The first attempt fails with a retryable error, then the caller aborts
+    // while retryAsync sits in its backoff sleep (>= 500ms minDelay). The
+    // sleep must reject on abort instead of riding out the full delay.
+    const controller = new AbortController();
+    agentSessionMocks.generateSummary.mockRejectedValueOnce(new Error("transient rate limit"));
+
+    const startedAt = Date.now();
+    const promise = summarizeWithFallback({
+      messages: [
+        {
+          role: "user",
+          content: "hello",
+          timestamp: 1,
+        } satisfies UserMessage,
+      ],
+      model: testModel,
+      apiKey: "test-key", // pragma: allowlist secret
+      signal: controller.signal,
+      reserveTokens: 1000,
+      maxChunkTokens: 50_000,
+      contextWindow: 200_000,
+    });
+    const rejection = expect(promise).rejects.toThrow("aborted");
+    setTimeout(() => controller.abort(), 50);
+    await rejection;
+    const elapsedMs = Date.now() - startedAt;
+
+    // Well under the 500ms minimum backoff — the abort interrupted the sleep.
+    expect(elapsedMs).toBeLessThan(400);
+    expect(agentSessionMocks.generateSummary).toHaveBeenCalledTimes(1);
+  });
+
   it("still attempts partial summarization when oversized messages were excluded", async () => {
     // Oversized-message fallback tries the safe subset so a huge attachment or
     // tool output does not prevent summarizing the rest of the transcript.

src/agents/compaction.ts

@@ -3,6 +3,7 @@
  */
 import type { AgentCompactionIdentifierPolicy } from "../config/types.agent-defaults.js";
 import { isAbortError } from "../infra/abort-signal.js";
+import { sleepWithAbort } from "../infra/backoff.js";
 import { formatErrorMessage } from "../infra/errors.js";
 import { retryAsync } from "../infra/retry.js";
 import { createSubsystemLogger } from "../logging/subsystem.js";
@@ -156,6 +157,9 @@ async function summarizeChunks(params: {
           maxDelayMs: 5000,
           jitter: 0.2,
           label: "compaction/generateSummary",
+          // Backoff must honor caller cancellation; otherwise an abort during
+          // the sleep would stall compaction until the full delay elapses.
+          sleep: (ms) => sleepWithAbort(ms, params.signal),
           shouldRetry: (err) => {
             // Stop retrying when the caller explicitly cancelled.
             if (params.signal.aborted) {