feat(monitoring): bot observability dashboards — Grafana + Persian RTL ERP page

Grafana (full use of the existing monitoring/ stack, per OSS policy): - datasources.yml: BotPostgres (uid bot-postgres) over the shared network; grafana service joins 'shared' + carries the datasource env - db_init.py: idempotent read-only role obs_reader (USAGE on schema public + SELECT on bot_traces/bot_trace_events; skipped when OBS_READER_PASSWORD empty; ALTER ROLE for rotation) + composite index ix_bot_trace_events_stage_time (stage, created_at DESC) — senior-engineer review applied with the column-name fix (event_type -> stage); (trace_id) already covered by the existing (trace_id, created_at) index - dashboards/bot-observability.json: KPI row (real traffic, events, Jev/ fallback errors, Jev p95, tokens, commits, archive guard), events/day, per-stage p50/p95, mention statuses (arbiter+fallback aggregates + legacy rows), DeepSeek arbiter actions, tokens/day, model-drift vs jev-1.13.0, user-choice A/B, STT + audio.enhance health, archive size guard - dashboards/bot-trace-explorer.json: trace table + full event drill-down via the trace_id variable ERP page (admin-only, Persian RTL, recharts@^3 for React 19): - backend bot_metrics.py: read-only require_role('admin') endpoints (summary/routing/quality/jev/stt/traces + trace detail); aggregates count real traffic only (calibration/demo/chat_id=0 excluded) - frontend BotMonitorPage at /bot-monitor (KPI cards, daily events, status donut, arbiter actions, user-choice A/B, STT/enhance tables, stage latency, tokens chart, traces table + detail drawer with payload viewer); entry link from the admin panel; MobileTabBar hidden there - api.ts: typed botMetrics client section Tooling/docs: - tools/seed_observability_demo.py: deterministic demo traces (source='demo', every current pipeline stage incl. audio.enhance/jev.failure/arbitration/ fallback/user choices) — excluded from real-traffic aggregates - monitoring/README.md + bot/docs/DEPLOY-AND-MONITORING.md: setup, seed and rollback notes; alerts deliberately deferred (owner decision 2026-10-24) Verified locally: obs_reader SELECT ok / INSERT denied; panels' SQL run as obs_reader with data; datasource health OK; both dashboards survive a grafana restart; endpoints 200 as admin and 403 without role; tsc + vite build clean; page renders in-browser (admin session); bot startup provisions role+index.
2026-10-24 15:09:01 +03:30
parent 787824a19e
commit 0adadd2cf8
19 changed files with 1912 additions and 11 deletions
@@ -0,0 +1,207 @@
+"""Seed DEMO observability traces so every dashboard panel has data (phase 5).
+
+    docker compose exec bot python tools/seed_observability_demo.py [--days 7]
+
+- Writes traces with source='demo' (excluded from real-traffic aggregates in
+  monitor.py / bot-metrics / Grafana by default; visible via the ERP page's
+  «نمایش ترافیک آزمایشی» checkbox or Grafana queries without the filter).
+- Covers every current pipeline stage (787824a): input.text/voice,
+  audio.enhance.result, stt.run/result, nlu.request/result, entities.extracted,
+  jev.request/result (+ one jev.failure), routing.decision (jev_scored),
+  matching.arbitration.result, matching.fallback.result, ambiguity.sent,
+  ambiguity.user_choice, user_choice.pick, draft.render, draft.commit.
+- Idempotent per run-day: re-running the same day replaces that day's demo
+  traces (delete + re-insert of source='demo' rows only — real data untouched,
+  no-deletion policy applies to real traffic only).
+"""
+import argparse
+import asyncio
+import random
+import sys
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from sqlalchemy import text
+
+from app.db import SessionLocal
+
+rng = random.Random(42)  # deterministic demo data
+
+FIELDS = ["project", "supplier", "material", "brand", "spec", "unit"]
+VARIANTS = {
+    "project": [["منشیانه"], ["دفتر"], ["باق اسپارت"], ["پردیس"]],
+    "supplier": [["مصطفوی", "مستفدی"], ["فولاد مبارکه"], ["راجین"]],
+    "material": [["گچ و خاک"], ["سیمان"], ["میلگرد"]],
+    "brand": [["ذوب آهن"], ["سیمان تهران"]],
+    "spec": [["سایز 14"], ["50 کیلویی"]],
+    "unit": [["کیسه"], ["کیلو"], ["تن"]],
+}
+NAMES = {
+    "project": ["منشانه", "دفتر", "باغ اسپارت", "پردیس کاشان"],
+    "supplier": ["بازرگانی مصطفوی", "نمایندگی فولاد مبارکه", "راجین"],
+    "material": ["گچ و خاک", "سیمان پرتلند", "میلگرد آجدار"],
+    "brand": ["ذوب آهن اصفهان", "سیمان تهران"],
+    "spec": ["سایز ۱۴", "۵۰ کیلویی"],
+    "unit": ["کیسه", "کیلوگرم", "تن"],
+}
+STAGES_F = ["سایز ۱۴", "۵۰ کیلویی"]
+
+
+def _mentions(n: int) -> list[dict]:
+    out = []
+    for field in (FIELDS[:n]):
+        out.append({"field": field, "variants": rng.choice(VARIANTS[field])})
+    return out
+
+
+async def one_trace(s, chat_id: int, t0: datetime, kind: str) -> None:
+    tid = uuid.uuid4()
+    n_ev = 0
+
+    async def ev(stage: str, *, minutes=0.0, ok=True, latency=None, model=None,
+                 field=None, payload=None) -> None:
+        nonlocal n_ev
+        n_ev += 1
+        await s.execute(text("""
+            INSERT INTO bot_trace_events
+              (trace_id, stage, ok, latency_ms, model, field, payload, created_at)
+            VALUES (:t, :stage, :ok, :lat, :model, :field, CAST(:payload AS jsonb), :ts)
+        """), {"t": tid, "stage": stage, "ok": ok, "lat": latency, "model": model,
+               "field": field, "payload": __import__("json").dumps(payload or {}, ensure_ascii=False),
+               "ts": t0 + timedelta(minutes=minutes)})
+
+    await s.execute(text("""
+        INSERT INTO bot_traces (id, chat_id, message_id, kind, source, status, created_at, updated_at)
+        VALUES (:id, :chat, :msg, :kind, 'demo', :status, :ts, :ts)
+    """), {"id": tid, "chat": chat_id, "msg": rng.randrange(1000, 99999), "kind": kind,
+           "status": "committed", "ts": t0})
+
+    m = 0.0
+    # ---- input
+    if kind == "voice":
+        await ev("input.voice", minutes=m, payload={
+            "path": f"data/voices/demo/{tid}.ogg", "size": 21400, "mime": "audio/ogg"})
+        # audio enhancement (new in 787824a)
+        ok_enh = rng.random() < 0.85
+        await ev("audio.enhance.result", minutes=(m := m + 0.1), ok=ok_enh,
+                 latency=rng.randrange(900, 2600), model="audio_enhancer",
+                 payload={"size": 18000, "original_size": 21400} if ok_enh
+                 else {"error": "HTTP 503", "fallback": True})
+        await ev("stt.run", minutes=(m := m + 0.1), payload={"providers": ["whisper", "vosk"]})
+        transcripts = []
+        for p in ("whisper", "vosk"):
+            ok = rng.random() < 0.92
+            t = "بیست و پنج کیسه گچ و خاک از مصطفوی برای منشیانه" if ok else None
+            await ev("stt.result", minutes=(m := m + 0.1), ok=ok, latency=rng.randrange(700, 2100),
+                     model="whisper-1" if p == "whisper" else "vosk-fa",
+                     payload={"provider": p, "text": t, "error": None if ok else "timeout"})
+            if ok:
+                transcripts.append({"name": p, "text": t})
+        payload_text = transcripts[0]["text"] if transcripts else ""
+    else:
+        await ev("input.text", minutes=m, payload={"text": "۲۵ کیسه گچ و خاک از مصطفوی برای منشیانه"})
+        payload_text = "۲۵ کیسه گچ و خاک از مصطفوی برای منشیانه"
+
+    # ---- NLU
+    await ev("nlu.request", minutes=(m := m + 0.05), payload={
+        "prompt_hash": "demo-hash", "prompt_version": "nlu-v2-entity-mentions"})
+    mentions = _mentions(rng.randint(3, 6))
+    await ev("nlu.result", minutes=(m := m + 0.9), latency=rng.randrange(1500, 3600),
+             model="deepseek-chat", payload={
+                 "intent": "register_quote", "consensus_text": payload_text})
+    await ev("entities.extracted", minutes=(m := m + 0.02), payload={"entity_mentions": mentions})
+
+    # ---- Jev scoring + arbitration/fallback (two-stage, 787824a)
+    scored: list = []
+    statuses = {"exact": 0, "ambiguity": 0, "new": 0}
+    actions: dict = {}
+    act_names = ["endorsed_jev", "endorsed_jev", "endorsed_jev",
+                 "demoted_to_ambiguity", "vetoed_to_new"]
+    for i, men in enumerate(mentions):
+        field = men["field"]
+        jev_failed = i == 0 and rng.random() < 0.25
+        if jev_failed:
+            await ev("jev.failure", minutes=(m := m + 0.1), field=field, ok=False,
+                     payload={"error": "jev_unavailable_or_error", "n_mentions": 1})
+            # crisis fallback path
+            status = rng.choice(["exact", "ambiguity", "new"])
+            statuses[status] += 1
+            continue
+        await ev("jev.request", minutes=(m := m + 0.05), field=field,
+                 payload={"n_mentions": 1, "n_candidates": rng.randint(6, 20)})
+        name = rng.choice(NAMES[field])
+        conf = rng.choice([0.55, 0.68, 0.75, 0.86, 0.9, 0.95, 0.99])
+        await ev("jev.result", minutes=(m := m + 0.9), field=field, model="jev-1.13.0",
+                 latency=rng.randrange(600, 1300),
+                 payload={"answers": {"mention_0": {"choice": name, "confidence": conf}},
+                          "usage": {"input_tokens": rng.randint(400, 2600),
+                                    "output_tokens": rng.randint(60, 1300)}})
+        scored.append({"variants": men["variants"], "jev_choice": name, "confidence": conf})
+    if scored:
+        await ev("routing.decision", minutes=(m := m + 0.02), payload={
+            "source": "jev_scored", "mentions": scored})
+        for _ in scored:
+            action = rng.choice(act_names)
+            actions[action] = actions.get(action, 0) + 1
+            statuses["exact" if action == "endorsed_jev"
+                     else "ambiguity" if action == "demoted_to_ambiguity" else "new"] += 1
+        await ev("matching.arbitration.result", minutes=(m := m + 1.1),
+                 latency=rng.randrange(1200, 3200), model="deepseek-chat",
+                 payload={"statuses": statuses, "actions": actions,
+                          "n_items": len(scored), "n_resolved": len(scored),
+                          "usage": {"input_tokens": rng.randint(800, 3000),
+                                    "output_tokens": rng.randint(100, 900)}})
+    else:
+        await ev("matching.fallback.result", minutes=(m := m + 1.2),
+                 latency=rng.randrange(1400, 3400), model="deepseek-chat",
+                 payload={"statuses": statuses, "n_mentions": len(mentions),
+                          "n_resolved": max(len(mentions) - 1, 0),
+                          "usage": {"input_tokens": rng.randint(900, 3200),
+                                    "output_tokens": rng.randint(120, 950)}})
+
+    # ---- ambiguity + user choice (the A/B proxy panels)
+    if statuses["ambiguity"] > 0:
+        await ev("ambiguity.sent", minutes=(m := m + 0.1), payload={
+            "amb_id": "em_demo", "kind": "entity_match",
+            "candidates": [{"id": "x1", "name": "منشانه", "prob": 0.62}]})
+        await ev("ambiguity.user_choice", minutes=(m := m + 2.5), payload={
+            "amb_id": "em_demo", "action": "pick", "arg": "0",
+            "candidates": [{"id": "x1", "name": "منشانه"}]})
+        await ev("user_choice.pick", minutes=(m := m + 0.05), payload={
+            "field": "project", "idx": 0, "choice": "منشانه"})
+
+    # ---- draft lifecycle
+    await ev("draft.render", minutes=(m := m + 0.2), payload={"fresh": True, "missing": []})
+    await ev("draft.commit", minutes=(m := m + 3.0), payload={
+        "quote_id": str(uuid.uuid4()), "items": len(mentions) - 3, "newly_created": [],
+        "only_resolved": False})
+
+
+async def run(days: int) -> None:
+    async with SessionLocal() as s:
+        # idempotent per run-day: wipe only this script's demo rows for those days
+        await s.execute(text("""
+            DELETE FROM bot_traces
+             WHERE source = 'demo'
+               AND created_at >= now() - (:days * interval '1 day')
+        """), {"days": days})
+        base = datetime.now(timezone.utc).replace(hour=10, minute=0, second=0, microsecond=0)
+        n = 0
+        for d in range(days):
+            for _ in range(rng.randint(2, 5)):  # 2-5 traces/day
+                day0 = base - timedelta(days=d)
+                t0 = day0 + timedelta(minutes=rng.randrange(0, 600))
+                await one_trace(s, rng.choice([1461166742, 1839839922]), t0,
+                                rng.choice(["voice", "text"]))
+                n += 1
+        await s.commit()
+        print(f"seeded {n} demo traces over {days} day(s) (source='demo')")
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--days", type=int, default=7)
+    asyncio.run(run(ap.parse_args().days))