feat(concurrency): add provider concurrency limits UI — settings + live status

3 files changed, 455 insertions, 0 deletions

diff --git a/src/app/App.svelte b/src/app/App.svelte
index 09be947..b40ed09 100644
--- a/src/app/App.svelte
+++ b/src/app/App.svelte
@@ -69,6 +69,14 @@
 		type HeartbeatRunsResult,
 		type HeartbeatStopResult,
 	} from "../features/heartbeat";
+	import {
+		ConcurrencyView,
+		manifest as concurrencyManifest,
+		type DeleteConcurrencyLimit,
+		type LoadConcurrencyLimits,
+		type LoadConcurrencyStatus,
+		type SaveConcurrencyLimit,
+	} from "../features/concurrency";
 	import type { ChatStore } from "../features/chat";
 	import {
 		SystemPromptBuilder,
@@ -107,6 +115,7 @@
 		{ id: "tasks", label: "Tasks" },
 		{ id: "compaction", label: "Compaction" },
 		{ id: "heartbeat", label: "Heartbeat" },
+		{ id: "concurrency", label: "Concurrency" },
 		{ id: "system-prompt", label: "System Prompt" },
 		{ id: "settings", label: "Settings" },
 	] as const;
@@ -143,6 +152,7 @@
 		settingsManifest,
 		systemPromptManifest,
 		heartbeatManifest,
+		concurrencyManifest,
 	].map((m) => [m.name, m.description] as const);
 
 	// Smart-scroll: keep the transcript pinned to the bottom while it streams,
@@ -412,6 +422,18 @@
 	function closeRunChat(conversationId: string): void {
 		store.unwatchConversation(conversationId);
 	}
+
+	// Adapt the store's concurrency results to the feature's ports. The store
+	// returns the feature's result types directly (the API is a plain REST surface
+	// under /concurrency, not a workspace/conversation-scoped one), so the adapter
+	// is a thin passthrough (kept for structural consistency — AGENTS.md "contracts
+	// are the cross-unit surface").
+	const loadConcurrencyLimits: LoadConcurrencyLimits = () => store.concurrencyLimits();
+	const saveConcurrencyLimit: SaveConcurrencyLimit = (providerId, limit) =>
+		store.setConcurrencyLimit(providerId, limit);
+	const deleteConcurrencyLimit: DeleteConcurrencyLimit = (providerId) =>
+		store.deleteConcurrencyLimit(providerId);
+	const loadConcurrencyStatus: LoadConcurrencyStatus = () => store.concurrencyStatus();
 </script>
 
 <main class="relative flex h-screen overflow-hidden">
@@ -690,5 +712,14 @@
 			loadNextRun={loadHeartbeatNextRun}
 			onOpenRun={(run) => (heartbeatRun = run)}
 		/>
+	{:else if kind === "concurrency"}
+		<!-- Per-provider concurrency limits + live status. GLOBAL (not workspace- or
+		     conversation-scoped), so the panel stays mounted across tab switches. -->
+		<ConcurrencyView
+			loadLimits={loadConcurrencyLimits}
+			saveLimit={saveConcurrencyLimit}
+			deleteLimit={deleteConcurrencyLimit}
+			loadStatus={loadConcurrencyStatus}
+		/>
 	{/if}
 {/snippet}
diff --git a/src/app/store.svelte.ts b/src/app/store.svelte.ts
index 78f6ede..8bc1080 100644
--- a/src/app/store.svelte.ts
+++ b/src/app/store.svelte.ts
@@ -53,6 +53,17 @@ import {
 } from "../core/protocol";
 import type { ChatStore, HistorySync, MetricsSync } from "../features/chat";
 import { createChatStore } from "../features/chat";
+import type {
+  ConcurrencyDeleteResult,
+  ConcurrencyLimitResult,
+  ConcurrencyLimitsResult,
+  ConcurrencyStatusResult,
+} from "../features/concurrency";
+import {
+  normalizeConcurrencyLimit,
+  normalizeConcurrencyLimits,
+  normalizeConcurrencyStatus,
+} from "../features/concurrency";
 import type { ConversationCache } from "../features/conversation-cache";
 import { createConversationCache } from "../features/conversation-cache";
 import type {
@@ -363,6 +374,38 @@ export interface AppStore {
   /** Dispose + unsubscribe a watch opened by {@link watchConversation}. */
   unwatchConversation(conversationId: string): void;
   /**
+   * Load all configured per-provider concurrency limits
+   * (`GET /concurrency/limits`). Global (not workspace-scoped). Returns an empty
+   * list when the concurrency extension isn't loaded (`{ limits: [] }`).
+   */
+  concurrencyLimits(): Promise<ConcurrencyLimitsResult>;
+  /**
+   * Fetch the configured limit for one provider
+   * (`GET /concurrency/limits/:providerId`). `404` (no limit configured) and
+   * `503` (extension not loaded) both surface as `ok: false`.
+   */
+  getConcurrencyLimit(providerId: string): Promise<ConcurrencyLimitResult>;
+  /**
+   * Set or update a provider's concurrency limit
+   * (`PUT /concurrency/limits/:providerId`, body `{ limit }`). `limit` must be a
+   * positive integer (a non-positive body is `400`). At the cap, further requests
+   * queue oldest-agent-first rather than being sent immediately.
+   */
+  setConcurrencyLimit(providerId: string, limit: number): Promise<ConcurrencyLimitResult>;
+  /**
+   * Remove a provider's concurrency limit (`DELETE /concurrency/limits/:providerId`),
+   * making it unlimited. `404` (not configured) and `503` (extension not loaded)
+   * both surface as `ok: false`.
+   */
+  deleteConcurrencyLimit(providerId: string): Promise<ConcurrencyDeleteResult>;
+  /**
+   * Fetch live concurrency status for every provider with a configured limit
+   * (`GET /concurrency/status`): in-flight slots held, agents queued, and a paused
+   * state with a `pausedUntil` epoch-ms when a 429 backoff is in effect. Returns
+   * an empty list when the extension isn't loaded (`{ providers: [] }`).
+   */
+  concurrencyStatus(): Promise<ConcurrencyStatusResult>;
+  /**
    * A critical error that blocks normal operation (e.g. the cross-device tab
    * restore fetch failed). When non-null, a full-screen modal is shown with the
    * error details. Cleared by `clearFatalError` (the modal's dismiss button).
@@ -1696,6 +1739,128 @@ export function createAppStore(opts?: CreateAppStoreOptions): AppStore {
       unwatchConversation(conversationId);
     },
 
+    // ── Concurrency (per-provider limits + live status; GLOBAL, not workspace-scoped)
+
+    async concurrencyLimits(): Promise<ConcurrencyLimitsResult> {
+      try {
+        const res = await fetchImpl(`${httpBase}/concurrency/limits`);
+        if (!res.ok) {
+          const errBody = (await res.json().catch(() => null)) as { error?: string } | null;
+          return {
+            ok: false,
+            error: errBody?.error ?? `Concurrency limits failed (HTTP ${res.status})`,
+          };
+        }
+        // Normalize the untyped JSON at the network seam (pure helper) so a
+        // malformed/partial response (e.g. the extension returning `{}`) can
+        // never crash the renderer.
+        const limits = normalizeConcurrencyLimits(await res.json());
+        return { ok: true, limits };
+      } catch (err) {
+        return {
+          ok: false,
+          error: err instanceof Error ? err.message : "Concurrency limits request failed",
+        };
+      }
+    },
+
+    async getConcurrencyLimit(providerId: string): Promise<ConcurrencyLimitResult> {
+      try {
+        const res = await fetchImpl(
+          `${httpBase}/concurrency/limits/${encodeURIComponent(providerId)}`,
+        );
+        if (!res.ok) {
+          const errBody = (await res.json().catch(() => null)) as { error?: string } | null;
+          return {
+            ok: false,
+            error: errBody?.error ?? `Concurrency limit failed (HTTP ${res.status})`,
+          };
+        }
+        const limit = normalizeConcurrencyLimit(await res.json());
+        if (limit === null) {
+          return { ok: false, error: "Malformed concurrency limit response" };
+        }
+        return { ok: true, providerId: limit.providerId, limit: limit.limit };
+      } catch (err) {
+        return {
+          ok: false,
+          error: err instanceof Error ? err.message : "Concurrency limit request failed",
+        };
+      }
+    },
+
+    async setConcurrencyLimit(providerId: string, limit: number): Promise<ConcurrencyLimitResult> {
+      try {
+        const res = await fetchImpl(
+          `${httpBase}/concurrency/limits/${encodeURIComponent(providerId)}`,
+          {
+            method: "PUT",
+            headers: { "content-type": "application/json" },
+            body: JSON.stringify({ limit }),
+          },
+        );
+        if (!res.ok) {
+          const errBody = (await res.json().catch(() => null)) as { error?: string } | null;
+          return {
+            ok: false,
+            error: errBody?.error ?? `Set concurrency limit failed (HTTP ${res.status})`,
+          };
+        }
+        const echoed = normalizeConcurrencyLimit(await res.json());
+        if (echoed === null) {
+          return { ok: false, error: "Malformed concurrency limit response" };
+        }
+        return { ok: true, providerId: echoed.providerId, limit: echoed.limit };
+      } catch (err) {
+        return {
+          ok: false,
+          error: err instanceof Error ? err.message : "Set concurrency limit request failed",
+        };
+      }
+    },
+
+    async deleteConcurrencyLimit(providerId: string): Promise<ConcurrencyDeleteResult> {
+      try {
+        const res = await fetchImpl(
+          `${httpBase}/concurrency/limits/${encodeURIComponent(providerId)}`,
+          { method: "DELETE" },
+        );
+        if (!res.ok) {
+          const errBody = (await res.json().catch(() => null)) as { error?: string } | null;
+          return {
+            ok: false,
+            error: errBody?.error ?? `Delete concurrency limit failed (HTTP ${res.status})`,
+          };
+        }
+        return { ok: true, providerId };
+      } catch (err) {
+        return {
+          ok: false,
+          error: err instanceof Error ? err.message : "Delete concurrency limit request failed",
+        };
+      }
+    },
+
+    async concurrencyStatus(): Promise<ConcurrencyStatusResult> {
+      try {
+        const res = await fetchImpl(`${httpBase}/concurrency/status`);
+        if (!res.ok) {
+          const errBody = (await res.json().catch(() => null)) as { error?: string } | null;
+          return {
+            ok: false,
+            error: errBody?.error ?? `Concurrency status failed (HTTP ${res.status})`,
+          };
+        }
+        const providers = normalizeConcurrencyStatus(await res.json());
+        return { ok: true, providers };
+      } catch (err) {
+        return {
+          ok: false,
+          error: err instanceof Error ? err.message : "Concurrency status request failed",
+        };
+      }
+    },
+
     async loadSystemPrompt(): Promise<SystemPromptLoadResult> {
       try {
         const res = await fetchImpl(`${httpBase}/system-prompt`);
diff --git a/src/app/store.test.ts b/src/app/store.test.ts
index 947a9b0..2074d9e 100644
--- a/src/app/store.test.ts
+++ b/src/app/store.test.ts
@@ -1375,4 +1375,263 @@ describe("createAppStore", () => {
 
     store.dispose();
   });
+
+  // ── Concurrency (per-provider limits + live status; GLOBAL REST surface) ─────
+  //
+  // The concurrency API is a plain REST surface under /concurrency (provided by
+  // the `concurrency` extension). These tests fake all five endpoints + verify
+  // the store coerces the untyped JSON and routes the right method/URL.
+
+  function concurrencyFetchImpl(opts?: {
+    limits?: Record<string, unknown>;
+    limit?: Record<string, unknown>;
+    status?: Record<string, unknown>;
+  }): typeof fetch {
+    const base = fakeFetchImpl();
+    const limits = opts?.limits ?? {
+      limits: [
+        { providerId: "umans", limit: 4 },
+        { providerId: "openai-compat", limit: 5 },
+      ],
+    };
+    const limit = opts?.limit ?? { providerId: "umans", limit: 4 };
+    const status = opts?.status ?? {
+      providers: [
+        { providerId: "umans", limit: 4, inFlight: 2, queued: 1, paused: false },
+        {
+          providerId: "openai-compat",
+          limit: 5,
+          inFlight: 5,
+          queued: 3,
+          paused: true,
+          pausedUntil: 1_719_408_000_000,
+        },
+      ],
+    };
+    return async (input, init) => {
+      const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
+      const method = init?.method ?? "GET";
+      if (url.endsWith("/concurrency/status") && method === "GET") {
+        return new Response(JSON.stringify(status), { status: 200 });
+      }
+      if (url.endsWith("/concurrency/limits") && method === "GET") {
+        return new Response(JSON.stringify(limits), { status: 200 });
+      }
+      if (url.includes("/concurrency/limits/") && method === "GET") {
+        return new Response(JSON.stringify(limit), { status: 200 });
+      }
+      if (url.includes("/concurrency/limits/") && method === "PUT") {
+        const seg = url.slice(
+          url.lastIndexOf("/concurrency/limits/") + "/concurrency/limits/".length,
+        );
+        const body = init?.body ? JSON.parse(init.body as string) : {};
+        return new Response(JSON.stringify({ providerId: seg, ...body }), { status: 200 });
+      }
+      if (url.includes("/concurrency/limits/") && method === "DELETE") {
+        return new Response(JSON.stringify({ ok: true, providerId: "umans" }), { status: 200 });
+      }
+      return base(input, init);
+    };
+  }
+
+  it("concurrencyLimits loads + coerces the limits list", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: concurrencyFetchImpl(),
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.concurrencyLimits();
+    expect(result.ok).toBe(true);
+    if (!result.ok) throw new Error("unreachable");
+    expect(result.limits).toEqual([
+      { providerId: "umans", limit: 4 },
+      { providerId: "openai-compat", limit: 5 },
+    ]);
+    store.dispose();
+  });
+
+  it("concurrencyLimits tolerates a malformed/empty body (extension not loaded)", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: concurrencyFetchImpl({ limits: {} }),
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.concurrencyLimits();
+    expect(result.ok).toBe(true);
+    if (!result.ok) throw new Error("unreachable");
+    expect(result.limits).toEqual([]);
+    store.dispose();
+  });
+
+  it("concurrencyLimits surfaces an HTTP error", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: async (input) => {
+        const url =
+          typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
+        if (url.endsWith("/concurrency/limits"))
+          return new Response(JSON.stringify({ error: "Concurrency service not available" }), {
+            status: 503,
+          });
+        return fakeFetchImpl()(input);
+      },
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.concurrencyLimits();
+    expect(result.ok).toBe(false);
+    if (result.ok) throw new Error("unreachable");
+    expect(result.error).toContain("Concurrency service not available");
+    store.dispose();
+  });
+
+  it("getConcurrencyLimit loads one provider's limit", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: concurrencyFetchImpl(),
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.getConcurrencyLimit("umans");
+    expect(result.ok).toBe(true);
+    if (!result.ok) throw new Error("unreachable");
+    expect(result.providerId).toBe("umans");
+    expect(result.limit).toBe(4);
+    store.dispose();
+  });
+
+  it("getConcurrencyLimit surfaces a 404 (not configured)", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: async (input, init) => {
+        const url =
+          typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
+        const method = init?.method ?? "GET";
+        if (url.includes("/concurrency/limits/") && method === "GET")
+          return new Response(
+            JSON.stringify({ error: "No concurrency limit configured for this provider" }),
+            { status: 404 },
+          );
+        return fakeFetchImpl()(input);
+      },
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.getConcurrencyLimit("anthropic");
+    expect(result.ok).toBe(false);
+    if (result.ok) throw new Error("unreachable");
+    expect(result.error).toContain("No concurrency limit configured");
+    store.dispose();
+  });
+
+  it("setConcurrencyLimit PUTs { limit } to the provider URL and returns the echoed limit", async () => {
+    const calls: { url: string; method: string; body: unknown }[] = [];
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: async (input, init) => {
+        const url =
+          typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
+        const method = init?.method ?? "GET";
+        if (url.includes("/concurrency/limits/") && method === "PUT") {
+          calls.push({ url, method, body: JSON.parse(init?.body as string) });
+        }
+        return concurrencyFetchImpl()(input, init);
+      },
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.setConcurrencyLimit("anthropic", 8);
+    expect(result.ok).toBe(true);
+    expect(calls).toHaveLength(1);
+    expect(calls[0]?.method).toBe("PUT");
+    expect(calls[0]?.url).toContain("/concurrency/limits/anthropic");
+    expect(calls[0]?.body).toEqual({ limit: 8 });
+    if (!result.ok) throw new Error("unreachable");
+    expect(result.providerId).toBe("anthropic"); // echoed by the fake
+    expect(result.limit).toBe(8);
+    store.dispose();
+  });
+
+  it("setConcurrencyLimit surfaces a 400 (non-positive body)", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: async (input, init) => {
+        const url =
+          typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
+        if (url.includes("/concurrency/limits/") && init?.method === "PUT")
+          return new Response(
+            JSON.stringify({ error: "Body must be { limit: <positive integer> }" }),
+            {
+              status: 400,
+            },
+          );
+        return fakeFetchImpl()(input);
+      },
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.setConcurrencyLimit("umans", 0);
+    expect(result.ok).toBe(false);
+    if (result.ok) throw new Error("unreachable");
+    expect(result.error).toContain("Body must be");
+    store.dispose();
+  });
+
+  it("deleteConcurrencyLimit DELETEs the provider URL and returns ok", async () => {
+    const calls: { url: string; method: string }[] = [];
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: async (input, init) => {
+        const url =
+          typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
+        const method = init?.method ?? "GET";
+        if (url.includes("/concurrency/limits/") && method === "DELETE") {
+          calls.push({ url, method });
+        }
+        return concurrencyFetchImpl()(input, init);
+      },
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.deleteConcurrencyLimit("umans");
+    expect(result.ok).toBe(true);
+    expect(calls).toHaveLength(1);
+    expect(calls[0]?.method).toBe("DELETE");
+    expect(calls[0]?.url).toContain("/concurrency/limits/umans");
+    if (!result.ok) throw new Error("unreachable");
+    expect(result.providerId).toBe("umans");
+    store.dispose();
+  });
+
+  it("concurrencyStatus loads + coerces the status list (incl. pausedUntil)", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: concurrencyFetchImpl(),
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.concurrencyStatus();
+    expect(result.ok).toBe(true);
+    if (!result.ok) throw new Error("unreachable");
+    expect(result.providers).toHaveLength(2);
+    expect(result.providers[0]).toEqual({
+      providerId: "umans",
+      limit: 4,
+      inFlight: 2,
+      queued: 1,
+      paused: false,
+    });
+    expect(result.providers[1]).toMatchObject({
+      providerId: "openai-compat",
+      paused: true,
+      pausedUntil: 1_719_408_000_000,
+    });
+    store.dispose();
+  });
+
+  it("concurrencyStatus tolerates a malformed body (extension not loaded)", async () => {
+    const store = createAppStore({
+      socketFactory: () => fakeSocket(),
+      fetchImpl: concurrencyFetchImpl({ status: {} }),
+      localStorage: createFakeStorage(),
+    });
+    const result = await store.concurrencyStatus();
+    expect(result.ok).toBe(true);
+    if (!result.ok) throw new Error("unreachable");
+    expect(result.providers).toEqual([]);
+    store.dispose();
+  });
 });