operator-plane: guarded command bus (flatten / pause kill switch)

Adds OperatorCommandEvent + a closed command vocabulary (flatten_all,
flatten_symbol, pause_entries, resume_entries). POST /api/command emits it
(token-required: 503 when DASHBOARD_TOKEN unset — commands never run
unauthenticated); the live loop's new 5s "command" step consumes them via
OperatorCommandConsumer with three replay-safety layers (in-session nonce set,
persisted operator_command_done reload, 10-min staleness window).

flatten_all routes through close_session DIRECTLY (never step_eod_flatten) so it
cannot trip plan 003's daily-restart exit. Pause is checked at the top of BOTH
request_approvals_for_ready_signals AND submit_breakout_now (missing either
leaks pause on that path). qabot gains an authorized-chat-only /flatten slash
command (prose "flatten" still refuses via _ACTION_WORDS). Snapshot exposes
broker truth + command_bus + a defensive experiment ledger.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

api/main.py

@@ -12,6 +12,7 @@ POST /api/trading/approvals/approve     {"order_id": ..., "approved_by"?: ...}
 POST /api/trading/approvals/reject      {"order_id": ..., "reason"?: ...}
 POST /api/trading/exit-order            {"symbol": ...}
 POST /api/watch/add                     {"symbol": ...}  (manual watchlist inject)
+POST /api/command                       {"command": ..., "symbol"?: ...}  (operator kill switch; requires DASHBOARD_TOKEN)
 
 The server holds an EventStore for reads; trade actions route through a
 TradingExecutionService when one is attached (the orchestrator does this).
@@ -26,11 +27,13 @@ import json
 import os
 import threading
 import time
+import uuid
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import parse_qs, urlparse
 
+from storage.event_schema import EventMode, OperatorCommandEvent, OPERATOR_COMMANDS
 from storage.event_store import EventStore
 from storage.projections import (
     query_account_orders_snapshot,
@@ -56,6 +59,27 @@ from storage.projections import (
 STATIC_DIR = Path(__file__).parent / "static"
 
 
+def _command_bus_enabled() -> bool:
+    """The operator command bus is live only when a token guards it AND it is
+    explicitly enabled — so a stray COMMAND_BUS_ENABLED never opens an
+    unauthenticated kill switch."""
+    return bool(os.environ.get("DASHBOARD_TOKEN")) and (
+        os.environ.get("COMMAND_BUS_ENABLED", "").strip().lower()
+        in {"1", "true", "yes", "on"}
+    )
+
+
+def _broker_block() -> dict:
+    """Broker truth for the LIVE badge, derived from the base URL the services
+    already have (not the execution_mode literal, which lied). Alpaca's paper host
+    is ``paper-api.alpaca.markets``; anything else is live money."""
+    from urllib.parse import urlparse as _urlparse
+
+    url = os.environ.get("ALPACA_BASE_URL") or "https://paper-api.alpaca.markets"
+    host = _urlparse(url).hostname or url
+    return {"base_url_host": host, "is_paper": host.startswith("paper-api")}
+
+
 def _eastern_session_date():
     """Today's date in US/Eastern — matches the ingestion pipeline's flags."""
     from zoneinfo import ZoneInfo
@@ -150,6 +174,44 @@ class _ChangeSignal:
             return self._stamp
 
 
+def _maybe_json(v):
+    """JSONB columns come back from psycopg2 as str OR already-parsed dict; make it
+    a dict either way so the ledger never double-encodes."""
+    if isinstance(v, str):
+        try:
+            return json.loads(v)
+        except (ValueError, TypeError):
+            return {}
+    return v or {}
+
+
+def _experiment_ledger(store) -> list[dict]:
+    """Latest verdict per experiment for the ledger panel (plan 006 tables).
+
+    Returns [] on ANY error — the experiments/experiment_verdicts tables live in a
+    separate schema file applied by the registry, so on a bare/test database they
+    simply don't exist and the panel shows its empty state."""
+    try:
+        rows = store.con.execute(
+            "SELECT v.experiment_id, v.as_of, v.n, v.verdict, v.metrics, "
+            "       e.hypothesis, e.bars, e.status "
+            "FROM (SELECT DISTINCT ON (experiment_id) experiment_id, as_of, n, verdict, metrics "
+            "      FROM experiment_verdicts ORDER BY experiment_id, as_of DESC) v "
+            "LEFT JOIN experiments e ON e.id = v.experiment_id "
+            "ORDER BY v.as_of DESC"
+        ).fetchall()
+    except Exception:  # noqa: BLE001 — tables absent on a bare DB
+        return []
+    out = []
+    for r in rows:
+        out.append({
+            "experiment": r[0], "as_of": str(r[1]), "n": r[2], "verdict": r[3],
+            "metrics": _maybe_json(r[4]), "hypothesis": r[5],
+            "bars": _maybe_json(r[6]), "status": r[7],
+        })
+    return out
+
+
 class DashboardState:
     """Shared state: event store path + optional live execution service."""
 
@@ -201,6 +263,11 @@ class DashboardState:
                     "generated_at": datetime.now(timezone.utc).isoformat(),
                     "execution_mode": self.execution_mode,
                     "has_execution": self.execution_service is not None,
+                    # broker truth (LIVE badge) + operator command-bus availability
+                    "broker": _broker_block(),
+                    "command_bus": _command_bus_enabled(),
+                    # experiment ledger (plan 006 tables) — [] until they exist
+                    "experiments": _experiment_ledger(store) if is_live else [],
                     "view_date": None if is_live else for_date,
                     "available_dates": _available_dates(store),
                     "pnl": query_session_pnl(store, for_date=for_date),
@@ -316,6 +383,28 @@ class DashboardState:
     def exit_position(self, symbol: str) -> dict | None:
         return self._with_service(lambda svc: svc.submit_exit_order(symbol))
 
+    def emit_command(self, command: str, symbol: str | None = None,
+                     requested_by: str = "dashboard") -> dict:
+        """Emit a guarded operator command event (the loop consumes + executes it).
+
+        Works even when execution_service is None — the standalone dashboard is a
+        thin emitter; the loop process is the single execution authority. Callers
+        gate on DASHBOARD_TOKEN before reaching here (commands never run
+        unauthenticated)."""
+        nonce = uuid.uuid4().hex
+        with self._lock:
+            store = self.open_store()
+            try:
+                store.emit(OperatorCommandEvent(
+                    timestamp=datetime.now(), mode=EventMode.PAPER,
+                    message=f"operator command {command}"
+                            + (f" {symbol}" if symbol else ""),
+                    command=command, symbol=(symbol or None),
+                    requested_by=requested_by, nonce=nonce))
+            finally:
+                store.close()
+        return {"ok": True, "command": command, "symbol": symbol, "nonce": nonce}
+
     def add_watch(self, symbol: str) -> dict | None:
         """Manually inject a symbol onto the watchlist (if supported)."""
         if not self.watch_inject or not symbol:
@@ -490,6 +579,29 @@ class DashboardHandler(BaseHTTPRequestHandler):
                 result = self.state.exit_position(str(body.get("symbol") or ""))
             elif route == "/api/watch/add":
                 result = self.state.add_watch(str(body.get("symbol") or ""))
+            elif route == "/api/command":
+                # The command bus is token-gated by construction: without
+                # DASHBOARD_TOKEN a command could run unauthenticated, so refuse
+                # outright (the top-of-do_POST gate only bites WHEN a token is set).
+                if not os.environ.get("DASHBOARD_TOKEN"):
+                    self._send_json(
+                        {"ok": False, "error": "command bus disabled: DASHBOARD_TOKEN not set"},
+                        503)
+                    return
+                command = str(body.get("command") or "")
+                if command not in OPERATOR_COMMANDS:
+                    self._send_json(
+                        {"ok": False, "error": f"unknown command {command!r}"}, 400)
+                    return
+                symbol = body.get("symbol")
+                symbol = str(symbol).upper() if symbol else None
+                if command == "flatten_symbol" and not symbol:
+                    self._send_json(
+                        {"ok": False, "error": "flatten_symbol requires a symbol"}, 400)
+                    return
+                self._send_json(
+                    self.state.emit_command(command, symbol=symbol), 200)
+                return
             else:
                 self._send_json({"error": "not found"}, 404)
                 return

run_live_paper.py

@@ -778,6 +778,22 @@ def main(argv: list[str] | None = None) -> int:
             f"{b['symbol']} ({b['reason']})" for b in backed
         )
 
+    # Operator command bus (plan 007): consume kill-switch commands emitted by the
+    # dashboard / qabot on the same 5s cadence as the guard. flatten_all routes
+    # through close_session DIRECTLY (never step_eod_flatten) so it can't trip the
+    # plan-003 daily-restart exit. None in dry mode (no execution service).
+    from runtime.operator_commands import OperatorCommandConsumer
+    command_consumer = (
+        OperatorCommandConsumer(rt["store"], rt["execution"], session_id=rt["session_id"])
+        if rt["execution"] is not None else None
+    )
+
+    def step_commands():
+        if command_consumer is None:
+            return None
+        done = command_consumer.poll()
+        return f"operator commands executed: {len(done)}" if done else None
+
     def step_discover():
         if not client:
             return "skipped (no keys)"
@@ -1272,6 +1288,9 @@ def main(argv: list[str] | None = None) -> int:
                   enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
     scheduler.add("guard", step_guard, float(os.environ.get("TRADING_ENTRY_GUARD_INTERVAL_SECONDS", "5")),
                   enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
+    scheduler.add("command", step_commands,
+                  float(os.environ.get("OPERATOR_COMMAND_INTERVAL_SECONDS", "5")),
+                  enabled=(rt["execution"] is not None))
     scheduler.add("exits", step_manage_exits, float(os.environ.get("TRADING_EXIT_MANAGE_INTERVAL_SECONDS", "12")),
                   enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
     scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")),

runtime/operator_commands.py

@@ -0,0 +1,130 @@
+"""Operator command bus consumer (plan 007).
+
+The dashboard / qabot emit ``operator_command`` events (a guarded kill switch);
+the live loop polls this consumer on its 5s guard cadence and executes each new
+command through the execution service's existing protective-close helpers, then
+echoes an ``operator_command_done`` event so a restart never re-runs it.
+
+Replay safety (three layers):
+  1. an in-session consumed-nonce set (deduped within the process),
+  2. persisted ``operator_command_done`` events reloaded on boot (deduped across
+     restarts),
+  3. a staleness window — commands older than ``stale_seconds`` are ignored, so a
+     command issued while the loop was down does not fire on the next boot.
+
+Deliberately minimal: a closed command vocabulary (OPERATOR_COMMANDS), no
+parameters beyond an optional symbol. flatten_all routes through
+``close_session`` DIRECTLY (never ``step_eod_flatten``) so it can never trip
+plan 003's daily-restart exit.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timedelta
+
+from storage.event_schema import EventMode, OperatorCommandEvent, EventType
+
+logger = logging.getLogger(__name__)
+
+STALE_SECONDS = 600  # ignore commands older than 10 minutes (boot-replay guard)
+
+
+class OperatorCommandConsumer:
+    def __init__(self, store, execution, *, session_id=None, now_fn=None,
+                 stale_seconds: float = STALE_SECONDS, mode: EventMode = EventMode.PAPER):
+        self.store = store
+        self.execution = execution
+        self.session_id = session_id
+        self._now = now_fn or datetime.now
+        self._stale = float(stale_seconds)
+        self._mode = mode
+        # nonces already handled — seeded from persisted _done events so a restart
+        # doesn't re-execute a command the previous process already ran.
+        self._consumed: set[str] = self._load_done_nonces()
+
+    def _cutoff_iso(self) -> str:
+        return (self._now() - timedelta(seconds=self._stale)).isoformat()
+
+    def _load_done_nonces(self) -> set[str]:
+        done: set[str] = set()
+        try:
+            for e in self.store.query_events(
+                event_type=EventType.OPERATOR_COMMAND_DONE.value,
+                since=self._cutoff_iso(), limit=None,
+            ):
+                nonce = json.loads(e.get("payload_json") or "{}").get("nonce")
+                if nonce:
+                    done.add(str(nonce))
+        except Exception:  # noqa: BLE001 — a rebuild failure must not block boot
+            logger.warning("operator-command done-nonce reload skipped", exc_info=True)
+        return done
+
+    def poll(self) -> list[str]:
+        """Execute any new, in-window operator commands. Returns nonces executed."""
+        try:
+            events = self.store.query_events(
+                event_type=EventType.OPERATOR_COMMAND.value,
+                since=self._cutoff_iso(), limit=None,
+            )
+        except Exception:  # noqa: BLE001
+            logger.warning("operator-command poll failed", exc_info=True)
+            return []
+        executed: list[str] = []
+        for e in events:
+            p = json.loads(e.get("payload_json") or "{}")
+            nonce = str(p.get("nonce") or "")
+            if not nonce or nonce in self._consumed:
+                continue
+            self._consumed.add(nonce)  # mark BEFORE executing so a raising command can't loop
+            result = self._execute(p)
+            self._emit_done(p, result)
+            executed.append(nonce)
+            logger.info("operator command %s (%s) -> %s",
+                        p.get("command"), p.get("symbol") or "-", result)
+        return executed
+
+    def _execute(self, p: dict) -> str:
+        cmd = p.get("command")
+        sym = p.get("symbol")
+        ex = self.execution
+        if ex is None:
+            return "error: no execution service"
+        try:
+            if cmd == "flatten_all":
+                # DIRECT close_session — never step_eod_flatten, so this cannot
+                # request plan 003's daily-restart exit. close_session also blocks
+                # re-entry for the session (the kill-switch semantic).
+                res = ex.close_session("operator_flatten")
+                errs = res.get("errors") if isinstance(res, dict) else None
+                closed = ", ".join(res.get("closed_positions") or []) if isinstance(res, dict) else ""
+                return f"error: {errs}" if errs else f"ok: flattened [{closed or 'none'}]"
+            if cmd == "flatten_symbol":
+                if not sym:
+                    return "error: flatten_symbol requires a symbol"
+                res = ex.flatten_symbol(sym)
+                return "ok" if res.get("ok") else f"error: {res.get('error')}"
+            if cmd == "pause_entries":
+                ex.set_operator_paused(True)
+                return "ok: entries paused"
+            if cmd == "resume_entries":
+                ex.set_operator_paused(False)
+                return "ok: entries resumed"
+            return f"error: unknown command {cmd!r}"
+        except Exception as exc:  # noqa: BLE001 — one bad command must not kill the loop
+            logger.warning("operator command %s failed", cmd, exc_info=True)
+            return f"error: {exc}"
+
+    def _emit_done(self, p: dict, result: str) -> None:
+        try:
+            self.store.emit(OperatorCommandEvent(
+                timestamp=self._now(), mode=self._mode, correlation_id=self.session_id,
+                event_type=EventType.OPERATOR_COMMAND_DONE,
+                message=f"operator command {p.get('command')} done: {result}",
+                command=str(p.get("command") or ""), symbol=p.get("symbol"),
+                requested_by=str(p.get("requested_by") or "operator"),
+                nonce=str(p.get("nonce") or ""), result=result,
+            ))
+        except Exception:  # noqa: BLE001
+            logger.warning("operator-command done-echo emit failed", exc_info=True)

storage/event_schema.py

@@ -44,6 +44,16 @@ class EventType(str, Enum):
     SYMBOL_EVALUATION_DETAIL = "symbol_evaluation_detail"
     ENTRY_GATE_EVALUATED = "entry_gate_evaluated"
     EXPERIMENT_VERDICT = "experiment_verdict"
+    OPERATOR_COMMAND = "operator_command"
+    OPERATOR_COMMAND_DONE = "operator_command_done"
+
+
+# The complete operator command vocabulary — the ONLY commands the bus accepts.
+# Deliberately closed: adding a parameterized command must redo the audit-trail +
+# replay-safety analysis (see plan 007 maintenance notes).
+OPERATOR_COMMANDS: frozenset[str] = frozenset(
+    {"flatten_all", "flatten_symbol", "pause_entries", "resume_entries"}
+)
 
 
 class BaseEvent(BaseModel):
@@ -294,6 +304,25 @@ class EntryGateEvaluatedEvent(BaseEvent):
     )
 
 
+class OperatorCommandEvent(BaseEvent):
+    """A guarded operator kill-switch command (plan 007). Emitted by the dashboard
+    command endpoint or the qabot ``/flatten``; consumed by the live loop's 5s
+    guard, which executes it through the existing protective-close helpers and
+    echoes an ``operator_command_done`` event (same schema, ``result`` set) so a
+    restart never re-runs it. ``nonce`` (a uuid) is the idempotency key."""
+
+    event_type: EventType = Field(
+        default=EventType.OPERATOR_COMMAND, description="Event type identifier"
+    )
+    command: str = Field(description="One of OPERATOR_COMMANDS")
+    symbol: str | None = Field(default=None, description="Target symbol (flatten_symbol only)")
+    requested_by: str = Field(default="operator", description="Who issued the command")
+    nonce: str = Field(description="Idempotency key (uuid) — deduped by the consumer")
+    result: str | None = Field(
+        default=None, description="Set on the operator_command_done echo (ok / error text)"
+    )
+
+
 class BrokerHealthChangedEvent(BaseEvent):
     """Event when broker health changes."""
 

telegram_bot.py

@@ -227,6 +227,33 @@ def answer(store, question: str, *, host: str, model: str, timeout: int,
     return f"(LLM unavailable — here's the raw snapshot)\n{context}"
 
 
+# --------------------------------------------------------- operator kill switch
+def parse_flatten_command(text: str):
+    """Parse the ONE action the qabot performs: the explicit ``/flatten`` slash
+    command. Returns ``(command, symbol)`` or None. Prose like "sell everything"
+    is NOT matched here — it still hits the _ACTION_WORDS refusal in answer(). This
+    is the deliberate carve-out: an explicit slash command from the authorized chat
+    is allowed; free-text trade requests are not."""
+    parts = text.strip().split()
+    if not parts or parts[0].lower() != "/flatten":
+        return None
+    if len(parts) >= 2 and parts[1].strip():
+        return ("flatten_symbol", parts[1].strip().upper())
+    return ("flatten_all", None)
+
+
+def emit_operator_command(store, command: str, symbol, requested_by: str) -> None:
+    """Emit the same OperatorCommandEvent the dashboard does — the live loop's 5s
+    consumer executes it (single execution authority)."""
+    import uuid
+    from storage.event_schema import EventMode, OperatorCommandEvent
+    store.emit(OperatorCommandEvent(
+        timestamp=datetime.now(), mode=EventMode.PAPER,
+        message=f"telegram operator command {command}" + (f" {symbol}" if symbol else ""),
+        command=command, symbol=symbol, requested_by=requested_by,
+        nonce=uuid.uuid4().hex))
+
+
 # ------------------------------------------------------------------- dead-man
 def _now_et() -> datetime:
     """ET wall clock, tz-NAIVE — events.timestamp is ET-naive (RUNBOOK), so the
@@ -369,6 +396,18 @@ def main() -> None:
             if not is_authorized(chat_id, allowed):
                 logger.warning("ignoring message from unauthorized chat %s", chat_id)
                 continue
+            # explicit /flatten slash command (authorized chat only) — the qabot's
+            # one action. Emits the guarded event; the loop executes it in ~5s.
+            cmd = parse_flatten_command(text)
+            if cmd is not None:
+                try:
+                    emit_operator_command(store, cmd[0], cmd[1],
+                                          requested_by=f"telegram:{chat_id}")
+                    _send(token, chat_id, "Command sent — loop executes within ~5s.")
+                except Exception as exc:  # noqa: BLE001
+                    logger.warning("flatten command emit failed", exc_info=True)
+                    _send(token, chat_id, f"(command failed to send: {exc})")
+                continue
             try:
                 reply = answer(store, text, host=host, model=model, timeout=timeout,
                                ollama_enabled=ollama_enabled)

tests/unit/api/test_dashboard_command.py

@@ -0,0 +1,109 @@
+"""POST /api/command — token gate, enum validation, event emission (plan 007).
+
+Drives DashboardHandler.do_POST directly (no loopback socket), against a real
+DashboardState pinned to an isolated Postgres schema so emit_command actually
+lands an event."""
+
+import io
+import json
+import os
+import uuid
+
+import psycopg2
+import pytest
+
+from api.main import DashboardHandler, DashboardState
+from storage.event_schema import EventType
+from storage.event_store import EventStore
+
+
+@pytest.fixture(autouse=True)
+def _isolated_pg_schema(monkeypatch):
+    schema = f"mem_{uuid.uuid4().hex[:12]}"
+    monkeypatch.setenv("MOMENTUM_PG_SCHEMA", schema)
+    yield
+    url = os.environ.get("DATABASE_URL")
+    if url:
+        conn = psycopg2.connect(url)
+        conn.autocommit = True
+        conn.cursor().execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
+        conn.close()
+
+
+class CaptureHandler(DashboardHandler):
+    def __init__(self, state, path, body=b"", headers=None):
+        self.state = state
+        self.path = path
+        self.rfile = io.BytesIO(body)
+        self.headers = headers or {}
+        self.captured = []
+
+    def _send_json(self, payload, status=200):
+        self.captured.append((status, payload))
+
+
+def _post(state, body, headers=None):
+    raw = json.dumps(body).encode()
+    hdrs = {"Content-Length": str(len(raw))}
+    if headers:
+        hdrs.update(headers)
+    h = CaptureHandler(state, "/api/command", raw, hdrs)
+    h.do_POST()
+    return h.captured[-1]
+
+
+def test_command_503_when_token_unset(monkeypatch):
+    monkeypatch.delenv("DASHBOARD_TOKEN", raising=False)
+    status, payload = _post(DashboardState("momentum"), {"command": "flatten_all"})
+    assert status == 503
+    assert payload["ok"] is False
+
+
+def test_command_unknown_rejected(monkeypatch):
+    monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
+    status, payload = _post(DashboardState("momentum"), {"command": "sell_everything"},
+                            headers={"X-Dashboard-Token": "s3cret"})
+    assert status == 400
+    assert "unknown command" in payload["error"]
+
+
+def test_command_flatten_symbol_requires_symbol(monkeypatch):
+    monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
+    status, payload = _post(DashboardState("momentum"), {"command": "flatten_symbol"},
+                            headers={"X-Dashboard-Token": "s3cret"})
+    assert status == 400
+
+
+def test_command_bad_token_401(monkeypatch):
+    monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
+    status, _ = _post(DashboardState("momentum"), {"command": "flatten_all"},
+                      headers={"X-Dashboard-Token": "wrong"})
+    assert status == 401
+
+
+def test_command_valid_emits_event(monkeypatch):
+    monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
+    state = DashboardState("momentum")
+    status, payload = _post(state, {"command": "flatten_all"},
+                            headers={"X-Dashboard-Token": "s3cret"})
+    assert status == 200 and payload["ok"] is True and payload["nonce"]
+    store = EventStore("momentum")
+    try:
+        events = store.query_events(event_type=EventType.OPERATOR_COMMAND.value, limit=None)
+        assert len(events) == 1
+        assert json.loads(events[0]["payload_json"])["command"] == "flatten_all"
+    finally:
+        store.close()
+
+
+def test_command_emits_even_without_execution_service(monkeypatch):
+    # the standalone dashboard (execution_service=None) is a thin emitter
+    monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
+    state = DashboardState("momentum", execution_service=None)
+    status, payload = _post(state, {"command": "pause_entries"},
+                            headers={"X-Dashboard-Token": "s3cret"})
+    assert status == 200 and payload["ok"] is True
+
+
+if __name__ == "__main__":  # pragma: no cover
+    pytest.main([__file__, "-q"])

tests/unit/test_operator_commands.py

@@ -0,0 +1,140 @@
+"""Operator command bus: consumer dedup/staleness, pause on BOTH entry paths,
+flatten-does-not-self-restart, and the qabot slash-command carve-out (plan 007)."""
+
+import uuid
+from datetime import datetime, timedelta
+
+from runtime.operator_commands import OperatorCommandConsumer
+from storage.event_schema import EventMode, EventType, OperatorCommandEvent
+from storage.event_store import EventStore
+from tests.unit.test_trading_execution import FakeExecutor, emit_signal, make_service
+
+
+class FakeExecution:
+    def __init__(self):
+        self.closed = []
+        self.flattened = []
+        self.paused = None
+
+    def close_session(self, reason="eod_flatten"):
+        self.closed.append(reason)
+        return {"closed_positions": ["AAA"], "cancelled_entries": 0, "errors": []}
+
+    def flatten_symbol(self, symbol, reason="operator_flatten_symbol"):
+        self.flattened.append(symbol)
+        return {"ok": True, "symbol": symbol}
+
+    def set_operator_paused(self, paused):
+        self.paused = bool(paused)
+
+
+def _emit_command(store, command, symbol=None, when=None, nonce=None):
+    store.emit(OperatorCommandEvent(
+        timestamp=when or datetime.now(), mode=EventMode.PAPER,
+        message=f"cmd {command}", command=command, symbol=symbol,
+        requested_by="test", nonce=nonce or uuid.uuid4().hex))
+
+
+def test_command_consumed_once_nonce_dedup():
+    store = EventStore(":memory:")
+    ex = FakeExecution()
+    _emit_command(store, "flatten_all")
+    consumer = OperatorCommandConsumer(store, ex)
+    assert len(consumer.poll()) == 1
+    assert ex.closed == ["operator_flatten"]
+    # second poll: already-consumed nonce -> no re-execution
+    assert consumer.poll() == []
+    assert ex.closed == ["operator_flatten"]
+    store.close()
+
+
+def test_done_event_prevents_reexecution_after_restart():
+    store = EventStore(":memory:")
+    _emit_command(store, "flatten_all")
+    OperatorCommandConsumer(store, FakeExecution()).poll()  # process A executes + echoes _done
+    ex2 = FakeExecution()
+    # process B boots fresh, reloads the _done nonce, must NOT re-run
+    assert OperatorCommandConsumer(store, ex2).poll() == []
+    assert ex2.closed == []
+    store.close()
+
+
+def test_stale_command_ignored_on_boot():
+    store = EventStore(":memory:")
+    ex = FakeExecution()
+    _emit_command(store, "flatten_all", when=datetime.now() - timedelta(minutes=20))
+    _emit_command(store, "flatten_symbol", symbol="AAA")  # fresh
+    consumer = OperatorCommandConsumer(store, ex, stale_seconds=600)
+    consumer.poll()
+    assert ex.closed == []          # the 20-min-old flatten_all was NOT run
+    assert ex.flattened == ["AAA"]  # the fresh one was
+    store.close()
+
+
+def test_flatten_all_routes_through_close_session_not_self_restart():
+    import run_live_paper
+    run_live_paper._clean_exit_requested.clear()
+    store = EventStore(":memory:")
+    ex = FakeExecution()
+    _emit_command(store, "flatten_all")
+    OperatorCommandConsumer(store, ex).poll()
+    assert ex.closed == ["operator_flatten"]
+    # the operator flatten must NOT trip plan 003's daily-restart exit
+    assert not run_live_paper._clean_exit_requested.is_set()
+    store.close()
+
+
+def test_pause_resume_dispatch():
+    store = EventStore(":memory:")
+    ex = FakeExecution()
+    _emit_command(store, "pause_entries")
+    OperatorCommandConsumer(store, ex).poll()
+    assert ex.paused is True
+    _emit_command(store, "resume_entries")
+    OperatorCommandConsumer(store, ex).poll()
+    assert ex.paused is False
+    store.close()
+
+
+def test_done_echo_records_result():
+    store = EventStore(":memory:")
+    _emit_command(store, "flatten_symbol", symbol="BBB")
+    OperatorCommandConsumer(store, FakeExecution()).poll()
+    done = store.query_events(event_type=EventType.OPERATOR_COMMAND_DONE.value, limit=None)
+    assert len(done) == 1
+
+
+# ---- pause blocks BOTH entry paths (the drift class plan 005 warns about) ----
+def test_pause_blocks_both_entry_paths():
+    store = EventStore(":memory:")
+    emit_signal(store)  # a would-otherwise-pass ready signal
+    svc = make_service(store, FakeExecutor())
+    svc.set_operator_paused(True)
+    assert svc.request_approvals_for_ready_signals() == []      # SLOW path blocked
+    fast = svc.submit_breakout_now("GOOD", 14.0, 13.45, last_price=14.1)
+    assert fast.get("skipped") == "paused"                       # FAST path blocked
+    # resume re-enables the slow path (the same signal now creates a request)
+    svc.set_operator_paused(False)
+    assert len(svc.request_approvals_for_ready_signals()) == 1
+    store.close()
+
+
+# ---- qabot slash-command carve-out + auth ----
+def test_qabot_parse_flatten_command():
+    from telegram_bot import parse_flatten_command
+    assert parse_flatten_command("/flatten") == ("flatten_all", None)
+    assert parse_flatten_command("/flatten nvda") == ("flatten_symbol", "NVDA")
+    assert parse_flatten_command("please flatten everything") is None  # prose still refuses
+    assert parse_flatten_command("how did we do?") is None
+
+
+def test_qabot_unauthorized_chat_refused():
+    from telegram_bot import is_authorized
+    assert is_authorized("999", "123") is False
+    assert is_authorized("123", "123") is True
+    assert is_authorized("123", None) is False  # fail closed when unconfigured
+
+
+if __name__ == "__main__":  # pragma: no cover
+    import pytest
+    pytest.main([__file__, "-q"])

trading_execution.py

@@ -445,6 +445,11 @@ class TradingExecutionService:
         self._data_halt = False
         # session closed (end-of-day flatten): no new entries for the rest of the day
         self._session_closed = False
+        # operator PAUSE (plan 007 command bus): blocks new entries on BOTH the
+        # slow auto path and the fast trigger path until a resume command clears
+        # it. In-memory + recoverable — unlike _session_closed it does not flatten
+        # and a fresh command re-enables entries.
+        self._operator_paused = False
         # today's EOD flatten already SUCCEEDED (rebuilt from events on boot) —
         # the daily-restart loop reads this so a post-flatten fresh process
         # neither re-flattens nor re-exits (restart-churn guard)
@@ -1082,11 +1087,20 @@ class TradingExecutionService:
             symbol=ctx.symbol, source=ctx.source, passed=decision.passed,
             reject_reason=decision.reason, gates=decision.gates))
 
+    def set_operator_paused(self, paused: bool) -> None:
+        """Operator PAUSE/RESUME (command bus). Blocks new entries on both paths."""
+        self._operator_paused = bool(paused)
+
     def request_approvals_for_ready_signals(self, rank_map: dict | None = None) -> list[str]:
         """Turn fresh ready signals into approval requests. Returns order ids."""
         if not self.settings.enabled:
             return []
 
+        # operator pause — slow-path half of the BOTH-paths check (submit_breakout_now
+        # holds the other half; missing either would let pause leak on that path).
+        if self._operator_paused:
+            return []
+
         if self._daily_loss_breach():
             return []
 
@@ -1350,6 +1364,10 @@ class TradingExecutionService:
         """
         if not self.settings.enabled:
             return {"ok": False, "skipped": "disabled"}
+        # operator pause — fast-path half of the BOTH-paths check (see
+        # request_approvals_for_ready_signals for the slow-path half).
+        if self._operator_paused:
+            return {"ok": False, "skipped": "paused"}
         if self._daily_loss_breach():
             return {"ok": False, "skipped": "halted"}
         if self._open_position_count() >= self.settings.max_concurrent_positions:
@@ -1596,6 +1614,29 @@ class TradingExecutionService:
         )
         return result
 
+    @_locked
+    def flatten_symbol(self, symbol: str, reason: str = "operator_flatten_symbol") -> dict:
+        """Operator single-name flatten (command bus): cancel the symbol's resting
+        protective legs, then market-close it (the naked-leg-safe helper shared with
+        the exit manager). Does NOT block re-entry / halt the session — that's the
+        whole-book ``close_session``'s job. Best-effort: never raises."""
+        client = getattr(self.executor, "client", None)
+        if client is None or not hasattr(client, "close_position"):
+            return {"ok": False, "symbol": symbol, "error": "no broker client to flatten"}
+        try:
+            self._release_and_close(client, symbol)
+        except Exception as exc:  # noqa: BLE001
+            return {"ok": False, "symbol": symbol, "error": str(exc)}
+        self.store.emit(
+            RiskRuleTriggeredEvent(
+                timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id,
+                message=f"Operator flatten {symbol} ({reason})",
+                rule_type="operator_flatten_symbol", rule_value=0.0,
+                current_state={"symbol": symbol}, action_taken="flatten_symbol",
+            )
+        )
+        return {"ok": True, "symbol": symbol}
+
     def flatten_overnight_carries(self, today_et, reason: str = "open_catchup") -> dict:
         """Catch-up flatten: close positions CARRIED from a prior session.