feat(observability): phase-1 silent pipeline recorder + phase-0 Jev validation

Phase 0 — Jev validated with the real key (5 Persian scenarios, api.typesafe.ai): stale project name «منشیانه»->«منشانه» conf 0.73 (ambiguity band), STT spelling variants «مصطفوی/مستفدی/مسطفی»->«بازرگانی مصطفوی» conf 0.93, exact «دفتر» 0.88, proper _none rejection for «کاغذ A4» 0.92, and two mentions in ONE request (0.77/0.64). Exact model version to pin: jev-1.13.0. Latency 0.7-1.1s per field request; ~450-650 in / 64-133 out tokens.

Phase 1 — Observability (DB-only, zero extra LLM tokens, nothing is ever deleted):

- new tables bot_traces + bot_trace_events (idempotent DDL in db_init.py; ORM in models.py)

- new app/observability.py: _jsonable (UUID/datetime/Decimal/set/enum/dataclass-safe), trace_start, event, log_resolution, link_draft, latest_trace_for_draft, set_trace_status, span() context manager, save_voice (Path.mkdir parents+exist_ok, day folders)

- capture points wired: input.text; input.voice (+voice persisted under OBS_VOICE_DIR); stt.run + per-provider stt.result (success AND failure, latency, model); nlu.request + nlu.result with prompt_hash/prompt_version; entities.extracted; routing.decision (difflib baseline for the phase-6 A/B); ambiguity.sent; ambiguity.user_choice (buttons + free-text reply path); user_choice.pick; draft.render; draft.edit (wizard-side edits detected by the sync loop); draft.commit (quote/supplier/material/project + trace status/quote id); draft.cancel

- compose: persistent ./bot/data:/srv/bot/data volume; bot/.gitignore: data/ (runtime data never committed, never deleted)

- tests: bot/tools/test_observability.py -> 8/8 pass (jsonable cases, trace lifecycle, JSONB round-trip, draft linking, status update, save_voice day-folder mkdir, span task). Verified live: voice file lands on the host volume and is gitignored; rows + files persist across bot restart.

bot/.gitignore

@@ -2,6 +2,9 @@
 .env
 *.local
 
+# ---- Runtime data (voices etc.) — never commit, never delete ----
+data/
+
 # ---- Python ----
 __pycache__/
 *.pyc

bot/app/config.py

@@ -133,6 +133,12 @@ class Settings(BaseSettings):
     # Drafts older than this are marked expired by the sweeper loop.
     DRAFT_TTL_HOURS: int = 24
 
+    # Observability (phase 1) — silent pipeline recorder (DB only; no external
+    # tools, no extra LLM tokens). Data is NEVER deleted.
+    OBS_ENABLED: bool = True
+    OBS_VOICE_DIR: str = "data/voices"   # persistent volume (./bot/data) — keep forever
+    OBS_PROMPT_VERSION: str = ""         # manual label for the NLU prompt in use
+
     LOG_LEVEL: str = "INFO"
 
     @property

bot/app/db_init.py

@@ -64,6 +64,40 @@ DDL = [
         updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
     )
     """,
+    # ---- Observability (phase 1) — silent pipeline recorder, DB-only -------
+    """
+    CREATE TABLE IF NOT EXISTS bot_traces (
+        id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+        chat_id BIGINT NOT NULL,
+        message_id BIGINT,
+        kind VARCHAR(8) NOT NULL,
+        source VARCHAR(16),
+        draft_id UUID,
+        draft_kind VARCHAR(16),
+        committed_quote_id UUID,
+        status VARCHAR(16) NOT NULL DEFAULT 'open',
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+        updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+    """,
+    "CREATE INDEX IF NOT EXISTS ix_bot_traces_chat ON bot_traces (chat_id, created_at)",
+    "CREATE INDEX IF NOT EXISTS ix_bot_traces_draft ON bot_traces (draft_id)",
+    """
+    CREATE TABLE IF NOT EXISTS bot_trace_events (
+        id BIGSERIAL PRIMARY KEY,
+        trace_id UUID NOT NULL REFERENCES bot_traces(id) ON DELETE CASCADE,
+        stage VARCHAR(32) NOT NULL,
+        ok BOOLEAN NOT NULL DEFAULT TRUE,
+        latency_ms INTEGER,
+        model VARCHAR(64),
+        field VARCHAR(32),
+        payload JSONB NOT NULL DEFAULT '{}'::jsonb,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+    """,
+    "CREATE INDEX IF NOT EXISTS ix_bot_trace_events_trace ON bot_trace_events (trace_id, created_at)",
+    "CREATE INDEX IF NOT EXISTS ix_bot_trace_events_stage ON bot_trace_events (stage)",
+    "CREATE INDEX IF NOT EXISTS ix_bot_trace_events_payload ON bot_trace_events USING GIN (payload jsonb_path_ops)",
 ]
 
 

bot/app/handlers/__init__.py

@@ -11,6 +11,7 @@ Pipeline per update:
 import logging
 
 from app import drafts as draft_store
+from app import observability as obs
 from app.config import settings
 from app.nlu import understand
 from app.runtime import BotContext, LinkedUser
@@ -48,6 +49,16 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
     voice_transcripts: list[dict] | None = None
     message_id = message.get("message_id")
 
+    # Observability (phase 1): open a trace for every incoming message.
+    _doc = message.get("document") or {}
+    _audio_like = bool(
+        message.get("voice") or message.get("audio")
+        or (_doc.get("mime_type") or "").lower().startswith("audio/")
+    )
+    trace_id = await obs.trace_start(chat_id, message_id, "voice" if _audio_like else "text")
+    if text:
+        await obs.event(trace_id, "input.text", payload={"text": text[:500]})
+
     if text.startswith("/"):
         await commands.handle_command(ctx, chat_id, text)
         return
@@ -79,11 +90,20 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
         file_id = audio_media["file_id"]
         try:
             audio_bytes = await ctx.bale.download_file(file_id)
+            voice_path = obs.save_voice(trace_id, message_id or 0, audio_bytes)
+            await obs.event(trace_id, "input.voice", payload={
+                "path": voice_path,
+                "size": len(audio_bytes),
+                "mime": audio_media.get("mime_type"),
+                "file_id": file_id,
+                "duration": audio_media.get("duration"),
+            })
             transcripts = await transcribe_multi(
                 settings,
                 audio_bytes,
                 mime_type=audio_media.get("mime_type"),
                 filename=audio_media.get("file_name"),
+                trace_id=trace_id,
             )
         except Exception as exc:
             detail = "هوش مصنوعی تنظیم نشده" if "not configured" in str(exc) else str(exc)[:200]
@@ -201,9 +221,10 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
 
     draft_summary = _draft_summary(target_draft) if target_draft else None
     input_payload = voice_transcripts if voice_transcripts else text
-    result = await understand(settings, input_payload, draft_summary)
+    result = await understand(settings, input_payload, draft_summary, trace_id=trace_id)
     intent, patch = result.get("intent"), result.get("patch") or {}
     consensus_text = result.get("consensus_text") or text
+    await obs.event(trace_id, "entities.extracted", payload={"patch": patch, "intent": intent})
 
     if source == "voice":
         await ctx._log(chat_id, "in", "voice", consensus_text)
@@ -216,23 +237,27 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
             ctx, chat_id, user, patch, source,
             incoming_msg_id=message_id,
             target_draft=target_draft,
+            trace_id=trace_id,
         )
     elif intent == "commit_draft":
         active = target_draft or await draft_store.get_active(chat_id)
         if active is None:
             await ctx.send(chat_id, NOTHING_TO_COMMIT, reply_to=message_id)
         elif active.kind == draft_store.KIND_QUOTE:
-            await quote.commit(ctx, chat_id, user, active)
+            await quote.commit(ctx, chat_id, user, active, trace_id=trace_id)
         elif active.kind == draft_store.KIND_SUPPLIER:
-            await supplier.commit(ctx, chat_id, user, active)
+            await supplier.commit(ctx, chat_id, user, active, trace_id=trace_id)
         elif active.kind == draft_store.KIND_MATERIAL:
-            await material.commit(ctx, chat_id, user, active)
+            await material.commit(ctx, chat_id, user, active, trace_id=trace_id)
         elif active.kind == draft_store.KIND_PROJECT:
-            await project.commit(ctx, chat_id, user, active)
+            await project.commit(ctx, chat_id, user, active, trace_id=trace_id)
     elif intent == "discard_draft":
         active = target_draft or await draft_store.get_active(chat_id)
         if active:
             await draft_store.set_status(active.id, "cancelled")
+            await obs.event(trace_id, "draft.cancel",
+                            payload={"draft_id": str(active.id), "kind": active.kind})
+            await obs.set_trace_status(trace_id, "cancelled")
             if active.review_message_id:
                 try:
                     await ctx.delete(chat_id, active.review_message_id)
@@ -253,11 +278,11 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
         elif active.kind == draft_store.KIND_PROJECT:
             await project.render(ctx, chat_id, active)
     elif intent == "add_supplier":
-        await supplier.handle_entities(ctx, chat_id, user, patch, source)
+        await supplier.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id)
     elif intent == "add_material":
-        await material.handle_entities(ctx, chat_id, user, patch, source)
+        await material.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id)
     elif intent == "add_project":
-        await project.handle_entities(ctx, chat_id, user, patch, source)
+        await project.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id)
     elif intent == "status_query":
         await queries.handle_query(ctx, chat_id, user, patch)
     elif intent == "accept_quote":
@@ -375,18 +400,24 @@ async def _handle_ambiguity_user_reply(
     draft, amb: dict, message: dict, incoming_msg_id: int | None
 ) -> None:
     """Handle text/audio correction sent directly as a reply to an ambiguity sub-message (Requirement 2.b)."""
+    trace_id = await obs.latest_trace_for_draft(draft.id)
     text = (message.get("text") or message.get("caption") or "").strip()
     if not text and (message.get("voice") or message.get("audio")):
         try:
             audio_media = message.get("voice") or message.get("audio")
             audio_bytes = await ctx.bale.download_file(audio_media["file_id"])
-            text = await transcribe(settings, audio_bytes)
+            obs.save_voice(trace_id, incoming_msg_id or 0, audio_bytes)
+            text = await transcribe(settings, audio_bytes, trace_id=trace_id)
         except Exception:
             text = ""
 
     if text:
+        await obs.event(trace_id, "ambiguity.user_choice", payload={
+            "amb_id": amb.get("id"), "action": "reply", "kind": amb.get("kind"),
+            "reply_text": text[:500],
+        })
         draft_summary = _draft_summary(draft)
-        result = await understand(settings, text, draft_summary)
+        result = await understand(settings, text, draft_summary, trace_id=trace_id)
         patch = result.get("patch") or {}
 
         # Apply user correction specifically to the target field
@@ -408,11 +439,11 @@ async def _handle_ambiguity_user_reply(
             draft_store.merge_quote_patch(draft.payload, patch)
 
         amb["status"] = "applied"
-        await quote.resolve_payload(ctx, draft.payload)
+        await quote.resolve_payload(ctx, draft.payload, trace_id)
         await draft_store.save_payload(draft, mark_bale_synced=True)
 
         # In-place edit main review card (Requirement 2.b)
-        await quote.render(ctx, chat_id, draft, fresh=False)
+        await quote.render(ctx, chat_id, draft, fresh=False, trace_id=trace_id)
 
     # Delete the ambiguity message (Requirement 2.b)
     if amb.get("message_id"):

bot/app/handlers/callbacks.py

@@ -6,6 +6,7 @@ Data grammar (≤64 bytes):  op:{arg}:{d8}  /  pk:{d8}:{field}:{idx}  /  qc:{idx
 import logging
 
 from app import drafts as draft_store
+from app import observability as obs
 from app.runtime import LinkedUser
 from app.texts import CANCELLED, UNLINKED, esc
 
@@ -77,6 +78,12 @@ async def _dispatch(ctx, chat_id: int, data: str, cq: dict) -> None:
         if not target_amb:
             return
 
+        tid = await obs.latest_trace_for_draft(draft.id)
+        await obs.event(tid, "ambiguity.user_choice", payload={
+            "amb_id": amb_id, "action": action, "kind": target_amb.get("kind"),
+            "title": target_amb.get("title"),
+        })
+
         if action == "ok":
             proposed = target_amb.get("proposed_patch") or {}
             draft_store.merge_quote_patch(payload, proposed)
@@ -126,6 +133,9 @@ async def _dispatch(ctx, chat_id: int, data: str, cq: dict) -> None:
             return
         if arg == "no":
             await draft_store.set_status(draft.id, "cancelled")
+            tid = await obs.latest_trace_for_draft(draft.id)
+            await obs.event(tid, "draft.cancel", payload={"draft_id": str(draft.id), "kind": draft.kind})
+            await obs.set_trace_status(tid, "cancelled")
             ctx.pending.pop(chat_id, None)
             if draft.review_message_id:
                 try:
@@ -151,8 +161,10 @@ async def _apply_pick(ctx, chat_id: int, user: LinkedUser, draft, field: str, id
         await ctx.send(chat_id, "گزینه نامعتبر است.")
         return
     choice = options[idx]
+    tid = await obs.latest_trace_for_draft(draft.id)
+    await obs.event(tid, "user_choice.pick", payload={"field": field, "idx": idx, "choice": choice})
     if draft.kind == draft_store.KIND_QUOTE:
-        await quote.apply_pick(ctx, chat_id, user, draft, field, choice)
+        await quote.apply_pick(ctx, chat_id, user, draft, field, choice, trace_id=tid)
     elif draft.kind == draft_store.KIND_SUPPLIER:
         await supplier.apply_pick(ctx, chat_id, user, draft, field, choice)
     elif draft.kind == draft_store.KIND_MATERIAL:

bot/app/handlers/material.py

@@ -2,6 +2,7 @@
 import logging
 
 from app import drafts as draft_store
+from app import observability as obs
 from app.api.client import ApiError
 from app.drafts import (
     KIND_MATERIAL, merge_material_patch, material_missing, new_material_payload,
@@ -13,7 +14,8 @@ from app.texts import ERR_GENERIC, esc, material_committed_report, material_revi
 logger = logging.getLogger("bot.material")
 
 
-async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None) -> None:
+async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None,
+                          trace_id: str | None = None) -> None:
     draft = await draft_store.get_active(chat_id, KIND_MATERIAL)
     if draft is not None:
         payload = draft.payload
@@ -22,15 +24,18 @@ async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None) -
         payload = new_material_payload()
         merge_material_patch(payload, patch)
         draft = await draft_store.create(chat_id, KIND_MATERIAL, payload, source=source)
-    await resolve_payload(ctx, payload)
+    await obs.link_draft(trace_id, draft.id, draft.kind)
+    await resolve_payload(ctx, payload, trace_id)
     await draft_store.save_payload(draft, missing={"fields": material_missing(payload)})
-    await render(ctx, chat_id, draft, fresh=draft.review_message_id is None)
+    await render(ctx, chat_id, draft, fresh=draft.review_message_id is None, trace_id=trace_id)
 
 
-async def resolve_payload(ctx, payload: dict) -> None:
+async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None:
     options = payload.setdefault("options", {})
     if not payload.get("canonical_unit_id") and payload.get("unit_name"):
         res = await safe_resolve(ctx.resolver, "unit", payload["unit_name"])
+        await obs.log_resolution(trace_id, "unit", payload["unit_name"], res.status,
+                                 [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]])
         if res.resolved:
             payload["canonical_unit_id"] = res.candidates[0].id
             payload["unit_name"] = res.candidates[0].name
@@ -58,7 +63,7 @@ def _keyboard(payload: dict, draft_id: str) -> dict:
     return simple_review_keyboard(draft_id, KIND_MATERIAL, rows)
 
 
-async def render(ctx, chat_id, draft, fresh: bool = True) -> None:
+async def render(ctx, chat_id, draft, fresh: bool = True, trace_id: str | None = None) -> None:
     payload = draft.payload
     missing = material_missing(payload)
     text = material_review_text(payload, missing)
@@ -71,11 +76,14 @@ async def render(ctx, chat_id, draft, fresh: bool = True) -> None:
     else:
         await ctx.edit(chat_id, draft.review_message_id, text, reply_markup=keyboard)
         await draft_store.save_payload(draft, missing={"fields": missing})
+    await obs.event(trace_id or await obs.latest_trace_for_draft(draft.id), "draft.render",
+                    payload={"fresh": fresh, "missing": [m.get("key") for m in missing if not m.get("ok")]})
 
 
-async def commit(ctx, chat_id, user, draft) -> None:
+async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None:
     payload = draft.payload
-    await resolve_payload(ctx, payload)
+    trace_id = trace_id or await obs.latest_trace_for_draft(draft.id)
+    await resolve_payload(ctx, payload, trace_id)
     missing = {m["key"]: m["ok"] for m in material_missing(payload)}
     problems = [label for key, label, ok in [
         ("name", "نام کالا", missing.get("name")),
@@ -94,5 +102,8 @@ async def commit(ctx, chat_id, user, draft) -> None:
         await ctx.send(chat_id, ERR_GENERIC.format(detail=esc(exc.message)))
         return
     await draft_store.set_status(draft.id, "committed")
+    await obs.event(trace_id, "draft.commit", payload={"kind": "material",
+                                                       "material_id": (created or {}).get("id")})
+    await obs.set_trace_status(trace_id, "committed")
     ctx.resolver.invalidate("materials")
     await ctx.send(chat_id, material_committed_report(created))

bot/app/handlers/project.py

@@ -2,6 +2,7 @@
 import logging
 
 from app import drafts as draft_store
+from app import observability as obs
 from app.api.client import ApiError
 from app.drafts import (
     KIND_PROJECT, merge_project_patch, new_project_payload, project_missing,
@@ -12,7 +13,8 @@ from app.texts import ERR_GENERIC, esc, project_committed_report, project_review
 logger = logging.getLogger("bot.project")
 
 
-async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None) -> None:
+async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None,
+                          trace_id: str | None = None) -> None:
     draft = await draft_store.get_active(chat_id, KIND_PROJECT)
     if draft is not None:
         payload = draft.payload
@@ -21,15 +23,16 @@ async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None) -
         payload = new_project_payload()
         merge_project_patch(payload, patch)
         draft = await draft_store.create(chat_id, KIND_PROJECT, payload, source=source)
+    await obs.link_draft(trace_id, draft.id, draft.kind)
     await draft_store.save_payload(draft, missing={"fields": project_missing(payload)})
-    await render(ctx, chat_id, draft, fresh=draft.review_message_id is None)
+    await render(ctx, chat_id, draft, fresh=draft.review_message_id is None, trace_id=trace_id)
 
 
 def _keyboard(payload: dict, draft_id: str) -> dict:
     return simple_review_keyboard(draft_id, KIND_PROJECT, [])
 
 
-async def render(ctx, chat_id, draft, fresh: bool = True) -> None:
+async def render(ctx, chat_id, draft, fresh: bool = True, trace_id: str | None = None) -> None:
     payload = draft.payload
     missing = project_missing(payload)
     text = project_review_text(payload, missing)
@@ -42,10 +45,13 @@ async def render(ctx, chat_id, draft, fresh: bool = True) -> None:
     else:
         await ctx.edit(chat_id, draft.review_message_id, text, reply_markup=keyboard)
         await draft_store.save_payload(draft, missing={"fields": missing})
+    await obs.event(trace_id or await obs.latest_trace_for_draft(draft.id), "draft.render",
+                    payload={"fresh": fresh, "missing": [m.get("key") for m in missing if not m.get("ok")]})
 
 
-async def commit(ctx, chat_id, user, draft) -> None:
+async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None:
     payload = draft.payload
+    trace_id = trace_id or await obs.latest_trace_for_draft(draft.id)
     missing = {m["key"]: m["ok"] for m in project_missing(payload)}
     if not missing.get("name"):
         await ctx.send(chat_id, "ثبت ممکن نیست؛ نام پروژه مشخص نشده است.")
@@ -63,5 +69,8 @@ async def commit(ctx, chat_id, user, draft) -> None:
         await ctx.send(chat_id, ERR_GENERIC.format(detail="خطای سرور"))
         return
     await draft_store.set_status(draft.id, "committed")
+    await obs.event(trace_id, "draft.commit", payload={"kind": "project",
+                                                       "project_id": (created or {}).get("id")})
+    await obs.set_trace_status(trace_id, "committed")
     ctx.resolver.invalidate("projects")
     await ctx.send(chat_id, project_committed_report(created))

bot/app/handlers/quote.py

@@ -2,6 +2,7 @@
 import logging
 
 from app import drafts as draft_store
+from app import observability as obs
 from app.api.client import ApiError
 from app.drafts import (
     KIND_QUOTE, build_quote_request, compute_quote_missing, merge_quote_patch,
@@ -20,6 +21,7 @@ async def handle_entities(
     ctx, chat_id, user, patch: dict, source: str | None,
     incoming_msg_id: int | None = None,
     target_draft = None,
+    trace_id: str | None = None,
 ) -> None:
     prev_review_msg_id = None
     if target_draft is not None:
@@ -47,7 +49,8 @@ async def handle_entities(
             chat_id, KIND_QUOTE, payload, source=source, source_message_id=incoming_msg_id
         )
 
-    await resolve_payload(ctx, payload)
+    await obs.link_draft(trace_id, draft.id, draft.kind)
+    await resolve_payload(ctx, payload, trace_id)
 
     if prev_review_msg_id:
         try:
@@ -55,18 +58,25 @@ async def handle_entities(
         except Exception as e:
             logger.warning("Could not delete previous review message %s: %s", prev_review_msg_id, e)
 
-    await render(ctx, chat_id, draft, fresh=True, reply_to=incoming_msg_id)
+    await render(ctx, chat_id, draft, fresh=True, reply_to=incoming_msg_id, trace_id=trace_id)
 
 
 # ---------------------------------------------------------------------------
 # resolution
 # ---------------------------------------------------------------------------
 
-async def resolve_payload(ctx, payload: dict) -> None:
+async def _log_res(trace_id, field: str, mention, res) -> None:
+    """Observability: record one difflib resolution attempt (baseline for phase 6)."""
+    await obs.log_resolution(trace_id, field, mention, res.status,
+                             [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]])
+
+
+async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None:
     options = payload.setdefault("options", {})
 
     if not payload.get("supplier_id") and payload.get("supplier_name"):
         res = await safe_resolve(ctx.resolver, "supplier", payload["supplier_name"])
+        await _log_res(trace_id, "supplier", payload["supplier_name"], res)
         if res.resolved:
             c = res.candidates[0]
             payload["supplier_id"] = c.id
@@ -85,6 +95,7 @@ async def resolve_payload(ctx, payload: dict) -> None:
 
     if not payload.get("project_id") and payload.get("project_name"):
         res = await safe_resolve(ctx.resolver, "project", payload["project_name"])
+        await _log_res(trace_id, "project", payload["project_name"], res)
         if res.resolved:
             c = res.candidates[0]
             payload["project_id"] = c.id
@@ -101,6 +112,7 @@ async def resolve_payload(ctx, payload: dict) -> None:
 
     for p_name in payload.get("project_names") or []:
         res = await safe_resolve(ctx.resolver, "project", p_name)
+        await _log_res(trace_id, "project", p_name, res)
         if res.resolved:
             c = res.candidates[0]
             p_id = c.id
@@ -114,6 +126,7 @@ async def resolve_payload(ctx, payload: dict) -> None:
     for i, item in enumerate(payload.get("items") or [], start=1):
         if not item.get("material_id") and item.get("material_name"):
             res = await safe_resolve(ctx.resolver, "material", item["material_name"])
+            await _log_res(trace_id, "material", item["material_name"], res)
             if res.resolved:
                 _set_material(item, res.candidates[0].id, res.candidates[0].name, res.candidates[0].extra)
                 options.pop(f"mat{i}", None)
@@ -122,7 +135,7 @@ async def resolve_payload(ctx, payload: dict) -> None:
             else:
                 options.pop(f"mat{i}", None)
         if not item.get("unit_id"):
-            await _resolve_unit(ctx, item)
+            await _resolve_unit(ctx, item, trace_id)
 
 
 def _set_material(item: dict, material_id: str, name: str, extra: dict | None = None) -> None:
@@ -130,11 +143,12 @@ def _set_material(item: dict, material_id: str, name: str, extra: dict | None =
     item["material_row"] = {"id": material_id, "canonical_unit_id": (extra or {}).get("canonical_unit_id")}
 
 
-async def _resolve_unit(ctx, item: dict) -> None:
+async def _resolve_unit(ctx, item: dict, trace_id: str | None = None) -> None:
     if item.get("unit_id"):
         return
     if item.get("unit_name"):
         res = await safe_resolve(ctx.resolver, "unit", item["unit_name"])
+        await _log_res(trace_id, "unit", item["unit_name"], res)
         if res.resolved:
             item["unit_id"] = res.candidates[0].id
             item["unit_explicit"] = True
@@ -146,8 +160,10 @@ async def _resolve_unit(ctx, item: dict) -> None:
         item["unit_explicit"] = False
 
 
-async def apply_pick(ctx, chat_id, user, draft, field: str, choice: dict) -> None:
+async def apply_pick(ctx, chat_id, user, draft, field: str, choice: dict,
+                     trace_id: str | None = None) -> None:
     payload = draft.payload
+    trace_id = trace_id or await obs.latest_trace_for_draft(draft.id)
     options = payload.setdefault("options", {})
     options.pop(field, None)
 
@@ -163,7 +179,7 @@ async def apply_pick(ctx, chat_id, user, draft, field: str, choice: dict) -> Non
         if 0 <= index < len(items):
             items[index]["material_name"] = choice["name"]
             _set_material(items[index], choice["id"], choice["name"])
-            await _resolve_unit(ctx, items[index])
+            await _resolve_unit(ctx, items[index], trace_id)
     elif field.startswith("unit"):
         index = int(field[4:]) - 1
         items = payload.get("items") or []
@@ -172,9 +188,9 @@ async def apply_pick(ctx, chat_id, user, draft, field: str, choice: dict) -> Non
             items[index]["unit_explicit"] = True
             items[index]["unit_name"] = choice["name"]
 
-    await resolve_payload(ctx, payload)
+    await resolve_payload(ctx, payload, trace_id)
     await draft_store.save_payload(draft, missing={"fields": compute_quote_missing(payload)})
-    await render(ctx, chat_id, draft, fresh=False)
+    await render(ctx, chat_id, draft, fresh=False, trace_id=trace_id)
 
 
 # ---------------------------------------------------------------------------
@@ -214,7 +230,7 @@ def _keyboard(payload: dict, draft_id: str, wizard_url: str | None = None) -> di
 
 
 async def render(ctx, chat_id, draft, fresh: bool = True, note: str | None = None,
-                 reply_to: int | None = None) -> None:
+                 reply_to: int | None = None, trace_id: str | None = None) -> None:
     from app.config import settings
 
     from app.security import create_draft_token
@@ -263,6 +279,9 @@ async def render(ctx, chat_id, draft, fresh: bool = True, note: str | None = Non
             logger.warning("Could not edit review message %s for draft %s", draft.review_message_id, draft.id)
             await draft_store.set_bale_updated_at(draft.id)
 
+    await obs.event(trace_id or await obs.latest_trace_for_draft(draft.id), "draft.render",
+                    payload={"fresh": fresh, "missing": [m.get("key") for m in missing if not m.get("ok")]})
+
 
 async def send_pending_ambiguities(ctx, chat_id: int, draft) -> None:
     from app.keyboards import ambiguity_keyboard
@@ -275,6 +294,7 @@ async def send_pending_ambiguities(ctx, chat_id: int, draft) -> None:
     payload = draft.payload or {}
     ambiguities = payload.get("ambiguities") or []
     updated = False
+    trace_id = await obs.latest_trace_for_draft(draft.id)
 
     for amb in ambiguities:
         if amb.get("status") == "pending" and not amb.get("message_id"):
@@ -289,6 +309,10 @@ async def send_pending_ambiguities(ctx, chat_id: int, draft) -> None:
             if sent and sent.get("message_id"):
                 amb["message_id"] = sent["message_id"]
                 updated = True
+                await obs.event(trace_id, "ambiguity.sent", payload={
+                    "amb_id": amb.get("id"), "title": amb.get("title"),
+                    "kind": amb.get("kind"), "message_id": sent["message_id"],
+                })
 
     if updated:
         await draft_store.save_payload(draft, mark_bale_synced=True)
@@ -298,9 +322,11 @@ async def send_pending_ambiguities(ctx, chat_id: int, draft) -> None:
 # commit
 # ---------------------------------------------------------------------------
 
-async def commit(ctx, chat_id, user, draft, only_resolved: bool = False) -> None:
+async def commit(ctx, chat_id, user, draft, only_resolved: bool = False,
+                 trace_id: str | None = None) -> None:
     payload = draft.payload
-    await resolve_payload(ctx, payload)
+    trace_id = trace_id or await obs.latest_trace_for_draft(draft.id)
+    await resolve_payload(ctx, payload, trace_id)
 
     newly_created: dict = {}
 
@@ -418,6 +444,13 @@ async def commit(ctx, chat_id, user, draft, only_resolved: bool = False) -> None
     ctx.resolver.invalidate()
 
     quote_id = (created or {}).get("id") or ""
+    await obs.event(trace_id, "draft.commit", payload={
+        "quote_id": quote_id,
+        "items": len(body.get("line_items") or []),
+        "newly_created": newly_created,
+        "only_resolved": only_resolved,
+    })
+    await obs.set_trace_status(trace_id, "committed", quote_id)
     # If a proforma file was attached, persist it to the created ERP quote
     proforma_info = payload.get("proforma")
     if quote_id and proforma_info and proforma_info.get("saved_path"):

bot/app/handlers/supplier.py

@@ -2,6 +2,7 @@
 import logging
 
 from app import drafts as draft_store
+from app import observability as obs
 from app.api.client import ApiError
 from app.drafts import (
     KIND_SUPPLIER, merge_supplier_patch, new_supplier_payload, supplier_missing,
@@ -13,7 +14,8 @@ from app.texts import ERR_GENERIC, esc, supplier_committed_report, supplier_revi
 logger = logging.getLogger("bot.supplier")
 
 
-async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None) -> None:
+async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None,
+                          trace_id: str | None = None) -> None:
     draft = await draft_store.get_active(chat_id, KIND_SUPPLIER)
     if draft is not None:
         payload = draft.payload
@@ -22,15 +24,18 @@ async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None) -
         payload = new_supplier_payload()
         merge_supplier_patch(payload, patch)
         draft = await draft_store.create(chat_id, KIND_SUPPLIER, payload, source=source)
-    await resolve_payload(ctx, payload)
+    await obs.link_draft(trace_id, draft.id, draft.kind)
+    await resolve_payload(ctx, payload, trace_id)
     await draft_store.save_payload(draft, missing={"fields": supplier_missing(payload)})
-    await render(ctx, chat_id, draft, fresh=draft.review_message_id is None)
+    await render(ctx, chat_id, draft, fresh=draft.review_message_id is None, trace_id=trace_id)
 
 
-async def resolve_payload(ctx, payload: dict) -> None:
+async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None:
     options = payload.setdefault("options", {})
     if not payload.get("category_id") and payload.get("category_name"):
         res = await safe_resolve(ctx.resolver, "category", payload["category_name"])
+        await obs.log_resolution(trace_id, "category", payload["category_name"], res.status,
+                                 [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]])
         if res.resolved:
             payload["category_id"] = res.candidates[0].id
             payload["category_name"] = res.candidates[0].name
@@ -58,7 +63,7 @@ def _keyboard(payload: dict, draft_id: str) -> dict:
     return simple_review_keyboard(draft_id, KIND_SUPPLIER, rows)
 
 
-async def render(ctx, chat_id, draft, fresh: bool = True) -> None:
+async def render(ctx, chat_id, draft, fresh: bool = True, trace_id: str | None = None) -> None:
     payload = draft.payload
     missing = supplier_missing(payload)
     text = supplier_review_text(payload, missing)
@@ -71,11 +76,14 @@ async def render(ctx, chat_id, draft, fresh: bool = True) -> None:
     else:
         await ctx.edit(chat_id, draft.review_message_id, text, reply_markup=keyboard)
         await draft_store.save_payload(draft, missing={"fields": missing})
+    await obs.event(trace_id or await obs.latest_trace_for_draft(draft.id), "draft.render",
+                    payload={"fresh": fresh, "missing": [m.get("key") for m in missing if not m.get("ok")]})
 
 
-async def commit(ctx, chat_id, user, draft) -> None:
+async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None:
     payload = draft.payload
-    await resolve_payload(ctx, payload)
+    trace_id = trace_id or await obs.latest_trace_for_draft(draft.id)
+    await resolve_payload(ctx, payload, trace_id)
     missing = {m["key"]: m["ok"] for m in supplier_missing(payload)}
     problems = [label for key, label, ok in [
         ("name", "نام تأمین‌کننده", missing.get("name")),
@@ -99,5 +107,8 @@ async def commit(ctx, chat_id, user, draft) -> None:
         await ctx.send(chat_id, ERR_GENERIC.format(detail=esc(exc.message)))
         return
     await draft_store.set_status(draft.id, "committed")
+    await obs.event(trace_id, "draft.commit", payload={"kind": "supplier",
+                                                       "supplier_id": (created or {}).get("id")})
+    await obs.set_trace_status(trace_id, "committed")
     ctx.resolver.invalidate("suppliers")
     await ctx.send(chat_id, supplier_committed_report(created))

bot/app/models.py

@@ -5,7 +5,7 @@ container and owns its own state (links, drafts, message audit log).
 """
 from datetime import datetime
 
-from sqlalchemy import BigInteger, DateTime, Index, Integer, String, Text, func
+from sqlalchemy import BigInteger, Boolean, DateTime, ForeignKey, Index, Integer, String, Text, func
 from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
@@ -83,3 +83,47 @@ class BotMeta(Base):
     key: Mapped[str] = mapped_column(String(64), primary_key=True)
     value: Mapped[str | None] = mapped_column(Text)
     updated_at: Mapped[datetime] = _timestamptz(onupdate=func.now())
+
+
+class BotTrace(Base):
+    """Observability: one row per incoming interaction (message) — the chain root.
+
+    Events of the whole pipeline hang off it (bot_trace_events). Linked to the
+    draft it produced so later button clicks can be attributed to the same chain.
+    """
+
+    __tablename__ = "bot_traces"
+
+    id: Mapped[str] = _uuid_pk()
+    chat_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    message_id: Mapped[int | None] = mapped_column(BigInteger)
+    kind: Mapped[str] = mapped_column(String(8), nullable=False)  # text | voice
+    source: Mapped[str | None] = mapped_column(String(16))  # bale | wizard | api
+    draft_id: Mapped[str | None] = mapped_column(UUID(as_uuid=False))
+    draft_kind: Mapped[str | None] = mapped_column(String(16))
+    committed_quote_id: Mapped[str | None] = mapped_column(UUID(as_uuid=False))
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="open")  # open|committed|cancelled
+    created_at: Mapped[datetime] = _timestamptz()
+    updated_at: Mapped[datetime] = _timestamptz(onupdate=func.now())
+
+
+class BotTraceEvent(Base):
+    """Observability: append-only pipeline event log (payload schemas by `stage`).
+
+    Written ONLY through app.observability (fire-and-forget, _jsonable-safe).
+    Nothing is ever deleted from here by the bot.
+    """
+
+    __tablename__ = "bot_trace_events"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    trace_id: Mapped[str] = mapped_column(
+        UUID(as_uuid=False), ForeignKey("bot_traces.id", ondelete="CASCADE"), nullable=False
+    )
+    stage: Mapped[str] = mapped_column(String(32), nullable=False)
+    ok: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    latency_ms: Mapped[int | None] = mapped_column(Integer)
+    model: Mapped[str | None] = mapped_column(String(64))
+    field: Mapped[str | None] = mapped_column(String(32))
+    payload: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    created_at: Mapped[datetime] = _timestamptz()

bot/app/nlu.py

@@ -4,12 +4,14 @@ One chat/completions call per message with JSON-mode output. The prompt embeds
 the current draft summary so the model can decide between "start a new
 activity" and "refine the open draft" (update_draft with a partial patch).
 """
+import hashlib
 import json
 import logging
 import re
 
 import httpx
 
+from app import observability as obs
 from app.config import Settings
 
 logger = logging.getLogger("bot.nlu")
@@ -143,7 +145,8 @@ def _extract_json(raw: str) -> dict | None:
         return None
 
 
-async def understand(settings: Settings, text: str | list[dict], draft_summary: str | None = None) -> dict:
+async def understand(settings: Settings, text: str | list[dict], draft_summary: str | None = None,
+                     trace_id: str | None = None) -> dict:
     """Return {"consensus_text": ..., "intent": ..., "patch": {...}, "confidence": ...}; never raises."""
     fallback = {"consensus_text": "", "intent": "unknown", "patch": {}, "confidence": 0.0}
     if not settings.AI_BASE_URL or not settings.AI_API_KEY:
@@ -172,6 +175,17 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
 
     fallback["consensus_text"] = first_text
 
+    system_prompt = _build_system_prompt(draft_summary)
+    prompt_hash = hashlib.sha256(
+        f"{settings.OBS_PROMPT_VERSION}|{system_prompt}".encode()
+    ).hexdigest()[:16]
+    await obs.event(trace_id, "nlu.request", model=settings.LLM_MODEL, payload={
+        "is_multi": isinstance(text, list),
+        "content": user_content[:2000],
+        "prompt_hash": prompt_hash,
+        "prompt_version": settings.OBS_PROMPT_VERSION,
+    })
+
     try:
         async with httpx.AsyncClient(timeout=settings.LLM_TIMEOUT_SECONDS) as client:
             resp = await client.post(
@@ -180,7 +194,7 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
                 json={
                     "model": settings.LLM_MODEL,
                     "messages": [
-                        {"role": "system", "content": _build_system_prompt(draft_summary)},
+                        {"role": "system", "content": system_prompt},
                         {"role": "user", "content": user_content},
                     ],
                     "temperature": 0.1,
@@ -189,16 +203,28 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
             )
         if resp.status_code != 200:
             logger.error("LLM HTTP %s: %s", resp.status_code, resp.text[:300])
+            await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL,
+                            payload={"error": f"HTTP {resp.status_code}", "body": resp.text[:300]})
             return fallback
         content = ((resp.json() or {}).get("choices") or [{}])[0].get("message", {}).get("content", "")
         parsed = _extract_json(content)
         if not parsed or parsed.get("intent") not in INTENTS:
             logger.warning("unparseable LLM output: %r", content[:200])
+            await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL,
+                            payload={"error": "unparseable", "raw": content[:500]})
             return fallback
         parsed.setdefault("patch", {})
         if not parsed.get("consensus_text"):
             parsed["consensus_text"] = first_text
+        await obs.event(trace_id, "nlu.result", ok=True, model=settings.LLM_MODEL, payload={
+            "intent": parsed.get("intent"),
+            "confidence": parsed.get("confidence"),
+            "patch": parsed.get("patch"),
+            "consensus_text": parsed.get("consensus_text"),
+        })
         return parsed
     except (httpx.HTTPError, json.JSONDecodeError, KeyError, TypeError):
         logger.exception("NLU failure")
+        await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL,
+                        payload={"error": "exception"})
         return fallback

bot/app/observability.py

@@ -0,0 +1,206 @@
+"""Silent observer — records the whole bot pipeline into Postgres (phase 1).
+
+Design rules (final plan v5):
+- DB is the single source of truth; no external tools, no dashboards here.
+- ZERO extra LLM tokens: we only persist values that already exist in code.
+- Never raises, never blocks the reply path (fire-and-forget like runtime._log).
+- Never deletes anything (no retention/cleanup tasks exist by policy).
+- `_jsonable` guarantees JSONB-safety for UUID/datetime/Decimal/... payloads.
+
+Stages captured (see plan §4.3): input.text, input.voice, stt.run, stt.result,
+nlu.request, nlu.result, entities.extracted, jev.request, jev.result,
+routing.decision, ambiguity.sent, ambiguity.user_choice, user_choice.pick,
+draft.render, draft.edit, draft.commit, draft.cancel.
+"""
+import asyncio
+import logging
+import uuid
+from datetime import date, datetime, time
+from decimal import Decimal
+from enum import Enum
+from pathlib import Path
+
+from sqlalchemy import select, update
+
+from app.config import settings
+from app.db import SessionLocal
+from app.models import BotTrace, BotTraceEvent
+
+logger = logging.getLogger("bot.obs")
+
+_bg_tasks: set = set()
+
+
+def _fire(coro) -> None:
+    """Schedule a coroutine without awaiting; keep a ref so it is not GC'd."""
+    try:
+        task = asyncio.get_running_loop().create_task(coro)
+        _bg_tasks.add(task)
+        task.add_done_callback(_bg_tasks.discard)
+    except RuntimeError:
+        # No running loop (should not happen inside the bot) — drop silently but log.
+        logger.warning("obs: no running loop; dropped %r", coro)
+
+
+def _jsonable(obj):
+    """Recursively make any Python object JSONB-safe (UUID/datetime/Decimal/…)."""
+    if obj is None or isinstance(obj, (bool, int, float, str)):
+        return obj
+    if isinstance(obj, (uuid.UUID, Path)):
+        return str(obj)
+    if isinstance(obj, (datetime, date, time)):
+        return obj.isoformat()
+    if isinstance(obj, Decimal):
+        return str(obj)  # دقت حفظ می‌شود؛ در تحلیل با ::numeric قابل تبدیل است
+    if isinstance(obj, Enum):
+        return _jsonable(obj.value)
+    if isinstance(obj, dict):
+        return {str(k): _jsonable(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple, set, frozenset)):
+        return [_jsonable(v) for v in obj]
+    if hasattr(obj, "model_dump"):          # pydantic v2
+        return _jsonable(obj.model_dump())
+    if hasattr(obj, "__dataclass_fields__"):
+        return _jsonable(vars(obj))
+    return str(obj)                          # تور نهایی — هرگز raise نمی‌کند
+
+
+# ---------------------------------------------------------------------------
+# traces
+# ---------------------------------------------------------------------------
+
+async def trace_start(chat_id: int, message_id: int | None, kind: str,
+                      source: str = "bale") -> str | None:
+    """Open a new trace; returns its id (str) or None on failure."""
+    try:
+        async with SessionLocal() as s:
+            t = BotTrace(chat_id=chat_id, message_id=message_id, kind=kind, source=source)
+            s.add(t)
+            await s.commit()
+            return str(t.id)
+    except Exception:
+        logger.exception("obs: trace_start failed")
+        return None
+
+
+async def link_draft(trace_id: str | None, draft_id, draft_kind: str | None) -> None:
+    """Attach a draft to its originating trace (so later button clicks can find it)."""
+    if not trace_id:
+        return
+    try:
+        async with SessionLocal() as s:
+            await s.execute(
+                update(BotTrace)
+                .where(BotTrace.id == trace_id)
+                .values(draft_id=str(draft_id), draft_kind=draft_kind)
+            )
+            await s.commit()
+    except Exception:
+        logger.exception("obs: link_draft failed")
+
+
+async def latest_trace_for_draft(draft_id) -> str | None:
+    """Newest trace that produced this draft — used by button/render flows."""
+    try:
+        async with SessionLocal() as s:
+            row = (await s.execute(
+                select(BotTrace.id)
+                .where(BotTrace.draft_id == str(draft_id))
+                .order_by(BotTrace.created_at.desc())
+                .limit(1)
+            )).scalar()
+            return str(row) if row else None
+    except Exception:
+        logger.exception("obs: latest_trace_for_draft failed")
+        return None
+
+
+async def set_trace_status(trace_id: str | None, status: str,
+                           committed_quote_id: str | None = None) -> None:
+    if not trace_id:
+        return
+    try:
+        async with SessionLocal() as s:
+            values = {"status": status}
+            if committed_quote_id:
+                values["committed_quote_id"] = str(committed_quote_id)
+            await s.execute(update(BotTrace).where(BotTrace.id == trace_id).values(**values))
+            await s.commit()
+    except Exception:
+        logger.exception("obs: set_trace_status failed")
+
+
+# ---------------------------------------------------------------------------
+# events
+# ---------------------------------------------------------------------------
+
+async def event(trace_id, stage: str, *, ok: bool = True, latency_ms: int | None = None,
+                model: str | None = None, field: str | None = None,
+                payload: dict | None = None) -> None:
+    """Append one pipeline event. Never raises; failures are loud in the log."""
+    if not trace_id or not settings.OBS_ENABLED:
+        return
+    try:
+        async with SessionLocal() as s:
+            s.add(BotTraceEvent(
+                trace_id=str(trace_id), stage=stage, ok=ok, latency_ms=latency_ms,
+                model=model, field=field, payload=_jsonable(payload or {}),
+            ))
+            await s.commit()
+    except Exception:
+        logger.exception("obs: event %s failed", stage)   # خطای خاموش نمی‌ماند
+
+
+async def log_resolution(trace_id, field: str, mention, status: str, candidates: list) -> None:
+    """Record one matching attempt (difflib baseline / fallback paths)."""
+    await event(trace_id, "routing.decision", field=field, payload={
+        "source": "difflib",
+        "mention": mention,
+        "status": status,
+        "candidates": candidates[:4],
+    })
+
+
+class span:
+    """with span('stt.run', trace_id=t, provider='whisper'): ... → one DB event with latency/ok."""
+
+    def __init__(self, name: str, trace_id: str | None = None, **attrs):
+        self.name, self.trace_id, self.attrs = name, trace_id, attrs
+        self._t0 = None
+
+    def __enter__(self):
+        import time as _t
+        self._t0 = _t.perf_counter_ns()
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        import time as _t
+        try:
+            _fire(event(
+                self.trace_id, self.name, ok=exc_type is None,
+                latency_ms=int((_t.perf_counter_ns() - self._t0) / 1e6),
+                model=self.attrs.get("model"), field=self.attrs.get("field"),
+                payload=self.attrs,
+            ))
+        except Exception:
+            logger.exception("obs: span %s failed", self.name)
+        return False   # استثنا هرگز بلعیده نمی‌شود
+
+
+# ---------------------------------------------------------------------------
+# voice persistence (never deleted)
+# ---------------------------------------------------------------------------
+
+def save_voice(trace_id: str | None, message_id: int, content: bytes,
+               ext: str = ".ogg") -> str | None:
+    """Persist a voice note under OBS_VOICE_DIR; creates the day folder safely."""
+    try:
+        day = datetime.utcnow().strftime("%Y%m%d")
+        base = Path(settings.OBS_VOICE_DIR) / day
+        base.mkdir(parents=True, exist_ok=True)     # الزام سخت‌سازی: مسیر همیشه ساخته می‌شود
+        path = base / f"{trace_id}_{message_id}{ext}"
+        path.write_bytes(content)
+        return str(path)
+    except Exception:
+        logger.warning("obs: save_voice failed for trace %s", trace_id, exc_info=True)
+        return None

bot/app/stt.py

@@ -9,10 +9,12 @@ import asyncio
 import logging
 import subprocess
 import tempfile
+import time
 from pathlib import Path
 
 import httpx
 
+from app import observability as obs
 from app.config import Settings
 
 logger = logging.getLogger("bot.stt")
@@ -37,7 +39,8 @@ def _to_mp3(raw: bytes, suffix: str) -> bytes:
         return dst.read_bytes()
 
 
-async def _call_single_stt(client: httpx.AsyncClient, provider: dict, converted_mp3: bytes) -> dict | None:
+async def _call_single_stt(client: httpx.AsyncClient, provider: dict, converted_mp3: bytes) -> dict:
+    """Call one provider; always returns {"name","ok","text"/"error","model","latency_ms"}."""
     name = provider.get("name") or "stt"
     base_url = (provider.get("base_url") or "").rstrip("/")
     api_key = provider.get("api_key") or ""
@@ -46,6 +49,7 @@ async def _call_single_stt(client: httpx.AsyncClient, provider: dict, converted_
     timeout = float(provider.get("timeout") or _STT_TIMEOUT)
 
     url = f"{base_url}/audio/transcriptions"
+    t0 = time.perf_counter()
     try:
         resp = await client.post(
             url,
@@ -54,24 +58,30 @@ async def _call_single_stt(client: httpx.AsyncClient, provider: dict, converted_
             data={"model": model, "language": language},
             timeout=timeout,
         )
+        ms = int((time.perf_counter() - t0) * 1000)
         if resp.status_code == 200:
             txt = str((resp.json() or {}).get("text") or "").strip()
             if txt:
                 logger.info("STT provider %s succeeded: %r", name, txt[:60])
-                return {"name": name, "text": txt, "model": model}
-        else:
-            logger.warning("STT provider %s returned HTTP %s: %s", name, resp.status_code, resp.text[:200])
+                return {"name": name, "ok": True, "text": txt, "model": model, "latency_ms": ms}
+            return {"name": name, "ok": False, "error": "empty transcript (HTTP 200)",
+                    "model": model, "latency_ms": ms}
+        logger.warning("STT provider %s returned HTTP %s: %s", name, resp.status_code, resp.text[:200])
+        return {"name": name, "ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:200]}",
+                "model": model, "latency_ms": ms}
     except Exception as exc:
         logger.warning("STT provider %s failed: %s", name, exc)
-    return None
+        return {"name": name, "ok": False, "error": str(exc)[:300],
+                "model": model, "latency_ms": int((time.perf_counter() - t0) * 1000)}
 
 
 async def transcribe_multi(settings: Settings, audio: bytes, mime_type: str | None = None,
-                           filename: str | None = None) -> list[dict]:
+                           filename: str | None = None, trace_id: str | None = None) -> list[dict]:
     """Transcribe audio with all configured STT providers in parallel.
 
     Returns a list of candidate dicts: [{"name": str, "text": str, "model": str}, ...]
-    Empty list if all failed or silence.
+    Empty list if all failed or silence. Every provider outcome (success AND
+    failure) is recorded in Observability when trace_id is given.
     """
     providers = settings.stt_providers
     if not providers:
@@ -83,20 +93,32 @@ async def transcribe_multi(settings: Settings, audio: bytes, mime_type: str | No
                   "audio/mp4": ".m4a", "audio/wav": ".wav"}.get(mime_type.split(";")[0].strip(), ".ogg")
     converted = await asyncio.to_thread(_to_mp3, audio, suffix)
 
+    await obs.event(trace_id, "stt.run", payload={
+        "providers": [p.get("name") or "stt" for p in providers],
+        "audio_bytes": len(converted),
+    })
+
     async with httpx.AsyncClient(timeout=_STT_TIMEOUT) as client:
         tasks = [_call_single_stt(client, p, converted) for p in providers]
         results = await asyncio.gather(*tasks, return_exceptions=True)
 
     valid_transcripts: list[dict] = []
     for r in results:
-        if isinstance(r, dict) and r.get("text"):
-            valid_transcripts.append(r)
+        if not isinstance(r, dict):
+            continue
+        await obs.event(trace_id, "stt.result", ok=bool(r.get("ok")),
+                        latency_ms=r.get("latency_ms"), model=r.get("model"),
+                        payload={"provider": r.get("name"), "text": r.get("text"),
+                                 "error": r.get("error")})
+        if r.get("ok") and r.get("text"):
+            valid_transcripts.append({"name": r["name"], "text": r["text"], "model": r.get("model")})
 
     return valid_transcripts
 
 
 async def transcribe(settings: Settings, audio: bytes, mime_type: str | None = None,
-                     filename: str | None = None) -> str:
+                     filename: str | None = None, trace_id: str | None = None) -> str:
     """Return the Persian transcript for a voice note (empty string on silence)."""
-    transcripts = await transcribe_multi(settings, audio, mime_type=mime_type, filename=filename)
+    transcripts = await transcribe_multi(settings, audio, mime_type=mime_type,
+                                         filename=filename, trace_id=trace_id)
     return transcripts[0]["text"] if transcripts else ""

bot/app/sync.py

@@ -8,6 +8,7 @@ If a draft was updated in the Wizard:
 import asyncio
 import logging
 from app import drafts as draft_store
+from app import observability as obs
 from app.handlers import quote
 
 logger = logging.getLogger("bot.sync")
@@ -40,6 +41,19 @@ async def sync_drafts_loop(ctx) -> None:
                         logger.info("Synced cancelled draft %s to Bale message %s", draft.id, draft.review_message_id)
 
                     elif draft.status == "active" and draft.kind == draft_store.KIND_QUOTE:
+                        import hashlib
+                        import json as _json
+                        await obs.event(
+                            await obs.latest_trace_for_draft(draft.id), "draft.edit",
+                            payload={
+                                "source": "wizard",
+                                "kind": draft.kind,
+                                "payload_hash": hashlib.sha1(
+                                    _json.dumps(draft.payload, ensure_ascii=False,
+                                                sort_keys=True, default=str).encode()
+                                ).hexdigest()[:16],
+                            },
+                        )
                         await quote.render(ctx, draft.chat_id, draft, fresh=False)
                         await draft_store.set_bale_updated_at(draft.id)
                         logger.info("Synced active draft %s edits to Bale message %s", draft.id, draft.review_message_id)

bot/tools/test_observability.py

@@ -0,0 +1,140 @@
+"""Phase-1 tests for the Observability subsystem (run INSIDE the bot container):
+
+    docker compose exec bot python tools/test_observability.py
+
+Covers: _jsonable safety (UUID/datetime/Decimal/set/enum/dataclass), trace
+lifecycle, event round-trip through JSONB, draft linking, status updates,
+save_voice (day-folder mkdir), and the span() context manager.
+
+NOTE (policy): test rows/files are tagged `_test` and are NEVER deleted —
+the observability subsystem has no retention/cleanup by design.
+"""
+import asyncio
+import dataclasses
+import sys
+import uuid
+from datetime import datetime, timezone
+from decimal import Decimal
+from enum import Enum
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from sqlalchemy import select
+
+from app import observability as obs
+from app.db import SessionLocal
+from app.db_init import ensure_schema
+from app.models import BotTrace, BotTraceEvent
+
+PASSED: list[str] = []
+FAILED: list[str] = []
+
+
+def check(name: str, ok: bool, extra: str = "") -> None:
+    (PASSED if ok else FAILED).append(name)
+    print(f"{'PASS' if ok else 'FAIL'}  {name}{'  — ' + str(extra) if extra else ''}")
+
+
+class SampleColor(Enum):
+    RED = "red"
+
+
+@dataclasses.dataclass
+class SampleRec:
+    a: int
+    b: str
+
+
+def test_jsonable() -> None:
+    payload = {
+        "uuid": uuid.uuid4(),
+        "dt": datetime(2026, 10, 12, 10, 30, tzinfo=timezone.utc),
+        "dec": Decimal("123.45"),
+        "set": {"x", "y"},
+        "enum": SampleColor.RED,
+        "dc": SampleRec(a=1, b="two"),
+        "nested": [{"inner": uuid.UUID(int=1)}],
+        "none": None,
+    }
+    out = obs._jsonable(payload)
+    ok = (
+        isinstance(out["uuid"], str)
+        and isinstance(out["dt"], str)
+        and isinstance(out["dec"], str)
+        and isinstance(out["set"], list)
+        and out["enum"] == "red"
+        and isinstance(out["dc"], dict)
+    )
+    check("_jsonable produces JSON-serializable output", ok,
+          f"dt={out['dt'][:19]}, dec={out['dec']!r}")
+
+
+async def main() -> None:
+    await ensure_schema()
+    test_jsonable()
+
+    # 1) trace lifecycle
+    tid = await obs.trace_start(chat_id=0, message_id=999001, kind="voice")
+    check("trace_start returns an id", bool(tid), str(tid))
+
+    # 2) events with tricky payloads (must not raise; must be stored)
+    await obs.event(tid, "input.voice", payload={
+        "_test": True, "size": 1234,
+        "when": datetime.now(timezone.utc),
+        "amount": Decimal("9.9"),
+    })
+    await obs.event(tid, "routing.decision", field="project", payload={
+        "_test": True,
+        "candidates": [{"id": uuid.uuid4(), "name": "منشانه"}],
+    })
+    await obs.log_resolution(tid, "supplier", "مصطفوی", "many",
+                             [{"id": "x", "name": "بازرگانی مصطفوی"}])
+
+    # 3) draft linking + latest lookup
+    draft_id = str(uuid.uuid4())
+    await obs.link_draft(tid, draft_id, "quote")
+    latest = await obs.latest_trace_for_draft(draft_id)
+    check("latest_trace_for_draft returns the linked trace", latest == tid, latest)
+
+    # 4) status update + JSONB round-trip
+    await obs.set_trace_status(tid, "committed", str(uuid.uuid4()))
+    async with SessionLocal() as s:
+        tr = (await s.execute(select(BotTrace).where(BotTrace.id == tid))).scalar_one()
+        check("trace status + quote id updated",
+              tr.status == "committed" and tr.committed_quote_id is not None)
+        rows = (await s.execute(
+            select(BotTraceEvent).where(BotTraceEvent.trace_id == tid)
+            .order_by(BotTraceEvent.id)
+        )).scalars().all()
+        check("events persisted", len(rows) >= 3, f"{len(rows)} rows")
+        proj = next((r for r in rows if r.stage == "routing.decision"), None)
+        check("JSONB payload round-trip keeps values as strings",
+              proj is not None and isinstance(proj.payload["candidates"][0]["id"], str))
+
+    # 5) save_voice (parents mkdir + day folder)
+    path = obs.save_voice(tid, 999001, b"OGGFAKE", ext=".ogg")
+    check("save_voice writes file under day folder",
+          bool(path) and Path(path).exists(), str(path))
+
+    # 6) span() context manager (fire-and-forget task)
+    with obs.span("span.test", trace_id=tid, note="unit"):
+        pass
+    await asyncio.sleep(0.4)
+    async with SessionLocal() as s:
+        n = (await s.execute(
+            select(BotTraceEvent).where(
+                BotTraceEvent.trace_id == tid, BotTraceEvent.stage == "span.test"
+            )
+        )).scalars().all()
+        check("span() recorded via background task", len(n) == 1, f"{len(n)} rows")
+
+    total = len(PASSED) + len(FAILED)
+    print(f"\n{len(PASSED)}/{total} passed")
+    if FAILED:
+        print("FAILED:", FAILED)
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())