diff --git a/CHANGELOG.md b/CHANGELOG.md
index b9360da..ce8668f 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/README.md b/README.md
index 850666f..901f9b4 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.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.
diff --git a/pyproject.toml b/pyproject.toml
index 45afecf..7d280f3 100644
--- a/pyproject.toml
+++ b/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"}
diff --git a/supervisor/app.py b/supervisor/app.py
index 07ecdc4..08c7abf 100644
--- a/supervisor/app.py
+++ b/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))
diff --git a/supervisor/daemon/client.py b/supervisor/daemon/client.py
index f59e482..97ab7d3 100644
--- a/supervisor/daemon/client.py
+++ b/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})
diff --git a/supervisor/daemon/server.py b/supervisor/daemon/server.py
index 7c66bc2..a223067 100644
--- a/supervisor/daemon/server.py
+++ b/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:
diff --git a/supervisor/operator/actions.py b/supervisor/operator/actions.py
index 9358bb8..bc2620f 100644
--- a/supervisor/operator/actions.py
+++ b/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.
 
diff --git a/supervisor/operator/command_dispatch.py b/supervisor/operator/command_dispatch.py
index 6d6daf8..c43ed0b 100644
--- a/supervisor/operator/command_dispatch.py
+++ b/supervisor/operator/command_dispatch.py
@@ -16,6 +16,7 @@
 from supervisor.operator.actions import (
     ActionUnavailable,
     OperatorJob,
+    do_escalate_clarification,
     do_exchange,
     do_inspect,
     do_note_add,
@@ -267,6 +268,7 @@ def format_notes_result(notes: list[dict[str, Any]]) -> str:
 /explain <id> - explain what the run is doing
 /drift <id> - assess drift from plan
 /ask <id> <question> - ask about the run
+/escalate <id> [question] - escalate last (or given) clarification to the worker
 /pause <id> - pause a run
 /resume <id> - resume a paused run
 /note <id> <text> - add operator note
@@ -308,6 +310,38 @@ def _require_run(args: list[str]) -> tuple[RunContext, dict[str, Any]] | Command
     return ctx, run
 
 
+def _latest_clarification(
+    ctx: RunContext, *, override: str = "",
+) -> tuple[str, float | None]:
+    """Pick the question + confidence to use for ``/escalate``.
+
+    When *override* is non-empty the operator supplied the question
+    explicitly, so we use it verbatim with unknown confidence. Otherwise
+    we scan the session log for the most recent ``clarification_response``
+    event and pull its ``question`` / ``confidence`` fields.
+    """
+    if override:
+        return override, None
+    log_path = ctx.session_log_path
+    if log_path is None or not log_path.exists():
+        return "", None
+    from supervisor.operator.api import timeline_from_session_log
+
+    events = timeline_from_session_log(log_path, limit=50)
+    for ev in events:
+        if ev.event_type == "clarification_response":
+            q = str(ev.payload.get("question", "")).strip()
+            if not q:
+                continue
+            conf = ev.payload.get("confidence")
+            try:
+                conf_val: float | None = float(conf) if conf is not None else None
+            except (TypeError, ValueError):
+                conf_val = None
+            return q, conf_val
+    return "", None
+
+
 def dispatch_command(
     cmd: str,
     args: list[str],
@@ -421,6 +455,46 @@ def dispatch_command(
         except ActionUnavailable as exc:
             return CommandResult(text=str(exc), error=True)
 
+    if cmd == "escalate":
+        if not args:
+            return CommandResult(
+                text="Usage: /escalate <run_id> [question]", error=True,
+            )
+        resolved = _require_run([args[0]])
+        if isinstance(resolved, CommandResult):
+            return resolved
+        ctx, run = resolved
+        explicit_question = " ".join(args[1:]).strip()
+        question, confidence = _latest_clarification(ctx, override=explicit_question)
+        if not question:
+            return CommandResult(
+                text=(
+                    "No prior clarification to escalate. "
+                    "Ask first with /ask, or provide a question: "
+                    "/escalate <run_id> <question>"
+                ),
+                error=True,
+            )
+        try:
+            resp = do_escalate_clarification(
+                ctx, question,
+                language=language,
+                reason="im_operator",
+                confidence=confidence,
+            )
+            esc_id = resp.get("escalation_id", "")[:12]
+            return CommandResult(
+                text=(
+                    f"Escalated to worker (id={esc_id}).\n"
+                    f"Question: {question[:200]}\n"
+                    "Transport lands in 0.3.8 — session log has the audit entry."
+                ),
+                data={"escalation_id": resp.get("escalation_id", "")},
+                buttons=_run_buttons(run["run_id"]),
+            )
+        except ActionUnavailable as exc:
+            return CommandResult(text=str(exc), error=True)
+
     if cmd == "pause":
         resolved = _require_run(args)
         if isinstance(resolved, CommandResult):
diff --git a/supervisor/operator/tui.py b/supervisor/operator/tui.py
index 72afa1c..5a3f395 100644
--- a/supervisor/operator/tui.py
+++ b/supervisor/operator/tui.py
@@ -19,6 +19,7 @@
 from supervisor.operator.actions import (
     ActionUnavailable,
     OperatorJob,
+    do_escalate_clarification,
     do_exchange,
     do_inspect,
     do_note_add,
@@ -213,6 +214,9 @@ def format_clarification(result: dict[str, Any]) -> list[str]:
     conf = result.get("confidence")
     if conf is not None:
         lines.append(f"  Confidence: {conf}")
+    if result.get("escalation_recommended"):
+        lines.append("")
+        lines.append("  ⚠  low confidence — press 'E' to escalate to worker")
     return lines
 
 
@@ -460,6 +464,11 @@ def _curses_main(stdscr):
     # Pending async job state (non-blocking)
     pending_job: dict[str, Any] | None = None  # {"job": OperatorJob, "ctx": RunContext, "label": ...}
 
+    # Last completed clarification (for 'E' escalate keybind).
+    # Shape: {"ctx": RunContext, "question": str, "confidence": float|None,
+    #         "escalation_recommended": bool, "language": str}
+    last_clarify: dict[str, Any] | None = None
+
     while True:
         h, w = stdscr.getmaxyx()
         if h < MIN_HEIGHT or w < MIN_WIDTH:
@@ -476,13 +485,25 @@ def _curses_main(stdscr):
             try:
                 result = poll_job(pending_job["ctx"], pending_job["job"])
                 if result.get("status") in ("completed", "failed"):
+                    status_msg = default_status
                     if result.get("status") == "failed":
                         right_lines = [f"Job failed: {result.get('error', 'unknown error')}"]
                     elif pending_job.get("label") == "clarification":
-                        right_lines = format_clarification(result.get("result", {}))
+                        payload = result.get("result", {}) or {}
+                        right_lines = format_clarification(payload)
+                        if payload.get("escalation_recommended"):
+                            last_clarify = {
+                                "ctx": pending_job["ctx"],
+                                "question": pending_job.get("question", ""),
+                                "confidence": payload.get("confidence"),
+                                "escalation_recommended": True,
+                                "language": pending_job.get("language", language),
+                            }
+                            status_msg = " Low-confidence answer — press 'E' to escalate to the worker "
+                        else:
+                            last_clarify = None
                     else:
                         right_lines = format_explanation(result.get("result", {}))
-                    status_msg = default_status
                     pending_job = None
                 # else: still pending, keep spinner
             except Exception as exc:
@@ -572,14 +593,25 @@ def _curses_main(stdscr):
             continue
 
         if key in (ord("j"), curses.KEY_DOWN) and runs:
+            prev_idx = selected_idx
             selected_idx = min(selected_idx + 1, len(runs) - 1)
             detail_lines = ["(loading...)"]
             right_lines = []
+            if selected_idx != prev_idx and last_clarify is not None:
+                # Escalation target is captured per-answer; clear the
+                # hint so the stale "press E" message doesn't survive
+                # a run change.
+                last_clarify = None
+                status_msg = default_status
 
         if key in (ord("k"), curses.KEY_UP) and runs:
+            prev_idx = selected_idx
             selected_idx = max(selected_idx - 1, 0)
             detail_lines = ["(loading...)"]
             right_lines = []
+            if selected_idx != prev_idx and last_clarify is not None:
+                last_clarify = None
+                status_msg = default_status
 
         # Enter or space: load snapshot + timeline
         if key in (10, 32, ord("i")) and runs:
@@ -656,9 +688,13 @@ def _curses_main(stdscr):
             if question:
                 try:
                     job = submit_clarification(ctx, question, language=language)
-                    pending_job = {"job": job, "ctx": ctx, "label": "clarification"}
+                    pending_job = {
+                        "job": job, "ctx": ctx, "label": "clarification",
+                        "question": question, "language": language,
+                    }
                     status_msg = " Asking... (waiting for answer) "
                     right_lines = ["(asking...)"]
+                    last_clarify = None
                 except ActionUnavailable as exc:
                     status_msg = f" {exc} "
                 except Exception as exc:
@@ -666,6 +702,26 @@ def _curses_main(stdscr):
             else:
                 status_msg = default_status
 
+        # E: escalate last low-confidence clarification to the worker.
+        # Only active when `last_clarify` was set by the most recent answer
+        # and no async job is in flight.
+        if key == ord("E") and last_clarify is not None and pending_job is None:
+            try:
+                resp = do_escalate_clarification(
+                    last_clarify["ctx"],
+                    last_clarify["question"],
+                    language=last_clarify.get("language", language),
+                    reason="tui_low_confidence",
+                    confidence=last_clarify.get("confidence"),
+                )
+                esc_id = resp.get("escalation_id", "")[:12]
+                status_msg = f" Escalated to worker (id={esc_id}) "
+                last_clarify = None
+            except ActionUnavailable as exc:
+                status_msg = f" Escalate unavailable: {exc} "
+            except Exception as exc:
+                status_msg = f" Escalate failed: {exc} "
+
         # p: pause
         if key == ord("p") and runs:
             run = runs[selected_idx]
diff --git a/tests/test_command_dispatch.py b/tests/test_command_dispatch.py
index c23d1e1..5ed4e39 100644
--- a/tests/test_command_dispatch.py
+++ b/tests/test_command_dispatch.py
@@ -251,6 +251,105 @@ def test_run_not_found(self, mock_resolve):
         assert "not found" in result.text
 
 
+# ── /escalate ────────────────────────────────────────────────────
+
+
+class TestEscalateCommand:
+    def _seed(self, tmp_path, run_id, events):
+        from supervisor.operator.api import append_timeline_event
+        run_dir = tmp_path / ".supervisor" / "runtime" / "runs" / run_id
+        run_dir.mkdir(parents=True)
+        (run_dir / "state.json").write_text('{"run_id": "' + run_id + '"}')
+        log = run_dir / "session_log.jsonl"
+        for et, payload in events:
+            append_timeline_event(log, run_id, et, payload)
+        return {
+            "run_id": run_id, "tag": "local", "top_state": "COMPLETED",
+            "pane_target": "", "worktree": str(tmp_path), "socket": "",
+        }
+
+    def test_missing_args(self):
+        result = dispatch_command("escalate", [])
+        assert result.error
+        assert "Usage" in result.text
+
+    @patch("supervisor.operator.command_dispatch.do_escalate_clarification")
+    @patch("supervisor.operator.command_dispatch.resolve_run")
+    def test_uses_last_clarification_from_log(
+        self, mock_resolve, mock_escalate, tmp_path,
+    ):
+        run = self._seed(tmp_path, "run_esc_a", [
+            ("clarification_request", {"question": "q1"}),
+            ("clarification_response", {
+                "question": "why is the gate failing?",
+                "confidence": 0.2,
+                "escalation_recommended": True,
+            }),
+        ])
+        mock_resolve.return_value = [run]
+        mock_escalate.return_value = {"escalation_id": "abcd1234efgh5678", "source": "local"}
+
+        result = dispatch_command("escalate", ["run_esc_a"])
+        assert not result.error
+        assert "Escalated" in result.text
+        assert "abcd1234efgh" in result.text  # id prefix
+        kwargs = mock_escalate.call_args.kwargs
+        args = mock_escalate.call_args.args
+        assert args[1] == "why is the gate failing?"
+        assert kwargs["confidence"] == 0.2
+        assert kwargs["reason"] == "im_operator"
+
+    @patch("supervisor.operator.command_dispatch.do_escalate_clarification")
+    @patch("supervisor.operator.command_dispatch.resolve_run")
+    def test_picks_newest_clarification_when_multiple(
+        self, mock_resolve, mock_escalate, tmp_path,
+    ):
+        # Regression guard: /escalate must resolve the *latest*
+        # clarification_response, not the earliest in the session log.
+        run = self._seed(tmp_path, "run_esc_multi", [
+            ("clarification_response", {"question": "old q", "confidence": 0.8}),
+            ("clarification_response", {"question": "mid q", "confidence": 0.5}),
+            ("clarification_response", {"question": "latest q", "confidence": 0.1}),
+        ])
+        mock_resolve.return_value = [run]
+        mock_escalate.return_value = {"escalation_id": "11112222aaaabbbb", "source": "local"}
+
+        result = dispatch_command("escalate", ["run_esc_multi"])
+        assert not result.error
+        args = mock_escalate.call_args.args
+        kwargs = mock_escalate.call_args.kwargs
+        assert args[1] == "latest q"
+        assert kwargs["confidence"] == 0.1
+
+    @patch("supervisor.operator.command_dispatch.do_escalate_clarification")
+    @patch("supervisor.operator.command_dispatch.resolve_run")
+    def test_explicit_question_overrides_log(
+        self, mock_resolve, mock_escalate, tmp_path,
+    ):
+        run = self._seed(tmp_path, "run_esc_b", [
+            ("clarification_response", {"question": "old", "confidence": 0.1}),
+        ])
+        mock_resolve.return_value = [run]
+        mock_escalate.return_value = {"escalation_id": "deadbeef00000000", "source": "local"}
+
+        result = dispatch_command(
+            "escalate", ["run_esc_b", "is", "this", "safe?"],
+        )
+        assert not result.error
+        args = mock_escalate.call_args.args
+        kwargs = mock_escalate.call_args.kwargs
+        assert args[1] == "is this safe?"
+        assert kwargs["confidence"] is None  # explicit override has no confidence
+
+    @patch("supervisor.operator.command_dispatch.resolve_run")
+    def test_no_prior_clarification_errors(self, mock_resolve, tmp_path):
+        run = self._seed(tmp_path, "run_esc_c", [])
+        mock_resolve.return_value = [run]
+        result = dispatch_command("escalate", ["run_esc_c"])
+        assert result.error
+        assert "No prior clarification" in result.text
+
+
 # ── AsyncJobPoller ───────────────────────────────────────────────
 
 
diff --git a/tests/test_daemon.py b/tests/test_daemon.py
index b725e4f..c40c291 100644
--- a/tests/test_daemon.py
+++ b/tests/test_daemon.py
@@ -973,6 +973,84 @@ def test_resume_rejects_legacy_state_without_spec_hash(self, tmp_path, monkeypat
         assert "no persisted spec hash" in result["error"]
 
 
+class TestDaemonEscalateClarification:
+    """Daemon IPC for operator-initiated escalation (0.3.7)."""
+
+    def _server(self, tmp_path):
+        return DaemonServer(runs_dir=str(tmp_path / "runs"))
+
+    def _seed_run_state(self, tmp_path, run_id: str) -> None:
+        run_dir = Path(tmp_path) / "runs" / run_id
+        run_dir.mkdir(parents=True)
+        (run_dir / "state.json").write_text(json.dumps({"run_id": run_id}))
+
+    def test_writes_audit_event_and_returns_id(self, tmp_path):
+        server = self._server(tmp_path)
+        self._seed_run_state(tmp_path, "run_esc1")
+
+        resp = server._do_escalate_clarification({
+            "run_id": "run_esc1",
+            "question": "is this migration safe?",
+            "language": "zh",
+            "reason": "tui_low_confidence",
+            "operator": "op1",
+            "confidence": 0.12,
+        })
+        assert resp["ok"] is True
+        assert len(resp["escalation_id"]) == 16
+
+        log_path = Path(tmp_path) / "runs" / "run_esc1" / "session_log.jsonl"
+        assert log_path.exists()
+        events = [json.loads(line) for line in log_path.read_text().strip().splitlines()]
+        assert len(events) == 1
+        ev = events[0]
+        assert ev["event_type"] == "clarification_escalated_to_worker"
+        assert ev["payload"]["question"] == "is this migration safe?"
+        assert ev["payload"]["reason"] == "tui_low_confidence"
+        assert ev["payload"]["operator"] == "op1"
+        assert ev["payload"]["confidence"] == 0.12
+        assert ev["payload"]["language"] == "zh"
+        assert ev["payload"]["transport"] == "pending_0_3_8"
+        assert ev["payload"]["escalation_id"] == resp["escalation_id"]
+
+    def test_rejects_missing_run(self, tmp_path):
+        server = self._server(tmp_path)
+        resp = server._do_escalate_clarification({
+            "run_id": "run_ghost", "question": "q?",
+        })
+        assert resp["ok"] is False
+        assert "not found" in resp["error"]
+
+    def test_rejects_empty_question(self, tmp_path):
+        server = self._server(tmp_path)
+        self._seed_run_state(tmp_path, "run_esc2")
+        resp = server._do_escalate_clarification({
+            "run_id": "run_esc2", "question": "",
+        })
+        assert resp["ok"] is False
+        assert "question" in resp["error"]
+
+    def test_client_roundtrip(self, client, tmp_path):
+        # End-to-end: client.escalate_clarification → daemon → event on disk.
+        # Daemon fixture (see `daemon_server`) uses `tmp_path / "runs"`,
+        # so seeding state.json there exposes the run through
+        # `_resolve_run_store`'s on-disk fallback.
+        self._seed_run_state(tmp_path, "run_cli")
+
+        resp = client.escalate_clarification(
+            "run_cli", "what happened?",
+            language="en", reason="im_operator",
+            operator="op2", confidence=0.3,
+        )
+        assert resp["ok"] is True
+        assert resp["escalation_id"]
+
+        log_path = Path(tmp_path) / "runs" / "run_cli" / "session_log.jsonl"
+        events = [json.loads(line) for line in log_path.read_text().strip().splitlines()]
+        assert events[0]["event_type"] == "clarification_escalated_to_worker"
+        assert events[0]["payload"]["operator"] == "op2"
+
+
 class TestDaemonExternalTask:
     """Daemon IPC for event-plane request/result/mailbox (Task 3)."""
 
diff --git a/tests/test_operator_actions.py b/tests/test_operator_actions.py
index c2900fa..97877a4 100644
--- a/tests/test_operator_actions.py
+++ b/tests/test_operator_actions.py
@@ -12,6 +12,7 @@
     OperatorJob,
     _local_jobs,
     build_explainer_context,
+    do_escalate_clarification,
     do_exchange,
     do_inspect,
     do_note_add,
@@ -492,6 +493,86 @@ def test_response_summarizer(self):
         assert "agent idle" in summary
 
 
+class TestEscalateClarification:
+    def test_daemon_path(self):
+        ctx = _make_ctx(tag="daemon")
+        mock_client = MagicMock()
+        mock_client.escalate_clarification.return_value = {
+            "ok": True, "escalation_id": "abc123def456abcd",
+        }
+        mock_client.is_running.return_value = True
+        with patch.object(ctx, "get_client", return_value=mock_client):
+            resp = do_escalate_clarification(
+                ctx, "why is verification stuck?",
+                language="en", reason="tui_low_confidence",
+                operator="op1", confidence=0.15,
+            )
+        assert resp["source"] == "daemon"
+        assert resp["escalation_id"] == "abc123def456abcd"
+        mock_client.escalate_clarification.assert_called_once_with(
+            "run_test123", "why is verification stuck?",
+            language="en", reason="tui_low_confidence",
+            operator="op1", confidence=0.15,
+        )
+
+    def test_rejects_empty_question_before_capability_check(self):
+        # Shared validation: both transports must refuse an empty /
+        # whitespace-only question.  Using a context with no capabilities
+        # proves the guard fires before we try to talk to the daemon or
+        # touch the filesystem.
+        ctx = _make_ctx(tag="daemon")
+        with pytest.raises(ActionUnavailable, match="question is required"):
+            do_escalate_clarification(ctx, "")
+        with pytest.raises(ActionUnavailable, match="question is required"):
+            do_escalate_clarification(ctx, "   \t  ")
+
+    def test_daemon_error_becomes_unavailable(self):
+        ctx = _make_ctx(tag="daemon")
+        mock_client = MagicMock()
+        mock_client.escalate_clarification.return_value = {
+            "ok": False, "error": "run not found",
+        }
+        mock_client.is_running.return_value = True
+        with patch.object(ctx, "get_client", return_value=mock_client):
+            with pytest.raises(ActionUnavailable, match="run not found"):
+                do_escalate_clarification(ctx, "q?")
+
+    def test_local_path_writes_event(self, tmp_path):
+        run_dir = tmp_path / ".supervisor" / "runtime" / "runs" / "run_e"
+        run_dir.mkdir(parents=True)
+        (run_dir / "state.json").write_text(json.dumps({"run_id": "run_e"}))
+
+        ctx = _make_ctx(
+            tag="completed", run_id="run_e", worktree=str(tmp_path),
+            socket="", state_dir=run_dir, state_path=run_dir / "state.json",
+            session_log_path=run_dir / "session_log.jsonl",
+        )
+        with patch.object(ctx, "_has_daemon", return_value=False):
+            resp = do_escalate_clarification(
+                ctx, "is the migration safe?",
+                language="zh", reason="im_operator",
+                operator="song", confidence=0.2,
+            )
+        assert resp["source"] == "local"
+        assert len(resp["escalation_id"]) == 16
+
+        events = [
+            json.loads(line) for line in
+            (run_dir / "session_log.jsonl").read_text().strip().splitlines()
+        ]
+        assert len(events) == 1
+        ev = events[0]
+        assert ev["event_type"] == "clarification_escalated_to_worker"
+        assert ev["run_id"] == "run_e"
+        assert ev["payload"]["question"] == "is the migration safe?"
+        assert ev["payload"]["language"] == "zh"
+        assert ev["payload"]["reason"] == "im_operator"
+        assert ev["payload"]["operator"] == "song"
+        assert ev["payload"]["confidence"] == 0.2
+        assert ev["payload"]["transport"] == "pending_0_3_8"
+        assert ev["payload"]["escalation_id"] == resp["escalation_id"]
+
+
 class TestAppendTimelineEvent:
     def test_writes_event(self, tmp_path):
         from supervisor.operator.api import append_timeline_event
diff --git a/tests/test_tui.py b/tests/test_tui.py
index b5d2d66..0c5d9bb 100644
--- a/tests/test_tui.py
+++ b/tests/test_tui.py
@@ -236,6 +236,25 @@ def test_empty_result(self):
         lines = format_clarification({})
         assert "(no answer)" in "\n".join(lines)
 
+    def test_escalation_hint_shown_when_recommended(self):
+        result = {
+            "answer": "unsure",
+            "confidence": 0.15,
+            "escalation_recommended": True,
+        }
+        text = "\n".join(format_clarification(result))
+        assert "escalate" in text.lower()
+        assert "E" in text  # keybind surfaced
+
+    def test_no_escalation_hint_when_confident(self):
+        result = {
+            "answer": "sure thing",
+            "confidence": 0.9,
+            "escalation_recommended": False,
+        }
+        text = "\n".join(format_clarification(result))
+        assert "escalate" not in text.lower()
+
 
 class TestCollectRunsLocal:
     def _patch(self, monkeypatch, *, known=(), daemons=(), panes=()):