"""Always-visible error surface + one-click diagnostics hand-off. Companion to :mod:`app.ui.api_usage_panel`, and deliberately governed by a different rule: * **API Usage Status** is a developer view of *everything* that happened — token counts, per-stage API-vs-local badges, the raw call table. It stays behind ``SHOW_DEBUG_PANEL`` because it is noise for a normal user. * **This panel** shows only *failures*, and it is **never** debug-gated. A user who hits an error is exactly the user who cannot turn on a debug flag, and hiding the explanation from them is what turns "I picked a retired model" into a support ticket that reads "the app is broken". It renders nothing at all when there are no failures, so a healthy run is visually unchanged. Privacy ------- Everything shown and shared comes from the metadata-only records in :mod:`app.services.diagnostics` — no API key, dossier text, screenshot, job description, or generated proposal. The full payload is shown to the user *before* any upload, because a share button people cannot inspect is a share button people do not press. """ from __future__ import annotations import re from typing import Optional import streamlit as st from app.services import diagnostics # Plain-English meaning per status code, with the actionable next step. The # wording deliberately separates "no model is set" (a configuration gap) # from "the model you set was rejected" (a wrong/retired name) — collapsing # those two into one message is what previously sent users to re-check a # dropdown that already had a value in it. _STATUS_HELP: dict[str, tuple[str, str]] = { "access_denied": ("Model access denied", "Check your provider permissions or choose a model your account can use."), "app_error": ("Unexpected app error", "Retry the step. If it repeats, share the diagnostics with the app team."), "invalid_key": ( "API key rejected", "Re-copy the whole key and confirm it belongs to the provider " "selected in Setup (an OpenAI key will not work with Anthropic).", ), "model_missing": ( "Model not accepted by the provider", "The model name was not recognised — it may be retired, misspelled, " "or unavailable on your plan. Pick a different model in Setup.", ), "quota": ( "Out of credit or rate limited", "Add credit in your provider dashboard, or wait a moment and retry.", ), "context_overflow": ( "Too much input for this model", "Shorten the dossier or screenshot text and try again.", ), "connection": ( "Could not reach the provider", "Check your internet connection, VPN, or firewall, then retry.", ), "parse_error": ( "Unreadable reply from the model", "The model did not return valid JSON. Retrying usually fixes it; if " "it repeats, try a different model.", ), "truncated": ( "Reply cut off by the token limit", "Raise the output-token limit in Setup and retry.", ), "no_api": ( "No API key configured", "Add your key in Setup and save the configuration.", ), "unsupported_provider": ( "Provider not configured", "Choose a supported provider in Setup.", ), "exception": ( "Unexpected provider error", "Share the diagnostics below with the app team.", ), } # HTTP status refines the story where the SDK gave us one. These are exact # signals, unlike the message-text heuristics used to bucket the status. _HTTP_HELP: dict[int, str] = { 400: "The provider rejected the request itself (bad parameter) — this is " "an app-side issue, not something you configured wrong.", 401: "The key was not accepted at all.", 403: "The key is valid but is not allowed to use this model.", 404: "The model name does not exist for this provider.", 429: "Rate limit or credit limit hit.", } # Characters a provider, model, id or timestamp keeps when shown below. The # values go into Markdown and some are free text (a custom model name, a # request id from whatever answered the call): a backtick would close the # code span around it and the rest would render as live Markdown, such as a # link in the app's own sidebar. Real names, ids and timestamps fit this set. _UNSAFE_CHARS = re.compile(r"[^A-Za-z0-9_.:/@+-]") _MAX_VALUE_LEN = 80 def _safe(value: object, default: str = "—") -> str: """``value`` cut down to plain id characters, safe inside Markdown.""" text = _UNSAFE_CHARS.sub("", str(value or ""))[:_MAX_VALUE_LEN] return text or default def _describe(event: dict) -> tuple[str, str]: """Return ``(headline, next_step)`` for one failed call.""" status = str(event.get("status") or "exception") headline, action = _STATUS_HELP.get( status, ("Something went wrong", "Share the diagnostics with the app team.") ) http_status = event.get("http_status") if isinstance(http_status, int) and http_status in _HTTP_HELP: action = f"{_HTTP_HELP[http_status]} {action}" return headline, action def _stage_label(event: dict) -> str: task = _safe(event.get("task_name"), "unknown") return task.replace("_", " ").strip().capitalize() _INTERVIEW_TASKS = frozenset({"interview_turn", "interview_check"}) def render(*, key_suffix: str = "", limit: int = 50) -> None: """Render the diagnostics panel, or nothing when there is no failure. Pass a unique ``key_suffix`` when calling from more than one screen so Streamlit widget keys stay unique. """ try: session_events = list(st.session_state.get("api_usage_log") or []) except Exception: # noqa: BLE001 - outside a script run session_events = [] # Disk first on a single-user install: it survives the page reload that # usually follows a failure and spans earlier sessions. On the shared # cloud host the disk log holds every visitor's calls, so only this # session's own log is used there (see diagnostics.visitor_events). events = diagnostics.visitor_events(session_events, limit=limit) # The guided interview reports on its own API where the user is, and it # can run on a different provider (an optional Groq key) from the rest of # the app. Its failures are left out here, or a rejected interview key # would read as "Problem detected" for the main API that is working. events = [ e for e in events if str((e or {}).get("task_name") or "") not in _INTERVIEW_TASKS ] # Only failures since the last successful call: a success is proof the # problem is over, and a warning that never clears is one users learn # to ignore. failures = diagnostics.unresolved_errors(events) if not failures: return latest = failures[-1] headline, action = _describe(latest) with st.expander(f"⚠️ Problem detected — {headline}", expanded=False): st.markdown(f"**{headline}**") st.caption(action) st.markdown( f"- **Stage:** {_stage_label(latest)}\n" f"- **Provider / model:** `{_safe(latest.get('provider'))}`" f" · `{_safe(latest.get('model'))}`\n" f"- **When:** {_safe(latest.get('timestamp'))}" ) # Correlation ids: meaningless to the user, decisive for support. req_id = latest.get("request_id") http_status = latest.get("http_status") err_type = latest.get("error_type") if req_id or http_status or err_type: bits = [] if http_status: bits.append(f"HTTP `{_safe(http_status)}`") if err_type: bits.append(f"type `{_safe(err_type)}`") if req_id: bits.append(f"request `{_safe(req_id)}`") st.caption("Reference: " + " · ".join(bits)) if len(failures) > 1: st.caption(f"{len(failures)} failed calls since the last success.") st.divider() _render_share(session_events, key_suffix=key_suffix) def _render_share(session_events: list[dict], *, key_suffix: str = "") -> None: """Build-and-share controls: inspect, download, and optional upload.""" st.markdown("**Share with the app team**") st.caption( "Metadata only — no API key, dossier text, screenshots, job posts, " "or generated proposals." ) # One report id per session so the download, the upload, and whatever the # user quotes in an email all refer to the same thing. id_key = f"diag_report_id_{key_suffix}" if id_key not in st.session_state: st.session_state[id_key] = diagnostics.new_report_id() report_id: str = st.session_state[id_key] report = diagnostics.build_report( report_id=report_id, session_events=session_events ) payload = diagnostics.report_to_json(report) st.download_button( "Download diagnostics", data=payload, file_name=f"{report_id}.json", mime="application/json", use_container_width=True, key=f"diag_dl_{key_suffix}", help="Save the report and attach it to an email or chat message.", ) # st.code ships a native copy button, so there is no need for custom JS # (which the app's CSP-tight embed would complicate anyway). with st.expander("View / copy report", expanded=False): st.code(payload, language="json") if diagnostics.is_upload_configured(): sent_key = f"diag_sent_{key_suffix}" if st.button( "Send to support", use_container_width=True, key=f"diag_send_{key_suffix}", help="Uploads the report above so the team can look it up by id.", ): st.session_state[sent_key] = diagnostics.upload_report(report) sent: Optional[bool] = st.session_state.get(sent_key) if sent is True: st.success(f"Sent. Quote reference **{report_id}** to the team.") elif sent is False: st.warning( "Upload failed. Use **Download diagnostics** above and send " "the file instead." ) else: st.caption(f"Reference id: **{report_id}**") __all__ = ["render"]