feat(steering): mesaje mid-tur + /stop pe turul în zbor

Un al doilea mesaj trimis cât Claude încă lucra aștepta până se termina
turul 1 — corecția „stai, nu în master" ajungea după ce greșeala era gata.
Verificat în producție înainte de commit: mesajul 2 stătea 25s blocat în
lock, apoi pornea ca tur separat.

Acum canalele de chat pot ține un proces `claude` viu per canal, cu stdin
deschis, și al doilea mesaj intră în ACELAȘI tur.

- `src/claude_runner.py` — ClaudeProcess (steering, respawn cu --resume,
  drenare stderr, respawn la comutarea OpenRouter) + RunnerRegistry
  (max_live, reaper pe inactivitate, stop_all la shutdown)
- `src/stream_json.py` — parser stream-json partajat cu `_run_claude`;
  pur, nu aruncă niciodată pe is_error (PlanningSession retrimite pe
  error_max_turns și depinde de asta)
- `src/sentinels.py` — un singur loc pentru __AUDIO__/__STEERED__, în loc
  de 4 verificări copiate; repară și bug-ul preexistent prin care
  WhatsApp posta literal `__AUDIO__:/cale`
- dispecer în `send_message`: lock.acquire(blocking=False) — eșecul de a
  lua lock-ul ESTE „rulează un tur", ceea ce elimină flagul inflight din
  decizie și cursa TOCTOU odată cu el
- `/stop` oprește turul, nu sesiunea — active.json rămâne valid
- rate limit prin proces persistent vine ca result.is_error, nu ca exit
  code; convertit înapoi în același RuntimeError, altfel fallback-ul
  local nu s-ar mai declanșa niciodată, în tăcere

Steering-ul nu face niciodată cross-adapter (un mesaj text nu intră
într-un tur voice: împart același channel_id). Mesajele steered dintr-un
tur care pică sunt re-livrate, nu pierdute.

Testat live cu CLI-ul real: corecție la secunda 10 dintr-un tur de 24s,
un singur result, num_turns=2. Notă: mesajele steered sunt împachetate în
[EXTERNAL CONTENT], deci o corecție formulată ca override agresiv poate
fi refuzată ca prompt injection — pentru oprire folosește /stop.

Suită: 1199 passed, 12 failed (toate pre-existente pe HEAD curat).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SiJGsZVSEGjRHZEJiXaxCC

CLAUDE.md

@@ -90,6 +90,13 @@ source .venv/bin/activate && pip install -r requirements.txt
 
 **Sesiuni** (`src/claude_session.py`): o sesiune persistentă per canal, `claude --resume <session_id>`. Mesajele externe împachetate în markeri `[EXTERNAL CONTENT]`.
 
+**Steering — turnuri persistente** (`src/claude_runner.py`): pe lângă calea one-shot de mai sus (`_run_claude` — un `claude -p` per tur, procesul iese la final), canalele de chat interactive (Discord/Telegram/WhatsApp) pot ține un proces `claude` **viu** per canal, cu stdin deschis, ca un al doilea mesaj trimis cât primul încă rulează să intre în ACELAȘI tur (`ClaudeProcess.steer()`) în loc să aștepte după el.
+- **Config** (`config.json → steering`): `{"enabled": false, "idle_minutes": 20, "max_live": 2}`. **Off implicit** — rollback e o linie (`enabled: false` + restart). Kill switch fără să atingi JSON versionat: variabila de mediu `ECHO_STEERING=off`.
+- `heartbeat.py`, `planning_session.py` și `scheduler.py` rămân **deliberat one-shot** — folosesc `_run_claude`/`_run_claude_extra` direct, nu importă router-ul: n-are cine corecta un job cron sau o conversație de planning la mijlocul turului, deci un proces viu acolo ar adăuga doar RAM (292-541 MB per proces) pentru o capabilitate nefolosită.
+- **`/stop`** oprește doar turul curent în zbor (`ClaudeProcess.stop()` / `stop_turn()` în `claude_session.py`), nu sesiunea — `sessions/active.json` rămâne valid, canalul răspunde normal la mesajul următor.
+- **Diagnostic:** `eco status` arată `steering: on/off · N procese vii` (flag citit din config la fiecare apel; numărătoarea e prin `pgrep -f "--input-format stream-json"`, nu prin registry-ul din proces — `eco` rulează separat de serviciu). `eco doctor` verifică suportul binarului pentru `--input-format stream-json` doar cât timp `steering.enabled` e pornit.
+- **Rețetă de reproducere manuală** (T15): cere-i lui Echo ceva cu un `sleep` de 30s+ în Bash pe canalul de test (ex. „rulează `sleep 40 && echo gata`, apoi zi-mi vremea"), apoi trimite al doilea mesaj pe același canal cât primul încă rulează — urmărește linia de log „steered N chars". **Contează:** primul spike de testare n-a dovedit nimic, pentru că Claude a mutat `sleep`-ul în `run_in_background`, iar turul s-a terminat în 7.8s înainte ca steering-ul să apuce să conteze — dacă turul se termină prea repede, cere explicit ca task-ul să blocheze în prim-plan, nu în fundal.
+
 **State:** `sessions/active.json` — channel ID → `{session_id, model, message_count, ...}`
 
 **Credențiale** (`src/credential_store.py`): keyring de sistem, serviciu `"echo-core"`. Niciodată secrete ca argumente CLI.
@@ -255,6 +262,9 @@ Fișierele Ralph (planning_session, planning_orchestrator, ralph.sh, ralph_dag,
 | `src/main.py` | Entry point — adaptoare + scheduler + heartbeat |
 | `src/router.py` | Comenzi vs mesaje Claude |
 | `src/claude_session.py` | Wrapper Claude CLI cu `--resume` |
+| `src/claude_runner.py` | Procese Claude persistente per canal ("steering") — vezi § Arhitectură |
+| `src/stream_json.py` | Parser stream-json partajat între `claude_session.py` și `claude_runner.py` |
+| `src/sentinels.py` | Markeri de protocol partajați (ex. `__AUDIO__:`, `__STEERED__`) pe cele 4 căi (Discord/Telegram/WhatsApp/voice) |
 | `src/local_fallback_tools.py` | Registry allowlist de unelte doar-citire pentru modelul local (vezi § Fallback local) |
 | `src/fallback_history.py` | Istoric conversație per canal pentru fallback (6 schimburi, TTL 30 min) |
 | `src/net_status.py` | Status read-only mașini Proxmox/LXC prin SSH paralel |

cli.py

@@ -94,6 +94,7 @@ def cmd_status(args):
         print(f"WA Bridge: OFFLINE ({bridge_active})")
 
     _print_session_count()
+    print(_steering_status_line())
 
 
 def _print_session_count():
@@ -103,6 +104,56 @@ def _print_session_count():
     print(f"Sessions: {count} active")
 
 
+def _count_steering_processes() -> int | None:
+    """Count live steering `claude` processes via the process table, not
+    the in-process registry — `eco` is its own short-lived process, so
+    `get_registry().live_count()` here would always be 0 regardless of what
+    the running echo-core.service actually has live (a diagnostic that
+    structurally cannot be non-zero is worse than none).
+
+    Works because `ClaudeProcess._build_cmd()` (claude_runner.py) always
+    spawns with `--input-format stream-json`, and the one-shot path
+    (`_run_claude` in claude_session.py) never passes that flag — so
+    counting processes with it in their command line counts exactly the
+    live steering procs, from any process, no shared state needed.
+
+    Returns None (unknown) only on a real failure to count. `pgrep`'s exit
+    code 1 means zero matches, not an error — must not be treated as one.
+    """
+    import subprocess
+    try:
+        result = subprocess.run(
+            ["pgrep", "-fc", "--", "--input-format stream-json"],
+            capture_output=True, text=True, timeout=5,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    if result.returncode not in (0, 1):
+        return None
+    try:
+        return int(result.stdout.strip())
+    except ValueError:
+        return None
+
+
+def _steering_status_line() -> str:
+    """A3: 'steering: on/off · N procese vii'.
+
+    Flag read through `Config()` at call time (never cached at module load —
+    the flag is meant to be flippable without a restart). Guarded (D1/D3):
+    a counting failure must not break `eco status` — a diagnostics command
+    that dies because the thing it diagnoses is broken is worthless.
+    """
+    try:
+        from src.config import Config
+        enabled = bool(Config().get("steering.enabled", False))
+    except Exception:
+        enabled = False
+    count = _count_steering_processes()
+    count_str = "necunoscut" if count is None else str(count)
+    return f"steering: {'on' if enabled else 'off'} · {count_str} procese vii"
+
+
 def _load_sessions_file() -> dict:
     """Load sessions/active.json, return {} on any error."""
     try:
@@ -325,6 +376,25 @@ def cmd_doctor(args):
     else:
         checks.append(("WhatsApp bridge (optional)", True))
 
+    # 13. Steering: verify the CLI binary supports stream-json input (D1) —
+    # only meaningful if the feature is actually turned on.
+    try:
+        from src.config import Config
+        steering_enabled = bool(Config().get("steering.enabled", False))
+    except Exception:
+        steering_enabled = False
+    if steering_enabled and claude_found:
+        try:
+            help_out = subprocess.run(
+                ["claude", "--help"], capture_output=True, text=True, timeout=10,
+            ).stdout
+            checks.append(("Claude CLI supports --input-format stream-json",
+                            "--input-format" in help_out))
+        except Exception:
+            checks.append(("Claude CLI supports --input-format stream-json", False))
+    else:
+        checks.append(("Steering (optional, currently off)", True))
+
     # ---- Voice stack checks (Pas 10) ----
     checks.extend(_voice_doctor_checks())
 

config.json

@@ -127,5 +127,10 @@
   "tts": {
     "default_engine": "pockettts",
     "pockettts_url": "http://127.0.0.1:7789"
+  },
+  "steering": {
+    "enabled": true,
+    "idle_minutes": 20,
+    "max_live": 2
   }
 }

src/adapters/discord_bot.py

@@ -29,6 +29,7 @@ from src.router import (
     start_planning_session,
 )
 from src.adapters._text_chunks import split_message
+from src.sentinels import audio_path as _sentinel_audio_path, is_steered as _sentinel_is_steered
 from src.adapters.discord_views import (
     RalphRootView,
     PlanningActiveView,
@@ -1036,8 +1037,8 @@ def create_bot(config: Config) -> discord.Client:
             total = len(chunks)
             for i, chunk in enumerate(chunks, 1):
                 result = await asyncio.to_thread(fast_dispatch, "audio", [voice, chunk])
-                if result and result.startswith("__AUDIO__:"):
+                wav_path = _sentinel_audio_path(result)
-                    wav_path = result[len("__AUDIO__:"):]
+                if wav_path:
                     ogg_path = await asyncio.to_thread(_wav_to_ogg, wav_path)
                     try:
                         ext = "ogg" if ogg_path.endswith(".ogg") else "wav"
@@ -1066,8 +1067,8 @@ def create_bot(config: Config) -> discord.Client:
         if rezumat:
             args.append("rezumat")
         result = await asyncio.to_thread(fast_dispatch, "audio", args)
-        if result and result.startswith("__AUDIO__:"):
+        wav_path = _sentinel_audio_path(result)
-            wav_path = result[len("__AUDIO__:"):]
+        if wav_path:
             ogg_path = await asyncio.to_thread(_wav_to_ogg, wav_path)
             try:
                 ext = "ogg" if ogg_path.endswith(".ogg") else "wav"
@@ -1298,11 +1299,18 @@ def create_bot(config: Config) -> discord.Client:
                     adapter_name="discord",
                 )
 
+            # Steered: message was injected into an in-flight turn — react,
+            # post nothing.
+            if _sentinel_is_steered(response):
+                try:
+                    await message.add_reaction("➡️")
+                except Exception:
+                    logger.warning("steered reaction failed", exc_info=True)
             # Only send the final combined response if no intermediates
             # were delivered (avoids duplicating content).
-            if sent_count == 0:
+            elif sent_count == 0:
-                if response.startswith("__AUDIO__:"):
+                wav_path = _sentinel_audio_path(response)
-                    wav_path = response[len("__AUDIO__:"):]
+                if wav_path:
                     await message.channel.send(
                         file=discord.File(wav_path, filename="echo-audio.wav")
                     )

src/adapters/telegram_bot.py

@@ -47,6 +47,7 @@ from src.router import (
     start_planning_session,
 )
 from src.planning_session import is_in_planning
+from src.sentinels import audio_path as _sentinel_audio_path, is_steered as _sentinel_is_steered
 
 WORKSPACE_DIR = Path("/home/moltbot/workspace")
 ADAPTER_NAME = "telegram"
@@ -747,8 +748,6 @@ async def callback_ralph(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
 
 # --- Audio helpers ---
 
-_AUDIO_PREFIX = "__AUDIO__:"
-
 
 async def _send_voice_telegram(update: Update, wav_path: str) -> None:
     """Convertește WAV→OGG (ffmpeg) și trimite ca voice note Telegram."""
@@ -787,8 +786,8 @@ async def _fast_cmd(update: Update, name: str, args: list[str]) -> None:
     await update.message.chat.send_action(ChatAction.TYPING)
     result = await asyncio.to_thread(fast_dispatch, name, args)
     if result:
-        if result.startswith(_AUDIO_PREFIX):
+        wav_path = _sentinel_audio_path(result)
-            wav_path = result[len(_AUDIO_PREFIX):]
+        if wav_path:
             await _send_voice_telegram(update, wav_path)
         else:
             for chunk in split_message(result):
@@ -1034,18 +1033,22 @@ async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
             adapter_name=ADAPTER_NAME,
         )
 
-        # Only send combined response if no intermediates were delivered
+        if _sentinel_is_steered(response):
-        if sent_count == 0:
+            # Injected into an in-flight turn — react, post nothing.
-            chunks = split_message(response)
+            reaction = "➡️"
-            for chunk in chunks:
+        else:
-                await message.reply_text(chunk)
+            # Only send combined response if no intermediates were delivered
+            if sent_count == 0:
+                chunks = split_message(response)
+                for chunk in chunks:
+                    await message.reply_text(chunk)
+            reaction = "✅"
 
-        # Emoji reaction: ✅ = răspuns trimis
         try:
             await context.bot.set_message_reaction(
                 chat_id=chat_id,
                 message_id=message.message_id,
-                reaction=[ReactionTypeEmoji(emoji="✅")],
+                reaction=[ReactionTypeEmoji(emoji=reaction)],
             )
         except Exception:
             pass

src/adapters/whatsapp.py

@@ -1,13 +1,16 @@
 """WhatsApp adapter for Echo Core — connects to Node.js bridge."""
 
 import asyncio
+import base64
 import logging
+import os
 
 import httpx
 
 from src.config import Config
 from src.router import route_message
 from src.claude_session import clear_session, get_active_session
+from src.sentinels import audio_path as _sentinel_audio_path, is_steered as _sentinel_is_steered
 
 log = logging.getLogger("echo-core.whatsapp")
 _security_log = logging.getLogger("echo-core.security")
@@ -104,6 +107,32 @@ async def send_whatsapp(client: httpx.AsyncClient, to: str, text: str) -> bool:
         return False
 
 
+async def send_whatsapp_audio(client: httpx.AsyncClient, to: str, wav_path: str) -> bool:
+    """Send a WAV file via the bridge's /send-document endpoint.
+
+    The Baileys bridge has no dedicated voice-note (PTT) route, only
+    generic document upload — so this arrives as a playable attachment,
+    not a mic-bubble voice note. Good enough to not lose the audio.
+    """
+    try:
+        with open(wav_path, "rb") as f:
+            data_b64 = base64.b64encode(f.read()).decode()
+        resp = await client.post(
+            f"{_bridge_url}/send-document",
+            json={
+                "to": to,
+                "filename": "echo-audio.wav",
+                "mimetype": "audio/wav",
+                "data_base64": data_b64,
+            },
+            timeout=30,
+        )
+        return resp.status_code == 200 and resp.json().get("ok", False)
+    except Exception as e:
+        log.error("Send audio error: %s", e)
+        return False
+
+
 async def react_whatsapp(
     client: httpx.AsyncClient, to: str, message_id: str, emoji: str,
     *, from_me: bool = False, participant: str | None = None,
@@ -222,23 +251,36 @@ async def handle_incoming(msg: dict, client: httpx.AsyncClient) -> None:
         )
         sent_count += 1
 
+    steered = False
     try:
         response, _is_cmd = await asyncio.to_thread(
             route_message, channel_id, user_id, text,
             on_text=on_text,
             adapter_name="whatsapp",
         )
-        # Only send combined response if no intermediates were delivered
+        if _sentinel_is_steered(response):
-        if sent_count == 0:
+            # Injected into an in-flight turn — react, post nothing.
-            await send_whatsapp(client, sender, response)
+            steered = True
+        elif sent_count == 0:
+            # Only send combined response if no intermediates were delivered
+            wav_path = _sentinel_audio_path(response)
+            if wav_path:
+                if not await send_whatsapp_audio(client, sender, wav_path):
+                    await send_whatsapp(client, sender, "Nu am putut trimite audio.")
+                try:
+                    os.unlink(wav_path)
+                except OSError:
+                    pass
+            else:
+                await send_whatsapp(client, sender, response)
     except Exception as e:
         log.error("Error handling message from %s: %s", user_id, e)
         await send_whatsapp(client, sender, "Sorry, an error occurred.")
     finally:
-        # Remove eyes reaction after responding
+        # Swap eyes for an arrow when steered, otherwise just clear it.
         if message_id:
             await react_whatsapp(
-                client, sender, message_id, "",
+                client, sender, message_id, "➡️" if steered else "",
                 from_me=from_me,
                 participant=msg.get("participant"),
             )

src/claude_runner.py

@@ -0,0 +1,548 @@
+"""Persistent Claude CLI processes ("steering") for Echo-Core.
+
+Why two paths exist
+--------------------
+Echo-Core has ALWAYS run the Claude CLI one-shot: `_run_claude` in
+`claude_session.py` spawns `claude -p "<prompt>"`, reads the response, and
+the process exits. That is simple and fine for a request/response turn, but
+it means a second message sent while turn 1 is still running can't reach
+Claude until turn 1 finishes — by then it's too late to say "wait, don't do
+that".
+
+`ClaudeProcess` below keeps one `claude` subprocess ALIVE per channel, with
+stdin held open, so a second message can be written into the SAME turn
+(`steer()`) instead of queueing behind it. `RunnerRegistry` owns the set of
+live processes: a cap (`max_live`), an idle reaper, and shutdown.
+
+`heartbeat.py` and `planning_session.py` deliberately stay on the one-shot
+path (`_run_claude` / `start_session` / `resume_session`, unchanged) — there
+is no human waiting to correct a 3am cron job or a planning conversation
+mid-turn, so a live process there would only add RAM (measured 292-541 MB
+per process) for a capability nobody uses. Only interactive chat channels
+(Discord/Telegram/WhatsApp) route through this module, and only when
+`steering.enabled` is on (that flag and the dispatch decision live in
+`claude_session.py`/`router.py` — this module doesn't read config itself).
+
+Manual repro recipe (T15): to exercise steering by hand, ask Echo for
+something with a 30s+ `sleep` in Bash ("run `sleep 40 && echo done` then
+tell me the weather"), then send a second message on the same channel
+while it's still running. Watch for the "steered N chars" log line below.
+
+Threading model
+----------------
+Thread-based, NOT asyncio: `send_message` (in `claude_session.py`) is sync
+code called via `asyncio.to_thread` from the async adapters, so a second
+event loop here would be unused complexity. Three kinds of threads touch a
+single `ClaudeProcess`: the caller's own thread running `run_turn()`
+(blocking read of stdout), a caller's thread calling `steer()` concurrently,
+and the shared reaper thread. `_stdin_lock` is the single lock serializing
+all of them around `self.inflight` and stdin writes — see `steer()`'s
+docstring for the race it closes and the one it deliberately does not (and
+cannot: it is an inter-process race, not a Python one).
+"""
+
+import atexit
+import dataclasses
+import enum
+import json
+import logging
+import shutil
+import subprocess
+import threading
+import time
+from collections import deque
+from pathlib import Path
+from typing import Callable
+
+from src.claude_session import (
+    CLAUDE_BIN,
+    DEFAULT_MODEL,
+    DEFAULT_TIMEOUT,
+    PROJECT_ROOT,
+    _safe_env,
+    build_system_prompt,
+    is_rate_limit_error,
+)
+from src.stream_json import consume_stream
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+MAX_LIVE_DEFAULT = 2  # measured 292-541 MB RSS per live `claude` process (8 GB host)
+IDLE_MINUTES_DEFAULT = 20
+
+# A2: `_safe_env()` checks this at spawn to decide whether to route through
+# OpenRouter. On a one-shot process that's re-evaluated every turn; on a
+# persistent one it would freeze at spawn time. Compared against its state
+# each turn (see `_respawn_if_needed`) — the ONLY justified respawn-on-change
+# (respawning on personality/ edits was explicitly rejected, see plan
+# "Riscul #2").
+OPENROUTER_SEMAPHORE = PROJECT_ROOT / ".use_openrouter"
+
+
+def _wrap_external_content(text: str) -> str:
+    """Same injection-protection wrapping `start_session`/`resume_session`
+    use — T1: a steered message MUST NOT bypass it just because it goes in
+    over stdin instead of argv (Section 3 finding S1, a security regression
+    otherwise)."""
+    return f"[EXTERNAL CONTENT]\n{text}\n[END EXTERNAL CONTENT]"
+
+
+def _user_turn_line(text: str) -> str:
+    wrapped = _wrap_external_content(text)
+    return json.dumps(
+        {"type": "user", "message": {"role": "user", "content": wrapped}},
+        ensure_ascii=False,
+    )
+
+
+# ---------------------------------------------------------------------------
+# steer() outcome — X2: enum, not bool, so the fallback logic can't be
+# skipped by accident.
+# ---------------------------------------------------------------------------
+
+
+class SteerStatus(enum.Enum):
+    STEERED = "steered"
+    # DANGER, not the safe case (C1): no turn was in flight when the write
+    # landed, so the CLI treated it as a brand-new turn N+1 whose stdout
+    # nobody else is reading. `SteerResult.turn` already holds that turn's
+    # full result — decision D4 ("consume the stream you just started").
+    RAN_AS_TURN = "ran_as_turn"
+    # T7: the stdin write itself raised (process was dead/dying). The text
+    # was re-dispatched as a fresh `run_turn()` (respawning if needed) —
+    # `SteerResult.turn` holds ITS result. Never lost, per CLAUDE.md's
+    # "turnul nu se pierde niciodată".
+    PROCESS_DEAD = "process_dead"
+
+
+@dataclasses.dataclass
+class SteerResult:
+    """Return value of `ClaudeProcess.steer()`.
+
+    `turn` is populated for every status except STEERED, and IS the
+    response for that text — bundling it here (instead of a bare enum)
+    makes it structurally impossible for a caller to see a non-STEERED
+    status and call `run_turn(text)` again "just in case": that would
+    double-send the message and bill it twice.
+    """
+
+    status: SteerStatus
+    turn: dict | None = None
+
+
+# ---------------------------------------------------------------------------
+# ClaudeProcess
+# ---------------------------------------------------------------------------
+
+
+class ClaudeProcess:
+    """One live `claude` subprocess for a single channel."""
+
+    def __init__(self, channel_id: str, model: str = DEFAULT_MODEL,
+                 session_id: str | None = None, cwd: Path | str | None = None):
+        self.channel_id = channel_id
+        self.model = model
+        self.session_id = session_id
+        self.cwd = cwd or PROJECT_ROOT
+        self.proc: subprocess.Popen | None = None
+        self.inflight = False
+        self.last_active = time.monotonic()
+
+        self._stdin_lock = threading.Lock()
+        self._lifecycle_lock = threading.Lock()
+        self._pending_steers: list[str] = []
+        self._stderr_buf: deque[str] = deque(maxlen=50)
+        self._openrouter_at_spawn = False
+
+    # -- lifecycle ---------------------------------------------------------
+
+    def alive(self) -> bool:
+        return self.proc is not None and self.proc.poll() is None
+
+    def _build_cmd(self) -> list[str]:
+        cmd = [
+            CLAUDE_BIN, "-p",
+            "--input-format", "stream-json",
+            "--output-format", "stream-json", "--verbose",
+            "--model", self.model,
+            "--system-prompt", build_system_prompt(),
+            "--dangerously-skip-permissions",
+            "--autocompact", "auto",
+        ]
+        if self.session_id:
+            cmd += ["--resume", self.session_id]
+        return cmd
+
+    def _spawn(self) -> None:
+        if not shutil.which(CLAUDE_BIN):
+            raise FileNotFoundError(
+                "Claude CLI not found. "
+                "Install: https://docs.anthropic.com/en/docs/claude-code"
+            )
+        self._openrouter_at_spawn = OPENROUTER_SEMAPHORE.exists()
+        self.proc = subprocess.Popen(
+            self._build_cmd(),
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1,  # line-buffered: a steer() write must reach the CLI promptly
+            env=_safe_env(),
+            cwd=str(self.cwd),
+        )
+        self.inflight = False
+        self.last_active = time.monotonic()
+        self._stderr_buf.clear()
+        threading.Thread(target=self._drain_stderr, args=(self.proc,), daemon=True).start()
+        logger.info("channel=%s: spawned claude pid=%s resume=%s",
+                    self.channel_id, self.proc.pid, bool(self.session_id))
+
+    def _drain_stderr(self, proc: subprocess.Popen) -> None:
+        """T3: MANDATORY. An undrained stderr pipe fills (~64KB) and
+        deadlocks the child mid-turn with no exception on our side —
+        this thread is the only thing preventing that."""
+        try:
+            for line in proc.stderr:
+                self._stderr_buf.append(line.rstrip("\n"))
+        except (ValueError, OSError):
+            pass  # pipe torn down under us — process is exiting, nothing left to drain
+
+    def _respawn_if_needed(self) -> None:
+        """Call with `_lifecycle_lock` held."""
+        if self.alive() and OPENROUTER_SEMAPHORE.exists() != self._openrouter_at_spawn:
+            logger.info("channel=%s: .use_openrouter changed since spawn — respawning (A2)",
+                        self.channel_id)
+            self.stop()
+        if not self.alive():
+            self._spawn()
+
+    def stop(self) -> None:
+        """Terminate the process. Safe to call repeatedly / when already dead.
+
+        Takes `_stdin_lock` (E-C2): a concurrent `steer()` must never write
+        into a pipe we're closing underneath it.
+        """
+        with self._stdin_lock:
+            proc = self.proc
+            if proc is None:
+                return
+            try:
+                proc.stdin.close()
+            except (BrokenPipeError, OSError, ValueError):
+                pass
+            try:
+                proc.terminate()
+                proc.wait(timeout=3)
+            except subprocess.TimeoutExpired:
+                try:
+                    proc.kill()
+                    proc.wait(timeout=3)
+                except OSError:
+                    pass
+            except OSError:
+                pass
+        logger.info("channel=%s: process stopped", self.channel_id)
+
+    # -- turns ---------------------------------------------------------
+
+    def run_turn(self, text: str, on_text: Callable[[str], None] | None = None,
+                 timeout: int = DEFAULT_TIMEOUT) -> dict:
+        """Run one turn. Returns exactly the dict `_run_claude` returns, and
+        raises the exact same exceptions for the same reasons
+        (FileNotFoundError, TimeoutError, `RuntimeError("Claude CLI error
+        (exit 1): ...")` on a rate limit) so `router.py`'s existing handling
+        — including the local-fallback trigger — keeps working unchanged.
+        """
+        with self._lifecycle_lock:
+            self._respawn_if_needed()
+
+        line = _user_turn_line(text)
+        with self._stdin_lock:
+            self.inflight = True
+            self._pending_steers = []
+            try:
+                self.proc.stdin.write(line + "\n")
+                self.proc.stdin.flush()
+            except (BrokenPipeError, ValueError, OSError) as exc:
+                self.inflight = False
+                raise RuntimeError(f"Claude CLI error: failed to start turn: {exc}") from exc
+
+        return self._read_until_result(on_text=on_text, timeout=timeout)
+
+    def steer(self, text: str) -> SteerResult:
+        """Write *text* into the current in-flight turn's stdin, without
+        opening a new turn — the common case.
+
+        C1/E-C1: the only race a lock can close is between OUR threads —
+        `steer()` checks `self.inflight` under the same `_stdin_lock` that
+        `run_turn()`/`_read_until_result()` flips it under, so nothing here
+        can observe a stale `True`. What no lock can close is the CLI's own
+        internal turn boundary: it may have already finished turn N and
+        gone idle before our write physically reaches its stdin pipe, in
+        which case the CLI treats the write as opening turn N+1 regardless
+        of what we believed. That is `RAN_AS_TURN` — see decision D4 on
+        `SteerStatus` above. (A permanent stdout-owning reader thread would
+        close this residual race too; deferred as a documented upgrade, not
+        required for the accepted fix.)
+        """
+        line = _user_turn_line(text)
+        wrote = False
+        became_new_turn = False
+        with self._stdin_lock:
+            was_inflight = self.inflight
+            if self.proc is not None:
+                try:
+                    self.proc.stdin.write(line + "\n")
+                    self.proc.stdin.flush()
+                    wrote = True
+                except (BrokenPipeError, ValueError, OSError):
+                    wrote = False
+            if wrote and was_inflight:
+                self._pending_steers.append(text)
+            elif wrote:
+                became_new_turn = True
+                self.inflight = True  # we now own reading this orphaned turn
+                self._pending_steers = []
+
+        if wrote and was_inflight:
+            logger.info("channel=%s: steered %d chars into in-flight turn",
+                        self.channel_id, len(text))
+            return SteerResult(SteerStatus.STEERED)
+
+        if became_new_turn:
+            logger.info(
+                "channel=%s: steer() found no in-flight turn — the write became "
+                "turn N+1, consuming its stream now (D4)", self.channel_id,
+            )
+            turn = self._read_until_result(on_text=None, timeout=DEFAULT_TIMEOUT)
+            return SteerResult(SteerStatus.RAN_AS_TURN, turn=turn)
+
+        # T7: the write itself failed — process dead/dying. Never lose the
+        # message: re-dispatch as a fresh (respawning) turn.
+        logger.info("channel=%s: steer() hit a dead process — falling back to run_turn (T7)",
+                    self.channel_id)
+        turn = self.run_turn(text)
+        return SteerResult(SteerStatus.PROCESS_DEAD, turn=turn)
+
+    def pop_pending_steers(self) -> list[str]:
+        """C3: texts steered into the turn that just failed (rate limit,
+        timeout, crash) — the caller MUST re-dispatch them ("turnul nu se
+        pierde niciodată"). Empty once the turn they belonged to succeeds."""
+        pending, self._pending_steers = self._pending_steers, []
+        return pending
+
+    def _read_until_result(self, on_text: Callable[[str], None] | None,
+                            timeout: int) -> dict:
+        proc = self.proc
+        timed_out = threading.Event()
+
+        def _watchdog():
+            try:
+                proc.wait(timeout=timeout)
+            except subprocess.TimeoutExpired:
+                timed_out.set()
+                try:
+                    proc.kill()  # M2: next run_turn() respawns with --resume
+                except OSError:
+                    pass
+
+        watchdog = threading.Thread(target=_watchdog, daemon=True)
+        watchdog.start()
+
+        def _on_init(sid: str) -> None:
+            self.session_id = sid
+
+        try:
+            result = consume_stream(proc.stdout, on_text=on_text, on_init=_on_init)
+        finally:
+            with self._stdin_lock:
+                self.inflight = False
+            self.last_active = time.monotonic()
+
+        if timed_out.is_set():
+            raise TimeoutError(f"Claude CLI timed out after {timeout}s")
+
+        if result is None:
+            stderr_tail = "\n".join(self._stderr_buf)[-500:]
+            raise RuntimeError(
+                f"Claude CLI error: no result line in stream. stderr: {stderr_tail}"
+            )
+
+        if result.get("session_id"):
+            # E-T4: refresh every turn, not just the first — a respawn after
+            # turn 5 must --resume turn 5's session, not turn 1's.
+            self.session_id = result["session_id"]
+
+        if result.get("is_error"):
+            detail = result.get("result", "")
+            if is_rate_limit_error(detail):
+                # T2/C4: persistent process surfaces a rate limit as
+                # `result.is_error`, never a nonzero exit code. Must raise
+                # this EXACT phrasing or `is_rate_limit_error()` in
+                # router.py/scheduler.py won't recognize it and the local
+                # fallback never fires.
+                raise RuntimeError(f"Claude CLI error (exit 1): {detail}")
+            # any other is_error: return it, don't raise (C4 — parser and
+            # this wrapper both stay non-throwing for non-rate-limit errors)
+
+        self._pending_steers = []
+        return result
+
+
+# ---------------------------------------------------------------------------
+# RunnerRegistry
+# ---------------------------------------------------------------------------
+
+
+class RunnerRegistry:
+    """Owns the set of live `ClaudeProcess` instances, one per channel.
+
+    Module-level global state (like `claude_session._session_locks`) — see
+    `get_registry()` below. X6: no LRU eviction; at `max_live` capacity a
+    new channel simply degrades to the one-shot path (T10/M1) rather than
+    evicting someone else's live process.
+    """
+
+    def __init__(self, max_live: int = MAX_LIVE_DEFAULT,
+                 idle_minutes: int = IDLE_MINUTES_DEFAULT,
+                 reap_interval_s: float = 60.0):
+        if max_live <= 0:
+            # X7: don't fail silently — a mis-set config value should be
+            # loud in the logs, even though the *behavior* (always degrade)
+            # is already correct without special-casing it below.
+            logger.warning(
+                "steering.max_live=%d <= 0 — steering effectively disabled, "
+                "every channel degrades to one-shot", max_live,
+            )
+        self.max_live = max_live
+        self.idle_minutes = max(1, idle_minutes)  # X7: clamp, never "reap instantly"
+        self._procs: dict[str, ClaudeProcess] = {}
+        self._lock = threading.Lock()
+        self._reap_interval_s = reap_interval_s
+        self._stopping = threading.Event()
+        self._reaper = threading.Thread(target=self._reap_loop, daemon=True)
+        self._reaper.start()
+        atexit.register(self.stop_all)
+
+    def get(self, channel_id: str, model: str = DEFAULT_MODEL,
+            session_id: str | None = None, cwd: Path | str | None = None) -> ClaudeProcess | None:
+        """Return the channel's live process, creating one if under
+        `max_live`. Returns None (T10) when at capacity and this channel
+        doesn't already have one — the caller must degrade to one-shot.
+
+        An existing entry for *channel_id* is always returned as-is,
+        regardless of capacity — which is also what rules out the M1
+        "two writers on one session_id" scenario: a channel already
+        holding a live process can never be told to fall back to one-shot
+        by this method.
+        """
+        with self._lock:
+            proc = self._procs.get(channel_id)
+            if proc is not None:
+                return proc
+            if len(self._procs) >= self.max_live:
+                logger.info(
+                    "channel=%s: max_live=%d reached, degrading to one-shot",
+                    channel_id, self.max_live,
+                )
+                return None
+            proc = ClaudeProcess(channel_id, model=model, session_id=session_id, cwd=cwd)
+            self._procs[channel_id] = proc
+            return proc
+
+    def stop(self, channel_id: str) -> bool:
+        with self._lock:
+            proc = self._procs.pop(channel_id, None)
+        if proc is None:
+            return False
+        proc.stop()
+        return True
+
+    def stop_all(self) -> None:
+        """Shutdown path (T4/A1) — zero orphaned `claude` processes after a
+        `systemctl restart`. Registered with `atexit` in `__init__`; a
+        caller in `main.py` should also invoke this explicitly on SIGTERM
+        for a synchronous shutdown (atexit is the fallback, not the primary
+        path)."""
+        self._stopping.set()
+        with self._lock:
+            procs = list(self._procs.values())
+            self._procs.clear()
+        for proc in procs:
+            try:
+                proc.stop()
+            except Exception:
+                logger.exception("channel=%s: error stopping during stop_all", proc.channel_id)
+
+    def live_count(self) -> int:
+        with self._lock:
+            return len(self._procs)
+
+    # -- reaper ---------------------------------------------------------
+
+    def _reap_loop(self) -> None:
+        while not self._stopping.wait(self._reap_interval_s):
+            self._reap_once()
+
+    def _reap_once(self) -> None:
+        """T11: resilient to any single process's `stop()` raising — the
+        loop (and the rest of the batch) must keep going, and every reap
+        must be logged (a silently-dead reaper is an invisible RAM leak)."""
+        cutoff = time.monotonic() - self.idle_minutes * 60
+        with self._lock:
+            candidates = [
+                cid for cid, p in self._procs.items()
+                if not p.inflight and p.last_active < cutoff
+            ]
+        for cid in candidates:
+            try:
+                with self._lock:
+                    proc = self._procs.get(cid)
+                    # Re-check under lock: it may have gone inflight, been
+                    # stopped, or been replaced since the snapshot above.
+                    if proc is None or proc.inflight or proc.last_active >= cutoff:
+                        continue
+                    del self._procs[cid]
+                proc.stop()
+                logger.info("reaper: stopped idle channel=%s (idle >= %d min)",
+                            cid, self.idle_minutes)
+            except Exception:
+                logger.exception("reaper: error stopping channel=%s — continuing", cid)
+
+
+# ---------------------------------------------------------------------------
+# Module-level singleton — mirrors `claude_session._session_locks`.
+# ---------------------------------------------------------------------------
+
+_registry: RunnerRegistry | None = None
+_registry_lock = threading.Lock()
+
+
+def get_registry(max_live: int = MAX_LIVE_DEFAULT,
+                  idle_minutes: int = IDLE_MINUTES_DEFAULT) -> RunnerRegistry:
+    """Lazy singleton. *max_live*/*idle_minutes* only apply on first call —
+    later calls just return the existing registry (config is read by
+    `router.py` per call and passed in here; this function doesn't re-read
+    config.json itself)."""
+    global _registry
+    if _registry is not None:
+        return _registry
+    with _registry_lock:
+        if _registry is None:
+            _registry = RunnerRegistry(max_live=max_live, idle_minutes=idle_minutes)
+        return _registry
+
+
+def reset_registry_for_tests() -> None:
+    """H2: tests must not leak live (fake) processes into each other. Stops
+    and drops the current singleton so the next `get_registry()` call
+    builds a fresh one."""
+    global _registry
+    with _registry_lock:
+        if _registry is not None:
+            _registry.stop_all()
+        _registry = None

src/claude_session.py

@@ -19,6 +19,8 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Callable
 
+from src.stream_json import consume_stream
+
 logger = logging.getLogger(__name__)
 _invoke_log = logging.getLogger("echo-core.invoke")
 _security_log = logging.getLogger("echo-core.security")
@@ -95,6 +97,53 @@ def _get_session_lock(channel_id: str) -> threading.Lock:
         return _session_locks.setdefault(channel_id, threading.Lock())
 
 
+# ---------------------------------------------------------------------------
+# In-flight turn registry — for /stop
+# ---------------------------------------------------------------------------
+#
+# Etapa 0 of the steering plan: `/stop` kills only the current turn's
+# subprocess, not the session (sessions/active.json is untouched). Maps
+# channel_id -> the live Popen for that channel's in-flight `_run_claude`
+# call. Registered at process start, removed in `_run_claude`'s `finally`
+# so it's cleaned up on every exit path (success, error, timeout).
+_live_procs: dict[str, subprocess.Popen] = {}
+_live_procs_lock = threading.Lock()
+
+
+def stop_turn(channel_id: str) -> bool:
+    """Kill the in-flight Claude CLI turn for *channel_id*, if any.
+
+    Returns True if a process was signalled, False if nothing was running.
+    Does NOT touch sessions/active.json — the session survives, only the
+    turn dies (killed proc makes _run_claude raise; route_message's
+    existing error handling reports that back to the user).
+
+    Must NOT acquire `_get_session_lock(channel_id)` — the turn thread
+    holds that lock for the whole call, so taking it here would deadlock.
+    """
+    with _live_procs_lock:
+        proc = _live_procs.get(channel_id)
+    if proc is None or proc.poll() is not None:
+        return False
+    try:
+        proc.terminate()
+        proc.wait(timeout=3)
+    except subprocess.TimeoutExpired:
+        try:
+            proc.kill()
+        except OSError:
+            pass
+    except OSError:
+        pass
+    # Also remove here (not just in _run_claude's finally): the turn thread
+    # may take a moment to notice the process died, and a repeated /stop
+    # in that window should see "nothing running", not re-signal a dead proc.
+    with _live_procs_lock:
+        if _live_procs.get(channel_id) is proc:
+            del _live_procs[channel_id]
+    return True
+
+
 PERSONALITY_FILES = [
     "IDENTITY.md",
     "SOUL.md",
@@ -286,6 +335,7 @@ def _run_claude(
     timeout: int,
     on_text: Callable[[str], None] | None = None,
     cwd: Path | str | None = None,
+    channel_id: str | None = None,
 ) -> dict:
     """Run a Claude CLI command and return parsed output.
 
@@ -300,6 +350,10 @@ def _run_claude(
     *cwd* — optional working directory override (default: PROJECT_ROOT).
     Used by PlanningSession to scope the subprocess to ``~/workspace/<slug>/``
     so artifacts land in the target repo.
+
+    *channel_id* — if given, registers the subprocess in `_live_procs` so
+    `/stop` can kill this turn. Callers that omit it (heartbeat, planning
+    sessions) simply aren't stoppable — unchanged behavior.
     """
     if not shutil.which(CLAUDE_BIN):
         raise FileNotFoundError(
@@ -316,6 +370,10 @@ def _run_claude(
         cwd=str(cwd) if cwd else PROJECT_ROOT,
     )
 
+    if channel_id is not None:
+        with _live_procs_lock:
+            _live_procs[channel_id] = proc
+
     # Watchdog thread: kill the process if it exceeds the timeout
     timed_out = threading.Event()
 
@@ -332,40 +390,18 @@ def _run_claude(
     watchdog = threading.Thread(target=_watchdog, daemon=True)
     watchdog.start()
 
-    # --- Parse stream-json output line by line ---
+    # --- Parse stream-json output (T5: shared with ClaudeProcess via
-    text_blocks: list[str] = []
+    # src.stream_json.consume_stream; see that module's docstring for its
-    result_obj: dict | None = None
+    # non-throwing contract — blocker C4. The raising below stays here. ---
-    intermediate_count = 0
-
     try:
-        for line in proc.stdout:
+        result = consume_stream(proc.stdout, on_text=on_text)
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = json.loads(line)
-            except json.JSONDecodeError:
-                continue
-
-            msg_type = obj.get("type")
-
-            if msg_type == "assistant":
-                message = obj.get("message", {})
-                for block in message.get("content", []):
-                    if block.get("type") == "text":
-                        text = block.get("text", "").strip()
-                        if text:
-                            text_blocks.append(text)
-                            if on_text:
-                                try:
-                                    on_text(text)
-                                    intermediate_count += 1
-                                except Exception:
-                                    logger.exception("on_text callback error")
-
-            elif msg_type == "result":
-                result_obj = obj
     finally:
+        # Deregister on every exit path (success, error, timeout) so a
+        # stale entry never outlives the process it points to.
+        if channel_id is not None:
+            with _live_procs_lock:
+                if _live_procs.get(channel_id) is proc:
+                    del _live_procs[channel_id]
         # Ensure process resources are cleaned up
         proc.stdout.close()
         try:
@@ -381,39 +417,24 @@ def _run_claude(
         raise TimeoutError(f"Claude CLI timed out after {timeout}s")
 
     if proc.returncode != 0:
-        stdout_tail = "\n".join(text_blocks[-3:]) if text_blocks else ""
+        stdout_tail = (result or {}).get("result", "") or ""
-        # Check if result_obj has an error
+        result_error = stdout_tail if (result and result.get("is_error")) else ""
-        result_error = ""
-        if result_obj and result_obj.get("is_error"):
-            result_error = result_obj.get("result", "") or result_obj.get("error", "")
         detail = stderr_output[:500] or result_error[:500] or stdout_tail[:500]
         logger.error("Claude CLI stderr: %s", stderr_output[:1000])
-        logger.error("Claude CLI result_obj: %s", result_obj)
+        logger.error("Claude CLI result_obj: %s", result)
         raise RuntimeError(
             f"Claude CLI error (exit {proc.returncode}): {detail}"
         )
 
-    if result_obj is None:
+    if result is None:
         raise RuntimeError(
             "Failed to parse Claude CLI output: no result line in stream"
         )
 
-    combined_text = "\n\n".join(text_blocks) if text_blocks else result_obj.get("result", "")
+    # `consume_stream` already returns exactly the dict shape below
-
+    # (result/session_id/usage/total_cost_usd/cost_usd/duration_ms/num_turns/
-    return {
+    # intermediate_count/subtype/is_error) — nothing left to rebuild.
-        "result": combined_text,
+    return result
-        "session_id": result_obj.get("session_id", ""),
-        "usage": result_obj.get("usage", {}),
-        "total_cost_usd": result_obj.get("total_cost_usd", 0),
-        "cost_usd": result_obj.get("cost_usd", 0),
-        "duration_ms": result_obj.get("duration_ms", 0),
-        "num_turns": result_obj.get("num_turns", 0),
-        "intermediate_count": intermediate_count,
-        # Surface subtype/is_error for callers that retry on `error_max_turns`
-        # (PlanningSession does this — spike findings recommended retry strategy).
-        "subtype": result_obj.get("subtype", ""),
-        "is_error": bool(result_obj.get("is_error", False)),
-    }
 
 
 # ---------------------------------------------------------------------------
@@ -497,7 +518,7 @@ def start_session(
     ]
 
     _t0 = time.monotonic()
-    data = _run_claude(cmd, timeout, on_text=on_text)
+    data = _run_claude(cmd, timeout, on_text=on_text, channel_id=channel_id)
     _elapsed_ms = int((time.monotonic() - _t0) * 1000)
 
     for field in ("result", "session_id"):
@@ -543,11 +564,15 @@ def resume_session(
     message: str,
     timeout: int = DEFAULT_TIMEOUT,
     on_text: Callable[[str], None] | None = None,
+    channel_id: str | None = None,
 ) -> str:
     """Resume an existing Claude session by ID. Returns response text.
 
     If *on_text* is provided, each intermediate Claude text block is passed
     to the callback as soon as it arrives.
+
+    *channel_id* — passed through to `_run_claude` so `/stop` can find this
+    turn's subprocess; optional (defaults to not stoppable).
     """
     # Find channel/model for logging and model selection
     sessions = _load_sessions()
@@ -571,7 +596,7 @@ def resume_session(
     ]
 
     _t0 = time.monotonic()
-    data = _run_claude(cmd, timeout, on_text=on_text)
+    data = _run_claude(cmd, timeout, on_text=on_text, channel_id=channel_id)
     _elapsed_ms = int((time.monotonic() - _t0) * 1000)
 
     if not data.get("result"):
@@ -608,6 +633,268 @@ def resume_session(
     return response_text
 
 
+# ---------------------------------------------------------------------------
+# Steering dispatcher (H1/C2/C3/M1) — see tasks/steering-plan.md.
+#
+# `claude_runner` imports FROM this module (parser, build_system_prompt,
+# _safe_env) — importing it back at module level here would be a cycle, so
+# every reference below is a lazy `from src import claude_runner` inside a
+# function body, as the plan requires.
+# ---------------------------------------------------------------------------
+
+# C2: which adapter owns the turn currently dispatched on a channel. Voice
+# and text share `channel_id` as session_key (router.py's `session_key =
+# channel_id`) — without this a text message could steer into a live VOICE
+# turn and its reply would be spoken aloud, never rendered as text. Plain
+# dict; a stale/racy read only affects the rare tie-break in the tiny
+# window between acquiring the lock and recording ownership below, which is
+# dwarfed by real subprocess/IO latency.
+_channel_adapter: dict[str, str] = {}
+
+
+# Same three namespaces router.py's channel config lives in (Discord,
+# Telegram, WhatsApp) — checked in order so a per-channel `steering` key
+# works regardless of which adapter the channel_id came from.
+_CHANNEL_NAMESPACES = ("channels", "telegram_channels", "whatsapp_channels")
+
+
+def _channel_steering_override(config, channel_id: str) -> bool | None:
+    """Etapa 6 (X16): `channels.<alias>.steering` (or its telegram/whatsapp
+    twins), matched by the channel's `id` field — same lookup router.py's
+    `_get_channel_config` uses. None if unset, so the caller falls through
+    to the global `steering.enabled` default."""
+    for namespace in _CHANNEL_NAMESPACES:
+        for ch in (config.get(namespace, {}) or {}).values():
+            if ch.get("id") == channel_id and "steering" in ch:
+                return bool(ch["steering"])
+    return None
+
+
+def _steering_config(channel_id: str | None = None) -> tuple[bool, int, int]:
+    """Read `steering.*` through `Config()` on EVERY call — never at import
+    time. A module-level read (like `ALLOWED_TOOLS` above) would mean the
+    flag can't be flipped without a restart, which defeats the point of a
+    rollback flag. `ECHO_STEERING=off` (kill switch, precedent: `CLAUDE_BIN`)
+    always wins over config.json. Returns (enabled, max_live, idle_minutes)
+    with the clamps already applied: `max_live <= 0` forces disabled;
+    `idle_minutes < 1` clamps to 1.
+
+    *channel_id* — if given, a per-channel `steering` override (see
+    `_channel_steering_override`) wins over the global `enabled` flag for
+    that one channel — the staged rollout the plan's Etapa 6 needs ("flip
+    on one channel, then global"). Unset on the channel -> global default.
+    """
+    if os.environ.get("ECHO_STEERING", "").strip().lower() == "off":
+        logger.debug("steering: disabled by ECHO_STEERING=off kill switch")
+        return False, 0, 20
+    try:
+        from src.config import Config
+        full_cfg = Config()
+        cfg = full_cfg.get("steering", {}) or {}
+    except (OSError, ValueError):
+        logger.exception("steering: failed to read config.json — treating as disabled")
+        return False, 0, 20
+    enabled = bool(cfg.get("enabled", False))
+    if channel_id is not None:
+        override = _channel_steering_override(full_cfg, channel_id)
+        if override is not None:
+            logger.debug("channel=%s: per-channel steering override=%s", channel_id, override)
+            enabled = override
+    max_live = cfg.get("max_live", 2)
+    if not isinstance(max_live, int):
+        max_live = 0
+    idle_minutes = cfg.get("idle_minutes", 20)
+    if not isinstance(idle_minutes, int) or idle_minutes < 1:
+        idle_minutes = 1
+    if max_live <= 0 and enabled:
+        logger.warning("steering: max_live=%r <= 0 — disabling steering", max_live)
+        enabled = False
+    if not enabled:
+        logger.debug("steering: disabled by config")
+    return enabled, max_live, idle_minutes
+
+
+def _persist_steering_session(channel_id: str, model: str, data: dict) -> None:
+    """Persist sessions/active.json after a steering-dispatched turn — same
+    fields/update semantics `start_session`/`resume_session` already use, so
+    `/status`, `/model`, and a later one-shot fallback all keep working."""
+    usage = data.get("usage", {})
+    now = datetime.now(timezone.utc).isoformat()
+    sessions = _load_sessions()
+    existing = sessions.get(channel_id)
+    if existing is None:
+        sessions[channel_id] = {
+            "session_id": data.get("session_id", ""),
+            "model": model,
+            "created_at": now,
+            "last_message_at": now,
+            "message_count": 1,
+            "total_input_tokens": usage.get("input_tokens", 0),
+            "total_output_tokens": usage.get("output_tokens", 0),
+            "total_cost_usd": data.get("total_cost_usd", 0),
+            "duration_ms": data.get("duration_ms", 0),
+            "context_tokens": usage.get("input_tokens", 0) + usage.get("output_tokens", 0),
+        }
+    else:
+        existing["session_id"] = data.get("session_id") or existing.get("session_id", "")
+        existing["model"] = model
+        existing["last_message_at"] = now
+        existing["message_count"] = existing.get("message_count", 0) + 1
+        existing["total_input_tokens"] = existing.get("total_input_tokens", 0) + usage.get("input_tokens", 0)
+        existing["total_output_tokens"] = existing.get("total_output_tokens", 0) + usage.get("output_tokens", 0)
+        existing["total_cost_usd"] = existing.get("total_cost_usd", 0) + data.get("total_cost_usd", 0)
+        existing["duration_ms"] = existing.get("duration_ms", 0) + data.get("duration_ms", 0)
+        existing["context_tokens"] = usage.get("input_tokens", 0) + usage.get("output_tokens", 0)
+    _save_sessions(sessions)
+    _invoke_log.info(
+        "channel=%s model=%s steered=1 session=%s",
+        channel_id, model, sessions[channel_id]["session_id"][:8],
+    )
+
+
+def _run_steering_turn(
+    proc, lock: threading.Lock, channel_id: str, message: str, model: str,
+    timeout: int, on_text: Callable[[str], None] | None, adapter_name: str,
+) -> str:
+    """Run *message* as a fresh turn on the already-acquired *lock*/*proc*.
+
+    Releases the lock on every exit path. On failure, any messages that got
+    steered into this same turn before it failed are left in
+    `proc._pending_steers` for `pop_pending_steers()` (C3) — router.py picks
+    them up from its exception handler and re-dispatches them, since their
+    own request threads already returned `__STEERED__` and are gone.
+    """
+    _channel_adapter[channel_id] = adapter_name
+    try:
+        data = proc.run_turn(message, on_text=on_text, timeout=timeout)
+    except Exception:
+        lock.release()
+        raise
+    lock.release()
+    _persist_steering_session(channel_id, model, data)
+    return data["result"]
+
+
+def _dispatch_steering(
+    channel_id: str,
+    message: str,
+    model: str,
+    timeout: int,
+    on_text: Callable[[str], None] | None,
+    adapter_name: str | None,
+    max_live: int,
+    idle_minutes: int,
+) -> str | None:
+    """Try to steer *message* into a live process for *channel_id*, or run
+    it as a fresh turn. Returns the response text, the `__STEERED__`
+    sentinel, or None to tell `send_message` to fall back to the unchanged
+    one-shot path (T10: registry is at capacity and this channel has no
+    live process of its own — the only case where that path is safe, see
+    M1 below).
+
+    T12: every branch logs why a message did or did not become a steer.
+    """
+    from src import claude_runner
+
+    registry = claude_runner.get_registry(max_live=max_live, idle_minutes=idle_minutes)
+    session = get_active_session(channel_id)
+    session_id = session.get("session_id") if session else None
+    effective_model = model
+    if session is not None and session.get("model"):
+        effective_model = session["model"]
+
+    proc = registry.get(channel_id, model=effective_model, session_id=session_id)
+    if proc is None:
+        # M1: safe only because a channel that ALREADY has a live process
+        # is always returned as-is by `registry.get()`, regardless of
+        # capacity — so we only ever land here for a channel with no live
+        # process, meaning the one-shot fallback below is the sole writer.
+        logger.info(
+            "channel=%s: steering registry at capacity — degrading to one-shot (T10)",
+            channel_id,
+        )
+        return None
+
+    lock = _get_session_lock(channel_id)
+    caller_adapter = adapter_name or "echo"
+
+    if lock.acquire(blocking=False):
+        # H1: acquiring the lock IS "no turn is running" — no separate
+        # `inflight` flag/check, no TOCTOU between them.
+        logger.debug("channel=%s: lock free — running as a full turn (not a steer)", channel_id)
+        return _run_steering_turn(
+            proc, lock, channel_id, message, effective_model, timeout, on_text, caller_adapter,
+        )
+
+    owner = _channel_adapter.get(channel_id)
+    if owner is not None and owner != caller_adapter:
+        # C2: never steer a message from a different adapter into someone
+        # else's in-flight turn (a text message hijacking a live voice
+        # turn, or vice versa — they share `channel_id`). M1 rules out
+        # falling back to the one-shot `resume_session` path here too: this
+        # channel already has a live process holding the session, and a
+        # concurrent one-shot `--resume` would be a second writer on the
+        # same session_id. So: wait for the current turn's lock like today,
+        # then take our turn through the SAME live process.
+        logger.info(
+            "channel=%s: turn owned by adapter=%s, message from adapter=%s — "
+            "waiting instead of steering (C2), staying on steering process (M1)",
+            channel_id, owner, caller_adapter,
+        )
+        lock.acquire(blocking=True)
+        return _run_steering_turn(
+            proc, lock, channel_id, message, effective_model, timeout, on_text, caller_adapter,
+        )
+
+    result = proc.steer(message)
+    if result.status is claude_runner.SteerStatus.STEERED:
+        logger.info("channel=%s: steered %d chars (adapter=%s)",
+                    channel_id, len(message), caller_adapter)
+        from src.sentinels import STEERED
+        return STEERED
+
+    # RAN_AS_TURN / PROCESS_DEAD: `result.turn` IS the response for this
+    # text (D4/T7) — never re-run it, that would double-send and double-bill.
+    logger.info("channel=%s: steer() resolved as %s — using its turn result directly",
+                channel_id, result.status.value)
+    _persist_steering_session(channel_id, effective_model, result.turn)
+    return result.turn["result"]
+
+
+def pop_pending_steers(channel_id: str) -> list[str]:
+    """C3 hook for router.py: texts steered into a turn on *channel_id*
+    that then failed (rate limit, timeout, crash) — the caller MUST
+    re-dispatch them, never drop them (CLAUDE.md: "turnul nu se pierde
+    niciodată"). Empty list if steering was never used on this process
+    (never spins up the registry as a side effect — that would be wrong
+    for a channel that never used steering) or the channel has no live
+    process, or nothing was pending.
+    """
+    from src import claude_runner
+
+    registry = claude_runner._registry
+    if registry is None:
+        return []
+    proc = registry._procs.get(channel_id)  # peek only — .get() would spawn a new one
+    if proc is None:
+        return []
+    return proc.pop_pending_steers()
+
+
+def _stop_steering_process(channel_id: str) -> None:
+    """T9: `/clear` and `/model` must not leave a live steering process
+    running with stale session/model config — it respawns on the next
+    turn (via `_respawn_if_needed`/`--resume`), picking up the new state.
+    No-op if steering was never used (doesn't spin up the registry)."""
+    from src import claude_runner
+
+    registry = claude_runner._registry
+    if registry is None:
+        return
+    if registry.stop(channel_id):
+        logger.info("channel=%s: stopped live steering process (T9)", channel_id)
+
+
 def send_message(
     channel_id: str,
     message: str,
@@ -615,6 +902,7 @@ def send_message(
     timeout: int = DEFAULT_TIMEOUT,
     on_text: Callable[[str], None] | None = None,
     voice_mode: bool = False,
+    adapter_name: str | None = None,
 ) -> str:
     """High-level convenience: auto start or resume based on channel state.
 
@@ -624,12 +912,35 @@ def send_message(
     parallel — each holds its own lock. Lock is acquired blocking; we rely
     on `timeout` (default 5 minutes) to bound the worst case rather than
     a non-blocking acquire (loss of fairness vs adapter-side queueing).
+
+    *adapter_name* — steering dispatcher only (C2). Identifies which
+    adapter's turn is in flight so a message from a DIFFERENT adapter on
+    the same channel_id never gets steered into it. Ignored entirely when
+    steering is off — this whole path is then bit-for-bit what it always
+    was.
+
+    When `steering.enabled` is on (config.json, per-call — see
+    `_steering_config`), this dispatches through `_dispatch_steering`
+    first: it returns a response, the `__STEERED__` sentinel, or None to
+    fall through to the unchanged path below.
     """
+    enabled, max_live, idle_minutes = _steering_config(channel_id)
+    if enabled:
+        dispatched = _dispatch_steering(
+            channel_id, message, model, timeout, on_text, adapter_name,
+            max_live, idle_minutes,
+        )
+        if dispatched is not None:
+            return dispatched
+
     with _get_session_lock(channel_id):
         session = get_active_session(channel_id)
         # Only resume if session has a valid session_id (not a pre-set model placeholder)
         if session is not None and session.get("session_id"):
-            return resume_session(session["session_id"], message, timeout, on_text=on_text)
+            return resume_session(
+                session["session_id"], message, timeout,
+                on_text=on_text, channel_id=channel_id,
+            )
         # Use model from pre-set session if available, otherwise use provided model
         effective_model = model
         if session is not None and session.get("model"):
@@ -643,6 +954,7 @@ def send_message(
 
 def clear_session(channel_id: str) -> bool:
     """Remove a channel's session entry. Returns True if removed."""
+    _stop_steering_process(channel_id)  # T9
     sessions = _load_sessions()
     if channel_id not in sessions:
         return False
@@ -663,6 +975,7 @@ def set_session_model(channel_id: str, model: str) -> bool:
         raise ValueError(
             f"Invalid model '{model}'. Must be one of: {', '.join(sorted(VALID_MODELS))}"
         )
+    _stop_steering_process(channel_id)  # T9
     sessions = _load_sessions()
     if channel_id not in sessions:
         return False

src/main.py

@@ -21,6 +21,21 @@ PID_FILE = PROJECT_ROOT / "echo-core.pid"
 LOG_DIR = PROJECT_ROOT / "logs"
 
 
+def _stop_steering_registry(logger: logging.Logger) -> None:
+    """T4: kill any live persistent `claude` processes before the process
+    exits. `RunnerRegistry.stop_all()` is already `atexit`-registered, but
+    `atexit` does NOT fire on SIGTERM — exactly how `systemctl --user
+    restart echo-core` stops this process — so it must also be called
+    explicitly from the shutdown path. A leaked live process is 292-541 MB
+    RSS on an 8 GB host. Imported lazily so a system without steering wired
+    up yet (or any import error in claude_runner) can't break shutdown."""
+    try:
+        from src.claude_runner import get_registry
+        get_registry().stop_all()
+    except Exception:
+        logger.exception("Error stopping claude_runner registry during shutdown")
+
+
 def setup_logging():
     LOG_DIR.mkdir(parents=True, exist_ok=True)
     fmt = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
@@ -148,6 +163,7 @@ def main():
         loop.run_until_complete(scheduler.stop())
         loop.run_until_complete(client.close())
     finally:
+        _stop_steering_registry(logger)
         PID_FILE.unlink(missing_ok=True)
         logger.info("Echo Core shut down.")
 

src/router.py

@@ -24,7 +24,10 @@ from src.claude_session import (
     rate_limit_detail as _rate_limit_detail,
     RATE_LIMIT_RE as _RATE_LIMIT_RE,
     VALID_MODELS,
+    stop_turn,
+    pop_pending_steers as _pop_pending_steers,
 )
+from src.sentinels import is_steered as _is_steered
 from src.jsonlock import read_locked, write_locked
 from src.planning_orchestrator import PlanningOrchestrator
 from src.planning_session import (
@@ -566,6 +569,11 @@ def route_message(
     if text.lower() == "/status":
         return _status(channel_id), True
 
+    if text.lower() == "/stop":
+        if stop_turn(channel_id):
+            return "⏹ Oprit.", True
+        return "Nu rulează nimic pe canalul ăsta.", True
+
     if text.lower().startswith("/model"):
         return _model_command(channel_id, text), True
 
@@ -602,15 +610,29 @@ def route_message(
     try:
         response = send_message(
             session_key, claude_text, model=model, on_text=on_text,
-            voice_mode=voice_mode,
+            voice_mode=voice_mode, adapter_name=adapter_name,
         )
+        if _is_steered(response):
+            # Same pattern as the existing __AUDIO__: sentinel — no
+            # _set_last_response, the adapter reacts instead of replying.
+            return response, False
         _set_last_response(channel_id, response)
         return response, False
     except Exception as e:
         log.error("Claude error for channel %s: %s", channel_id, e)
+        # C3: texts steered into this same turn before it failed — their own
+        # request threads already returned __STEERED__ and are gone, so the
+        # only way left to answer them is `on_text`, the same real-time
+        # channel already used for intermediate assistant text.
+        pending_steers = _pop_pending_steers(channel_id)
         if _is_rate_limit_error(e):
             log.warning("Rate limit detected for channel %s — trying local fallback", channel_id)
             fallback = _local_fallback_reply(text, channel_id=channel_id)
+            for steered_text in pending_steers:
+                _redeliver_steered_reply(
+                    steered_text, channel_id, on_text,
+                    _local_fallback_reply(steered_text, channel_id=channel_id),
+                )
             if fallback is not None:
                 _set_last_response(channel_id, fallback)
                 return fallback, False
@@ -622,9 +644,37 @@ def route_message(
                 "⚠️ Claude e la limită, iar modelul local nu a răspuns.\n"
                 f"{_rate_limit_detail(e)}"
             ), False
+        for steered_text in pending_steers:
+            _redeliver_steered_reply(steered_text, channel_id, on_text, f"Error: {e}")
         return f"Error: {e}", False
 
 
+def _redeliver_steered_reply(
+    steered_text: str,
+    channel_id: str,
+    on_text: Callable[[str], None] | None,
+    reply: str | None,
+) -> None:
+    """C3 — a message steered into a turn that then failed must still get
+    an answer. Its own request thread already returned `__STEERED__` and is
+    gone, so the only way left to reach the user is `on_text` (the same
+    real-time channel adapters already use for intermediate assistant
+    text). Logs instead of dropping silently when there's no `on_text` to
+    push through (T12 — "it ignored my message" must stay diagnosable)."""
+    if reply is None:
+        reply = "⚠️ Claude e la limită — mesajul tău steered nu a primit răspuns."
+    if on_text is None:
+        log.warning(
+            "channel=%s: steered message lost — no on_text to redeliver it: %r",
+            channel_id, steered_text[:80],
+        )
+        return
+    try:
+        on_text(reply)
+    except Exception:
+        log.exception("channel=%s: failed to redeliver steered reply via on_text", channel_id)
+
+
 def _status(channel_id: str) -> str:
     """Build status message for a channel."""
     session = get_active_session(channel_id)

src/sentinels.py

@@ -0,0 +1,40 @@
+"""Single place to recognise router sentinels.
+
+``route_message()`` normally returns text to post back to the user. Two
+special values are protocol instead of a reply, and every adapter must
+catch them before sending anything:
+
+- ``__AUDIO__:<path>`` — a TTS result; the payload is a WAV path to attach.
+- ``__STEERED__`` — the message was injected into an in-flight turn; the
+  adapter should react (not reply) and post nothing.
+
+Zero dependencies on adapters or router — importable standalone.
+"""
+from __future__ import annotations
+
+AUDIO_PREFIX = "__AUDIO__:"
+STEERED = "__STEERED__"
+
+
+def audio_path(response: str | None) -> str | None:
+    """Return the WAV path if `response` is an audio sentinel, else None."""
+    if response and response.startswith(AUDIO_PREFIX):
+        return response[len(AUDIO_PREFIX):]
+    return None
+
+
+def is_steered(response: str | None) -> bool:
+    """True if `response` is the steering-acknowledgement sentinel."""
+    return response == STEERED
+
+
+if __name__ == "__main__":
+    assert audio_path("__AUDIO__:/tmp/x.wav") == "/tmp/x.wav"
+    assert audio_path("hello") is None
+    assert audio_path("") is None
+    assert audio_path(None) is None
+    assert is_steered("__STEERED__") is True
+    assert is_steered("__STEERED__ ") is False
+    assert is_steered("") is False
+    assert is_steered(None) is False
+    print("ok")

src/stream_json.py

@@ -0,0 +1,153 @@
+"""Pure stream-json parsing, shared by the one-shot (`_run_claude`) and
+persistent (`ClaudeProcess`) read loops.
+
+Leaf module: imports nothing from `claude_session.py` or `claude_runner.py`,
+so neither can create an import cycle through it.
+
+Contract (blocker C4 in tasks/steering-plan.md): this module NEVER raises on
+`is_error` / `subtype`. It only parses stream-json events into the dict shape
+`_run_claude` has always returned. `PlanningSession` depends on `_run_claude`
+returning an error *result* rather than throwing (it retries on
+`error_max_turns` — see `planning_session.py`'s `RETRY_MAX_TURNS`); any
+future caller that wants to raise on a particular error (e.g. rate limits on
+a persistent process, done in `claude_runner.ClaudeProcess.run_turn`) must
+do so itself, on top of the dict this module returns.
+"""
+
+import json
+import logging
+from typing import Callable, Iterable
+
+logger = logging.getLogger(__name__)
+
+
+def parse_event(line: str) -> dict | None:
+    """Parse one stream-json line into an event dict, or None if blank/invalid."""
+    line = line.strip()
+    if not line:
+        return None
+    try:
+        return json.loads(line)
+    except json.JSONDecodeError:
+        return None
+
+
+def is_init_event(event: dict) -> bool:
+    """True for the `system`/`init` event a fresh Claude CLI process emits
+    once, at the start of its very first turn, carrying `session_id` before
+    any assistant text or the turn's `result` line exists.
+
+    Needed for `ClaudeProcess`: if a turn times out before its `result`
+    ever arrives, the session_id captured here is the only way the *next*
+    turn can `--resume` the same conversation (see plan finding M2) — the
+    one-shot path doesn't need this, since it always gets `session_id` off
+    the `result` line of a turn that, by definition, already completed.
+    """
+    return event.get("type") == "system" and event.get("subtype") == "init"
+
+
+def consume_stream(
+    lines: Iterable[str],
+    on_text: Callable[[str], None] | None = None,
+    on_init: Callable[[str], None] | None = None,
+) -> dict | None:
+    """Read stream-json events from *lines* until (and including) a `result`
+    event, then stop — never read past it.
+
+    Stopping exactly at the turn boundary is what lets a persistent process
+    reuse this for one turn at a time: reading further would eat the *next*
+    turn's events off the same stdout.
+
+    *on_text* is called with each intermediate assistant text block as soon
+    as it arrives (same as `_run_claude` today). *on_init* is called once,
+    with the session_id, when a `system`/`init` event is seen (may never
+    fire — the one-shot caller doesn't need it and passes None).
+
+    Returns the same dict `_run_claude` has always returned, or None if the
+    stream ended (EOF) without ever producing a `result` line — mirroring
+    `_run_claude`'s own `if result_obj is None: raise RuntimeError(...)`
+    check. Raising on that is the CALLER's job, not this function's (C4).
+    """
+    text_blocks: list[str] = []
+    result_obj: dict | None = None
+    intermediate_count = 0
+
+    for raw_line in lines:
+        event = parse_event(raw_line)
+        if event is None:
+            continue
+
+        msg_type = event.get("type")
+
+        if msg_type == "assistant":
+            message = event.get("message", {})
+            for block in message.get("content", []):
+                if block.get("type") == "text":
+                    text = block.get("text", "").strip()
+                    if text:
+                        text_blocks.append(text)
+                        if on_text:
+                            try:
+                                on_text(text)
+                                intermediate_count += 1
+                            except Exception:
+                                logger.exception("on_text callback error")
+
+        elif on_init is not None and is_init_event(event):
+            session_id = event.get("session_id")
+            if session_id:
+                try:
+                    on_init(session_id)
+                except Exception:
+                    logger.exception("on_init callback error")
+
+        elif msg_type == "result":
+            result_obj = event
+            break  # never read past the turn boundary
+
+    if result_obj is None:
+        return None
+
+    combined_text = "\n\n".join(text_blocks) if text_blocks else result_obj.get("result", "")
+
+    return {
+        "result": combined_text,
+        "session_id": result_obj.get("session_id", ""),
+        "usage": result_obj.get("usage", {}),
+        "total_cost_usd": result_obj.get("total_cost_usd", 0),
+        "cost_usd": result_obj.get("cost_usd", 0),
+        "duration_ms": result_obj.get("duration_ms", 0),
+        "num_turns": result_obj.get("num_turns", 0),
+        "intermediate_count": intermediate_count,
+        "subtype": result_obj.get("subtype", ""),
+        "is_error": bool(result_obj.get("is_error", False)),
+    }
+
+
+def demo() -> None:
+    """Smallest runnable self-check (ponytail: no test framework needed)."""
+    lines = [
+        json.dumps({"type": "system", "subtype": "init", "session_id": "s1"}),
+        json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": "hi"}]}}),
+        json.dumps({"type": "result", "session_id": "s1", "result": "hi", "is_error": False}),
+        json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": "SHOULD NOT APPEAR"}]}}),
+    ]
+    seen_init = []
+    out = consume_stream(lines, on_init=seen_init.append)
+    assert out is not None
+    assert out["result"] == "hi"
+    assert out["session_id"] == "s1"
+    assert out["is_error"] is False
+    assert seen_init == ["s1"]
+
+    # No result line at all -> None, does not raise.
+    assert consume_stream([json.dumps({"type": "assistant", "message": {"content": []}})]) is None
+
+    # Blank/garbage lines are skipped, not fatal.
+    assert consume_stream(["", "not json", json.dumps({"type": "result", "result": "ok"})])["result"] == "ok"
+
+    print("stream_json self-check OK")
+
+
+if __name__ == "__main__":
+    demo()

src/voice/pipeline.py

@@ -34,6 +34,7 @@ from typing import Any, Callable, Optional
 import numpy as np
 
 from src.router import _strip_leading_voice_tokens
+from src.sentinels import is_steered
 from src.voice._discord_voice_adapter import AudioSink, VoiceData
 from src.voice.voice_commands import detect_voice_change
 
@@ -434,7 +435,7 @@ class VoiceSession:
                 voice_stream_callback,      # on_text
                 "discord-voice",            # adapter_name
             )
-            if isinstance(result, tuple) and result:
+            if isinstance(result, tuple) and result and not is_steered(result[0]):
                 response_text = _strip_leading_voice_tokens(result[0] or "")
         except Exception as e:  # noqa: BLE001
             log.error("route_message voice path failed: %s", e)

tests/fake_claude.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Fake `claude` CLI binary for tests/test_claude_runner.py.
+
+Runs as a REAL subprocess (spawned by `subprocess.Popen`, exactly like the
+real CLI) so steering genuinely exercises stdin/stdout pipe timing that a
+mock `Popen` can't reproduce. Speaks the subset of stream-json that
+`src/claude_runner.py` depends on:
+
+  - one `system`/`init` event with a `session_id`, on the first turn only
+  - one `assistant` event per response, with a text block
+  - one `result` event per turn
+
+Controlled entirely via environment variables (this process has no access
+to the test's Python objects):
+
+  FAKE_CLAUDE_SCENARIO      normal | steer | timeout | rate_limit |
+                            big_stderr | crash_immediately   (default: normal)
+  FAKE_CLAUDE_SESSION_ID    session_id to report                (default: fake-session-1)
+
+The `system`/`init` event also carries the process's own `argv` — tests use
+this to assert `--resume <sid>` was actually passed on respawn, without
+needing a separate side channel.
+"""
+import json
+import os
+import sys
+import time
+
+
+def _emit(obj: dict) -> None:
+    print(json.dumps(obj), flush=True)
+
+
+def main() -> None:
+    scenario = os.environ.get("FAKE_CLAUDE_SCENARIO", "normal")
+    session_id = os.environ.get("FAKE_CLAUDE_SESSION_ID", "fake-session-1")
+
+    if scenario == "crash_immediately":
+        sys.exit(1)
+
+    if scenario == "big_stderr":
+        # T3: without a stderr-draining thread on the caller's side, this
+        # fills the OS pipe buffer (~64KB) and blocks forever, before this
+        # process ever gets to read a turn off stdin.
+        for _ in range(4000):
+            print("x" * 40, file=sys.stderr, flush=True)
+
+    first_turn = True
+    for raw_line in sys.stdin:
+        raw_line = raw_line.strip()
+        if not raw_line:
+            continue
+        try:
+            msg = json.loads(raw_line)
+        except json.JSONDecodeError:
+            continue
+        text = msg.get("message", {}).get("content", "")
+
+        if first_turn:
+            _emit({
+                "type": "system", "subtype": "init",
+                "session_id": session_id, "argv": sys.argv,
+            })
+            first_turn = False
+
+        if scenario == "timeout":
+            time.sleep(600)  # never responds — caller's own watchdog must kill us
+            return
+
+        if scenario == "rate_limit":
+            _emit({
+                "type": "result", "subtype": "error", "session_id": session_id,
+                "result": "You've hit your session limit · resets 10:50am (UTC)",
+                "is_error": True,
+            })
+            continue
+
+        if scenario == "generic_error":
+            # is_error=True but NOT a rate limit — C4: must be returned,
+            # never raised, so a future PlanningSession-style caller can
+            # still retry on `subtype` instead of catching an exception.
+            _emit({
+                "type": "result", "subtype": "error_max_turns", "session_id": session_id,
+                "result": "hit max turns for this task", "is_error": True,
+            })
+            continue
+
+        if scenario == "steer":
+            # Simulate a turn in progress that then reads a SECOND stdin
+            # line (the steer) before finishing — exactly what a real
+            # steering exchange looks like (spike: one `result`, num_turns=2).
+            _emit({"type": "assistant", "message": {"content": [
+                {"type": "text", "text": f"got:{text}"},
+            ]}})
+            second_raw = sys.stdin.readline().strip()
+            if second_raw:
+                try:
+                    second_msg = json.loads(second_raw)
+                    second_text = second_msg.get("message", {}).get("content", "")
+                except json.JSONDecodeError:
+                    second_text = ""
+                _emit({"type": "assistant", "message": {"content": [
+                    {"type": "text", "text": f"steered:{second_text}"},
+                ]}})
+            _emit({
+                "type": "result", "subtype": "success", "session_id": session_id,
+                "result": "done", "is_error": False, "num_turns": 2,
+                "usage": {"input_tokens": 1, "output_tokens": 1},
+            })
+            continue
+
+        # normal
+        _emit({"type": "assistant", "message": {"content": [
+            {"type": "text", "text": f"echo:{text}"},
+        ]}})
+        _emit({
+            "type": "result", "subtype": "success", "session_id": session_id,
+            "result": f"echo:{text}", "is_error": False, "num_turns": 1,
+            "usage": {"input_tokens": 1, "output_tokens": 1},
+        })
+
+
+if __name__ == "__main__":
+    main()

tests/test_claude_runner.py

@@ -0,0 +1,351 @@
+"""Tests for src/claude_runner.py (steering — persistent Claude processes).
+
+Fully offline: `tests/fake_claude.py` is a real subprocess (no mocked
+`subprocess.Popen`) speaking a minimal stream-json dialect, so steering
+timing over real OS pipes is genuinely exercised rather than assumed.
+"""
+
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+from src import claude_runner, claude_session
+from src.claude_runner import SteerStatus
+
+FAKE_CLAUDE = Path(__file__).parent / "fake_claude.py"
+
+
+def _wait_until(predicate, timeout=5.0, interval=0.02):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(interval)
+    raise AssertionError("condition not met within timeout")
+
+
+@pytest.fixture(autouse=True)
+def _reset_registry():
+    """H2: the registry is module-level global state (like
+    claude_session._session_locks) — clear it around every test so fake
+    processes from one test never leak into the next."""
+    claude_runner.reset_registry_for_tests()
+    yield
+    claude_runner.reset_registry_for_tests()
+
+
+@pytest.fixture
+def make_proc(monkeypatch):
+    """Factory for a ClaudeProcess wired to tests/fake_claude.py, with
+    automatic cleanup of every process it creates."""
+    created: list[claude_runner.ClaudeProcess] = []
+    monkeypatch.setattr(claude_runner, "CLAUDE_BIN", str(FAKE_CLAUDE))
+
+    def _factory(scenario: str, channel_id: str = "test-channel",
+                 session_id_env: str = "fake-session-1", model: str = "sonnet"):
+        monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", scenario)
+        monkeypatch.setenv("FAKE_CLAUDE_SESSION_ID", session_id_env)
+        proc = claude_runner.ClaudeProcess(channel_id, model=model)
+        created.append(proc)
+        return proc
+
+    yield _factory
+
+    for p in created:
+        try:
+            p.stop()
+        except Exception:
+            pass
+
+
+# ---------------------------------------------------------------------------
+# T1 — normal turn shape + rate limit + timeout (Etapa 1 contract)
+# ---------------------------------------------------------------------------
+
+
+def test_run_turn_matches_run_claude_dict_shape(make_proc):
+    proc = make_proc("normal")
+    result = proc.run_turn("hello")
+    expected_keys = {
+        "result", "session_id", "usage", "total_cost_usd", "cost_usd",
+        "duration_ms", "num_turns", "intermediate_count", "subtype", "is_error",
+    }
+    assert set(result.keys()) == expected_keys
+    assert result["is_error"] is False
+    assert result["session_id"] == "fake-session-1"
+
+
+def test_run_turn_wraps_external_content(make_proc):
+    """T1 (-k external_content selects this test and the steer one below)."""
+    proc = make_proc("normal")
+    result = proc.run_turn("plain text from Marius")
+    assert "[EXTERNAL CONTENT]" in result["result"]
+    assert "[END EXTERNAL CONTENT]" in result["result"]
+    assert "plain text from Marius" in result["result"]
+
+
+def test_timeout_raises_exact_message(make_proc):
+    proc = make_proc("timeout")
+    with pytest.raises(TimeoutError, match=r"Claude CLI timed out after 1s"):
+        proc.run_turn("hi", timeout=1)
+
+
+def test_rate_limit_raises_runtime_error_router_can_detect(make_proc):
+    """T2/C4/T6: the exact failure mode that must keep `_local_fallback_reply`
+    alive — router.py/scheduler.py both gate on `is_rate_limit_error`."""
+    proc = make_proc("rate_limit")
+    with pytest.raises(RuntimeError) as excinfo:
+        proc.run_turn("hi")
+    assert claude_session.is_rate_limit_error(str(excinfo.value))
+
+
+def test_non_rate_limit_error_is_returned_not_raised(make_proc):
+    """C4: the parser/wrapper must stay non-throwing for any `is_error` that
+    isn't a rate limit (PlanningSession-style callers retry on subtype)."""
+    proc = make_proc("generic_error")
+    result = proc.run_turn("hi")  # must NOT raise
+    assert result["is_error"] is True
+    assert result["subtype"] == "error_max_turns"
+
+
+# ---------------------------------------------------------------------------
+# Steering — mid-turn message lands on the same stdin
+# ---------------------------------------------------------------------------
+
+
+def test_steering_mid_turn_reflects_second_message(make_proc):
+    proc = make_proc("steer")
+    texts: list[str] = []
+    holder: dict = {}
+
+    def _go():
+        holder["result"] = proc.run_turn("first message", on_text=texts.append)
+
+    t = threading.Thread(target=_go)
+    t.start()
+    # Wait for the fake's first assistant block — proves it's now blocked
+    # on its own stdin.readline(), genuinely waiting for the steer.
+    _wait_until(lambda: len(texts) >= 1)
+    assert proc.inflight is True
+
+    outcome = proc.steer("second message")
+    t.join(timeout=5)
+
+    assert outcome.status == SteerStatus.STEERED
+    assert outcome.turn is None
+    result = holder["result"]
+    assert "steered:" in result["result"]
+    assert result["num_turns"] == 2
+    assert proc.inflight is False
+
+
+def test_steer_wraps_external_content(make_proc):
+    """T1 (-k external_content)."""
+    proc = make_proc("steer")
+    texts: list[str] = []
+    holder: dict = {}
+
+    def _go():
+        holder["result"] = proc.run_turn("first", on_text=texts.append)
+
+    t = threading.Thread(target=_go)
+    t.start()
+    _wait_until(lambda: len(texts) >= 1)
+    proc.steer("stai, schimbare de plan")
+    t.join(timeout=5)
+
+    assert "[EXTERNAL CONTENT]" in holder["result"]["result"]
+    assert "stai, schimbare de plan" in holder["result"]["result"]
+
+
+def test_steer_pending_texts_available_on_turn_failure(make_proc):
+    """C3: if the turn a steer landed in fails, the steered text must be
+    exposed for re-dispatch, never silently dropped."""
+    proc = make_proc("timeout")  # sleeps forever after init — never sends a result
+
+    def _go():
+        try:
+            proc.run_turn("first", timeout=1)
+        except TimeoutError:
+            pass
+
+    t = threading.Thread(target=_go)
+    t.start()
+    # session_id arrives (via the init event) right after spawn, well
+    # before the fake's 600s sleep would ever resolve — a reliable
+    # "the turn is genuinely in flight" signal for this scenario.
+    _wait_until(lambda: proc.session_id == "fake-session-1")
+    assert proc.inflight is True
+
+    outcome = proc.steer("please don't get lost")
+    assert outcome.status == SteerStatus.STEERED
+    t.join(timeout=5)  # watchdog kills the process after 1s -> TimeoutError
+
+    pending = proc.pop_pending_steers()
+    assert pending == ["please don't get lost"]
+    # popped once -> empty on a second call
+    assert proc.pop_pending_steers() == []
+
+
+# ---------------------------------------------------------------------------
+# C1 — RAN_AS_TURN: steer() finds no in-flight turn
+# ---------------------------------------------------------------------------
+
+
+def test_steer_without_inflight_turn_becomes_ran_as_turn(make_proc):
+    proc = make_proc("normal")
+    first = proc.run_turn("warm up")
+    assert first["is_error"] is False
+    assert proc.inflight is False
+
+    outcome = proc.steer("out of band message")
+
+    assert outcome.status == SteerStatus.RAN_AS_TURN
+    assert outcome.turn is not None
+    assert "out of band message" in outcome.turn["result"]
+    # D4: the caller must treat `turn` as the response and never re-dispatch
+    # by calling run_turn() again for the same text — nothing to assert
+    # mechanically here beyond the API shape making that the only sane path.
+
+
+# ---------------------------------------------------------------------------
+# T7 — BrokenPipeError on steer falls back to a normal turn
+# ---------------------------------------------------------------------------
+
+
+def test_steer_broken_pipe_falls_back_to_turn(make_proc, monkeypatch):
+    proc = make_proc("normal")
+    proc.run_turn("warm up")  # spawns the process
+    proc.proc.kill()
+    proc.proc.wait()
+
+    monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "normal")
+    outcome = proc.steer("please deliver me")
+
+    assert outcome.status == SteerStatus.PROCESS_DEAD
+    assert outcome.turn is not None
+    assert "please deliver me" in outcome.turn["result"]
+
+
+# ---------------------------------------------------------------------------
+# M2 — timeout kills the process; next turn respawns with --resume
+# ---------------------------------------------------------------------------
+
+
+def test_respawn_with_resume_after_timeout_kill(make_proc, monkeypatch):
+    proc = make_proc("timeout")
+    with pytest.raises(TimeoutError):
+        proc.run_turn("hi", timeout=1)
+
+    # session_id came from the `system`/`init` event, captured before the
+    # turn ever got its (never-arriving) result line.
+    assert proc.session_id == "fake-session-1"
+    assert not proc.alive()
+    dead_pid = proc.proc.pid
+
+    cmd = proc._build_cmd()
+    assert "--resume" in cmd
+    assert cmd[cmd.index("--resume") + 1] == "fake-session-1"
+
+    monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "normal")
+    result = proc.run_turn("again", timeout=5)
+
+    assert result["is_error"] is False
+    assert proc.proc.pid != dead_pid  # actually respawned, not reused
+
+
+# ---------------------------------------------------------------------------
+# T3 — stderr pipe must never deadlock a turn
+# ---------------------------------------------------------------------------
+
+
+def test_big_stderr_does_not_deadlock(make_proc):
+    proc = make_proc("big_stderr")
+    result = proc.run_turn("hello", timeout=15)
+    assert result["is_error"] is False
+    assert len(proc._stderr_buf) > 0
+    assert len(proc._stderr_buf) <= 50  # deque(maxlen=50)
+
+
+# ---------------------------------------------------------------------------
+# RunnerRegistry — max_live degradation, reaper resilience
+# ---------------------------------------------------------------------------
+
+
+def test_max_live_degrades_second_channel_to_one_shot():
+    registry = claude_runner.RunnerRegistry(max_live=1, idle_minutes=20)
+    proc_a = registry.get("channel-a")
+    assert proc_a is not None
+    proc_b = registry.get("channel-b")
+    assert proc_b is None  # T10: degrade, don't evict (X6) and don't block
+    # the same channel is never punished by its own occupied slot:
+    assert registry.get("channel-a") is proc_a
+    registry.stop_all()
+
+
+def test_reaper_never_stops_inflight_process():
+    registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=1)
+    proc = registry.get("channel-a")
+    proc.inflight = True
+    proc.last_active = time.monotonic() - 3600
+    registry._reap_once()
+    assert registry.live_count() == 1
+    registry.stop_all()
+
+
+def test_reaper_stops_idle_process():
+    registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=1)
+    proc = registry.get("channel-a")
+    proc.inflight = False
+    proc.last_active = time.monotonic() - 3600
+    registry._reap_once()
+    assert registry.live_count() == 0
+
+
+def test_reaper_survives_stop_exception(monkeypatch):
+    """T11: a dead reaper is a silent RAM leak — one process's stop()
+    raising must not kill the reap loop or the process."""
+    registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=1)
+    proc = registry.get("channel-a")
+    proc.inflight = False
+    proc.last_active = time.monotonic() - 3600
+
+    def _boom():
+        raise RuntimeError("stop() exploded")
+
+    monkeypatch.setattr(proc, "stop", _boom)
+    registry._reap_once()  # must not raise
+    assert registry.live_count() == 0  # bookkeeping removed regardless
+
+
+def test_stop_all_clears_registry(make_proc):
+    proc = make_proc("normal")
+    registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=20)
+    registry._procs["test-channel"] = proc
+    proc.run_turn("hello")  # actually spawns
+    assert proc.alive()
+    registry.stop_all()
+    assert registry.live_count() == 0
+    assert not proc.alive()
+
+
+def test_get_registry_singleton_and_reset():
+    r1 = claude_runner.get_registry(max_live=3, idle_minutes=5)
+    r2 = claude_runner.get_registry(max_live=99)  # ignored once created
+    assert r1 is r2
+    assert r1.max_live == 3
+
+    claude_runner.reset_registry_for_tests()
+
+    r3 = claude_runner.get_registry(max_live=7)
+    assert r3 is not r1
+    assert r3.max_live == 7
+
+
+def test_max_live_non_positive_logs_and_always_degrades(caplog):
+    with caplog.at_level("WARNING"):
+        registry = claude_runner.RunnerRegistry(max_live=0, idle_minutes=20)
+    assert any("max_live" in rec.message for rec in caplog.records)
+    assert registry.get("any-channel") is None

tests/test_claude_session_mutex.py

@@ -25,12 +25,14 @@ from unittest.mock import patch
 
 import pytest
 
-from src import claude_session
+from src import claude_runner, claude_session
+from src.claude_runner import SteerResult, SteerStatus
 from src.claude_session import (
     _get_session_lock,
     _session_locks,
     send_message,
 )
+from src.sentinels import is_steered
 
 
 # ---------------------------------------------------------------------------
@@ -40,10 +42,19 @@ from src.claude_session import (
 
 @pytest.fixture(autouse=True)
 def _clear_session_locks():
-    """Each test starts with a fresh lock map so we don't share state."""
+    """Each test starts with a fresh lock map so we don't share state.
+
+    H2: also resets the steering registry singleton and adapter-ownership
+    map — both module-level globals — so a fake `ClaudeProcess` installed
+    by one test (steering-ON variants below) never leaks into the next.
+    """
     _session_locks.clear()
+    claude_session._channel_adapter.clear()
+    claude_runner.reset_registry_for_tests()
     yield
     _session_locks.clear()
+    claude_session._channel_adapter.clear()
+    claude_runner.reset_registry_for_tests()
 
 
 @pytest.fixture
@@ -68,7 +79,7 @@ def _slow_run_claude(sleep_seconds: float, in_critical: threading.Event,
     """
     state = {"active": 0, "lock": threading.Lock()}
 
-    def fake(cmd, timeout, on_text=None, cwd=None):
+    def fake(cmd, timeout, on_text=None, cwd=None, channel_id=None):
         with state["lock"]:
             state["active"] += 1
             if state["active"] > 1:
@@ -268,7 +279,7 @@ class TestAcquisitionBehavior:
 
         call_count = {"n": 0}
 
-        def flaky(cmd, timeout, on_text=None, cwd=None):
+        def flaky(cmd, timeout, on_text=None, cwd=None, channel_id=None):
             call_count["n"] += 1
             if call_count["n"] == 1:
                 raise RuntimeError("simulated subprocess crash")
@@ -305,3 +316,98 @@ class TestAcquisitionBehavior:
             )
             t.join(timeout=1.0)
             assert result_box == ["Hello from Claude!"]
+
+
+# ---------------------------------------------------------------------------
+# H2 — twin of TestAcquisitionBehavior with steering ON: a contending
+# caller must STEER, not block-then-run. Pins that the flag actually
+# changes this contract (the flag-off tests above must stay unchanged).
+# ---------------------------------------------------------------------------
+
+
+class _FakeSteeringProc:
+    """Minimal `ClaudeProcess` stand-in — just enough of the contract
+    `_dispatch_steering` relies on (`inflight`, `run_turn`, `steer`,
+    `pop_pending_steers`) to prove the dispatch decision, without a real
+    subprocess (that's covered by tests/test_claude_runner.py)."""
+
+    def __init__(self):
+        self.inflight = False
+        self.session_id = "fake-sid"
+        self.steer_calls: list[str] = []
+        self._release = threading.Event()
+
+    def run_turn(self, text, on_text=None, timeout=300):
+        self._release.clear()
+        self.inflight = True
+        released = self._release.wait(timeout=5.0)
+        self.inflight = False
+        assert released, "test never released the fake turn"
+        return {
+            "result": f"done:{text}", "session_id": self.session_id, "usage": {},
+            "total_cost_usd": 0, "cost_usd": 0, "duration_ms": 1, "num_turns": 1,
+            "intermediate_count": 0, "subtype": "success", "is_error": False,
+        }
+
+    def steer(self, text):
+        self.steer_calls.append(text)
+        return SteerResult(SteerStatus.STEERED)
+
+    def pop_pending_steers(self):
+        return []
+
+    def release(self):
+        self._release.set()
+
+
+class _FakeSteeringRegistry:
+    def __init__(self, proc, channel_id):
+        self._procs = {channel_id: proc}
+
+    def get(self, channel_id, model=None, session_id=None, cwd=None):
+        return self._procs.get(channel_id)
+
+    def stop(self, channel_id):
+        return False
+
+
+class TestAcquisitionBehaviorSteering:
+    """Twin of `TestAcquisitionBehavior.test_contested_acquire_blocks_then_proceeds`
+    with `steering.enabled` on: contention must STEER instead of blocking."""
+
+    def test_contested_acquire_steers_when_steering_enabled(
+        self, temp_sessions, monkeypatch
+    ):
+        proc = _FakeSteeringProc()
+        registry = _FakeSteeringRegistry(proc, "ch-contend-steer")
+        monkeypatch.setattr(claude_runner, "_registry", registry)
+        monkeypatch.setattr(claude_runner, "get_registry", lambda **kw: registry)
+        monkeypatch.setattr(claude_session, "_steering_config", lambda channel_id=None: (True, 2, 20))
+
+        outcome: dict[str, str] = {}
+
+        def run(label: str):
+            outcome[label] = send_message(
+                "ch-contend-steer", label, adapter_name="discord",
+            )
+
+        t1 = threading.Thread(target=run, args=("first",))
+        t1.start()
+
+        deadline = time.monotonic() + 2.0
+        while time.monotonic() < deadline and not proc.inflight:
+            time.sleep(0.01)
+        assert proc.inflight, "first call never entered the turn"
+
+        t2 = threading.Thread(target=run, args=("second",))
+        t2.start()
+        t2.join(timeout=5.0)
+
+        assert is_steered(outcome["second"]), (
+            "With steering on, a contended caller must STEER, not block."
+        )
+        assert proc.steer_calls == ["second"]
+
+        proc.release()
+        t1.join(timeout=5.0)
+        assert outcome["first"] == "done:first"

tests/test_cli.py

@@ -87,6 +87,54 @@ class TestStatus:
         assert "WA Bridge: ONLINE" in out
 
 
+# ---------------------------------------------------------------------------
+# eco status — steering line (T4/A3)
+# ---------------------------------------------------------------------------
+
+
+class TestSteeringStatusLine:
+    def _mock_service(self, active="inactive"):
+        return {"ActiveState": active, "MainPID": "0", "ActiveEnterTimestamp": ""}
+
+    def test_steering_off_no_procs(self, iso, capsys):
+        """pgrep's own no-match case: exit code 1, stdout '0' — a
+        legitimate zero, not a counting failure."""
+        mock_cfg = MagicMock()
+        mock_cfg.get.return_value = False
+        mock_pgrep = MagicMock(returncode=1, stdout="0\n")
+        with patch("cli._get_service_status", return_value=self._mock_service()), \
+             patch("src.config.Config", return_value=mock_cfg), \
+             patch("subprocess.run", return_value=mock_pgrep):
+            cli.cmd_status(_args())
+        out = capsys.readouterr().out
+        assert "steering: off · 0 procese vii" in out
+
+    def test_steering_counts_live_procs(self, iso, capsys):
+        """A real match: pgrep -fc finds 2 live steering processes."""
+        mock_cfg = MagicMock()
+        mock_cfg.get.return_value = True
+        mock_pgrep = MagicMock(returncode=0, stdout="2\n")
+        with patch("cli._get_service_status", return_value=self._mock_service()), \
+             patch("src.config.Config", return_value=mock_cfg), \
+             patch("subprocess.run", return_value=mock_pgrep):
+            cli.cmd_status(_args())
+        out = capsys.readouterr().out
+        assert "steering: on · 2 procese vii" in out
+
+    def test_survives_pgrep_failure(self, iso, capsys):
+        """D1/D3 guard: a failed process-table scan must not break the
+        rest of `eco status`, and must not silently read as a wrong zero."""
+        mock_cfg = MagicMock()
+        mock_cfg.get.return_value = False
+        with patch("cli._get_service_status", return_value=self._mock_service()), \
+             patch("src.config.Config", return_value=mock_cfg), \
+             patch("subprocess.run", side_effect=OSError("pgrep not found")):
+            cli.cmd_status(_args())
+        out = capsys.readouterr().out
+        assert "steering: off · necunoscut procese vii" in out
+        assert "Sessions:" in out
+
+
 # ---------------------------------------------------------------------------
 # cmd_doctor
 # ---------------------------------------------------------------------------
@@ -194,6 +242,22 @@ class TestDoctor:
         out, code = self._run_doctor(iso, capsys, setup_full=True)
         assert "Claude CLI functional" in out
 
+    def test_steering_off_by_default(self, iso, capsys):
+        iso["config_file"].write_text('{"bot":{}}')
+        mock_cfg = MagicMock()
+        mock_cfg.get.return_value = False
+        with patch("src.config.Config", return_value=mock_cfg):
+            out, code = self._run_doctor(iso, capsys, setup_full=True)
+        assert "[PASS] Steering (optional, currently off)" in out
+
+    def test_steering_checks_stream_json_when_enabled(self, iso, capsys):
+        iso["config_file"].write_text('{"bot":{}}')
+        mock_cfg = MagicMock()
+        mock_cfg.get.return_value = True
+        with patch("src.config.Config", return_value=mock_cfg):
+            out, code = self._run_doctor(iso, capsys, setup_full=True)
+        assert "Claude CLI supports --input-format stream-json" in out
+
 
 # ---------------------------------------------------------------------------
 # cmd_restart

tests/test_local_fallback.py

@@ -1,15 +1,18 @@
 """Tests for the local LLM fallback: history, tools, net status, web search."""
 
 import time
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from src import fallback_history, net_status, web_search
+from src import claude_runner, claude_session, fallback_history, net_status, router, web_search
 from src import local_fallback_tools as lft
 from src.router import (_forced_tool, _is_creative_request, _is_text_task,
                         _parse_tool_args, _run_fallback_tools)
 
+FAKE_CLAUDE = Path(__file__).parent / "fake_claude.py"
+
 
 @pytest.fixture(autouse=True)
 def clean_history():
@@ -623,3 +626,217 @@ class TestFallbackNeverDropsTurn:
         assert "răspuns direct" in reply
         conv.assert_called_once()
 
+
+# ---------------------------------------------------------------------------
+# T6 — persistent-process rate limit must still reach the fallback
+# ---------------------------------------------------------------------------
+#
+# The one-shot `claude -p` dies with a nonzero exit code at the limit; a
+# persistent steering process (src/claude_runner.py) reports the same limit
+# as `result.is_error` in the stream instead — it never exits. `ClaudeProcess`
+# converts that back into the exact `RuntimeError` the one-shot path raises
+# (covered in tests/test_claude_runner.py). What isn't covered anywhere else
+# is the far end of the wire: that error genuinely reaching router.py's
+# handler and firing `_local_fallback_reply` for real. Untested, this is how
+# the single most-worked-on subsystem in the repo goes dead in silence
+# (tasks/steering-plan.md, blockers C3/C4, task T6 — "the 2am Friday test").
+
+
+def _local_fallback_cfg():
+    """`_get_config()` double: only `local_fallback` is stubbed — every
+    other key falls through to its caller-supplied default, same as the
+    real `Config().get(key, default)`. A blanket `return_value` (as used
+    for the router-only tests above) would also hijack route_message's own
+    `_get_config().get("bot.default_model", "sonnet")` lookup and feed a
+    dict into `--model`."""
+    cfg = MagicMock()
+    cfg.get.side_effect = lambda key, default=None: (
+        {"enabled": True, "url": "http://x"} if key == "local_fallback" else default
+    )
+    return cfg
+
+
+@pytest.fixture(autouse=True)
+def _reset_steering_registry():
+    """H2: the steering registry is module-level global state — never let
+    a live (fake) process from one test leak into the next."""
+    claude_runner.reset_registry_for_tests()
+    yield
+    claude_runner.reset_registry_for_tests()
+
+
+@pytest.fixture
+def temp_sessions(tmp_path, monkeypatch):
+    """Isolated sessions/active.json so these tests never touch the real one."""
+    sessions_dir = tmp_path / "sessions"
+    sessions_dir.mkdir()
+    sf = sessions_dir / "active.json"
+    sf.write_text("{}")
+    monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
+    monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
+    return sf
+
+
+@pytest.fixture
+def steering_on(monkeypatch, temp_sessions):
+    """Steering enabled, wired to the real tests/fake_claude.py subprocess —
+    a genuine persistent process, not a mock, so the rate-limit path fires
+    the way it actually does in production."""
+    monkeypatch.setattr(claude_runner, "CLAUDE_BIN", str(FAKE_CLAUDE))
+    monkeypatch.setattr(claude_session, "_steering_config", lambda *a, **kw: (True, 2, 20))
+
+
+class TestPersistentRateLimitReachesFallback:
+    """T6 (`pytest -k persistent`): a rate limit arriving as `result.is_error`
+    from a PERSISTENT process must still invoke the local model — asserting
+    the RESCUE happened, not merely that some exception was raised. A test
+    that only checks "an exception was raised" is exactly the bug this task
+    exists to catch: that's also true the day the fallback silently stops
+    firing."""
+
+    def test_persistent_rate_limit_invokes_real_fallback(self, steering_on, monkeypatch):
+        monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "rate_limit")
+
+        with patch("src.router._get_config", return_value=_local_fallback_cfg()), \
+             patch("src.router.set_channel_context"), \
+             patch("src.router._call_local_llm", return_value={"content": "raspuns local"}), \
+             patch("src.router._local_fallback_reply",
+                   wraps=router._local_fallback_reply) as fallback_spy:
+            result, is_cmd = router.route_message("ch-t6-persist", "user-1", "salut")
+
+        fallback_spy.assert_called_once_with("salut", channel_id="ch-t6-persist")
+        assert "raspuns local" in result
+        assert is_cmd is False
+
+
+class _RateLimitAfterSteerProc:
+    """Minimal ClaudeProcess double: the turn raises the rate-limit
+    RuntimeError while leaving one already-steered text behind in
+    `_pending_steers` — the shape of a real ClaudeProcess whose turn a
+    `steer()` landed in, then died (C3). `ClaudeProcess`'s own
+    thread-timing for how a steer lands mid-turn is exercised for real in
+    tests/test_claude_runner.py; this double exists only to drive
+    router.py's redelivery path deterministically."""
+
+    def __init__(self, channel_id, model=None, session_id=None, cwd=None):
+        self.channel_id = channel_id
+        self.session_id = session_id or "fake-sid"
+        self.inflight = False
+        self._pending_steers = ["steered while you were away"]
+
+    def run_turn(self, text, on_text=None, timeout=300):
+        self.inflight = True
+        raise RuntimeError(
+            "Claude CLI error (exit 1): You've hit your session limit · resets 10am (UTC)"
+        )
+
+    def pop_pending_steers(self):
+        pending, self._pending_steers = self._pending_steers, []
+        return pending
+
+
+class _SingleProcRegistry:
+    """RunnerRegistry double that always hands back the one proc it was
+    built with."""
+
+    def __init__(self, proc):
+        self._procs = {proc.channel_id: proc}
+
+    def get(self, channel_id, model=None, session_id=None, cwd=None):
+        return self._procs.get(channel_id)
+
+    def stop(self, channel_id):
+        return self._procs.pop(channel_id, None) is not None
+
+
+def _echo_llm(url, messages, tools=None, temperature=0.0):
+    """`_call_local_llm` double whose reply names the user text it saw, so
+    the original turn's reply and the steered turn's reply stay
+    distinguishable however many passes `_local_fallback_reply` makes."""
+    return {"content": f"echo:{messages[-1]['content']}"}
+
+
+class TestSteeredTextDeliveredOnRateLimit:
+    """C3: the plan is explicit that T6 must test DELIVERY, not just
+    detection. A message steered into a turn that then dies on a rate
+    limit must still produce its own answer to the user — its request
+    thread already returned `__STEERED__` and is gone, so `on_text` is the
+    only channel left."""
+
+    def test_steered_reply_delivered_via_on_text(self, monkeypatch, temp_sessions):
+        proc = _RateLimitAfterSteerProc("ch-t6-c3")
+        registry = _SingleProcRegistry(proc)
+        # Both names: `_dispatch_steering` calls `get_registry()`, but
+        # `claude_session.pop_pending_steers` (C3's redelivery hook) peeks
+        # the module-level `_registry` directly rather than calling
+        # `get_registry()` again (that would spin one up as a side effect
+        # for a channel that never used steering).
+        monkeypatch.setattr(claude_runner, "get_registry", lambda **kw: registry)
+        monkeypatch.setattr(claude_runner, "_registry", registry)
+        monkeypatch.setattr(claude_session, "_steering_config", lambda *a, **kw: (True, 2, 20))
+
+        streamed = []
+        with patch("src.router._get_config", return_value=_local_fallback_cfg()), \
+             patch("src.router.set_channel_context"), \
+             patch("src.router._call_local_llm", side_effect=_echo_llm):
+            result, is_cmd = router.route_message(
+                "ch-t6-c3", "user-1", "mesaj original", on_text=streamed.append,
+            )
+
+        # The original message's own answer still comes back as the
+        # function's return value.
+        assert "echo:mesaj original" in result
+        assert is_cmd is False
+        # The steered message never had a request thread of its own left to
+        # return to — its answer must have gone out through on_text.
+        assert len(streamed) == 1
+        assert "echo:steered while you were away" in streamed[0]
+
+
+class TestNonRateLimitErrorStaysQuiet:
+    """C4 regression guard: verified in tasks/steering-plan.md that
+    `PlanningSession` retries on `error_max_turns`, which depends on the
+    runner RETURNING (never raising) a non-rate-limit `is_error`. If a
+    future change 'simplifies' that into a raise, this test catches it by
+    failing on the wrong side: the fallback would fire when it must not."""
+
+    def test_generic_is_error_does_not_raise_or_call_fallback(self, steering_on, monkeypatch):
+        monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "generic_error")
+
+        with patch("src.router._local_fallback_reply") as fallback:
+            result, is_cmd = router.route_message("ch-t6-c4", "user-1", "salut")
+
+        fallback.assert_not_called()
+        assert "hit max turns" in result
+        assert is_cmd is False
+
+
+class TestSteeringOffRollback:
+    """The flag is the rollback mechanism (steering-plan.md Etapa 2) — it
+    has to actually roll the rate-limit -> fallback path back to today's
+    behavior, not merely skip steering-specific code paths."""
+
+    def test_steering_off_still_reaches_fallback_via_one_shot_path(
+        self, monkeypatch, temp_sessions,
+    ):
+        monkeypatch.setattr(claude_session, "_steering_config", lambda *a, **kw: (False, 0, 20))
+        monkeypatch.setattr(
+            claude_session, "_run_claude",
+            lambda *a, **kw: (_ for _ in ()).throw(RuntimeError(
+                "Claude CLI error (exit 1): You've hit your session limit · resets 10am (UTC)"
+            )),
+        )
+
+        with patch("src.router._get_config", return_value=_local_fallback_cfg()), \
+             patch("src.router.set_channel_context"), \
+             patch("src.router._call_local_llm", return_value={"content": "raspuns local"}), \
+             patch("src.router._local_fallback_reply",
+                   wraps=router._local_fallback_reply) as fallback_spy:
+            result, is_cmd = router.route_message("ch-t6-off", "user-1", "salut")
+
+        fallback_spy.assert_called_once_with("salut", channel_id="ch-t6-off")
+        assert "raspuns local" in result
+        assert is_cmd is False
+        # Steering played no role at all — the registry was never even created.
+        assert claude_runner._registry is None
+

tests/test_router.py

@@ -193,7 +193,7 @@ class TestRegularMessage:
         response, is_cmd = route_message("ch-1", "user-1", "hello")
         assert response == "Hello from Claude!"
         assert is_cmd is False
-        mock_send.assert_called_once_with("ch-1", "hello", model="sonnet", on_text=None, voice_mode=False)
+        mock_send.assert_called_once_with("ch-1", "hello", model="sonnet", on_text=None, voice_mode=False, adapter_name=None)
 
     @patch("src.router.send_message")
     def test_model_override(self, mock_send):
@@ -201,7 +201,7 @@ class TestRegularMessage:
         response, is_cmd = route_message("ch-1", "user-1", "hello", model="opus")
         assert response == "Response"
         assert is_cmd is False
-        mock_send.assert_called_once_with("ch-1", "hello", model="opus", on_text=None, voice_mode=False)
+        mock_send.assert_called_once_with("ch-1", "hello", model="opus", on_text=None, voice_mode=False, adapter_name=None)
 
     @patch("src.router._get_channel_config")
     @patch("src.router._get_config")
@@ -274,7 +274,7 @@ class TestRegularMessage:
 
         cb = lambda t: None
         route_message("ch-1", "user-1", "hello", on_text=cb)
-        mock_send.assert_called_once_with("ch-1", "hello", model="sonnet", on_text=cb, voice_mode=False)
+        mock_send.assert_called_once_with("ch-1", "hello", model="sonnet", on_text=cb, voice_mode=False, adapter_name=None)
 
 
 # --- _get_channel_config ---
@@ -316,7 +316,7 @@ class TestModelResolution:
         mock_chan_cfg.return_value = {"id": "ch-1", "default_model": "haiku"}
 
         route_message("ch-1", "user-1", "hello")
-        mock_send.assert_called_once_with("ch-1", "hello", model="haiku", on_text=None, voice_mode=False)
+        mock_send.assert_called_once_with("ch-1", "hello", model="haiku", on_text=None, voice_mode=False, adapter_name=None)
 
     @patch("src.router._get_channel_config")
     @patch("src.router._get_config")
@@ -330,7 +330,7 @@ class TestModelResolution:
         mock_get_config.return_value = mock_cfg
 
         route_message("ch-1", "user-1", "hello")
-        mock_send.assert_called_once_with("ch-1", "hello", model="opus", on_text=None, voice_mode=False)
+        mock_send.assert_called_once_with("ch-1", "hello", model="opus", on_text=None, voice_mode=False, adapter_name=None)
 
     @patch("src.router._get_channel_config")
     @patch("src.router._get_config")
@@ -344,7 +344,7 @@ class TestModelResolution:
         mock_get_config.return_value = mock_cfg
 
         route_message("ch-1", "user-1", "hello")
-        mock_send.assert_called_once_with("ch-1", "hello", model="sonnet", on_text=None, voice_mode=False)
+        mock_send.assert_called_once_with("ch-1", "hello", model="sonnet", on_text=None, voice_mode=False, adapter_name=None)
 
     @patch("src.router.get_active_session")
     @patch("src.router.send_message")
@@ -354,7 +354,7 @@ class TestModelResolution:
         mock_get_session.return_value = {"model": "opus", "session_id": "abc"}
 
         route_message("ch-1", "user-1", "hello")
-        mock_send.assert_called_once_with("ch-1", "hello", model="opus", on_text=None, voice_mode=False)
+        mock_send.assert_called_once_with("ch-1", "hello", model="opus", on_text=None, voice_mode=False, adapter_name=None)
 
 
 # --- Voice/text unify regression guards ---

tests/test_sentinels.py

@@ -0,0 +1,101 @@
+"""src.sentinels is the single place that recognises router sentinels
+(__AUDIO__:<path> and __STEERED__) so adapters don't each hand-roll the
+prefix check. Offline, no subprocess/network involved.
+"""
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from src.sentinels import AUDIO_PREFIX, STEERED, audio_path, is_steered
+
+
+def test_audio_path_extracts_payload():
+    assert audio_path(f"{AUDIO_PREFIX}/tmp/echo-audio.wav") == "/tmp/echo-audio.wav"
+
+
+def test_audio_path_none_for_plain_text():
+    assert audio_path("salut, ce mai faci?") is None
+
+
+def test_audio_path_none_for_empty_and_none():
+    assert audio_path("") is None
+    assert audio_path(None) is None
+
+
+def test_audio_path_none_for_steered():
+    assert audio_path(STEERED) is None
+
+
+def test_is_steered_true_only_for_exact_sentinel():
+    assert is_steered(STEERED) is True
+    assert is_steered("__STEERED__ ") is False
+    assert is_steered("some text") is False
+
+
+def test_is_steered_false_for_empty_and_none():
+    assert is_steered("") is False
+    assert is_steered(None) is False
+
+
+def test_is_steered_false_for_audio_sentinel():
+    assert is_steered(f"{AUDIO_PREFIX}/tmp/x.wav") is False
+
+
+# --- Voice pipeline: a steered response must not be mirrored into the
+# text channel (blocker C2 in tasks/steering-plan.md). Reuses the fakes
+# from tests/test_pipeline_mirror.py.
+
+from unittest.mock import AsyncMock, MagicMock
+
+from src.voice.pipeline import VoiceSession
+
+
+def _make_text_channel(send_mock: AsyncMock) -> MagicMock:
+    tc = MagicMock(name="text_channel")
+    tc.send = send_mock
+    return tc
+
+
+def _make_session(*, reply_text: str, text_channel) -> VoiceSession:
+    bot = MagicMock(name="bot")
+    bot.get_channel = MagicMock(return_value=text_channel)
+    bot.get_user = MagicMock(return_value=None)
+    ttsq = MagicMock(name="ttsq")
+    ttsq.push_text = MagicMock()
+    ttsq.clear = MagicMock()
+    route_mock = MagicMock(name="route_message", return_value=(reply_text, False))
+    return VoiceSession(
+        text_channel_id=1001,
+        voice_channel_id=2002,
+        guild_id=42,
+        voice_client=MagicMock(name="voice_client"),
+        bot=bot,
+        ttsq=ttsq,
+        whitelist=set(),
+        record_enabled=False,
+        mirror_enabled=True,
+        transcripts_jsonl_path=None,
+        loop=asyncio.get_event_loop_policy().new_event_loop(),
+        router_route_message=route_mock,
+    )
+
+
+def _reply_chunks(send_mock: AsyncMock) -> list[str]:
+    return [
+        call.args[0]
+        for call in send_mock.call_args_list
+        if not call.args[0].startswith("\U0001f3a4")
+    ]
+
+
+@pytest.mark.asyncio
+async def test_voice_path_does_not_mirror_steered_sentinel():
+    send_mock = AsyncMock(name="text_send")
+    text_channel = _make_text_channel(send_mock)
+    session = _make_session(reply_text=STEERED, text_channel=text_channel)
+
+    await session.on_segment_done(speaker_id=123, text="salut", no_speech_prob=0.1)
+
+    assert _reply_chunks(send_mock) == []

tests/test_steering_dispatch.py

@@ -0,0 +1,433 @@
+"""Tests for the steering dispatcher (H1/C2/C3/T9/T12) in
+``src/claude_session.py`` and its ``__STEERED__`` handling in
+``src/router.py``.
+
+`ClaudeProcess`/`RunnerRegistry` internals (spawn, stdin/stdout framing,
+`steer()`'s own race handling) are already covered by
+``tests/test_claude_runner.py`` against the real ``tests/fake_claude.py``
+subprocess. This file is about the DISPATCHER's decisions — which only need
+a process double that behaves the right way at the right time, synchronised
+via ``threading.Event`` rather than real subprocess timing (the plan
+explicitly warns sleep-based subprocess races are the flaky way to test
+this).
+"""
+
+import threading
+import time
+from unittest.mock import patch
+
+import pytest
+
+from src import claude_runner, claude_session, router
+from src.claude_runner import SteerResult, SteerStatus
+from src.claude_session import send_message
+from src.sentinels import is_steered
+
+
+def _wait_until(predicate, timeout=5.0, interval=0.01):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(interval)
+    raise AssertionError("condition not met within timeout")
+
+
+# ---------------------------------------------------------------------------
+# Scriptable ClaudeProcess/RunnerRegistry doubles
+# ---------------------------------------------------------------------------
+
+
+class FakeProc:
+    """Stand-in for `ClaudeProcess`. `run_turn` blocks on a per-call Event
+    so a test can deterministically control exactly when a turn "finishes"."""
+
+    def __init__(self, channel_id="ch", session_id="fake-sid"):
+        self.channel_id = channel_id
+        self.session_id = session_id
+        self.inflight = False
+        self._pending_steers: list[str] = []
+        self.steer_calls: list[str] = []
+        self.run_turn_calls: list[str] = []
+        self.stopped = False
+        self.turn_error: Exception | None = None
+        self._release = threading.Event()
+
+    def run_turn(self, text, on_text=None, timeout=300):
+        self._release.clear()
+        self.inflight = True
+        self.run_turn_calls.append(text)
+        released = self._release.wait(timeout=5.0)
+        self.inflight = False
+        assert released, "test never released the fake turn"
+        if self.turn_error is not None:
+            err, self.turn_error = self.turn_error, None
+            raise err
+        return {
+            "result": f"done:{text}", "session_id": self.session_id, "usage": {},
+            "total_cost_usd": 0, "cost_usd": 0, "duration_ms": 1, "num_turns": 1,
+            "intermediate_count": 0, "subtype": "success", "is_error": False,
+        }
+
+    def steer(self, text):
+        self.steer_calls.append(text)
+        assert self.inflight, "dispatcher should only steer an inflight turn"
+        self._pending_steers.append(text)
+        return SteerResult(SteerStatus.STEERED)
+
+    def pop_pending_steers(self):
+        pending, self._pending_steers = self._pending_steers, []
+        return pending
+
+    def release(self):
+        self._release.set()
+
+
+class FakeRegistry:
+    """Stand-in for `RunnerRegistry` — one proc per channel, created
+    on demand, degrading to None past `capacity` (T10)."""
+
+    def __init__(self, capacity=2):
+        self.capacity = capacity
+        self._procs: dict[str, FakeProc] = {}
+
+    def get(self, channel_id, model=None, session_id=None, cwd=None):
+        proc = self._procs.get(channel_id)
+        if proc is not None:
+            return proc
+        if len(self._procs) >= self.capacity:
+            return None
+        proc = FakeProc(channel_id, session_id=session_id or "fake-sid")
+        self._procs[channel_id] = proc
+        return proc
+
+    def stop(self, channel_id):
+        proc = self._procs.pop(channel_id, None)
+        if proc is None:
+            return False
+        proc.stopped = True
+        return True
+
+    def live_count(self):
+        return len(self._procs)
+
+
+@pytest.fixture(autouse=True)
+def _clean_state():
+    """Dispatcher state (`_session_locks`, `_channel_adapter`) and the
+    steering registry singleton are module-level globals — clear them
+    around every test so nothing leaks between tests."""
+    claude_session._session_locks.clear()
+    claude_session._channel_adapter.clear()
+    claude_runner.reset_registry_for_tests()
+    yield
+    claude_session._session_locks.clear()
+    claude_session._channel_adapter.clear()
+    claude_runner.reset_registry_for_tests()
+
+
+@pytest.fixture
+def temp_sessions(tmp_path, monkeypatch):
+    """Isolated active.json per test — keeps real session state untouched."""
+    sessions_dir = tmp_path / "sessions"
+    sessions_dir.mkdir()
+    sf = sessions_dir / "active.json"
+    sf.write_text("{}")
+    monkeypatch.setattr(claude_session, "SESSIONS_DIR", sessions_dir)
+    monkeypatch.setattr(claude_session, "_SESSIONS_FILE", sf)
+    return sf
+
+
+@pytest.fixture
+def fake_registry(monkeypatch, temp_sessions):
+    """Steering ON, backed by a FakeRegistry instead of real subprocesses."""
+    registry = FakeRegistry()
+    monkeypatch.setattr(claude_runner, "_registry", registry)
+    monkeypatch.setattr(claude_runner, "get_registry", lambda **kw: registry)
+    monkeypatch.setattr(claude_session, "_steering_config", lambda channel_id=None: (True, 2, 20))
+    return registry
+
+
+class _FakeConfig:
+    """Stand-in for `src.config.Config` used by `_steering_config`. Takes
+    the full raw dict (`steering`, `channels`, etc.) like the real thing."""
+
+    def __init__(self, data: dict):
+        self._data = data
+
+    def __call__(self, *a, **kw):
+        return self
+
+    def get(self, key, default=None):
+        return self._data.get(key, default)
+
+
+# ---------------------------------------------------------------------------
+# `_steering_config` — flag off, kill switch, clamps (X7/X8)
+# ---------------------------------------------------------------------------
+
+
+class TestSteeringConfig:
+    def test_disabled_by_default_config(self, monkeypatch):
+        monkeypatch.setattr("src.config.Config", _FakeConfig({"steering": {"enabled": False}}))
+        enabled, _, _ = claude_session._steering_config()
+        assert enabled is False
+
+    def test_kill_switch_wins(self, monkeypatch):
+        monkeypatch.setenv("ECHO_STEERING", "off")
+        monkeypatch.setattr(
+            "src.config.Config",
+            _FakeConfig({"steering": {"enabled": True, "max_live": 2, "idle_minutes": 20}}),
+        )
+        enabled, _, _ = claude_session._steering_config()
+        assert enabled is False
+
+    def test_max_live_zero_disables(self, monkeypatch):
+        monkeypatch.setattr(
+            "src.config.Config",
+            _FakeConfig({"steering": {"enabled": True, "max_live": 0, "idle_minutes": 20}}),
+        )
+        enabled, _, _ = claude_session._steering_config()
+        assert enabled is False
+
+    def test_idle_minutes_clamped_to_one(self, monkeypatch):
+        monkeypatch.setattr(
+            "src.config.Config",
+            _FakeConfig({"steering": {"enabled": True, "max_live": 2, "idle_minutes": 0}}),
+        )
+        _, _, idle_minutes = claude_session._steering_config()
+        assert idle_minutes == 1
+
+
+# ---------------------------------------------------------------------------
+# Etapa 6 (X16) — per-channel `channels.<alias>.steering` override
+# ---------------------------------------------------------------------------
+
+
+class TestPerChannelSteeringOverride:
+    def test_channel_on_overrides_global_off(self, monkeypatch):
+        monkeypatch.setattr("src.config.Config", _FakeConfig({
+            "steering": {"enabled": False, "max_live": 2, "idle_minutes": 20},
+            "channels": {"echo-core": {"id": "chan-1", "steering": True}},
+        }))
+        enabled, _, _ = claude_session._steering_config("chan-1")
+        assert enabled is True
+
+    def test_channel_off_overrides_global_on(self, monkeypatch):
+        monkeypatch.setattr("src.config.Config", _FakeConfig({
+            "steering": {"enabled": True, "max_live": 2, "idle_minutes": 20},
+            "channels": {"echo-core": {"id": "chan-1", "steering": False}},
+        }))
+        enabled, _, _ = claude_session._steering_config("chan-1")
+        assert enabled is False
+
+    def test_channel_unset_falls_through_to_global(self, monkeypatch):
+        monkeypatch.setattr("src.config.Config", _FakeConfig({
+            "steering": {"enabled": True, "max_live": 2, "idle_minutes": 20},
+            "channels": {"echo-core": {"id": "chan-1"}},  # no "steering" key
+        }))
+        enabled, _, _ = claude_session._steering_config("chan-1")
+        assert enabled is True
+
+        enabled_other, _, _ = claude_session._steering_config("chan-unrelated")
+        assert enabled_other is True  # not present at all -> global default too
+
+
+# ---------------------------------------------------------------------------
+# Flag off -> unchanged blocking path
+# ---------------------------------------------------------------------------
+
+
+def test_steering_off_uses_unchanged_blocking_path(monkeypatch, temp_sessions):
+    monkeypatch.setattr(claude_session, "_steering_config", lambda channel_id=None: (False, 0, 20))
+    calls = []
+
+    def fake_run_claude(cmd, timeout, on_text=None, cwd=None, channel_id=None):
+        calls.append(cmd)
+        return {
+            "result": "hi", "session_id": "sid-1", "usage": {}, "total_cost_usd": 0,
+            "cost_usd": 0, "duration_ms": 1, "num_turns": 1, "intermediate_count": 0,
+            "subtype": "success", "is_error": False,
+        }
+
+    with patch.object(claude_session, "_run_claude", side_effect=fake_run_claude):
+        result = send_message("ch-off", "hello")
+
+    assert result == "hi"
+    assert len(calls) == 1
+    # The dispatcher never touched the registry — disabled path is a no-op
+    # with respect to steering, not just "didn't steer".
+    assert claude_runner._registry is None
+
+
+# ---------------------------------------------------------------------------
+# H1 — contention steers, no separate `inflight` check/TOCTOU
+# ---------------------------------------------------------------------------
+
+
+def test_contention_steers(fake_registry):
+    outcome = {}
+
+    def first():
+        outcome["first"] = send_message("ch-1", "first", adapter_name="discord")
+
+    t1 = threading.Thread(target=first)
+    t1.start()
+    _wait_until(lambda: fake_registry._procs.get("ch-1") is not None
+                and fake_registry._procs["ch-1"].inflight)
+
+    second = send_message("ch-1", "second", adapter_name="discord")
+    assert is_steered(second)
+
+    proc = fake_registry._procs["ch-1"]
+    assert proc.steer_calls == ["second"]
+    proc.release()
+    t1.join(timeout=5)
+    assert outcome["first"] == "done:first"
+
+
+# ---------------------------------------------------------------------------
+# C2/M1 — adapter mismatch never steers, and never spawns a second writer
+# ---------------------------------------------------------------------------
+
+
+def test_adapter_mismatch_waits_instead_of_steering(fake_registry):
+    voice_outcome = {}
+
+    def voice_turn():
+        voice_outcome["r"] = send_message("ch-2", "voice text", adapter_name="discord-voice")
+
+    t1 = threading.Thread(target=voice_turn)
+    t1.start()
+    _wait_until(lambda: fake_registry._procs.get("ch-2") is not None
+                and fake_registry._procs["ch-2"].inflight)
+
+    text_outcome = {}
+
+    def text_turn():
+        text_outcome["r"] = send_message("ch-2", "text message", adapter_name="discord")
+
+    t2 = threading.Thread(target=text_turn)
+    t2.start()
+
+    proc = fake_registry._procs["ch-2"]
+    time.sleep(0.1)  # give t2 a chance to reach the mismatch branch
+    assert proc.steer_calls == [], "a mismatched adapter must never be steered (C2)"
+
+    proc.release()  # finish the voice turn -> releases the lock
+    t1.join(timeout=5)
+    assert voice_outcome["r"] == "done:voice text"
+
+    _wait_until(lambda: proc.inflight)  # t2 now got the lock and started its own turn
+    proc.release()
+    t2.join(timeout=5)
+
+    assert text_outcome["r"] == "done:text message"
+    assert not is_steered(text_outcome["r"])
+    assert proc.run_turn_calls == ["voice text", "text message"]
+    assert fake_registry.live_count() == 1, "must never spawn a second writer on the session (M1)"
+
+
+# ---------------------------------------------------------------------------
+# T10 — registry at capacity degrades to one-shot for a NEW channel
+# ---------------------------------------------------------------------------
+
+
+def test_capacity_degrades_new_channel_to_one_shot(fake_registry, temp_sessions):
+    fake_registry.capacity = 1
+    fake_registry._procs["ch-existing"] = FakeProc("ch-existing")
+
+    calls = []
+
+    def fake_run_claude(cmd, timeout, on_text=None, cwd=None, channel_id=None):
+        calls.append(cmd)
+        return {
+            "result": "one-shot reply", "session_id": "sid-2", "usage": {},
+            "total_cost_usd": 0, "cost_usd": 0, "duration_ms": 1, "num_turns": 1,
+            "intermediate_count": 0, "subtype": "success", "is_error": False,
+        }
+
+    with patch.object(claude_session, "_run_claude", side_effect=fake_run_claude):
+        result = send_message("ch-new", "hello", adapter_name="discord")
+
+    assert result == "one-shot reply"
+    assert len(calls) == 1
+    assert "ch-new" not in fake_registry._procs
+
+
+# ---------------------------------------------------------------------------
+# T9 — /clear and /model stop the live process
+# ---------------------------------------------------------------------------
+
+
+def test_clear_session_stops_live_process(fake_registry):
+    proc = fake_registry.get("ch-3")
+    fake_registry._procs["ch-3"] = proc
+    claude_session._save_sessions({"ch-3": {"session_id": "sid", "model": "sonnet"}})
+
+    assert claude_session.clear_session("ch-3") is True
+    assert proc.stopped is True
+    assert "ch-3" not in fake_registry._procs
+
+
+def test_set_session_model_stops_live_process(fake_registry):
+    proc = fake_registry.get("ch-4")
+    fake_registry._procs["ch-4"] = proc
+    claude_session._save_sessions({"ch-4": {"session_id": "sid", "model": "sonnet"}})
+
+    assert claude_session.set_session_model("ch-4", "opus") is True
+    assert proc.stopped is True
+    assert "ch-4" not in fake_registry._procs
+
+
+# ---------------------------------------------------------------------------
+# C3 — a steered message must never be lost when its turn then fails
+# ---------------------------------------------------------------------------
+
+
+def test_pending_steers_redispatched_on_failed_turn(monkeypatch):
+    """router.route_message's exception handler must pop pending steers
+    (claude_session.pop_pending_steers) and redeliver each one via
+    on_text — their own request threads already returned __STEERED__ and
+    are gone, so on_text is the only way left to answer them."""
+    pending = ["steered one", "steered two"]
+    monkeypatch.setattr(router, "_pop_pending_steers", lambda channel_id: list(pending))
+    monkeypatch.setattr(
+        router, "send_message",
+        lambda *a, **kw: (_ for _ in ()).throw(RuntimeError(
+            "Claude CLI error (exit 1): You've hit your session limit · resets 10am (UTC)"
+        )),
+    )
+    fallback_calls = []
+
+    def fake_fallback(text, channel_id=None, manual=False):
+        fallback_calls.append(text)
+        return f"fallback:{text}"
+
+    monkeypatch.setattr(router, "_local_fallback_reply", fake_fallback)
+
+    streamed = []
+    result, is_cmd = router.route_message(
+        "ch-c3", "user-1", "original message", on_text=streamed.append,
+    )
+
+    assert result == "fallback:original message"
+    assert is_cmd is False
+    assert fallback_calls == ["original message", "steered one", "steered two"]
+    assert streamed == ["fallback:steered one", "fallback:steered two"]
+
+
+def test_pending_steers_logged_without_on_text(monkeypatch, caplog):
+    """No on_text means no way to redeliver — must be logged (T12), not
+    silently dropped."""
+    monkeypatch.setattr(router, "_pop_pending_steers", lambda channel_id: ["lost message"])
+    monkeypatch.setattr(
+        router, "send_message",
+        lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("boom")),
+    )
+
+    with caplog.at_level("WARNING"):
+        result, is_cmd = router.route_message("ch-c3b", "user-1", "hello")
+
+    assert result == "Error: boom"
+    assert any("steered message lost" in rec.message for rec in caplog.records)

tests/test_stop_command.py

@@ -0,0 +1,58 @@
+"""Tests for /stop — Etapa 0 of the steering plan (one-shot architecture).
+
+`stop_turn(channel_id)` kills the in-flight Claude CLI subprocess registered
+for a channel, without touching sessions/active.json (the session survives,
+only the turn dies).
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.claude_session import _live_procs, stop_turn
+from src.router import route_message
+
+
+@pytest.fixture(autouse=True)
+def _clear_live_procs():
+    """Fresh registry per test — fake processes must not leak between tests."""
+    _live_procs.clear()
+    yield
+    _live_procs.clear()
+
+
+class TestStopTurn:
+    def test_idle_channel_returns_false(self):
+        assert stop_turn("ch-1") is False
+
+    def test_terminates_registered_process_and_cleans_up_registry(self):
+        fake_proc = MagicMock()
+        fake_proc.poll.return_value = None  # still running
+        _live_procs["ch-1"] = fake_proc
+
+        assert stop_turn("ch-1") is True
+        fake_proc.terminate.assert_called_once()
+        fake_proc.wait.assert_called_once()
+        assert "ch-1" not in _live_procs
+
+    def test_already_exited_process_returns_false(self):
+        fake_proc = MagicMock()
+        fake_proc.poll.return_value = 0  # exited before /stop ran
+        _live_procs["ch-1"] = fake_proc
+        assert stop_turn("ch-1") is False
+
+
+class TestStopCommand:
+    def test_stop_nothing_running(self):
+        response, is_cmd = route_message("ch-1", "user-1", "/stop")
+        assert response == "Nu rulează nimic pe canalul ăsta."
+        assert is_cmd is True
+
+    def test_stop_kills_in_flight_turn(self):
+        fake_proc = MagicMock()
+        fake_proc.poll.return_value = None
+        _live_procs["ch-1"] = fake_proc
+        response, is_cmd = route_message("ch-1", "user-1", "/stop")
+        assert response == "⏹ Oprit."
+        assert is_cmd is True
+        fake_proc.terminate.assert_called_once()