diff --git a/backend/app/agent/copilot.py b/backend/app/agent/copilot.py index 5ff1cf0..1370b5e 100644 --- a/backend/app/agent/copilot.py +++ b/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({ diff --git a/backend/app/agent/purchasing.py b/backend/app/agent/purchasing.py index f77db6e..1914016 100644 --- a/backend/app/agent/purchasing.py +++ b/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,9 +408,38 @@ 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 @@ -416,9 +447,14 @@ def _compute_bundles(db: Session, period_days: int) -> tuple[dict[str, list[dict 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]: diff --git a/backend/tests/test_capacity_diagnosis.py b/backend/tests/test_capacity_diagnosis.py index 658a3a8..7c4dd0f 100644 --- a/backend/tests/test_capacity_diagnosis.py +++ b/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=[], diff --git a/backend/tests/test_decision_log.py b/backend/tests/test_decision_log.py index eb6d712..eccfa84 100644 --- a/backend/tests/test_decision_log.py +++ b/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", diff --git a/backend/tests/test_purchasing_run.py b/backend/tests/test_purchasing_run.py index fe661cd..7224240 100644 --- a/backend/tests/test_purchasing_run.py +++ b/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( diff --git a/backend/tests/test_requisitions.py b/backend/tests/test_requisitions.py index 8f38c51..cf4359d 100644 --- a/backend/tests/test_requisitions.py +++ b/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={}, diff --git a/frontend/app.js b/frontend/app.js index 1296795..bfcfd3f 100644 --- a/frontend/app.js +++ b/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; diff --git a/frontend/features.js b/frontend/features.js index 39304ef..7dad594 100644 --- a/frontend/features.js +++ b/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() { `; }).join("") || `
Nothing needs surfacing right now.
`; } catch (e) { - $("#agent-insights").innerHTML = `
${esc(e.message)}
`; + $("#agent-insights").innerHTML = `
${esc(e.message)}
+ `; + const b = $("#agent-insights-btn"); if (b) b.addEventListener("click", loadInsights); } } async function runPurchasing() { const host = $("#agent-run"); - host.innerHTML = `
Computing the run…
`; + host.innerHTML = `
Computing the run… (the agent reasons over each buy — this can take up to a minute)
`; 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 | " +// 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 `
${plainPill(tier.label, tier.tone)}${esc(prod)}${euro(d.total)}
${esc(sup)} · ${d.qty} × ${euro(d.unit_price)}
-
${esc(d.rationale || "")}
+
${esc(cleanRationale(d.rationale))}
${esc((d.trigger || {}).type || "").replace(/_/g, " ")} · ${Math.round((d.confidence || 0) * 100)}%