feat: operator-initiated clarification escalation — 0.3.7

CHANGELOG.md

@@ -2,6 +2,16 @@
 
 ## Unreleased
 
+## 0.3.7 (2026-04-22)
+
+### Operator-initiated clarification escalation
+
+- Added `thin-supervisor clarify <run_id> <question> [--escalate] [--operator <name>]` — the first CLI entry point that runs a clarification and, on operator opt-in, records an escalation decision against the run's session log.
+- Added a daemon IPC handler (`escalate_clarification`) and matching `DaemonClient.escalate_clarification` / `operator.actions.do_escalate_clarification` so every surface (CLI, TUI, IM) routes through the same canonical action and capability model. Works through the daemon when one is attached; falls back to writing directly to the run's session log for local/completed runs.
+- Wired a TUI `E` keybind: when a clarification answer comes back below the escalation threshold, `format_clarification` surfaces an "escalate" hint and pressing `E` records the escalation against the selected run without leaving the channel.
+- Added a `/escalate <run_id> [question]` IM command to `command_dispatch`. With no explicit question it pulls the most recent `clarification_response` from the run's session log; with an override it uses the operator's text verbatim.
+- Emits a single `clarification_escalated_to_worker` timeline event — `transport="pending_0_3_8"` — so the audit trail makes clear that the actual side-instruction transport to the worker is deferred to 0.3.8 while the operator decision is already durable.
+
 ## 0.3.6 (2026-04-22)
 
 ### Operator Channel polish

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.6)
+## Current Status (0.3.7)
 
+- **Operator-initiated clarification escalation.** When a clarification answer comes back below the configured confidence threshold, operators can escalate the question to the worker from any channel: `thin-supervisor clarify <run_id> <question> --escalate`, the TUI `E` keybind, or `/escalate <run_id> [question]` in IM. Each escalation writes a single `clarification_escalated_to_worker` timeline event so the decision is durably auditable. Actual side-instruction transport to the worker is deferred to 0.3.8; 0.3.7 ships the end-to-end operator UX and the audit surface.
 - **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.

pyproject.toml

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

supervisor/app.py

@@ -833,6 +833,107 @@ def cmd_observe(args):
     return 0
 
 
+def cmd_clarify(args):
+    """Ask the explainer a free-form question about a run, optionally escalate.
+
+    The answer is served by ``ExplainerClient.request_clarification`` — same
+    async job that powers TUI/IM clarification. When ``--escalate`` is set,
+    records a ``clarification_escalated_to_worker`` session event for
+    operator audit (actual worker transport ships in 0.3.8).
+    """
+    from supervisor.operator.actions import (
+        ActionUnavailable,
+        do_escalate_clarification,
+        poll_job,
+        submit_clarification,
+    )
+    from supervisor.operator.run_context import RunContext
+    from supervisor.operator.session_index import find_session
+
+    rec = find_session(args.run_id)
+    if rec is None:
+        print(f"Error: run not found: {args.run_id}")
+        return 1
+
+    ctx = RunContext.from_run_dict({
+        "run_id": rec.run_id,
+        "worktree": rec.worktree_root,
+        "tag": rec.tag or "local",
+        "top_state": rec.top_state,
+        "pane_target": rec.pane_target,
+        "socket": rec.daemon_socket,
+    })
+
+    question = " ".join(args.question).strip()
+    if not question:
+        print("Error: question is required")
+        return 1
+
+    try:
+        job = submit_clarification(ctx, question, language=args.language)
+    except ActionUnavailable as e:
+        print(f"Error: clarify unavailable: {e}")
+        return 1
+    except (ConnectionRefusedError, FileNotFoundError, OSError) as e:
+        # Daemon may die between run lookup and socket connect; same
+        # narrow race handled in cmd_observe.
+        print(f"Error: daemon unreachable for {rec.run_id}: {e}")
+        return 1
+
+    deadline = _time.time() + max(1, args.timeout)
+    result: dict = {}
+    while _time.time() < deadline:
+        try:
+            status = poll_job(ctx, job)
+        except (ConnectionRefusedError, FileNotFoundError, OSError) as e:
+            print(f"Error: daemon unreachable for {rec.run_id}: {e}")
+            return 1
+        s = status.get("status", "")
+        if s == "completed":
+            result = status.get("result", {}) or {}
+            break
+        if s == "failed":
+            print(f"Error: clarify failed: {status.get('error', 'unknown')}")
+            return 1
+        _time.sleep(0.2)
+    else:
+        print(f"Error: clarify timed out after {args.timeout}s")
+        return 1
+
+    answer = result.get("answer", "")
+    confidence = result.get("confidence")
+    escalation_recommended = result.get("escalation_recommended", False)
+
+    print(f"Run:       {rec.run_id}")
+    print(f"Q:         {question}")
+    print(f"A:         {answer}")
+    if confidence is not None:
+        print(f"Confidence: {confidence}")
+    print(f"Escalate?  {'yes (recommended)' if escalation_recommended else 'no'}")
+
+    if args.escalate:
+        try:
+            esc = do_escalate_clarification(
+                ctx, question,
+                language=args.language,
+                reason=(
+                    "low_confidence" if escalation_recommended
+                    else "operator_initiated"
+                ),
+                operator=args.operator,
+                confidence=confidence,
+            )
+        except ActionUnavailable as e:
+            print(f"Error: escalate unavailable: {e}")
+            return 1
+        except (ConnectionRefusedError, FileNotFoundError, OSError) as e:
+            print(f"Error: daemon unreachable for {rec.run_id}: {e}")
+            return 1
+        print(f"Escalated: id={esc['escalation_id']} via={esc['source']}")
+        print("  (audit-only in 0.3.7; worker-side routing ships in 0.3.8)")
+    return 0
+
+
 def cmd_note(args):
     """Shared notes for cross-run collaboration."""
     from supervisor.daemon.client import DaemonClient
@@ -2945,6 +3046,30 @@ def build_runtime_parser() -> argparse.ArgumentParser:
     p_observe = sub.add_parser("observe", help="Read-only observation of a run")
     p_observe.add_argument("run_id", help="Run ID to observe")
 
+    p_clarify = sub.add_parser(
+        "clarify",
+        help="Ask the explainer about a run; optionally escalate to the worker",
+    )
+    p_clarify.add_argument("run_id", help="Run ID")
+    p_clarify.add_argument("question", nargs="+", help="Free-form question")
+    p_clarify.add_argument(
+        "--language", default="en", choices=["en", "zh"],
+        help="Answer language",
+    )
+    p_clarify.add_argument(
+        "--escalate", action="store_true",
+        help="Record an operator decision to escalate this to the worker "
+             "(audit-only in 0.3.7; worker transport in 0.3.8)",
+    )
+    p_clarify.add_argument(
+        "--operator", default="",
+        help="Optional operator identifier recorded in the escalation event",
+    )
+    p_clarify.add_argument(
+        "--timeout", type=int, default=30,
+        help="Clarify poll timeout (seconds)",
+    )
+
     p_note = sub.add_parser("note", help="Shared notes for cross-run collaboration")
     note_sub = p_note.add_subparsers(dest="note_action")
     p_note_add = note_sub.add_parser("add", help="Add a note")
@@ -3202,6 +3327,8 @@ def main():
         sys.exit(cmd_pane_owner(args))
     elif args.command == "observe":
         sys.exit(cmd_observe(args))
+    elif args.command == "clarify":
+        sys.exit(cmd_clarify(args))
     elif args.command == "note":
         if args.note_action in ("add", "list"):
             sys.exit(cmd_note(args))

supervisor/daemon/client.py

@@ -160,6 +160,31 @@ def request_clarification(self, run_id: str, question: str, *,
             "language": language,
         })
 
+    def escalate_clarification(
+        self,
+        run_id: str,
+        question: str,
+        *,
+        language: str = "en",
+        reason: str = "operator_initiated",
+        operator: str = "",
+        confidence: float | None = None,
+    ) -> dict:
+        """Record an operator's decision to escalate a clarification.
+
+        Audit-only in 0.3.7; the actual side-instruction transport to the
+        worker is wired in 0.3.8. Returns ``{ok, escalation_id}``.
+        """
+        return self._request({
+            "action": "escalate_clarification",
+            "run_id": run_id,
+            "question": question,
+            "language": language,
+            "reason": reason,
+            "operator": operator,
+            "confidence": confidence,
+        })
+
     def get_job(self, job_id: str) -> dict:
         """Poll for async job result."""
         return self._request({"action": "get_job", "job_id": job_id})

supervisor/daemon/server.py

@@ -373,6 +373,8 @@ def _handle_connection(self, conn: socket.socket) -> None:
             response = self._do_assess_drift(request)
         elif action == "request_clarification":
             response = self._do_request_clarification(request)
+        elif action == "escalate_clarification":
+            response = self._do_escalate_clarification(request)
         elif action == "get_job":
             response = self._do_get_job(request.get("job_id", ""))
         elif action == "external_task_create":
@@ -1006,6 +1008,52 @@ def _job():
         job_id = self._job_tracker.submit("clarification", _job)
         return {"ok": True, "job_id": job_id}
 
+    def _do_escalate_clarification(self, request: dict) -> dict:
+        """Record an operator's decision to escalate a clarification to the worker.
+
+        Audit-only in 0.3.7 — actual transport of the question to the worker
+        (via a side-instruction queue drained by SupervisorLoop) and
+        worker-reply capture ship in 0.3.8. The event written here is the
+        durable record that an operator chose to escalate.
+        """
+        run_id = request.get("run_id", "")
+        question = request.get("question", "")
+        language = request.get("language", "en")
+        reason = request.get("reason", "operator_initiated")
+        operator = request.get("operator", "")
+        confidence = request.get("confidence")
+
+        state, session_log = self._resolve_run_store(run_id)
+        if state is None:
+            return {"ok": False, "error": f"run {run_id} not found"}
+        if not question:
+            return {"ok": False, "error": "question is required"}
+
+        escalation_id = uuid.uuid4().hex[:16]
+        payload = {
+            "escalation_id": escalation_id,
+            "question": question,
+            "language": language,
+            "reason": reason,
+            "operator": operator,
+            "confidence": confidence,
+            "transport": "pending_0_3_8",
+        }
+
+        with self._lock:
+            entry = self._runs.get(run_id)
+        store = entry.store if entry else None
+        if store:
+            store.append_session_event(
+                run_id, "clarification_escalated_to_worker", payload,
+            )
+        elif session_log:
+            from supervisor.operator.api import append_timeline_event
+            append_timeline_event(
+                session_log, run_id, "clarification_escalated_to_worker", payload,
+            )
+        return {"ok": True, "escalation_id": escalation_id}
+
     def _do_get_job(self, job_id: str) -> dict:
         job = self._job_tracker.get(job_id)
         if not job:

supervisor/operator/actions.py

@@ -428,6 +428,68 @@ def _write(event_type: str, payload: dict[str, Any]) -> None:
     return OperatorJob(job_id=job_id, source="local")
 
 
+def do_escalate_clarification(
+    ctx: RunContext,
+    question: str,
+    *,
+    language: str = "en",
+    reason: str = "operator_initiated",
+    operator: str = "",
+    confidence: float | None = None,
+) -> dict[str, Any]:
+    """Record an operator's decision to escalate a clarification to the worker.
+
+    Audit-only in 0.3.7. Emits ``clarification_escalated_to_worker`` into
+    the session log. Actual worker transport + reply capture ship in 0.3.8.
+
+    Returns ``{"escalation_id": <hex>, "source": "daemon"|"local"}``.
+    """
+    import uuid
+
+    question = question.strip()
+    if not question:
+        raise ActionUnavailable("question is required")
+
+    caps = ctx.capabilities()
+    # Piggy-back on explain capability — same preconditions (need a
+    # resolvable run_id + session_log).
+    if caps.explain == ActionMode.UNAVAILABLE:
+        raise ActionUnavailable(caps.unavailable_reasons.get("explain", "unavailable"))
+
+    if caps.explain == ActionMode.ASYNC_DAEMON:
+        client = ctx.get_client()
+        resp = client.escalate_clarification(
+            ctx.run_id, question,
+            language=language, reason=reason,
+            operator=operator, confidence=confidence,
+        )
+        if not resp.get("ok", False):
+            raise ActionUnavailable(resp.get("error", "escalate failed"))
+        return {
+            "escalation_id": resp.get("escalation_id", ""),
+            "source": "daemon",
+        }
+
+    # ASYNC_LOCAL — write directly to the session log
+    from supervisor.operator.api import append_timeline_event
+
+    escalation_id = uuid.uuid4().hex[:16]
+    append_timeline_event(
+        ctx.session_log_path, ctx.run_id,
+        "clarification_escalated_to_worker",
+        {
+            "escalation_id": escalation_id,
+            "question": question,
+            "language": language,
+            "reason": reason,
+            "operator": operator,
+            "confidence": confidence,
+            "transport": "pending_0_3_8",
+        },
+    )
+    return {"escalation_id": escalation_id, "source": "local"}
+
+
 def poll_job(ctx: RunContext, job: OperatorJob) -> dict[str, Any]:
     """Poll for an async job result. Non-blocking.