feat: operator channel polish — 0.3.6

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 ## Unreleased
 
+## 0.3.6 (2026-04-22)
+
+### Operator Channel polish
+
+- Added `deep_explainer_model` (+ `deep_explainer_temperature`, `deep_explainer_max_tokens`) so `assess_drift` can opt into a heavier reasoner while routine `explain_run` / `explain_exchange` / `request_clarification` keep using the cheaper `explainer_model`. Falls back to the routine model, then to the structured stub, when the deep model is unset or the call fails.
+- Added frozen `DriftAssessment` and `ExchangeView` dataclasses (`supervisor.operator.models`) with `from_dict()` classmethods that accept both the legacy `last_*_summary` exchange shape and the post-explainer shape. Unknown drift statuses collapse to `watch`; non-numeric confidences collapse to `None`. Channel adapters can consume type-safe projections instead of stringifying raw dicts.
+- Emitted a dedicated `explainer_answer` timeline event alongside the existing `clarification_response`, tagged with `source="explainer"` so future worker-side clarification flows can be distinguished without touching existing consumers.
+- Added a confidence-gated escalation recommendation: when a clarification answer's confidence falls below `clarification_escalation_confidence` (default `0.4`), the result now carries `escalation_recommended=True` and a `clarification_escalation_recommended` event is written to the session log. Escalation is advisory — the operator still chooses whether to route the question to the worker.
+
 ## 0.3.5 (2026-04-21)
 
 ### JSONL injection via Stop hook

README.md

@@ -18,8 +18,9 @@ thin-supervisor fixes this. It's an acceptance-centered run supervisor that sits
 > - [docs/reviews/2026-04-11-deep-code-review.md](docs/reviews/2026-04-11-deep-code-review.md) — latest deep code review log and remaining-risk audit
 > - [docs/reviews/2026-04-12-amp-supervisor-capability-review.md](docs/reviews/2026-04-12-amp-supervisor-capability-review.md) — Amp-vs-thin-supervisor capability review and oracle-layer roadmap
 
-## Current Status (0.3.5)
+## Current Status (0.3.6)
 
+- **Operator channel polish.** `assess_drift` can opt into a heavier reasoner via `deep_explainer_model` while routine explanations stay on the cheap `explainer_model`. Clarification answers now emit a dedicated `explainer_answer` timeline event (tagged `source="explainer"`) and carry `escalation_recommended=True` when confidence falls below `clarification_escalation_confidence` (default `0.4`) so operators know when to route the question to the worker. `DriftAssessment` and `ExchangeView` frozen dataclasses give TUI / IM channels a type-safe projection instead of raw dicts.
 - **JSONL runs close the loop via Stop hook.** `thin-supervisor hook install` wires a Claude Code / Codex Stop hook that reads the supervisor's per-session handoff file, returns the next instruction as the agent's stop `reason`, and writes an ACK so the supervisor can confirm delivery. Observation-only runs no longer dead-end at "pause for human" when nothing else is wrong.
 - **A2A inbound is live.** `thin-supervisor a2a serve` exposes a stdlib-only Google A2A JSON-RPC listener (`tasks/send` / `tasks/get`) behind a shared inbound boundary guard — bearer auth with localhost fallback, per-IP rate-limit, injection scan, and secret redaction with SHA-256 audit hashes. `task_id` equals the client `request_id`, so task identity survives adapter and daemon restarts.
 - **Layered system observability is live.** `thin-supervisor overview` (with `--json` / `--watch`) renders a whole-system view — daemon counts, live/orphaned/completed sessions, event-plane backlog, actionable alerts, and a cross-run timeline — that stays consistent with `status`, `observe`, and the TUI global mode (`g` to toggle).

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "thin-supervisor"
-version = "0.3.5"
+version = "0.3.6"
 description = "Thin tmux sidecar supervisor for long-running AI coding agent workflows"
 requires-python = ">=3.10"
 license = {text = "MIT"}

supervisor/config.py

@@ -19,6 +19,8 @@
     "notification_channels", "pause_handling_mode", "max_auto_interventions",
     "poll_interval_sec", "read_lines",
     "explainer_model", "explainer_temperature", "explainer_max_tokens",
+    "deep_explainer_model", "deep_explainer_temperature", "deep_explainer_max_tokens",
+    "clarification_escalation_confidence",
 })
 
 
@@ -103,9 +105,25 @@ class RuntimeConfig:
     judge_max_tokens: int = 512
 
     # -- LLM Explainer (operator-facing, separate from judge) --
+    #
+    # Two tiers: routine `explainer_model` is the cheap/fast default used for
+    # explain_run / explain_exchange / request_clarification. Optional
+    # `deep_explainer_model` is used only for drift/codebase-heavy analysis
+    # (assess_drift) — set to the stronger model you're willing to spend on
+    # when operators ask "is this run still on track?" If `deep_explainer_model`
+    # is None, drift assessment falls back to the routine explainer.
     explainer_model: str | None = None  # None = stub mode (cheap/fast default)
     explainer_temperature: float = 0.3
     explainer_max_tokens: int = 1024
+    deep_explainer_model: str | None = None  # None = reuse explainer_model
+    deep_explainer_temperature: float = 0.2
+    deep_explainer_max_tokens: int = 2048
+
+    # Clarification routing: when the explainer's self-reported confidence
+    # is below this threshold, the channel surfaces `escalation_recommended`
+    # so the operator can explicitly request a worker follow-up. Escalation
+    # is never automatic — the operator remains in the loop.
+    clarification_escalation_confidence: float = 0.4
 
     # -- Runtime paths --
     runtime_dir: str = ".supervisor/runtime"
@@ -247,6 +265,16 @@ def default_config_yaml(self) -> str:
             f"explainer_temperature: {self.explainer_temperature}\n"
             f"explainer_max_tokens: {self.explainer_max_tokens}\n"
             "\n"
+            "# Optional heavier explainer for drift/codebase analysis (assess_drift).\n"
+            "# Leave null to reuse explainer_model.\n"
+            f"deep_explainer_model: null\n"
+            f"deep_explainer_temperature: {self.deep_explainer_temperature}\n"
+            f"deep_explainer_max_tokens: {self.deep_explainer_max_tokens}\n"
+            "\n"
+            "# Confidence below which clarification surfaces an escalation hint\n"
+            "# (does not auto-escalate — the operator decides).\n"
+            f"clarification_escalation_confidence: {self.clarification_escalation_confidence}\n"
+            "\n"
             "# Runtime\n"
             f"runtime_dir: \"{self.runtime_dir}\"\n"
             "\n"

supervisor/daemon/server.py

@@ -160,6 +160,9 @@ def __init__(self, config: RuntimeConfig | None = None, *,
             model=self.config.explainer_model,
             temperature=self.config.explainer_temperature,
             max_tokens=self.config.explainer_max_tokens,
+            deep_model=self.config.deep_explainer_model,
+            deep_temperature=self.config.deep_explainer_temperature,
+            deep_max_tokens=self.config.deep_explainer_max_tokens,
         )
         self._job_tracker = JobTracker()
         # Command channels are per-credential-set singletons with
@@ -980,7 +983,11 @@ def _write_event(event_type: str, payload: dict) -> None:
                 from supervisor.operator.api import append_timeline_event
                 append_timeline_event(session_log, run_id, event_type, payload)
 
+        escalation_threshold = self.config.clarification_escalation_confidence
+
         def _job():
+            from supervisor.operator.clarification import finalize_clarification
+
             _write_event("clarification_request", {
                 "question": question, "language": language,
             })
@@ -989,12 +996,12 @@ def _job():
             ctx["question"] = question
             result = self._explainer.request_clarification(ctx)
 
-            _write_event("clarification_response", {
-                "question": question,
-                "answer": result.get("answer", ""),
-                "confidence": result.get("confidence"),
-            })
-            return result
+            return finalize_clarification(
+                result,
+                question=question,
+                escalation_threshold=escalation_threshold,
+                write_event=_write_event,
+            )
 
         job_id = self._job_tracker.submit("clarification", _job)
         return {"ok": True, "job_id": job_id}

supervisor/llm/explainer_client.py

@@ -49,23 +49,36 @@ class ExplainerClient:
     Parameters
     ----------
     model : str | None
-        A LiteLLM model identifier for routine explanations.
-        Set to ``None`` for stub mode.
+        LiteLLM model identifier for routine explanations (explain_run,
+        explain_exchange, request_clarification). ``None`` → stub mode.
     temperature : float
         Sampling temperature (higher than judge — explanations are softer).
     max_tokens : int
         Max response tokens (larger than judge — explanations are longer).
+    deep_model : str | None
+        Optional heavier model used only for drift / codebase-aware analysis
+        (``assess_drift``). When ``None``, drift falls back to ``model``.
+        Lets operators pay for a stronger analysis only when it matters.
+    deep_temperature, deep_max_tokens
+        Sampling params applied when ``deep_model`` is used.
     """
 
     def __init__(
         self,
         model: str | None = None,
         temperature: float = 0.3,
         max_tokens: int = 1024,
+        *,
+        deep_model: str | None = None,
+        deep_temperature: float = 0.2,
+        deep_max_tokens: int = 2048,
     ):
         self.model = model
         self.temperature = temperature
         self.max_tokens = max_tokens
+        self.deep_model = deep_model
+        self.deep_temperature = deep_temperature
+        self.deep_max_tokens = deep_max_tokens
 
     # ------------------------------------------------------------------
     # Public methods
@@ -96,15 +109,32 @@ def explain_exchange(self, context: dict[str, Any]) -> dict[str, Any]:
         return self._call(prompt, context, fallback=fallback)
 
     def assess_drift(self, context: dict[str, Any]) -> dict[str, Any]:
-        """Assess whether a run is drifting from its approved plan."""
-        if self.model is None:
+        """Assess whether a run is drifting from its approved plan.
+
+        Prefers ``deep_model`` when configured — drift assessment benefits
+        from the stronger reasoner. Falls back to the routine model, then
+        to the stub analysis.
+        """
+        model = self.deep_model or self.model
+        if model is None:
             return self._stub_assess_drift(context)
         fallback = self._stub_assess_drift(context)
         try:
             prompt = _load_prompt("assess_drift.txt")
         except FileNotFoundError:
             logger.warning("assess_drift prompt not found, falling back to stub")
             return fallback
+        # Use falsy check (not `is not None`) so an empty-string
+        # ``SUPERVISOR_DEEP_EXPLAINER_MODEL=`` env var falls through to
+        # the routine model without keeping deep_temperature/deep_max_tokens.
+        if self.deep_model:
+            return self._call(
+                prompt, context,
+                fallback=fallback,
+                model=self.deep_model,
+                temperature=self.deep_temperature,
+                max_tokens=self.deep_max_tokens,
+            )
         return self._call(prompt, context, fallback=fallback)
 
     def request_clarification(self, context: dict[str, Any]) -> dict[str, Any]:
@@ -129,23 +159,30 @@ def _call(
         context: dict[str, Any],
         *,
         fallback: dict[str, Any],
+        model: str | None = None,
+        temperature: float | None = None,
+        max_tokens: int | None = None,
     ) -> dict[str, Any]:
         try:
             import litellm
         except ImportError:
             logger.warning("litellm not installed, falling back to stub")
             return fallback
 
+        effective_model = model or self.model
+        effective_temperature = temperature if temperature is not None else self.temperature
+        effective_max_tokens = max_tokens if max_tokens is not None else self.max_tokens
+
         messages = [
             {"role": "system", "content": system_prompt},
             {"role": "user", "content": json.dumps(context, ensure_ascii=False)},
         ]
         try:
             response = litellm.completion(
-                model=self.model,
+                model=effective_model,
                 messages=messages,
-                temperature=self.temperature,
-                max_tokens=self.max_tokens,
+                temperature=effective_temperature,
+                max_tokens=effective_max_tokens,
             )
             text = response.choices[0].message.content
             result = _parse_json(text)

supervisor/operator/actions.py

@@ -198,6 +198,9 @@ def _make_explainer(ctx: RunContext):
         model=config.explainer_model,
         temperature=config.explainer_temperature,
         max_tokens=config.explainer_max_tokens,
+        deep_model=config.deep_explainer_model,
+        deep_temperature=config.deep_explainer_temperature,
+        deep_max_tokens=config.deep_explainer_max_tokens,
     )
 
 
@@ -397,32 +400,29 @@ def submit_clarification(ctx: RunContext, question: str, *,
 
     # ASYNC_LOCAL
     explainer = _make_explainer(ctx)
+    escalation_threshold = ctx.load_config().clarification_escalation_confidence
 
     def _job() -> dict:
         from supervisor.operator.api import append_timeline_event
+        from supervisor.operator.clarification import finalize_clarification
 
-        # Record the question in the timeline
-        append_timeline_event(
-            ctx.session_log_path, ctx.run_id,
-            "clarification_request",
-            {"question": question, "language": language},
-        )
+        def _write(event_type: str, payload: dict[str, Any]) -> None:
+            append_timeline_event(
+                ctx.session_log_path, ctx.run_id, event_type, payload,
+            )
+
+        _write("clarification_request", {"question": question, "language": language})
 
         context = build_explainer_context(ctx, language=language)
         context["question"] = question
         result = explainer.request_clarification(context)
 
-        # Record the answer in the timeline
-        append_timeline_event(
-            ctx.session_log_path, ctx.run_id,
-            "clarification_response",
-            {
-                "question": question,
-                "answer": result.get("answer", ""),
-                "confidence": result.get("confidence"),
-            },
+        return finalize_clarification(
+            result,
+            question=question,
+            escalation_threshold=escalation_threshold,
+            write_event=_write,
         )
-        return result
 
     job_id = _local_jobs.submit("clarification", _job)
     return OperatorJob(job_id=job_id, source="local")

supervisor/operator/clarification.py

@@ -0,0 +1,59 @@
+"""Shared post-processing for ``ExplainerClient.request_clarification`` results.
+
+Used by both the local async path (``supervisor.operator.actions``) and
+the daemon path (``supervisor.daemon.server``). Keeping the escalation
+rule + event payloads here ensures both code paths stay on the same
+wire contract.
+"""
+from __future__ import annotations
+
+from typing import Any, Callable
+
+from supervisor.operator.models import coerce_confidence
+
+EventWriter = Callable[[str, dict[str, Any]], None]
+
+
+def finalize_clarification(
+    result: dict[str, Any],
+    *,
+    question: str,
+    escalation_threshold: float,
+    write_event: EventWriter,
+) -> dict[str, Any]:
+    """Annotate *result* with ``escalation_recommended`` and emit timeline events.
+
+    Emits in order:
+      - ``explainer_answer`` — source-tagged answer, for channel adapters.
+      - ``clarification_response`` — legacy event, preserved for back-compat.
+      - ``clarification_escalation_recommended`` — only when escalation fires.
+
+    Mutates *result* in place and returns it.
+    """
+    confidence = result.get("confidence")
+    conf_value = coerce_confidence(confidence)
+    escalate = conf_value is not None and conf_value < escalation_threshold
+    result["escalation_recommended"] = escalate
+
+    answer = result.get("answer", "")
+
+    write_event("explainer_answer", {
+        "source": "explainer",
+        "question": question,
+        "answer": answer,
+        "confidence": confidence,
+    })
+    write_event("clarification_response", {
+        "question": question,
+        "answer": answer,
+        "confidence": confidence,
+        "source": "explainer",
+        "escalation_recommended": escalate,
+    })
+    if escalate:
+        write_event("clarification_escalation_recommended", {
+            "question": question,
+            "confidence": confidence,
+            "threshold": escalation_threshold,
+        })
+    return result