diff --git a/bot/.gitignore b/bot/.gitignore index bef9ac6..51a2f85 100644 --- a/bot/.gitignore +++ b/bot/.gitignore @@ -2,6 +2,9 @@ .env *.local +# ---- Runtime data (voices etc.) — never commit, never delete ---- +data/ + # ---- Python ---- __pycache__/ *.pyc diff --git a/bot/app/config.py b/bot/app/config.py index 10f9d5c..5d592c1 100644 --- a/bot/app/config.py +++ b/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 diff --git a/bot/app/db_init.py b/bot/app/db_init.py index 5a051cc..1b42e4d 100644 --- a/bot/app/db_init.py +++ b/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)", ] diff --git a/bot/app/handlers/__init__.py b/bot/app/handlers/__init__.py index 0e33b33..878e23d 100644 --- a/bot/app/handlers/__init__.py +++ b/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"): diff --git a/bot/app/handlers/callbacks.py b/bot/app/handlers/callbacks.py index a60b129..5622d8c 100644 --- a/bot/app/handlers/callbacks.py +++ b/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: diff --git a/bot/app/handlers/material.py b/bot/app/handlers/material.py index 15adb20..8311ce8 100644 --- a/bot/app/handlers/material.py +++ b/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)) diff --git a/bot/app/handlers/project.py b/bot/app/handlers/project.py index e6e147b..c5cb3de 100644 --- a/bot/app/handlers/project.py +++ b/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)) diff --git a/bot/app/handlers/quote.py b/bot/app/handlers/quote.py index d5e3b2c..79d104b 100644 --- a/bot/app/handlers/quote.py +++ b/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"): diff --git a/bot/app/handlers/supplier.py b/bot/app/handlers/supplier.py index 0c1fb1a..4073ed6 100644 --- a/bot/app/handlers/supplier.py +++ b/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)) diff --git a/bot/app/models.py b/bot/app/models.py index f58dbd6..33fec27 100644 --- a/bot/app/models.py +++ b/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() diff --git a/bot/app/nlu.py b/bot/app/nlu.py index b4f2332..9f4dea4 100644 --- a/bot/app/nlu.py +++ b/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 diff --git a/bot/app/observability.py b/bot/app/observability.py new file mode 100644 index 0000000..34fcc5a --- /dev/null +++ b/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 diff --git a/bot/app/stt.py b/bot/app/stt.py index e72a174..1dc7f7e 100644 --- a/bot/app/stt.py +++ b/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 "" diff --git a/bot/app/sync.py b/bot/app/sync.py index 996d0ba..4617e25 100644 --- a/bot/app/sync.py +++ b/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) diff --git a/bot/tools/test_observability.py b/bot/tools/test_observability.py new file mode 100644 index 0000000..8b678df --- /dev/null +++ b/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()) diff --git a/docker-compose.yml b/docker-compose.yml index 2c01cb0..6095bc5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -322,6 +322,7 @@ services: volumes: - ./bot/app:/srv/bot/app + - ./bot/data:/srv/bot/data environment: PROCUREMENT_BOT_BOT_TOKEN: ${BALE_BOT_TOKEN:-1354444845:yhVMnKZBFuKaU3UW5fP1bHmknEHx99mipWw}