fix(agent): faster purchasing run, clean rationale, no silent hangs

backend/app/agent/copilot.py

@@ -140,11 +140,14 @@ def _raw_is_error(raw: str) -> bool:
     return raw.startswith("[agent-error]")
 
 
-def recommend_sourcing(db: Session, product_id: str,
-                       desired_qty: Optional[int] = None) -> SourcingRecommendation:
-    sig = gather_sourcing_signals(db, product_id, desired_qty)
-    user = json.dumps({"signals": sig}, default=str)
+def _recommend_sourcing_from_signals(sig: dict) -> SourcingRecommendation:
+    """The LLM half of recommend_sourcing — pure network, NO database access.
 
+    Split out so a batch run can gather every product's signals on the main
+    thread (DB-bound, sequential, session-safe) and then fire these LLM calls
+    concurrently (network-bound). Takes pre-gathered signals; touches no Session.
+    """
+    user = json.dumps({"signals": sig}, default=str)
     for attempt in range(2):
         system = prompts.SOURCING_SYSTEM + (_RETRY_NUDGE if attempt else "")
         raw = call_claude(system, user)
@@ -162,6 +165,17 @@ def recommend_sourcing(db: Session, product_id: str,
     raise AgentError("sourcing recommendation failed")  # unreachable
 
 
+def recommend_sourcing(db: Session, product_id: str,
+                       desired_qty: Optional[int] = None,
+                       *, signals: Optional[dict] = None) -> SourcingRecommendation:
+    """Sourcing verdict for a product. Pass pre-gathered ``signals`` to skip the
+    DB read (used by the concurrent batch path, which gathers signals on the main
+    thread). This stays the single mockable seam — tests patch THIS function, so
+    the batch path goes through it too."""
+    sig = signals if signals is not None else gather_sourcing_signals(db, product_id, desired_qty)
+    return _recommend_sourcing_from_signals(sig)
+
+
 def generate_insights(db: Session, min_count: int = 5) -> list[AgentInsight]:
     sig = gather_insight_signals(db)
     user = json.dumps({

backend/app/agent/purchasing.py

@@ -20,6 +20,7 @@
 import logging
 import math
 from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
 from datetime import date, datetime, timedelta, timezone
 from typing import Optional
 
@@ -365,6 +366,7 @@ def _compute_bundles(db: Session, period_days: int) -> tuple[dict[str, list[dict
 
     bundles: dict[str, list[dict]] = defaultdict(list)
     orphans: list[dict] = []
+    pending: list[dict] = []   # lines awaiting their (concurrent) LLM verdict
     for pid, info in net_needs.items():
         ranked = sourcing.suggest_sources(db, pid)
         if not ranked:
@@ -406,19 +408,53 @@ def _compute_bundles(db: Session, period_days: int) -> tuple[dict[str, list[dict
             "product_id": pid, "product_supplier_id": src["product_supplier_id"],
             "qty": qty, "unit_price": unit_price, "line_total": qty * unit_price,
             "trigger": info, "storage_capped": storage_capped,
+            "supplier_id": src["supplier_id"],
+            "_pid": pid, "_qty": qty,
+            # Signals gathered now, on this (main) thread — DB-bound and
+            # session-safe. The LLM call that consumes them is fired concurrently
+            # below (network-bound, no DB), turning N serial round-trips into one
+            # parallel batch.
+            "_signals": copilot.gather_sourcing_signals(db, pid, qty),
         }
+        pending.append(line)
+
+    _attach_recommendations(db, pending)
+    for line in pending:
+        bundles[line.pop("supplier_id")].append(line)
+
+    return bundles, orphans
+
+
+def _attach_recommendations(db: Session, lines: list[dict]) -> None:
+    """Fire the per-line sourcing LLM calls concurrently and attach the verdicts.
+
+    Goes through the public ``copilot.recommend_sourcing`` (the mockable seam),
+    passing pre-gathered ``signals`` so the worker thread does NO database access
+    — only the network call runs in the pool. A per-line failure degrades that
+    line to escalate; it never fails the run.
+    """
+    if not lines:
+        return
+
+    def _judge(line: dict) -> None:
         try:
-            rec = copilot.recommend_sourcing(db, pid, qty)
+            rec = copilot.recommend_sourcing(
+                db, line["_pid"], line["_qty"], signals=line["_signals"])
             line["confidence"] = rec.confidence
             line["agent_decision"] = rec.decision
             line["agent_rationale"] = rec.rationale
         except copilot.AgentError as exc:
             line["confidence"] = 0.0
             line["agent_decision"] = "escalate"
             line["agent_rationale"] = f"copilot unavailable: {exc}"
-        bundles[src["supplier_id"]].append(line)
-
-    return bundles, orphans
+        finally:
+            for k in ("_signals", "_pid", "_qty"):
+                line.pop(k, None)
+
+    # Cap concurrency so we don't hammer the API; small fleets just run in one wave.
+    workers = min(len(lines), 8)
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        list(pool.map(_judge, lines))
 
 
 def _tier_bundle(lines: list[dict], bundle_total: float) -> tuple[float, str]:

backend/tests/test_capacity_diagnosis.py

@@ -123,7 +123,7 @@ def test_order_capped_at_storage_headroom(db_session, monkeypatch):
     from app.agent import copilot, purchasing
     from app.agent.schemas import SourcingRecommendation
 
-    def fake(db, pid, q=None):
+    def fake(db, pid, q=None, *, signals=None):
         return SourcingRecommendation(
             product_id=pid, recommended_source_id="x", recommended_qty=q or 1,
             rationale="m", signals={}, assumptions=[], uncertainties=[],
@@ -153,7 +153,7 @@ def test_weekly_run_also_respects_storage_cap(db_session, monkeypatch):
     from app.agent import copilot, purchasing
     from app.agent.schemas import SourcingRecommendation
 
-    def fake(db, pid, q=None):
+    def fake(db, pid, q=None, *, signals=None):
         return SourcingRecommendation(
             product_id=pid, recommended_source_id="x", recommended_qty=q or 1,
             rationale="m", signals={}, assumptions=[], uncertainties=[],

backend/tests/test_decision_log.py

@@ -20,7 +20,7 @@
 
 
 def _mock_copilot(monkeypatch, *, decision="act", confidence=0.95):
-    def fake(db, product_id, desired_qty=None):
+    def fake(db, product_id, desired_qty=None, *, signals=None):
         return SourcingRecommendation(
             product_id=product_id, recommended_source_id="x",
             recommended_qty=desired_qty or 1, rationale="mock",

backend/tests/test_purchasing_run.py

@@ -41,7 +41,9 @@ def _source(client, product_id, supplier_id, *, price, moq=1, rank=1, lead=21):
 
 
 def _mock_copilot(monkeypatch, *, decision="act", confidence=0.9):
-    def fake(db, product_id, desired_qty=None):
+    # signals=... is accepted because the concurrent batch path passes pre-gathered
+    # signals through the same recommend_sourcing seam.
+    def fake(db, product_id, desired_qty=None, *, signals=None):
         return SourcingRecommendation(
             product_id=product_id, recommended_source_id="x",
             recommended_qty=desired_qty or 1,
@@ -246,7 +248,7 @@ def test_bundle_escalates_if_any_line_escalates(client, db_session, monkeypatch)
     _decommission_assets(db_session, ssd["id"], 4)
 
     # one line low-confidence -> bundle confidence is the weakest link
-    def fake(db, product_id, desired_qty=None):
+    def fake(db, product_id, desired_qty=None, *, signals=None):
         conf = 0.95 if product_id == srv["id"] else 0.3
         dec = "act" if product_id == srv["id"] else "escalate"
         return SourcingRecommendation(

backend/tests/test_requisitions.py

@@ -53,7 +53,7 @@ def _source(client, product_id, supplier_id, *, price, moq=1, rank=1, lead=21):
 
 
 def _mock_copilot(monkeypatch, *, decision="act", confidence=0.9):
-    def fake(db, product_id, desired_qty=None):
+    def fake(db, product_id, desired_qty=None, *, signals=None):
         return SourcingRecommendation(
             product_id=product_id, recommended_source_id="x",
             recommended_qty=desired_qty or 1, rationale="mock", signals={},

frontend/app.js

@@ -115,13 +115,31 @@ const plainPill = (label, tone) => {
 const capTone = (u, over) => over ? "var(--ts-negative)" : u >= 0.9 ? "var(--ts-warning)" : u >= 0.7 ? "var(--ts-brand-gold)" : "var(--ts-positive)";
 
 /* ── API layer ─────────────────────────────────────────────────────── */
-async function api(path, { method = "GET", body, form } = {}) {
+async function api(path, { method = "GET", body, form, timeout } = {}) {
   const headers = {};
   if (token) headers.Authorization = `Bearer ${token}`;
   let payload;
   if (form) { payload = new URLSearchParams(form); }
   else if (body) { headers["Content-Type"] = "application/json"; payload = JSON.stringify(body); }
-  const res = await fetch(API + path, { method, headers, body: payload });
+  // Optional timeout so a hung backend (e.g. a slow/stalled LLM call) surfaces
+  // as a clear error instead of spinning forever. Off by default.
+  let signal, timer;
+  if (timeout) {
+    const ctrl = new AbortController();
+    signal = ctrl.signal;
+    timer = setTimeout(() => ctrl.abort(), timeout);
+  }
+  let res;
+  try {
+    res = await fetch(API + path, { method, headers, body: payload, signal });
+  } catch (e) {
+    if (e && e.name === "AbortError") {
+      throw new Error(`Timed out after ${Math.round(timeout / 1000)}s — the server took too long to respond.`);
+    }
+    throw e;
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
   if (res.status === 401) { logout(); throw new Error("Session expired — sign in again"); }
   if (!res.ok) {
     let detail = res.statusText;

frontend/features.js

@@ -384,7 +384,9 @@ function closeAgent() { if (drawerEl) { drawerEl.classList.remove("open"); scrim
 
 async function loadInsights() {
   try {
-    const list = await api("/agent/insights");
+    // AI insights are an LLM call (can be slow / occasionally 502 on the demo).
+    // Bound it so the panel shows an error+retry instead of spinning forever.
+    const list = await api("/agent/insights", { timeout: 60000 });
     $("#agent-insights").innerHTML = list.map((it) => {
       const sev = SEVERITY[it.severity] || SEVERITY.info;
       const t = TONE[sev.tone];
@@ -397,15 +399,19 @@ async function loadInsights() {
       </div>`;
     }).join("") || `<div class="insight__finding" style="color:var(--ts-ink-faint)">Nothing needs surfacing right now.</div>`;
   } catch (e) {
-    $("#agent-insights").innerHTML = `<div class="insight__finding" style="color:var(--ts-negative)">${esc(e.message)}</div>`;
+    $("#agent-insights").innerHTML = `<div class="insight__finding" style="color:var(--ts-negative)">${esc(e.message)}</div>
+      <button class="btn btn--ink btn--sm" id="agent-insights-btn" style="margin-top:10px">${icon("refresh", 13)} Retry</button>`;
+    const b = $("#agent-insights-btn"); if (b) b.addEventListener("click", loadInsights);
   }
 }
 
 async function runPurchasing() {
   const host = $("#agent-run");
-  host.innerHTML = `<div class="state"><div class="state__sub">Computing the run…</div></div>`;
+  host.innerHTML = `<div class="state"><div class="state__sub">Computing the run… <span class="muted">(the agent reasons over each buy — this can take up to a minute)</span></div></div>`;
   try {
-    const res = await api("/agent/purchasing-run", { method: "POST", body: { dry_run: true, period_days: 7 } });
+    // The run makes one LLM call per justified product, so it's slow; give it a
+    // bounded 90s rather than letting a stalled call spin the panel forever.
+    const res = await api("/agent/purchasing-run", { method: "POST", body: { dry_run: true, period_days: 7 }, timeout: 90000 });
     lastRun = res;
     renderRun(res);
   } catch (e) {
@@ -415,6 +421,17 @@ async function runPurchasing() {
   }
 }
 
+// The backend's `rationale` packs machine evidence ahead of the human text:
+//   "[trigger] {evidence dict} | net_need=N | bundle_tier=X | <readable rationale>"
+// The audit log keeps the full string; the UI shows only the readable part —
+// everything after the last "bundle_tier=…|" segment. Falls back to the raw
+// string if the prefix isn't present (older/other shapes).
+function cleanRationale(s) {
+  if (!s) return "";
+  const m = String(s).match(/bundle_tier=[^|]*\|\s*([\s\S]+)$/);
+  return (m ? m[1] : s).trim();
+}
+
 function renderRun(res) {
   const decisions = res.decisions || [];
   $("#agent-run").innerHTML = decisions.map((d) => {
@@ -425,7 +442,7 @@ function renderRun(res) {
     return `<div class="decision">
       <div class="decision__top">${plainPill(tier.label, tier.tone)}<span class="decision__name">${esc(prod)}</span><span class="decision__total money">${euro(d.total)}</span></div>
       <div class="decision__sup" style="margin-bottom:6px">${esc(sup)} · ${d.qty} × ${euro(d.unit_price)}</div>
-      <div class="decision__rat">${esc(d.rationale || "")}</div>
+      <div class="decision__rat">${esc(cleanRationale(d.rationale))}</div>
       <div class="decision__foot">
         <span class="decision__trigger">${esc((d.trigger || {}).type || "").replace(/_/g, " ")} · ${Math.round((d.confidence || 0) * 100)}%</span>
         <label class="decision__check">${approvable ? `<input type="checkbox" data-sup="${esc(d.supplier_id)}" ${d.tier === "act" ? "checked" : ""}/> approve` : `<span style="color:var(--ts-ink-faint)">needs sign-off</span>`}</label>