File size: 4,904 Bytes
6303ae6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
"""Reusable "API Usage Status" panel.

Renders an expandable block that summarises how the configured LLM was
used during the current session. The panel reads
:data:`st.session_state.api_usage_log`, which is appended to by
:mod:`app.services.llm_client`.

The panel must NEVER show the API key, raw dossier text, screenshots,
or full generated proposals. It only shows metadata.
"""

from __future__ import annotations

from typing import Iterable, Optional

import streamlit as st

from app.config import LLM_TASK_NAMES, get_settings


TRACKED_TASKS: tuple[str, ...] = LLM_TASK_NAMES


_TASK_LABELS: dict[str, str] = {
    "api_check": "API capability check",
    "screenshot_extraction": "Screenshot extraction",
    "evidence_index_generation": "Evidence index generation",
    "opportunity_matching": "Opportunity matching",
    "missing_info_labeling": "Missing-info labeling",
    "recommendation_generation": "Recommendation reasoning",
    "proposal_generation": "Proposal generation",
    "verification_pass": "Proposal verification pass",
}


def _last_entry(log: Iterable[dict], *, task_name: Optional[str] = None) -> Optional[dict]:
    last: Optional[dict] = None
    for entry in log:
        if task_name and entry.get("task_name") != task_name:
            continue
        last = entry
    return last


def render(*, expanded: bool = False, key_suffix: str = "") -> None:
    """Render the API Usage Status panel.

    Pass a unique ``key_suffix`` if you call this from more than one
    screen so the Streamlit widget keys stay unique.

    Hidden entirely from the main UI when ``SHOW_DEBUG_PANEL`` is false
    (the production default). Internal API logging still runs β€” this
    only controls the on-screen surface.
    """
    settings = get_settings()
    if not getattr(settings, "show_debug_panel", False):
        return
    log: list[dict] = list(st.session_state.get("api_usage_log") or [])

    with st.expander("API Usage Status", expanded=expanded):
        # Environment / configuration line
        col1, col2, col3 = st.columns(3)
        col1.metric("Provider", settings.llm_provider or "β€”")
        col2.metric("Active model", settings.active_model or "β€”")
        col3.metric(
            "Local placeholders allowed",
            "yes" if settings.allow_local_placeholders else "no",
        )

        real_calls_enabled = settings.has_api_key
        st.caption(
            "Real API calls enabled: "
            + ("βœ… yes" if real_calls_enabled else "❌ no (missing API key)")
            + " β€” API key is never displayed or logged."
        )

        total_api_calls = sum(1 for e in log if e.get("used_api"))
        total_local = sum(1 for e in log if not e.get("used_api"))
        m1, m2, m3 = st.columns(3)
        m1.metric("API calls this session", total_api_calls)
        m2.metric("Local placeholder calls", total_local)
        last_any = _last_entry(log)
        m3.metric(
            "Last API call task",
            (last_any or {}).get("task_name") or "β€”",
        )

        if last_any:
            st.caption(
                f"Last call: `{last_any.get('task_name')}` β€’ "
                f"status `{last_any.get('status')}` β€’ "
                f"used_api `{str(bool(last_any.get('used_api'))).lower()}` β€’ "
                f"{last_any.get('timestamp')}"
            )

        st.markdown("**Per-stage status**")
        for task in TRACKED_TASKS:
            label = _TASK_LABELS.get(task, task)
            entries = [e for e in log if e.get("task_name") == task]
            api_hits = sum(1 for e in entries if e.get("used_api"))
            local_hits = sum(1 for e in entries if not e.get("used_api"))
            last = entries[-1] if entries else None

            if not entries:
                badge = "βšͺ not run"
            elif api_hits and not local_hits:
                badge = "🟒 used API"
            elif api_hits and local_hits:
                badge = "🟑 mixed (API + local)"
            else:
                badge = "πŸ”΄ LOCAL PLACEHOLDER β€” API NOT USED"

            line = f"- **{label}** (`{task}`) β€” {badge}"
            if last:
                line += (
                    f"  \n  status `{last.get('status')}` β€’ "
                    f"provider `{last.get('provider') or 'β€”'}` β€’ "
                    f"model `{last.get('model') or 'β€”'}`"
                )
                if last.get("error_message"):
                    line += f"  \n  reason: {last['error_message']}"
            st.markdown(line)

        if log:
            with st.expander("Raw call log (metadata only)", expanded=False):
                # Streamlit's dataframe renders dict lists cleanly.
                st.dataframe(log, use_container_width=True, hide_index=True)
        else:
            st.caption(
                "No API calls or local placeholder calls recorded yet."
            )