From 56524b9249a2d8b7571fa74f0398fabd5a9d3fd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=8B=E4=BC=A0=E8=83=9C?= Date: Wed, 22 Apr 2026 16:41:45 -0700 Subject: [PATCH 1/2] feat: operator channel polish for 0.3.6 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - deep_explainer_model (+ temperature/max_tokens) opts drift assessment into a heavier reasoner while routine explanations stay on explainer_model. Falls back to the routine model, then the structured stub, when the deep model is unset or the call fails. - DriftAssessment and ExchangeView frozen dataclasses (from_dict accepts legacy last_*_summary and post-explainer shapes). Unknown drift status collapses to "watch"; non-numeric confidence collapses to None. - Dedicated explainer_answer session-log event with source="explainer", emitted alongside the existing clarification_response so future worker-side clarification flows can be distinguished without touching existing consumers. - Confidence-gated escalation: when answer confidence < clarification_escalation_confidence (default 0.4), the result carries escalation_recommended=True and a clarification_escalation_recommended event is written. Advisory only — operator still routes. --- CHANGELOG.md | 9 ++ README.md | 3 +- pyproject.toml | 2 +- supervisor/config.py | 28 ++++++ supervisor/daemon/server.py | 29 ++++++- supervisor/llm/explainer_client.py | 48 +++++++++-- supervisor/operator/actions.py | 41 ++++++++- supervisor/operator/models.py | 121 ++++++++++++++++++++++++++ tests/test_operator_actions.py | 20 ++++- tests/test_operator_models.py | 132 +++++++++++++++++++++++++++++ 10 files changed, 418 insertions(+), 15 deletions(-) create mode 100644 tests/test_operator_models.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 16a2bf1..b9360da 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 3a4cc30..850666f 100644 --- a/README.md +++ b/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). diff --git a/pyproject.toml b/pyproject.toml index 2364f8a..45afecf 100644 --- a/pyproject.toml +++ b/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"} diff --git a/supervisor/config.py b/supervisor/config.py index 790255c..786d089 100644 --- a/supervisor/config.py +++ b/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" diff --git a/supervisor/daemon/server.py b/supervisor/daemon/server.py index b004576..f4a6672 100644 --- a/supervisor/daemon/server.py +++ b/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,6 +983,8 @@ 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(): _write_event("clarification_request", { "question": question, "language": language, @@ -989,11 +994,33 @@ def _job(): ctx["question"] = question result = self._explainer.request_clarification(ctx) + confidence = result.get("confidence") + try: + conf_value = float(confidence) if confidence is not None else None + except (TypeError, ValueError): + conf_value = None + escalate = conf_value is not None and conf_value < escalation_threshold + result["escalation_recommended"] = escalate + + _write_event("explainer_answer", { + "source": "explainer", + "question": question, + "answer": result.get("answer", ""), + "confidence": confidence, + }) _write_event("clarification_response", { "question": question, "answer": result.get("answer", ""), - "confidence": result.get("confidence"), + "confidence": confidence, + "source": "explainer", + "escalation_recommended": escalate, }) + if escalate: + _write_event("clarification_escalation_recommended", { + "question": question, + "confidence": confidence, + "threshold": escalation_threshold, + }) return result job_id = self._job_tracker.submit("clarification", _job) diff --git a/supervisor/llm/explainer_client.py b/supervisor/llm/explainer_client.py index 6f81bc6..a7209c1 100644 --- a/supervisor/llm/explainer_client.py +++ b/supervisor/llm/explainer_client.py @@ -49,12 +49,18 @@ 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__( @@ -62,10 +68,17 @@ def __init__( 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,8 +109,14 @@ 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: @@ -105,6 +124,14 @@ def assess_drift(self, context: dict[str, Any]) -> dict[str, Any]: except FileNotFoundError: logger.warning("assess_drift prompt not found, falling back to stub") return fallback + if self.deep_model is not None: + 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,6 +156,9 @@ 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 @@ -136,16 +166,20 @@ def _call( 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) diff --git a/supervisor/operator/actions.py b/supervisor/operator/actions.py index 86c1f15..524eec0 100644 --- a/supervisor/operator/actions.py +++ b/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,6 +400,7 @@ 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 @@ -412,16 +416,49 @@ def _job() -> dict: context["question"] = question result = explainer.request_clarification(context) - # Record the answer in the timeline + confidence = result.get("confidence") + try: + conf_value = float(confidence) if confidence is not None else None + except (TypeError, ValueError): + conf_value = None + escalate = conf_value is not None and conf_value < escalation_threshold + result["escalation_recommended"] = escalate + + # Emit explainer_answer — explicit source marker distinguishes + # this from a future worker-side response. + append_timeline_event( + ctx.session_log_path, ctx.run_id, + "explainer_answer", + { + "source": "explainer", + "question": question, + "answer": result.get("answer", ""), + "confidence": confidence, + }, + ) + # Record the answer in the timeline (legacy event kept for + # backward compatibility with existing consumers). append_timeline_event( ctx.session_log_path, ctx.run_id, "clarification_response", { "question": question, "answer": result.get("answer", ""), - "confidence": result.get("confidence"), + "confidence": confidence, + "source": "explainer", + "escalation_recommended": escalate, }, ) + if escalate: + append_timeline_event( + ctx.session_log_path, ctx.run_id, + "clarification_escalation_recommended", + { + "question": question, + "confidence": confidence, + "threshold": escalation_threshold, + }, + ) return result job_id = _local_jobs.submit("clarification", _job) diff --git a/supervisor/operator/models.py b/supervisor/operator/models.py index eca5442..f12693e 100644 --- a/supervisor/operator/models.py +++ b/supervisor/operator/models.py @@ -194,6 +194,127 @@ def to_dict(self) -> dict[str, Any]: } +@dataclass(frozen=True) +class ExchangeView: + """A human-readable "what just happened between supervisor and worker" view. + + Built from the last checkpoint + last instruction in a run's timeline. + Channel adapters (TUI, Telegram, Lark) consume this instead of + stringifying raw dicts. The explainer-filled fields (``explanation_*``) + are optional — they're populated only when the operator asks for a + natural-language translation of the exchange. + """ + run_id: str + window_start: str # ISO timestamp of earliest event shown + window_end: str # ISO timestamp of latest event shown + worker_text_excerpt: str # last checkpoint summary/content + supervisor_instruction_excerpt: str + checkpoint_excerpt: str # raw checkpoint marker text, if any + explanation_zh: str = "" + explanation_en: str = "" + confidence: float | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any], *, run_id: str = "") -> "ExchangeView": + """Wrap a raw exchange dict (from ``operator.api.recent_exchange``). + + Accepts both the pre-explainer shape (``last_checkpoint_summary`` + + ``last_instruction_summary``) and the post-explainer shape + (``explanation``, ``confidence``, …). + """ + conf = data.get("confidence") + if conf is not None: + try: + conf = float(conf) + except (TypeError, ValueError): + conf = None + return cls( + run_id=str(data.get("run_id") or run_id or ""), + window_start=str(data.get("window_start", "")), + window_end=str(data.get("window_end", "")), + worker_text_excerpt=str( + data.get("worker_text_excerpt") + or data.get("last_checkpoint_summary", "") + ), + supervisor_instruction_excerpt=str( + data.get("supervisor_instruction_excerpt") + or data.get("last_instruction_summary", "") + ), + checkpoint_excerpt=str(data.get("checkpoint_excerpt", "")), + explanation_zh=str(data.get("explanation_zh", "")), + explanation_en=str(data.get("explanation_en", "")), + confidence=conf, + ) + + def to_dict(self) -> dict[str, Any]: + return { + "run_id": self.run_id, + "window_start": self.window_start, + "window_end": self.window_end, + "worker_text_excerpt": self.worker_text_excerpt, + "supervisor_instruction_excerpt": self.supervisor_instruction_excerpt, + "checkpoint_excerpt": self.checkpoint_excerpt, + "explanation_zh": self.explanation_zh, + "explanation_en": self.explanation_en, + "confidence": self.confidence, + } + + +_DRIFT_STATUSES = frozenset({"on_track", "watch", "drifting", "blocked"}) + + +@dataclass(frozen=True) +class DriftAssessment: + """Structured answer to "is this run still on track?" + + Built from ``ExplainerClient.assess_drift`` output. The raw dict is + kept as the wire format across daemon IPC; this dataclass is for + type-safe consumption by TUI / IM channels. + """ + run_id: str + status: str # on_track | watch | drifting | blocked + reasons: list[str] + evidence: list[str] + codebase_signals: list[str] + recommended_action: str + confidence: float | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any], *, run_id: str = "") -> "DriftAssessment": + status = str(data.get("status", "") or "").strip().lower() + if status not in _DRIFT_STATUSES: + status = "watch" # unknown → conservative bucket + conf = data.get("confidence") + if conf is not None: + try: + conf = float(conf) + except (TypeError, ValueError): + conf = None + return cls( + run_id=str(data.get("run_id") or run_id or ""), + status=status, + reasons=[str(x) for x in (data.get("reasons") or [])], + evidence=[str(x) for x in (data.get("evidence") or [])], + codebase_signals=[str(x) for x in (data.get("codebase_signals") or [])], + recommended_action=str( + data.get("recommended_action") + or data.get("recommended_operator_action", "") + ), + confidence=conf, + ) + + def to_dict(self) -> dict[str, Any]: + return { + "run_id": self.run_id, + "status": self.status, + "reasons": list(self.reasons), + "evidence": list(self.evidence), + "codebase_signals": list(self.codebase_signals), + "recommended_action": self.recommended_action, + "confidence": self.confidence, + } + + @dataclass(frozen=True) class SystemSnapshot: """Top-level projection consumed by `overview`, `status`, `tui`.""" diff --git a/tests/test_operator_actions.py b/tests/test_operator_actions.py index 8fecbe6..c2900fa 100644 --- a/tests/test_operator_actions.py +++ b/tests/test_operator_actions.py @@ -438,6 +438,10 @@ def test_local_path(self, tmp_path): explainer_model=None, explainer_temperature=0.3, explainer_max_tokens=1024, + deep_explainer_model=None, + deep_explainer_temperature=0.2, + deep_explainer_max_tokens=2048, + clarification_escalation_confidence=0.4, ) job = submit_clarification(ctx, "what happened?", language="zh") assert job.source == "local" @@ -452,16 +456,26 @@ def test_local_path(self, tmp_path): assert j.status == "completed" # Stub mode should include the question in the answer assert "what happened?" in j.result.get("answer", "") + # Stub confidence (0.1) is below default threshold (0.4) → escalation + assert j.result.get("escalation_recommended") is True # Verify clarification events written to session_log session_log = run_dir / "session_log.jsonl" assert session_log.exists() events = [json.loads(line) for line in session_log.read_text().strip().splitlines()] - assert len(events) == 2 - assert events[0]["event_type"] == "clarification_request" + types = [e["event_type"] for e in events] + assert types == [ + "clarification_request", + "explainer_answer", + "clarification_response", + "clarification_escalation_recommended", + ] assert events[0]["payload"]["question"] == "what happened?" - assert events[1]["event_type"] == "clarification_response" + assert events[1]["payload"]["source"] == "explainer" assert "what happened?" in events[1]["payload"]["answer"] + assert events[2]["payload"]["source"] == "explainer" + assert events[2]["payload"]["escalation_recommended"] is True + assert events[3]["payload"]["threshold"] == 0.4 class TestClarificationSummarizers: diff --git a/tests/test_operator_models.py b/tests/test_operator_models.py new file mode 100644 index 0000000..2f747d7 --- /dev/null +++ b/tests/test_operator_models.py @@ -0,0 +1,132 @@ +"""Tests for supervisor.operator.models — frozen projections. + +Covers the new DriftAssessment + ExchangeView dataclasses added in 0.3.6. +RunSnapshot / RunTimelineEvent / RunEventPlaneSummary are exercised by +test_operator_api.py. +""" +from __future__ import annotations + +import pytest + +from supervisor.operator.models import DriftAssessment, ExchangeView + + +class TestDriftAssessment: + def test_from_dict_wraps_explainer_output(self): + raw = { + "status": "drifting", + "reasons": ["retry budget near cap", "node mismatch"], + "evidence": ["retries_used=4"], + "codebase_signals": ["supervisor/loop.py touched"], + "recommended_action": "Pause and review", + "confidence": 0.72, + } + d = DriftAssessment.from_dict(raw, run_id="run-1") + assert d.run_id == "run-1" + assert d.status == "drifting" + assert d.reasons == ["retry budget near cap", "node mismatch"] + assert d.evidence == ["retries_used=4"] + assert d.codebase_signals == ["supervisor/loop.py touched"] + assert d.recommended_action == "Pause and review" + assert d.confidence == pytest.approx(0.72) + + def test_unknown_status_falls_back_to_watch(self): + d = DriftAssessment.from_dict({"status": "wild_guess"}) + assert d.status == "watch" + + def test_missing_lists_default_to_empty(self): + d = DriftAssessment.from_dict({"status": "on_track"}) + assert d.reasons == [] + assert d.evidence == [] + assert d.codebase_signals == [] + assert d.recommended_action == "" + assert d.confidence is None + + def test_recommended_action_plan_alias(self): + d = DriftAssessment.from_dict( + {"status": "watch", "recommended_operator_action": "monitor"}, + ) + assert d.recommended_action == "monitor" + + def test_non_numeric_confidence_coerced_to_none(self): + d = DriftAssessment.from_dict({"status": "watch", "confidence": "maybe"}) + assert d.confidence is None + + def test_to_dict_roundtrip(self): + raw = { + "status": "on_track", + "reasons": ["r1"], + "evidence": [], + "codebase_signals": [], + "recommended_action": "none", + "confidence": 0.9, + } + d = DriftAssessment.from_dict(raw, run_id="r") + out = d.to_dict() + assert out["run_id"] == "r" + assert out["status"] == "on_track" + assert out["confidence"] == 0.9 + # Lists are copied, not shared — operator consumers can mutate safely. + out["reasons"].append("mutated") + assert d.reasons == ["r1"] + + def test_frozen(self): + d = DriftAssessment.from_dict({"status": "watch"}) + with pytest.raises(Exception): + d.status = "on_track" # type: ignore[misc] + + +class TestExchangeView: + def test_from_legacy_recent_exchange_shape(self): + """api.recent_exchange() returns a dict with last_*_summary keys.""" + raw = { + "last_checkpoint_summary": "worker finished step A", + "last_instruction_summary": "proceed to step B", + } + ex = ExchangeView.from_dict(raw, run_id="run-x") + assert ex.run_id == "run-x" + assert ex.worker_text_excerpt == "worker finished step A" + assert ex.supervisor_instruction_excerpt == "proceed to step B" + assert ex.explanation_zh == "" + assert ex.explanation_en == "" + assert ex.confidence is None + + def test_from_explainer_shape_preserves_explanations(self): + raw = { + "run_id": "run-y", + "window_start": "2026-04-22T10:00:00Z", + "window_end": "2026-04-22T10:05:00Z", + "worker_text_excerpt": "reading file", + "supervisor_instruction_excerpt": "keep going", + "checkpoint_excerpt": "...", + "explanation_zh": "worker 正在读取文件", + "explanation_en": "worker is reading the file", + "confidence": 0.82, + } + ex = ExchangeView.from_dict(raw) + assert ex.run_id == "run-y" + assert ex.explanation_zh == "worker 正在读取文件" + assert ex.explanation_en == "worker is reading the file" + assert ex.confidence == pytest.approx(0.82) + + def test_empty_input_yields_empty_view(self): + ex = ExchangeView.from_dict({}) + assert ex.run_id == "" + assert ex.worker_text_excerpt == "" + assert ex.confidence is None + + def test_non_numeric_confidence_coerced_to_none(self): + ex = ExchangeView.from_dict({"confidence": "n/a"}) + assert ex.confidence is None + + def test_to_dict_roundtrip(self): + raw = {"last_checkpoint_summary": "x", "last_instruction_summary": "y"} + ex = ExchangeView.from_dict(raw, run_id="r") + out = ex.to_dict() + assert out["worker_text_excerpt"] == "x" + assert out["supervisor_instruction_excerpt"] == "y" + + def test_frozen(self): + ex = ExchangeView.from_dict({}) + with pytest.raises(Exception): + ex.run_id = "tampered" # type: ignore[misc] From bf15c3a92737d7faae2e916f272119eff8e8851f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=8B=E4=BC=A0=E8=83=9C?= Date: Wed, 22 Apr 2026 16:48:50 -0700 Subject: [PATCH 2/2] review: address 0.3.6 round-1 findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix Devin finding: assess_drift used `is not None` on deep_model while line 118 used a falsy check, so an empty-string SUPERVISOR_DEEP_EXPLAINER_MODEL env var fell into the deep branch and kept deep_temperature/deep_max_tokens even though the effective model dropped back to the routine one. Switch to a consistent falsy check so empty-string is treated as unset. - Extract coerce_confidence() helper in operator.models — single source of truth for non-numeric → None coercion, now used by ExchangeView, DriftAssessment, and the clarification pipeline. - Extract finalize_clarification() helper in operator.clarification so the daemon and local-async paths share the same event sequence (explainer_answer + clarification_response + optional escalation) and escalation threshold semantics. Eliminates the two copies flagged by gemini-code-assist. --- supervisor/daemon/server.py | 36 ++++------------ supervisor/llm/explainer_client.py | 5 ++- supervisor/operator/actions.py | 61 ++++++---------------------- supervisor/operator/clarification.py | 59 +++++++++++++++++++++++++++ supervisor/operator/models.py | 29 +++++++------ 5 files changed, 100 insertions(+), 90 deletions(-) create mode 100644 supervisor/operator/clarification.py diff --git a/supervisor/daemon/server.py b/supervisor/daemon/server.py index f4a6672..7c66bc2 100644 --- a/supervisor/daemon/server.py +++ b/supervisor/daemon/server.py @@ -986,6 +986,8 @@ def _write_event(event_type: str, payload: dict) -> None: escalation_threshold = self.config.clarification_escalation_confidence def _job(): + from supervisor.operator.clarification import finalize_clarification + _write_event("clarification_request", { "question": question, "language": language, }) @@ -994,34 +996,12 @@ def _job(): ctx["question"] = question result = self._explainer.request_clarification(ctx) - confidence = result.get("confidence") - try: - conf_value = float(confidence) if confidence is not None else None - except (TypeError, ValueError): - conf_value = None - escalate = conf_value is not None and conf_value < escalation_threshold - result["escalation_recommended"] = escalate - - _write_event("explainer_answer", { - "source": "explainer", - "question": question, - "answer": result.get("answer", ""), - "confidence": confidence, - }) - _write_event("clarification_response", { - "question": question, - "answer": result.get("answer", ""), - "confidence": confidence, - "source": "explainer", - "escalation_recommended": escalate, - }) - if escalate: - _write_event("clarification_escalation_recommended", { - "question": question, - "confidence": confidence, - "threshold": escalation_threshold, - }) - 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} diff --git a/supervisor/llm/explainer_client.py b/supervisor/llm/explainer_client.py index a7209c1..b04ff06 100644 --- a/supervisor/llm/explainer_client.py +++ b/supervisor/llm/explainer_client.py @@ -124,7 +124,10 @@ def assess_drift(self, context: dict[str, Any]) -> dict[str, Any]: except FileNotFoundError: logger.warning("assess_drift prompt not found, falling back to stub") return fallback - if self.deep_model is not None: + # 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, diff --git a/supervisor/operator/actions.py b/supervisor/operator/actions.py index 524eec0..9358bb8 100644 --- a/supervisor/operator/actions.py +++ b/supervisor/operator/actions.py @@ -404,62 +404,25 @@ def submit_clarification(ctx: RunContext, question: str, *, 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) - confidence = result.get("confidence") - try: - conf_value = float(confidence) if confidence is not None else None - except (TypeError, ValueError): - conf_value = None - escalate = conf_value is not None and conf_value < escalation_threshold - result["escalation_recommended"] = escalate - - # Emit explainer_answer — explicit source marker distinguishes - # this from a future worker-side response. - append_timeline_event( - ctx.session_log_path, ctx.run_id, - "explainer_answer", - { - "source": "explainer", - "question": question, - "answer": result.get("answer", ""), - "confidence": confidence, - }, - ) - # Record the answer in the timeline (legacy event kept for - # backward compatibility with existing consumers). - append_timeline_event( - ctx.session_log_path, ctx.run_id, - "clarification_response", - { - "question": question, - "answer": result.get("answer", ""), - "confidence": confidence, - "source": "explainer", - "escalation_recommended": escalate, - }, + return finalize_clarification( + result, + question=question, + escalation_threshold=escalation_threshold, + write_event=_write, ) - if escalate: - append_timeline_event( - ctx.session_log_path, ctx.run_id, - "clarification_escalation_recommended", - { - "question": question, - "confidence": confidence, - "threshold": escalation_threshold, - }, - ) - return result job_id = _local_jobs.submit("clarification", _job) return OperatorJob(job_id=job_id, source="local") diff --git a/supervisor/operator/clarification.py b/supervisor/operator/clarification.py new file mode 100644 index 0000000..d471549 --- /dev/null +++ b/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 diff --git a/supervisor/operator/models.py b/supervisor/operator/models.py index f12693e..6daf529 100644 --- a/supervisor/operator/models.py +++ b/supervisor/operator/models.py @@ -10,6 +10,21 @@ from typing import Any +def coerce_confidence(value: Any) -> float | None: + """Parse a confidence value from raw explainer/IPC payloads. + + Returns ``None`` for missing, non-numeric, or unparseable values. + Shared by ``ExchangeView``, ``DriftAssessment``, and the clarification + pipeline so every consumer applies the same coercion rule. + """ + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + @dataclass(frozen=True) class RunSnapshot: """Current state of a run for operator display. @@ -222,12 +237,7 @@ def from_dict(cls, data: dict[str, Any], *, run_id: str = "") -> "ExchangeView": ``last_instruction_summary``) and the post-explainer shape (``explanation``, ``confidence``, …). """ - conf = data.get("confidence") - if conf is not None: - try: - conf = float(conf) - except (TypeError, ValueError): - conf = None + conf = coerce_confidence(data.get("confidence")) return cls( run_id=str(data.get("run_id") or run_id or ""), window_start=str(data.get("window_start", "")), @@ -284,12 +294,7 @@ def from_dict(cls, data: dict[str, Any], *, run_id: str = "") -> "DriftAssessmen status = str(data.get("status", "") or "").strip().lower() if status not in _DRIFT_STATUSES: status = "watch" # unknown → conservative bucket - conf = data.get("confidence") - if conf is not None: - try: - conf = float(conf) - except (TypeError, ValueError): - conf = None + conf = coerce_confidence(data.get("confidence")) return cls( run_id=str(data.get("run_id") or run_id or ""), status=status,