fix(daemon): queue limit, discovery coalescing, body release

apps/daemon/src/host/bun-session-repository.ts

@@ -116,7 +116,7 @@ export function usageDateKey(timestamp: number): string {
   return `${year}-${month}-${day}`;
 }
 
-const MAX_ACTIVE_PENDING_INPUTS = 5;
+const MAX_ACTIVE_PENDING_INPUTS = 10;
 
 /**
  * Coerce a stored session status into a value the renderer's SessionStatusSchema

apps/daemon/src/host/daemonConfigPresenter.ts

@@ -100,6 +100,9 @@ const defaultProviders: LLM_PROVIDER[] = DEFAULT_PROVIDERS.map((provider) => ({
 
 export class DaemonConfigPresenter {
   private store: Store;
+  /** In-flight model-discovery requests, coalesced per provider (DeepChat #2248 pattern). */
+  private inFlightModelRefreshes = new Map<string, Promise<MODEL_META[]>>();
+  private inFlightOllamaFetches = new Map<string, Promise<OllamaModel[]>>();
   private filePath: string;
   private readonly acpConfig: DaemonAcpConfig;
   private readonly mcpConfig: DaemonMcpConfig;
@@ -655,16 +658,31 @@ export class DaemonConfigPresenter {
     if (!provider.apiKey) {
       throw new Error(`Provider ${providerId} has no API key configured`);
     }
-
-    const definition = resolveAiSdkProviderDefinition(provider);
-    const modelSource = definition?.modelSource ?? "openai";
-
-    const models =
-      modelSource === "provider-db"
-        ? await this.fetchProviderModelsFromCatalog(provider)
-        : await this.fetchProviderModels(provider);
-    this.setProviderModels(providerId, models);
-    return models;
+    // Coalesce concurrent discovery for the same provider (model picker,
+    // store init, per-agent surfaces all call this) into one upstream
+    // request. The key includes the settings fingerprint so a mid-flight
+    // settings change starts a fresh discovery instead of handing back
+    // results fetched with stale credentials.
+    const key = `${providerId}:${provider.apiType}:${provider.baseUrl}:${provider.apiKey}`;
+    const inFlight = this.inFlightModelRefreshes.get(key);
+    if (inFlight) {
+      return inFlight;
+    }
+    const promise = (async () => {
+      const definition = resolveAiSdkProviderDefinition(provider);
+      const modelSource = definition?.modelSource ?? "openai";
+
+      const models =
+        modelSource === "provider-db"
+          ? await this.fetchProviderModelsFromCatalog(provider)
+          : await this.fetchProviderModels(provider);
+      this.setProviderModels(providerId, models);
+      return models;
+    })().finally(() => {
+      this.inFlightModelRefreshes.delete(key);
+    });
+    this.inFlightModelRefreshes.set(key, promise);
+    return promise;
   }
 
   private async fetchProviderModelsFromCatalog(provider: LLM_PROVIDER): Promise<MODEL_META[]> {
@@ -709,7 +727,7 @@ export class DaemonConfigPresenter {
       return [];
     }
 
-    return this.fetchOllamaModels(provider.baseUrl, provider.apiKey, "/api/tags");
+    return this.fetchOllamaModelsCoalesced(providerId, provider.baseUrl, provider.apiKey, "/api/tags");
   }
 
   async listOllamaRunningModels(providerId: string): Promise<OllamaModel[]> {
@@ -718,7 +736,26 @@ export class DaemonConfigPresenter {
       return [];
     }
 
-    return this.fetchOllamaModels(provider.baseUrl, provider.apiKey, "/api/ps");
+    return this.fetchOllamaModelsCoalesced(providerId, provider.baseUrl, provider.apiKey, "/api/ps");
+  }
+
+  /** Coalesce concurrent identical Ollama lookups into one upstream request. */
+  private fetchOllamaModelsCoalesced(
+    providerId: string,
+    baseUrl: string,
+    apiKey: string,
+    suffix: "/api/tags" | "/api/ps",
+  ): Promise<OllamaModel[]> {
+    const key = `${providerId}:${suffix}`;
+    const inFlight = this.inFlightOllamaFetches.get(key);
+    if (inFlight) {
+      return inFlight;
+    }
+    const promise = this.fetchOllamaModels(baseUrl, apiKey, suffix).finally(() => {
+      this.inFlightOllamaFetches.delete(key);
+    });
+    this.inFlightOllamaFetches.set(key, promise);
+    return promise;
   }
 
   async pullOllamaModel(providerId: string, modelName: string): Promise<boolean> {

apps/daemon/test/upstreamHygieneSweep.test.ts

@@ -0,0 +1,167 @@
+import { describe, expect, it, beforeEach, afterEach, vi } from "bun:test";
+import { Database } from "bun:sqlite";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { BunSessionRepository } from "../src/host/bun-session-repository";
+import { DaemonConfigPresenter } from "../src/host/daemonConfigPresenter";
+import { AcpLaunchSpecService } from "@argos/acp-runtime/config/acpLaunchSpecService";
+
+/**
+ * Hygiene sweep (docs/issues/upstream-hygiene-sweep):
+ * - pending-input queue limit raised to 10 (DeepChat #2236)
+ * - concurrent model discovery coalesces into one upstream request (#2248)
+ * - failed downloads release the response body before throwing (#2251)
+ */
+
+describe("pending input queue limit", () => {
+  let repo: BunSessionRepository;
+  let db: Database;
+
+  beforeEach(() => {
+    db = new Database(":memory:");
+    repo = new BunSessionRepository(db as never);
+    db.run(
+      `INSERT INTO daemon_sessions (id, agent_id, title, status, generation_status, created_at, updated_at)
+       VALUES (?, 'argos', 'Test', 'idle', 'idle', ?, ?)`,
+      ["session-1", Date.now(), Date.now()],
+    );
+  });
+
+  it("accepts a queue of 10 and rejects the 11th", async () => {
+    for (let index = 0; index < 10; index += 1) {
+      await expect(
+        repo.queuePendingInput("session-1", `message ${index}`, { source: "queue" } as never),
+      ).resolves.toBeTruthy();
+    }
+
+    await expect(repo.queuePendingInput("session-1", "message 11", { source: "queue" } as never)).rejects.toThrow(
+      "Pending input limit reached",
+    );
+  });
+});
+
+describe("model discovery coalescing", () => {
+  const originalFetch = globalThis.fetch;
+  const roots: string[] = [];
+
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+    while (roots.length > 0) {
+      const root = roots.pop();
+      if (root) fs.rmSync(root, { recursive: true, force: true });
+    }
+  });
+
+  const tempRoot = (): string => {
+    const dir = fs.mkdtempSync(path.join(os.tmpdir(), "argos-hygiene-"));
+    roots.push(dir);
+    return dir;
+  };
+
+  it("coalesces concurrent refreshes for the same provider into one upstream call", async () => {
+    const dataDir = tempRoot();
+    const presenter = new DaemonConfigPresenter(dataDir, dataDir);
+    presenter.setProviders([
+      {
+        id: "prov-1",
+        name: "Coalesced Provider",
+        type: "openai",
+        apiType: "openai",
+        baseUrl: "https://models.example.invalid/v1",
+        apiKey: "sk-test",
+      } as never,
+    ]);
+
+    let upstreamCalls = 0;
+    globalThis.fetch = (async () => {
+      upstreamCalls += 1;
+      await new Promise((resolve) => setTimeout(resolve, 25));
+      return new Response(JSON.stringify({ data: [{ id: "model-a", name: "Model A" }] }), {
+        status: 200,
+        headers: { "content-type": "application/json" },
+      });
+    }) as typeof fetch;
+
+    const [first, second, third] = await Promise.all([
+      presenter.refreshProviderModels("prov-1"),
+      presenter.refreshProviderModels("prov-1"),
+      presenter.refreshProviderModels("prov-1"),
+    ]);
+
+    expect(upstreamCalls).toBe(1);
+    expect(first.map((model) => model.id)).toEqual(["model-a"]);
+    expect(second).toBe(first);
+    expect(third).toBe(first);
+
+    // After the coalesced call settles, a new call is a fresh refresh.
+    const fourth = await presenter.refreshProviderModels("prov-1");
+    expect(upstreamCalls).toBe(2);
+    expect(fourth).not.toBe(first);
+  });
+
+  it("does not coalesce different providers", async () => {
+    const dataDir = tempRoot();
+    const presenter = new DaemonConfigPresenter(dataDir, dataDir);
+    presenter.setProviders([
+      {
+        id: "prov-1",
+        name: "P1",
+        type: "openai",
+        apiType: "openai",
+        baseUrl: "https://a.example.invalid/v1",
+        apiKey: "k",
+      } as never,
+      {
+        id: "prov-2",
+        name: "P2",
+        type: "openai",
+        apiType: "openai",
+        baseUrl: "https://b.example.invalid/v1",
+        apiKey: "k",
+      } as never,
+    ]);
+
+    let upstreamCalls = 0;
+    globalThis.fetch = (async () => {
+      upstreamCalls += 1;
+      return new Response(JSON.stringify({ data: [{ id: "m" }] }), {
+        status: 200,
+        headers: { "content-type": "application/json" },
+      });
+    }) as typeof fetch;
+
+    await Promise.all([presenter.refreshProviderModels("prov-1"), presenter.refreshProviderModels("prov-2")]);
+    expect(upstreamCalls).toBe(2);
+  });
+});
+
+describe("download archive response release", () => {
+  const originalFetch = globalThis.fetch;
+
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+  });
+
+  it("cancels the error body and throws on a failed archive download", async () => {
+    const registryRoot = fs.mkdtempSync(path.join(os.tmpdir(), "argos-hygiene-acp-"));
+    const service = new AcpLaunchSpecService(registryRoot);
+    const cancel = vi.fn(async () => undefined);
+
+    globalThis.fetch = (async () => {
+      return {
+        ok: false,
+        status: 500,
+        statusText: "Internal Server Error",
+        body: { cancel },
+      } as unknown as Response;
+    }) as typeof fetch;
+
+    await expect(
+      (service as any).downloadArchive("https://mirror.example.invalid/agent.zip", { id: "agent-1" }),
+    ).rejects.toThrow("Failed to download archive");
+    expect(cancel).toHaveBeenCalled();
+
+    fs.rmSync(registryRoot, { recursive: true, force: true });
+  });
+});

docs/issues/upstream-hygiene-sweep/plan.md

@@ -0,0 +1,18 @@
+# Plan: Upstream hygiene sweep
+
+1. `apps/daemon/src/host/bun-session-repository.ts`: `MAX_ACTIVE_PENDING_INPUTS` 5 → 10.
+2. `apps/daemon/src/host/daemonConfigPresenter.ts`: add
+   `inFlightModelRefreshes: Map<string, Promise<MODEL_META[]>>` wrapping `refreshProviderModels`,
+   keyed by a provider settings fingerprint (id + apiType + baseUrl + apiKey) so a mid-flight
+   settings change starts a fresh discovery instead of returning results fetched with stale
+   credentials, and `inFlightOllamaFetches: Map<string, Promise<OllamaModel[]>>`
+   (providerId+path key) wrapping `fetchOllamaModels`. Entries delete on settle; failures
+   propagate to every waiter and clear the key (next call retries).
+3. Release unconsumed fetch error bodies (best-effort `response.body?.cancel()`) in:
+   - `packages/acp-runtime/src/config/acpLaunchSpecService.ts` (downloadArchive),
+   - `packages/mcp-runtime/src/config/mcprouterManager.ts` (list + get),
+   - `packages/backend-core/src/provider/providerDbLoader.ts` (refresh),
+   - `packages/acp-runtime/src/config/acpRegistryService.ts` (icon fetch, found during the sweep).
+   Done inline per file — no new cross-package dependency for a three-line helper.
+4. Tests: pending-input limit (daemon), refresh coalescing (daemon), downloadArchive body
+   release (daemon test dir hosts it against the exported class).

docs/issues/upstream-hygiene-sweep/spec.md

@@ -0,0 +1,54 @@
+# Spec: Upstream hygiene sweep (queue limit, model-discovery coalescing, response-body release)
+
+Three small, independent fixes identified while sweeping recent DeepChat work
+(#2236, #2248, #2251 — plus #1946 assessed as already-satisfied).
+
+## 1. Pending-input queue limit 5 → 10 (DeepChat #2236)
+
+`MAX_ACTIVE_PENDING_INPUTS` in `bun-session-repository.ts` caps queued messages per session at
+5. Users who queue several messages during a long agent turn hit the limit with a bare error.
+Raise the cap to 10 (matching upstream's post-fix behavior) and add regression coverage that 10
+succeed and the 11th rejects.
+
+## 2. Coalesce model discovery requests (DeepChat #2248)
+
+Two daemon surfaces fetch model lists over the network with no in-flight dedup, so N concurrent
+UI callers (model picker, model store init, per-agent config surfaces) produce N identical
+upstream requests:
+
+- `DaemonConfigPresenter.refreshProviderModels(providerId)` — the OpenAI-compatible `/models`
+  fetch, lazily triggered by `models.getProviderCatalog` whenever a provider has credentials but
+  no stored catalog yet;
+- `DaemonConfigPresenter.fetchOllamaModels` via `listOllamaModels`/`listOllamaRunningModels` —
+  `/api/tags` and `/api/ps` per Ollama provider.
+
+Coalesce with a per-key in-flight promise map (keyed by providerId, and providerId+path for
+Ollama): concurrent callers share one upstream request; the map entry clears on settle so the
+next call after completion is a fresh refresh.
+
+## 3. Release unconsumed fetch response bodies (DeepChat #2251)
+
+Five sites throw/return on `!response.ok` without consuming the error body, which keeps the
+underlying socket busy until GC:
+
+- `AcpLaunchSpecService.downloadArchive` (`packages/acp-runtime`);
+- `mcprouterManager` list + get (`packages/mcp-runtime`);
+- `providerDbLoader` refresh (`packages/backend-core`);
+- `AcpRegistryService` icon fetch (`packages/acp-runtime`, found during the sweep).
+
+Fix: cancel the body (`response.body?.cancel()`, best-effort try/catch) before each
+error return. Success paths already consume. Done inline per file — no new cross-package
+dependency for a three-line helper.
+
+## Non-goals
+
+- No UI changes (the pending-input lane already reflects live queue contents; no rendered
+  static limit text exists to update).
+- No changes to the ACP registry refresh or Ollama pull (both already consume bodies).
+
+## Tests
+
+- Daemon: pending-input limit test (10 succeed, 11th throws) against the in-memory SQLite repo.
+- Daemon: `refreshProviderModels` coalescing test — two concurrent calls with a counting fetch
+  stub produce exactly one upstream request and identical results.
+- acp-runtime: `downloadArchive` non-ok response releases the body (cancel spy) and throws.

docs/issues/upstream-hygiene-sweep/tasks.md

@@ -0,0 +1,15 @@
+# Tasks: Upstream hygiene sweep
+
+- [x] T1 Pending-input queue limit 5 → 10 + limit regression test (10 OK, 11th rejects)
+- [x] T2 Coalesce `refreshProviderModels` (per provider) and Ollama tag/ps lookups (per
+      provider+suffix) + coalescing tests (same-provider → 1 upstream call; different providers
+      → separate calls; post-settle calls refresh)
+- [x] T3 Release unconsumed fetch error bodies: `downloadArchive`, `mcprouterManager` list+get,
+      `providerDbLoader` refresh, `acpRegistryService` icon fetch + downloadArchive release test
+- [x] T4 `bun run format` + `bun run lint` + `bun run typecheck` + `bun run test`
+
+## Verification results
+
+- Daemon: 409 tests pass (4 new); `tsc --noEmit` clean.
+- Desktop + UI: typechecks clean; full `bun run test` green.
+- `bun run lint`: all architecture guards + oxlint clean.

packages/acp-runtime/src/config/acpLaunchSpecService.ts

@@ -503,6 +503,12 @@ export class AcpLaunchSpecService {
     const archivePath = path.join(tempDir, path.basename(new URL(url).pathname));
     const response = await fetch(url);
     if (!response.ok) {
+      // Release the error body so the underlying connection returns to the pool.
+      try {
+        await response.body?.cancel();
+      } catch {
+        // best-effort
+      }
       throw new Error(`Failed to download archive: ${response.status} ${response.statusText}`);
     }
 

packages/acp-runtime/src/config/acpRegistryService.ts

@@ -541,6 +541,12 @@ export class AcpRegistryService {
       });
 
       if (!response.ok) {
+        // Release the error body so the underlying connection returns to the pool.
+        try {
+          await response.body?.cancel();
+        } catch {
+          // best-effort
+        }
         throw new Error(`Failed to fetch icon: ${response.status} ${response.statusText}`);
       }
 

packages/backend-core/src/provider/providerDbLoader.ts

@@ -283,6 +283,12 @@ export class ProviderDbLoader {
       }
 
       if (!res.ok) {
+        // Release the error body so the underlying connection returns to the pool.
+        try {
+          await res.body?.cancel();
+        } catch {
+          // best-effort
+        }
         const meta = this.createAttemptMeta(prevMeta, url, now);
         if (meta) this.writeMeta(meta);
         return this.createResult("error", meta, `Request failed with status ${res.status}`);