STT-Assistant/src/stt_daemon/tui/state.py

"""Pure reduction of assistant/STT events into TUI view state.

Kept free of Textual so the behaviour is unit-testable: attach, mode
self-correction from the heartbeat, command editing, bounded buffers.
"""

from dataclasses import dataclass, field

MAX_TRANSCRIPTS = 200
MAX_CHAT = 200
MAX_LOWER = 500

MODE_STYLES = {
    "idle": "green",
    "command_armed": "yellow",
    "executing": "cyan",
    "chat": "magenta",
}


def _num(value) -> float | None:
    """Best-effort float coercion; None when missing/non-numeric."""
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _fmt_gate(item: dict) -> str:
    """Format the gating confidence defensively; '?' when absent."""
    for key in ("answer_confidence", "confidence"):
        value = _num(item.get(key))
        if value is not None:
            return f"{value:.2f}"
    return "?"


@dataclass
class ViewState:
    """Everything the TUI renders."""

    mode: str = "idle"
    stt_ok: bool = False
    device_ok: bool = True
    connected: bool = True
    speech_open: bool = False
    worker_stall_s: float = 0.0
    queue_depth: int = 0
    inference_depth: int = 0
    inference_in_flight: bool = False
    ready: bool = False
    phase: str = "idle"
    command_text: str = ""
    command_edited: bool = False
    in_flight: bool = False
    transcripts: list[dict] = field(default_factory=list)
    transcripts_total: int = 0
    chat: list[dict] = field(default_factory=list)
    lower: list[tuple[str, bool]] = field(default_factory=list)
    tool: dict | None = None
    tool_result: dict | None = None
    last_error: dict | None = None
    chat_epoch: int = 0
    triage: dict | None = None
    triage_state: str = "disabled"
    triage_reason: str = ""
    triage_epoch: int = 0

    def ingest(self, event: dict) -> None:
        """Fold one event into the view state."""
        etype = event.get("type")

        if etype == "mode":
            self._set_mode(event.get("mode", self.mode))
        elif etype == "status":
            self._set_mode(event.get("mode", self.mode))
            self.ready = bool(event.get("ready", self.ready))
            self.phase = event.get("phase", self.phase)
            self.command_text = event.get("command_text", self.command_text)
            self.stt_ok = bool(event.get("stt_ok", self.stt_ok))
            if "speech_open" in event:
                self.speech_open = bool(event["speech_open"])
            changed = False
            if "triage" in event:
                self.triage = event["triage"]
                changed = True
            if "triage_state" in event:
                self.triage_state = event["triage_state"]
                changed = True
            if "triage_reason" in event:
                self.triage_reason = event["triage_reason"]
                changed = True
            if changed:
                self.triage_epoch += 1
        elif etype == "triage":
            stage = event.get("stage")
            if stage == "cleared":
                self.triage = None
            else:
                # A triage payload can only come from a ready lane: authoritative.
                self.triage_state = "ok"
                self.triage_reason = ""
                self.triage = event.get("triage")
            self.triage_epoch += 1
        elif etype == "speech_start":
            self.speech_open = True
        elif etype == "speech_stop":
            self.speech_open = False
        elif etype == "heartbeat":
            # The heartbeat re-states the mode: a missed `mode` event self-corrects.
            self._set_mode(event.get("mode", self.mode))
            self.stt_ok = bool(event.get("stt_ok", self.stt_ok))
            self.in_flight = bool(event.get("in_flight", self.in_flight))
            self.worker_stall_s = float(event.get("worker_stall_s", self.worker_stall_s) or 0.0)
            if "queue_depth" in event:
                self.queue_depth = int(event["queue_depth"] or 0)
            if "inference_depth" in event:
                self.inference_depth = int(event["inference_depth"] or 0)
            if "inference_in_flight" in event:
                self.inference_in_flight = bool(event["inference_in_flight"])
            if "device_ok" in event:
                self.device_ok = bool(event["device_ok"])
        elif etype == "command_text":
            self.command_text = event.get("text", "")
            self.command_edited = bool(event.get("edited", False))
            self.phase = event.get("phase", self.phase)
        elif etype == "transcript":
            self.transcripts.append(event)
            del self.transcripts[:-MAX_TRANSCRIPTS]
            self.transcripts_total += 1
        elif etype == "chat_message":
            self.chat.append(event)
            del self.chat[:-MAX_CHAT]
            self.append_lower(
                f"{event.get('role', '?')}: {event.get('text', '')}", from_chat=True
            )
        elif etype == "tool_call":
            self.tool = event
            args = event.get("args") or {}
            suffix = f" {args}" if args else ""
            self.append_lower(f"-> tool {event.get('tool')}{suffix}")
        elif etype == "tool_result":
            self.tool_result = event
            verdict = "ok" if event.get("ok") else "failed"
            self.append_lower(f"<- {verdict}: {event.get('result', '')}")
        elif etype == "error":
            self.last_error = event
            self.append_lower(f"! {event.get('code')}: {event.get('message')}")
        elif etype == "cleared":
            self.clear_output()

    def append_lower(self, line: str, from_chat: bool = False) -> None:
        """Append one line to the lower pane log (chat turns, tool calls, results, errors)."""
        self.lower.append((line, from_chat))
        del self.lower[:-MAX_LOWER]

    def clear_chat(self) -> None:
        """Forget chat turns and their rendered lines, keeping tool/error output."""
        self.chat = []
        self.lower = [(line, from_chat) for line, from_chat in self.lower if not from_chat]
        self.chat_epoch += 1

    def clear_output(self) -> None:
        """Clear conversation/command output: chat, tool calls/results, errors."""
        self.chat        = []
        self.lower       = []
        self.tool        = None
        self.tool_result = None
        self.last_error  = None
        self.triage      = None
        self.chat_epoch += 1
        self.triage_epoch += 1

    def _set_mode(self, mode: str) -> None:
        if self.mode == "chat" and mode != "chat":
            self.clear_chat()
        self.mode = mode

    def lower_lines(self) -> list[str]:
        return [line for line, _ in self.lower]

    def health_ok(self) -> bool:
        """False when the header should go red."""
        return self.stt_ok and self.device_ok

    def mode_style(self) -> str:
        return MODE_STYLES.get(self.mode, "white")

    def transcript_lines(self) -> list[str]:
        return [f"[{e.get('lang', '?')}] {e.get('text', '')}" for e in self.transcripts]

    def chat_lines(self) -> list[str]:
        return [f"{m.get('role', '?')}: {m.get('text', '')}" for m in self.chat]

    def triage_lines(self) -> list[str]:
        """Render the triage panel lines. Pure and defensive: never raises."""
        if self.triage_state != "ok":
            reason = self.triage_reason or self.triage_state or "unavailable"
            return [f"triage: unavailable — {reason}"]
        if not isinstance(self.triage, dict):
            return ["triage: unavailable — no data"]

        order = self.triage.get("order")
        items = self.triage.get("items")
        lines: list[str] = []
        if isinstance(order, list) and isinstance(items, dict):
            for qid in order:
                item = items.get(qid)
                if not isinstance(item, dict):
                    lines.append(f"{qid} ⚠" if item else str(qid))
                    continue
                kind = item.get("kind")
                flagged = bool(item.get("flagged"))
                mark = " ⚠" if flagged else ""
                if kind == "choice":
                    line = f"{qid}: {item.get('choice', '?')} ({_fmt_gate(item)}){mark}"
                elif kind == "noul":
                    p = _num(item.get("p"))
                    bar = "█" * int(p * 10) if p is not None else ""
                    bar = bar.ljust(10)
                    ptxt = f"{p:.2f}" if p is not None else "?"
                    line = f"{bar} {qid}: {ptxt}{mark}"
                elif kind == "score":
                    level = _num(item.get("level"))
                    bar = ("█" * int(level)).ljust(10) if level is not None else " " * 10
                    label = item.get("label", "?")
                    line = f"{bar} {qid}: {label}{mark}"
                else:
                    line = f"{qid}{mark}" if flagged else str(qid)
                lines.append(line)

        content = self.triage.get("content")
        if isinstance(content, dict):
            if lines:
                lines.append("")
            summary = content.get("summary")
            if summary:
                lines.append(f"summary: {summary}")
            for key in ("opportunities", "risks", "action_items"):
                values = content.get(key)
                if not isinstance(values, list):
                    continue
                for value in values[:3]:
                    lines.append(f"{key[:-1]}: {value}")

        if not lines:
            return ["triage: no signals"]
        return lines