"""AutoSource dashboard โ AG-UI-aligned event stream viewer + human approval gate.
Modes:
Live follow โ follows a run being written right now (start one from the sidebar).
Replay โ plays any recorded run from traces/ turn by turn (zero API calls).
Tabs:
๐ Mission control โ the animated run view + AP2 approval gate.
๐ฌ Under the hood โ agent system prompts, parsed transcripts, raw events,
and what every module contributes.
Run: streamlit run ui/app.py
"""
from __future__ import annotations
import html
import json
import os
import subprocess
import sys
import time
import uuid
from datetime import datetime
from pathlib import Path
import streamlit as st
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from core.config import PROJECT_ROOT, api_key, db_path, settings, traces_dir # noqa: E402
from core.hosted_runtime import ( # noqa: E402
acquire_live_slot,
cleanup_expired_sessions,
get_workspace,
is_hosted,
live_slot_status,
new_session_id,
release_live_slot,
)
st.set_page_config(page_title="AutoSource", page_icon="๐ค", layout="wide")
# ---------------------------------------------------------------- css / anims
# All custom components use EXPLICIT text colors on light backgrounds so they
# stay readable in both Streamlit themes (dark theme made inherited text white).
st.markdown("""
""", unsafe_allow_html=True)
# ---------------------------------------------------------------- data access
HOSTED = is_hosted()
BUNDLED_TRACES = traces_dir()
if HOSTED:
if "runtime_session_id" not in st.session_state:
st.session_state.runtime_session_id = new_session_id()
cleanup_expired_sessions()
WORKSPACE = get_workspace(st.session_state.runtime_session_id)
LIVE_TRACES = WORKSPACE.traces
LIVE_DB = WORKSPACE.db
CHILD_ENV = WORKSPACE.child_env()
else:
WORKSPACE = None
LIVE_TRACES = BUNDLED_TRACES
LIVE_DB = db_path()
CHILD_ENV = os.environ.copy()
def reset_warehouse() -> subprocess.CompletedProcess:
"""Seed this visitor's warehouse without affecting any other session."""
return subprocess.run(
[sys.executable, "-m", "scripts.init_db"], cwd=str(PROJECT_ROOT),
env=CHILD_ENV, capture_output=True, text=True, timeout=30,
)
if HOSTED and not LIVE_DB.exists():
init_result = reset_warehouse()
if init_result.returncode != 0:
st.error("Could not initialize your private demo warehouse. Please reload.\n\n"
+ (init_result.stderr or init_result.stdout)[-1200:])
st.stop()
def list_runs() -> list[str]:
paths = list(LIVE_TRACES.glob("*.jsonl"))
if LIVE_TRACES != BUNDLED_TRACES:
paths.extend(BUNDLED_TRACES.glob("*.jsonl"))
by_id = {p.stem: p for p in paths
if not p.stem.startswith(".") and p.stem != "llm_calls"}
return sorted(by_id, key=lambda r: by_id[r].stat().st_mtime, reverse=True)
def trace_root_for_run(run_id: str) -> Path:
if (LIVE_TRACES / f"{run_id}.jsonl").exists():
return LIVE_TRACES
return BUNDLED_TRACES
def load_events(run_id: str) -> list[dict]:
path = trace_root_for_run(run_id) / f"{run_id}.jsonl"
if not path.exists():
return []
events = []
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if line:
try:
events.append(json.loads(line))
except json.JSONDecodeError:
pass # tail of a line still being written
return events
def load_mandates(run_id: str) -> dict | None:
path = trace_root_for_run(run_id) / f"{run_id}.mandates.json"
if path.exists():
try:
return json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return None
return None
def write_decision(run_id: str, payload: dict) -> None:
if not (LIVE_TRACES / f"{run_id}.jsonl").exists():
raise PermissionError("recorded replay runs are read-only")
tmp = LIVE_TRACES / f".{run_id}.decision.tmp"
tmp.write_text(json.dumps(payload), encoding="utf-8")
tmp.replace(LIVE_TRACES / f"{run_id}.decision.json")
def start_live_run(sku: str | None) -> str:
"""Launch scripts.run_demo as a detached subprocess; returns its run_id."""
session_tag = (st.session_state.runtime_session_id[:8] if HOSTED else "local")
run_id = ("live-" + session_tag + "-" + datetime.now().strftime("%Y%m%d-%H%M%S")
+ "-" + uuid.uuid4().hex[:4])
args = [sys.executable, "-u", "-m", "scripts.run_demo", "--run-id", run_id]
if sku:
args += ["--sku", sku]
token = None
if HOSTED:
token = acquire_live_slot(st.session_state.runtime_session_id, run_id)
if token is None:
raise RuntimeError("another visitor is currently negotiating")
args += ["--live-lock-token", token]
console = open(LIVE_TRACES / f"{run_id}.console.log", "w", encoding="utf-8")
try:
subprocess.Popen(args, cwd=str(PROJECT_ROOT), env=CHILD_ENV,
stdout=console, stderr=subprocess.STDOUT)
console.close()
except Exception:
console.close()
if token:
release_live_slot(token)
raise
return run_id
def pending_gate_exists() -> bool:
for path in LIVE_TRACES.glob("*.mandates.json"):
try:
if json.loads(path.read_text(encoding="utf-8"))["context"]["status"] == "pending":
return True
except (KeyError, json.JSONDecodeError, OSError):
continue
return False
# ---------------------------------------------------------------- sidebar
st.sidebar.title("๐ค AutoSource")
st.sidebar.caption("Autonomous procurement & negotiation network")
qp = st.query_params
runs = list_runs()
default_mode = 0 if qp.get("mode", "live") == "live" else 1
mode = st.sidebar.radio("Mode", ["Live follow", "Replay"], index=default_mode)
requested_run = qp.get("run")
if requested_run and runs and requested_run not in runs:
if HOSTED:
own_prefix = f"live-{st.session_state.runtime_session_id[:8]}-"
pending_console = LIVE_TRACES / f"{requested_run}.console.log"
if requested_run.startswith(own_prefix) and pending_console.exists():
st.sidebar.info(f"โณ starting {requested_run} โฆ")
time.sleep(2)
st.rerun()
else:
st.sidebar.warning("That run belongs to an expired or different "
"browser session. Visitor runs are private.")
del st.query_params["run"]
st.rerun()
else:
st.sidebar.info(f"โณ waiting for {requested_run} โฆ")
if st.sidebar.button("Cancel waiting"):
del st.query_params["run"]
st.rerun()
time.sleep(2)
st.rerun()
if not runs:
st.info("No runs recorded yet. Start one below or run: "
"`python -m scripts.run_demo`")
run_id = None
else:
default_run = requested_run if requested_run in runs else runs[0]
run_id = st.sidebar.selectbox("Run", runs, index=runs.index(default_run))
# ---- live demo controls ---------------------------------------------------
st.sidebar.divider()
st.sidebar.subheader("๐ฎ Live demo")
if HOSTED:
st.sidebar.caption("Private visitor session ยท "
f"`{st.session_state.runtime_session_id[:8]}`")
if not api_key("GROQ_API_KEY"):
st.sidebar.warning("Live mode is temporarily unavailable because the Space "
"owner has not configured `GROQ_API_KEY`. Replay still works.")
else:
sku_options = (["SEED-MAIZE-01", "FERT-NPK-01", "(auto: first low-stock item)"]
if HOSTED else
["(auto: first low-stock item)", "SEED-MAIZE-01", "FERT-NPK-01"])
sku_choice = st.sidebar.selectbox("Item to restock", sku_options)
if HOSTED and sku_choice == "FERT-NPK-01":
st.sidebar.caption("Stress-test item: vendor floors exceed its budget, so "
"a safe walk-away is the expected result.")
hosted_cfg = settings().get("hosted", {})
launches = int(st.session_state.get("live_launches", 0))
max_launches = int(hosted_cfg.get("max_runs_per_session", 3))
last_launch = float(st.session_state.get("last_live_launch", 0))
cooldown_left = max(0, int(hosted_cfg.get("cooldown_s", 120)
- (time.time() - last_launch)))
slot = live_slot_status() if HOSTED else None
at_limit = HOSTED and launches >= max_launches
has_pending_gate = pending_gate_exists()
disabled_reason = None
if has_pending_gate:
disabled_reason = "Approve or reject your pending mandate before another run."
elif slot:
disabled_reason = "Another live negotiation is using the shared free-tier slot."
elif cooldown_left > 0:
disabled_reason = f"This session can launch again in {cooldown_left}s."
elif at_limit:
disabled_reason = (f"This browser session used its {max_launches} live runs. "
"Replay remains available.")
if disabled_reason:
st.sidebar.info(disabled_reason)
if st.sidebar.button("๐ Trigger stock check & negotiate", type="primary",
use_container_width=True, disabled=bool(disabled_reason)):
sku = None if sku_choice.startswith("(") else sku_choice
try:
new_run = start_live_run(sku)
except Exception as exc: # noqa: BLE001
st.sidebar.error(f"Could not start: {str(exc)[:240]}")
else:
st.session_state.live_launches = launches + 1
st.session_state.last_live_launch = time.time()
st.query_params["run"] = new_run
st.query_params["mode"] = "live"
st.sidebar.success(f"started {new_run}")
time.sleep(2)
st.rerun()
reset_blocked = bool(slot) or has_pending_gate
if st.sidebar.button("๐ Reset warehouse (seed data)",
use_container_width=True, disabled=reset_blocked):
out = reset_warehouse()
st.sidebar.success("warehouse reset" if out.returncode == 0
else f"failed: {out.stdout}{out.stderr}")
if reset_blocked:
st.sidebar.caption("Finish the active negotiation/approval before resetting.")
elif HOSTED:
st.sidebar.caption(f"{launches}/{max_launches} live runs used in this session. "
"Data expires automatically after 6 hours.")
else:
st.sidebar.caption("Live runs negotiate on real free-tier APIs โ see "
"docs/LIVE_DEMO.md")
events_all = load_events(run_id) if run_id else []
TRACES = trace_root_for_run(run_id) if run_id else LIVE_TRACES
run_is_owned = bool(run_id and (LIVE_TRACES / f"{run_id}.jsonl").exists())
if mode == "Replay" and run_id:
st.sidebar.divider()
speed = st.sidebar.slider("Replay speed (events/tick)", 1, 10, 2)
if "replay_idx" not in st.session_state or st.session_state.get("replay_run") != run_id:
st.session_state.replay_idx = 0
st.session_state.replay_run = run_id
st.session_state.playing = False
c1, c2, c3 = st.sidebar.columns(3)
if c1.button("โถ Play"):
st.session_state.playing = True
if c2.button("โธ Pause"):
st.session_state.playing = False
if c3.button("โฎ Reset"):
st.session_state.replay_idx = 0
st.session_state.playing = False
st.session_state.replay_idx = st.sidebar.slider(
"Position", 0, max(len(events_all), 1), st.session_state.replay_idx)
events = events_all[:st.session_state.replay_idx]
else:
events = events_all
if run_id:
st.sidebar.caption(f"{len(events)} events ยท auto-refreshing")
if run_id is None:
st.stop()
# ---------------------------------------------------------------- derive state
AGENTS = [("auditor", "๐ Auditor"), ("sourcing", "๐ Sourcing"),
("buyer", "๐ค Buyer"), ("vendor", "๐ช Vendors"),
("settlement", "โ๏ธ Settlement")]
def agent_key(actor: str) -> str:
return "vendor" if actor.startswith("vendor") else actor
last_status: dict[str, str] = {}
for e in events:
k = agent_key(e["actor"])
if e["type"] == "STATE_UPDATE" and "status" in e["data"]:
last_status[k] = str(e["data"]["status"]).replace("_", " ")
elif e["type"] == "TEXT_MESSAGE":
last_status[k] = "speakingโฆ"
active = agent_key(events[-1]["actor"]) if events else None
pipeline_running = bool(events) and not any(
e["type"] == "STATE_UPDATE" and e["data"].get("status") in
("complete", "rejected", "pipeline_done", "approval_timeout")
for e in events[-6:])
meta = {"budget": None, "best_ask": None, "round": None, "vendor": None}
deals, winner, requests_seen, active_sku, transcript_refs = [], None, [], None, []
for e in events:
d = e["data"]
if e["type"] == "STATE_UPDATE":
if "buyer_offer" in d:
meta["round"], meta["best_ask"] = d.get("round"), d.get("vendor_ask")
meta["vendor"] = d.get("vendor")
if "winner" in d and d.get("status") == "pipeline_done":
winner = d["winner"]
if "shortlist" in d:
meta["budget"] = d["shortlist"].get("max_unit_budget")
active_sku = d["shortlist"].get("sku")
if d.get("status") == "negotiation_done":
deals.append(d)
if d.get("transcript_ref"):
transcript_refs.append(d["transcript_ref"])
if e["type"] == "TEXT_MESSAGE" and "request" in d:
requests_seen.append(d["request"])
request = next((r for r in requests_seen if r["sku"] == active_sku),
requests_seen[0] if requests_seen else None)
# ---------------------------------------------------------------- tabs
tab_main, tab_tech = st.tabs(["๐ Mission control", "๐ฌ Under the hood"])
with tab_main:
cols = st.columns(len(AGENTS))
for col, (key, label) in zip(cols, AGENTS):
is_active = key == active and (mode == "Replay" or pipeline_running)
col.markdown(
f"""
{label}
{last_status.get(key, 'idle')}
""",
unsafe_allow_html=True)
st.divider()
m1, m2, m3, m4, m5 = st.columns(5)
list_price = request["list_price"] if request else None
m1.metric("Item", request["sku"] if request else "โ")
m2.metric("List price", f"{list_price}" if list_price else "โ")
m3.metric("Hidden budget ๐คซ", f"{request['max_unit_budget']}" if request else "โ")
if winner and winner.get("agreed_unit_price") and list_price:
saved = round((list_price - winner["agreed_unit_price"]) / list_price * 100, 1)
m4.metric("Winning price", winner["agreed_unit_price"],
delta=f"-{saved}% vs list", delta_color="inverse")
else:
m4.metric("Current ask", meta["best_ask"] or "โ")
m5.metric("Deals settled",
f"{sum(1 for d in deals if d.get('settled'))}/{len(deals)}"
if deals else "โ")
left, right = st.columns([3, 2])
with left:
st.subheader("Negotiation timeline")
box = st.container(height=520)
with box:
for e in events:
d, actor = e["data"], e["actor"]
if e["type"] == "TEXT_MESSAGE":
cls = ("buyer" if actor == "buyer" else
"vendor" if actor.startswith("vendor") else
"human" if actor == "human" else "system")
price = f'@{d["price"]}' \
if d.get("price") is not None else ""
flags = "".join(
f'๐ก {html.escape(str(f))}'
for f in d.get("flags", [])
if f.startswith(("BLOCKED", "clamped", "non_monotonic",
"premature")))
who = html.escape(actor.upper())
if d.get("round"):
who += f" ยท R{d['round']}"
st.markdown(
f'{who}
'
f'{html.escape(str(d["text"]))}{price}{flags}
๐ Winner: {winner["vendor_id"]} @ '
f'{winner["agreed_unit_price"]}/unit ยท total '
f'{winner["landed_cost"]}
', unsafe_allow_html=True)
mandates = load_mandates(run_id)
if mandates:
ctx = mandates["context"]
st.subheader("AP2 mandate")
if ctx["status"] == "pending":
st.warning(
f"**Approval required** โ {ctx['vendor_name']} ยท "
f"{mandates['cart']['qty']} ร {mandates['cart']['sku']} @ "
f"{mandates['cart']['unit_price']} = "
f"**{mandates['cart']['total']}** "
f"(saves {ctx['savings_pct']}% vs list)")
with st.expander("Intent ยท Cart ยท Payment (AP2 trio)"):
st.json(mandates["intent"])
st.json(mandates["cart"])
st.json(mandates["payment"])
if run_is_owned:
b1, b2 = st.columns(2)
if b1.button("โ
Approve", type="primary", use_container_width=True):
write_decision(run_id, {"decision": "approve"})
st.toast("Approved โ signing cart mandate")
time.sleep(1)
st.rerun()
if b2.button("โ Reject", use_container_width=True):
write_decision(run_id, {"decision": "reject"})
st.toast("Rejected")
time.sleep(1)
st.rerun()
new_price = st.number_input(
"โฆor change max unit price",
value=float(mandates["intent"]["max_unit_price"]),
step=0.05, format="%.2f")
if st.button("๐ฐ Change budget & re-evaluate",
use_container_width=True):
write_decision(run_id, {"decision": "change_budget",
"new_max_unit_price": new_price})
st.rerun()
else:
st.info("This is a read-only recorded run. Start a new live run "
"to make an approval decision.")
elif ctx["status"] == "approved":
po = ctx.get("po", {})
st.success(f"โ
PO **{po.get('po_id', 'โ')}** written via MCP ยท "
f"total {po.get('total', 'โ')} ยท MOCK payment "
f"authorized. Saved {ctx['savings_pct']}% vs list price.")
st.balloons()
elif ctx["status"] == "rejected":
st.error("โ Purchase rejected by the human. Nothing was written.")
# ---------------------------------------------------------------- tech tab
with tab_tech:
sub_arch, sub_llm, sub_console, sub_events, sub_db, sub_health = st.tabs(
["๐งญ Architecture & prompts", "๐ง LLM calls", "๐ฅ Console & files",
"๐ก Events & transcripts", "๐ Warehouse DB", "๐ฉบ Health & config"])
# ---------------- architecture + prompts ----------------
with sub_arch:
st.markdown("### How AutoSource fits together")
st.markdown("""
| Stage | Module | What it contributes |
|---|---|---|
| ๐ Warehouse | `mcp_server/server.py` | FastMCP server โ the **only** code that touches SQLite. 5 tools; `cost_floor` never leaves it |
| ๐ Detect | `agents/auditor/auditor.py` | Deterministic shortage scan over MCP โ `ProcurementRequest` (budget = list ร 0.9) |
| ๐ก Hand-off | `core/a2a_layer.py` | A2A protocol: Agent Cards + JSON-RPC `message/send` over HTTP (official a2a-sdk types) |
| ๐ Rank | `agents/sourcing/` | Deterministic vendor score + Gemini re-rank + `vendor_memory` learning |
| ๐ค Negotiate | `core/negotiation.py` | The referee: state machine, **hard validators** (budget/floor/monotonicity), transcripts |
| ๐ฃ Speak | `agents/buyer/` ยท `agents/vendor/` | LLM negotiators with **hidden thresholds** in system prompts (below) |
| โ๏ธ Settle | `agents/settlement/` | Deterministic winner (landed cost โ reliability โ lead time) + memory write-back |
| ๐ Gate | `protocols/ap2/` + `orchestrator/gate.py` | IntentโCartโPayment mandates, hash-bound; human approve โ verify โ PO |
| ๐ฆ Router | `core/router.py` | Token-bucket RPM limit, backoff on 429, Groq key rotation, provider failover |
| ๐ Prove | `eval/` | 20+ scenario harness, savings chart, LLM-judge; asserts 0 violations |
""")
st.markdown("### Agent system prompts (the real ones)")
st.caption("Live imports from the agent source โ what the models actually "
"see. The CONFIDENTIAL lines are the hidden thresholds; "
"deterministic validators enforce them even if a model ignores "
"its prompt.")
try:
from agents.buyer.buyer import SYSTEM_TMPL as BUYER_TMPL
from agents.vendor.vendor import PERSONA_TACTICS
from agents.vendor.vendor import SYSTEM_TMPL as VENDOR_TMPL
with st.expander("๐ค Buyer โ 'shrewd but fair procurement officer'"):
st.code(BUYER_TMPL, language="text")
with st.expander("๐ช Vendor โ persona-driven seller"):
st.code(VENDOR_TMPL, language="text")
st.markdown("**Persona tactics injected per vendor:**")
for persona, tactics in PERSONA_TACTICS.items():
st.markdown(f"- **{persona}** โ {tactics}")
except ImportError:
st.info("Agent sources not bundled in this deployment.")
try:
from agents.sourcing.ranking import RERANK_SYSTEM, WEIGHTS
with st.expander("๐ Sourcing โ deterministic score + LLM re-rank"):
st.markdown(f"**Score weights (code):** `{WEIGHTS}`")
st.code(RERANK_SYSTEM, language="text")
except ImportError:
pass
try:
from eval.judge import SYSTEM as JUDGE_SYSTEM
with st.expander("โ๏ธ LLM-judge โ transcript auditor"):
st.code(JUDGE_SYSTEM, language="text")
except ImportError:
pass
# ---------------- LLM call log ----------------
with sub_llm:
st.markdown("### Every LLM call the router made")
st.caption("Full system prompt, user message, and raw response per call โ "
"logged by `core/router.py` to `traces/llm_calls.jsonl`. "
"Failures (429s, failovers) appear in red.")
call_log = TRACES / "llm_calls.jsonl"
calls: list[dict] = []
if call_log.exists():
for line in call_log.read_text(encoding="utf-8").splitlines():
try:
calls.append(json.loads(line))
except json.JSONDecodeError:
pass
# scope to this run's time window when possible
if events and calls:
t_start, t_end = events[0]["ts"] - 5, events[-1]["ts"] + 90
run_calls = [c for c in calls if t_start <= c.get("ts", 0) <= t_end]
else:
run_calls = calls
c1, c2, c3 = st.columns(3)
scope = c1.radio("Scope", ["This run's window", "All calls"], horizontal=True)
shown_calls = run_calls if scope == "This run's window" else calls
roles = sorted({c.get("role", "?") for c in shown_calls})
role_pick = c2.multiselect("Agent role", roles, default=roles)
only_fail = c3.checkbox("Failures only")
shown_calls = [c for c in shown_calls if c.get("role") in role_pick
and (not only_fail or not c.get("ok", True))]
st.caption(f"{len(shown_calls)} calls shown (log has {len(calls)} total)")
for i, c in enumerate(reversed(shown_calls[-80:])):
ok = c.get("ok", True)
ts = datetime.fromtimestamp(c["ts"]).strftime("%H:%M:%S") if c.get("ts") else "โ"
head = (f'{"โ
" if ok else "โ"} {ts} ยท {c.get("role")} โ '
f'{c.get("provider")} ยท {c.get("latency_ms", "?")} ms'
+ ("" if ok else " ยท FAILED"))
with st.expander(head):
st.markdown("**System prompt**")
st.code(c.get("system", ""), language="text")
st.markdown("**User message**")
st.code(json.dumps(c.get("messages", []), indent=2,
ensure_ascii=False), language="json")
st.markdown("**Response**" if ok else "**Error**")
st.code(c.get("response", c.get("error", "")), language="text")
# ---------------- console + file browser ----------------
with sub_console:
st.markdown("### Run console (subprocess stdout/stderr)")
console_path = TRACES / f"{run_id}.console.log"
if console_path.exists():
lines = console_path.read_text(encoding="utf-8",
errors="replace").splitlines()
body = "\n".join(lines[-120:]) or "(empty)"
st.markdown(f'{html.escape(body)}
',
unsafe_allow_html=True)
st.caption(f"last 120 of {len(lines)} lines ยท traces/{run_id}.console.log")
else:
st.info("No console capture for this run โ console logs exist only for "
"runs started from the ๐ฎ Live demo button. Terminal-started "
"runs print to your terminal.")
st.markdown("### Trace file browser")
files = sorted(TRACES.rglob("*"), key=lambda p: p.stat().st_mtime,
reverse=True)
files = [f for f in files if f.is_file() and not f.name.startswith(".")]
names = [str(f.relative_to(TRACES)) for f in files[:60]]
pick = st.selectbox("File", names) if names else None
if pick:
fpath = TRACES / pick
size_kb = fpath.stat().st_size / 1024
st.caption(f"{size_kb:.1f} KB ยท modified "
f"{datetime.fromtimestamp(fpath.stat().st_mtime):%H:%M:%S}")
raw = fpath.read_text(encoding="utf-8", errors="replace")
if fpath.suffix == ".json":
try:
st.json(json.loads(raw))
except json.JSONDecodeError:
st.code(raw[-8000:], language="text")
else:
tail = "\n".join(raw.splitlines()[-150:])
st.markdown(f'{html.escape(tail)}
',
unsafe_allow_html=True)
# ---------------- events + transcripts ----------------
with sub_events:
st.markdown("### Negotiation transcripts (parsed)")
st.caption("Each turn stores the raw message AND the parsed price, so "
"fabricated numbers can be caught. ๐ก flags = validator actions.")
if transcript_refs:
for ref in transcript_refs:
ref_path = Path(ref)
tpath = (ref_path if ref_path.is_absolute()
else TRACES.parent / ref_path)
if not tpath.exists():
tpath = TRACES / "negotiations" / ref_path.name
if not tpath.exists():
continue
t = json.loads(tpath.read_text(encoding="utf-8"))
m, res = t["meta"], t["result"]
label = (f"{m['vendor_id']} โ {res['outcome']} "
f"{'@ ' + str(res['agreed_unit_price']) if res['agreed_unit_price'] else ''} "
f"(budget {m['budget']} ยท floor {m['floor']} ๐คซ)")
with st.expander(label):
st.dataframe(
[{"actor": turn["actor"], "action": turn["action"],
"price": turn["price"], "message": turn["raw"],
"๐ก flags": ", ".join(turn["flags"])}
for turn in t["turns"]],
use_container_width=True, hide_index=True)
else:
st.caption("No transcripts in this run yet.")
st.markdown("### Raw AG-UI event stream")
type_filter = st.multiselect(
"Event types",
["TEXT_MESSAGE", "STATE_UPDATE", "TOOL_CALL", "UI_COMPONENT"],
default=["TOOL_CALL", "UI_COMPONENT"])
shown = [e for e in events if e["type"] in type_filter]
st.caption(f"{len(shown)} of {len(events)} events (traces/{run_id}.jsonl)")
for e in shown[-60:]:
with st.expander(
f'{e["type"]} ยท {e["actor"]} ยท {e.get("iso_ts", "")[11:19]}'):
st.json(e["data"])
st.markdown("### Router telemetry (from this run)")
stats_events = [e["data"]["router_stats"] for e in events
if e["type"] == "STATE_UPDATE"
and "router_stats" in e["data"]]
if stats_events:
s = stats_events[-1]
r1, r2, r3, r4 = st.columns(4)
r1.metric("LLM calls", s.get("calls", 0))
r2.metric("Tokens", f'{s.get("prompt_tokens", 0) + s.get("completion_tokens", 0):,}')
r3.metric("Failovers", s.get("failovers", 0))
r4.metric("Providers", ", ".join(s.get("by_provider", {}).keys()) or "โ")
if s.get("last_errors"):
st.error("Last provider errors: "
+ json.dumps(s["last_errors"], indent=2))
else:
st.caption("No router telemetry events in this run "
"(recorded before this feature, or no LLM calls yet).")
# ---------------- warehouse DB inspector ----------------
with sub_db:
st.markdown("### Warehouse DB (read-only debug view)")
st.caption("โ Debug panel only. Agents NEVER touch this DB directly โ "
"they go through the MCP server's five tools. This inspector "
"reads the same SQLite file so you can verify what the tools did.")
try:
import sqlite3
if LIVE_DB.exists():
conn = sqlite3.connect(f"file:{LIVE_DB}?mode=ro", uri=True)
for table in ("inventory", "purchase_orders", "vendor_memory",
"suppliers"):
with st.expander(f"table: {table}",
expanded=(table == "purchase_orders")):
import pandas as pd
df = pd.read_sql_query(f"SELECT * FROM {table}", conn) # noqa: S608 โ fixed table names
if table == "suppliers":
df = df.drop(columns=["cost_floor"])
st.caption("cost_floor hidden โ same rule as the MCP "
"server: buyers must never see it.")
st.dataframe(df, use_container_width=True, hide_index=True)
conn.close()
else:
st.info("Your private warehouse has not been initialized yet.")
except Exception as e: # noqa: BLE001
st.warning(f"DB inspector unavailable: {e}")
# ---------------- health & config ----------------
with sub_health:
st.markdown("### API keys (presence only โ values never shown)")
for key_name in ("GROQ_API_KEY", "GROQ_API_KEY_2", "GEMINI_API_KEY",
"OPENROUTER_API_KEY", "HF_TOKEN"):
st.markdown(f"- `{key_name}`: "
+ ("โ
set" if api_key(key_name) else "โ missing"))
if api_key("GROQ_API_KEY") and not HOSTED:
st.markdown("### Provider health check")
st.caption("Sends one tiny live request per provider.")
if st.button("๐ฉบ Test providers now"):
from core.router import _PROVIDER_CLASSES
for pname, cls in _PROVIDER_CLASSES.items():
try:
provider = cls()
provider.chat("Reply with JSON only.",
[{"role": "user",
"content": 'reply with {"ok": true}'}],
0.1, True)
st.success(f"{pname}: OK")
except Exception as e: # noqa: BLE001
st.error(f"{pname}: {str(e)[:220]}")
elif HOSTED:
st.caption("Live provider pings are disabled on the public Space to "
"protect the shared free-tier quota. Pipeline calls still "
"report provider health in Router telemetry.")
st.markdown("### Runtime configuration (config/settings.yaml)")
cfg_path = PROJECT_ROOT / "config" / "settings.yaml"
if cfg_path.exists():
st.code(cfg_path.read_text(encoding="utf-8"), language="yaml")
# ---------------------------------------------------------------- refresh loop
if mode == "Live follow" and pipeline_running:
time.sleep(2)
st.rerun()
elif mode == "Replay" and st.session_state.get("playing"):
if st.session_state.replay_idx < len(events_all):
st.session_state.replay_idx = min(len(events_all),
st.session_state.replay_idx + speed)
time.sleep(0.5)
st.rerun()
else:
st.session_state.playing = False