refactor(dashboard): I1 — move Step-10 auto-escalate classifier → codec_chat_pipeline (SR-60) (#167)

codec_dashboard.py: 1,310 → 1,201 LOC (-109).
Cumulative since pre-B6 baseline: 3,912 → 1,201 (-69.3%).

The Phase 3 Step 10 auto-escalation classifier cluster (9 names:
_AUTO_ESCALATE_SYSTEM_PROMPT, _qwen_chat_classify, _classify_chat_message,
_AUTOESCALATE_SILENCE_LOCK, _autoescalate_silence_set,
ESCALATE_CHECKPOINTS_THRESHOLD, silence_session_autoescalate,
_reset_autoescalate_silence_for_test, _should_escalate_to_project) moves
verbatim to codec_chat_pipeline.py — its natural home alongside _StepBudget /
_is_conversational. The cluster is latent (no production caller yet; the
chat→project "Promote?" prompt is deferred to Phase 3.5) but test-covered.

codec_dashboard re-exports all 9 names identity-equal for back-compat.

Test surface:
  - test_chat_escalation: monkeypatch target swapped codec_dashboard →
    codec_chat_pipeline. The functions call each other through the pipeline
    module namespace, so patching the in-module chain (_qwen_chat_classify →
    _classify_chat_message → _should_escalate_to_project) must target where
    they're DEFINED, not the re-export. All 8 monkeypatch tests updated.
  - test_dashboard_llm: codec_llm.call count >=2 → >=1 (classifier moved;
    only /api/command's Flash fallback remains in codec_dashboard).
  - new TestI1EscalationExtraction (3 pins): cluster-in-pipeline, 9-name
    identity-equal re-export, dashboard-no-longer-defines.
  - drive-by ruff F401: dropped now-unused `import threading` from codec_dashboard
    (the only consumer was _AUTOESCALATE_SILENCE_LOCK, which moved).

codec_chat_pipeline gains `import threading` + `import codec_llm` for the
moved code. test_dashboard_llm's 3 _qwen_chat_classify behavior tests are
unaffected (they patch codec_llm.call at source, which the re-export honors).

Full suite: 2,055 passed / 77 skipped (was 2,052 in H1). +3 net.

Co-authored-by: Mickael Farina <farina.mickael@gmail.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: 3 people

codec_chat_pipeline.py

@@ -20,9 +20,11 @@
 import json
 import logging
 import os
+import threading
 from typing import Optional
 
 from codec_audit import STEP_BUDGET_EXHAUSTED, log_event
+import codec_llm  # I1 / SR-60: auto-escalation classifier LLM call
 
 log = logging.getLogger("codec_chat_pipeline")
 
@@ -170,3 +172,137 @@ def _emit_exhausted(self, kind: str):
             )
         except Exception as e:
             log.warning("[step_budget] emit failed: %s", e)
+
+
+# ═══════════════════════════════════════════════════════════════
+# Phase 3 Step 10 — Auto-escalation classifier (I1 / SR-60)
+# Moved verbatim from codec_dashboard.py. Latent (no production caller
+# yet — the chat→project 'Promote?' prompt is deferred to Phase 3.5),
+# but exercised by tests. The functions call each other in-module, so
+# tests must monkeypatch codec_chat_pipeline.* (not the codec_dashboard
+# re-export). codec_dashboard re-exports all names for back-compat.
+# ═══════════════════════════════════════════════════════════════
+
+# ── Phase 3 Step 10 — Auto-escalation classifier ──────────────────────────
+
+_AUTO_ESCALATE_SYSTEM_PROMPT = """You are CODEC's chat-input classifier. \
+Given the user's chat message, decide if it represents a "project" — \
+multi-step work that would benefit from autonomous execution by an agent \
+(file writes, browser automation, multi-checkpoint plan) — or a "quick \
+question" suitable for single-shot LLM answer.
+
+Return ONLY a JSON object:
+{
+  "is_project": <bool>,
+  "estimated_checkpoints": <int — best guess of plan size; 0 if not project>,
+  "reason": <short string explaining the verdict>
+}
+
+Rules:
+- Single-shot factual / conversational / explanatory questions → is_project=false.
+- "Build me X", "Set up Y", "Watch Z and tell me when W", "Plan launch of A" → is_project=true.
+- Be honest about checkpoint estimates; under 3 means not worth promoting.
+"""
+
+
+def _qwen_chat_classify(user_text: str, max_tokens: int = 300) -> str:
+    """Call Qwen-3.6 with the auto-escalation classifier prompt. Returns
+    raw response string. Caller handles JSON parsing + error fallback.
+
+    Hotfix: URL + model resolved from codec_config (was hardcoded to the
+    wrong dashboard port 8090; LLM lives at 8083 per ~/.codec/config.json)."""
+    try:
+        from codec_config import QWEN_BASE_URL, QWEN_MODEL as _qmodel
+        # A-12 (PR-3E-dashboard): canonical codec_llm.call (never-raises -> "").
+        # Now strips <think> + enable_thinking=False -> cleaner JSON for the
+        # downstream _classify_chat_message parse.
+        return codec_llm.call(
+            [
+                {"role": "system", "content": _AUTO_ESCALATE_SYSTEM_PROMPT},
+                {"role": "user", "content": user_text[:2000]},
+            ],
+            base_url=QWEN_BASE_URL, model=_qmodel,
+            max_tokens=max_tokens, temperature=0.1, timeout=15,
+        )
+    except Exception as e:
+        log.debug(f"_qwen_chat_classify failed: {e}")
+        return ""
+
+
+def _classify_chat_message(user_text: str) -> tuple[bool, int, str]:
+    """Returns (is_project, estimated_checkpoints, reason). Falls back to
+    (False, 0, reason) on any failure."""
+    raw = _qwen_chat_classify(user_text)
+    if not raw:
+        return (False, 0, "qwen unavailable")
+
+    raw = raw.strip()
+    if raw.startswith("```"):
+        import re as _re
+        raw = _re.sub(r"^```(?:json)?\s*", "", raw)
+        raw = _re.sub(r"\s*```\s*$", "", raw)
+
+    try:
+        d = json.loads(raw)
+    except json.JSONDecodeError:
+        return (False, 0, "qwen returned non-JSON")
+
+    return (
+        bool(d.get("is_project", False)),
+        int(d.get("estimated_checkpoints", 0)),
+        str(d.get("reason", ""))[:200],
+    )
+
+
+# ── Auto-escalation gate (in-memory session silence per Q11) ──────────────
+
+_AUTOESCALATE_SILENCE_LOCK = threading.Lock()
+_autoescalate_silence_set: set[str] = set()  # session_ids that said "no" once
+
+ESCALATE_CHECKPOINTS_THRESHOLD = 3
+
+
+def silence_session_autoescalate(session_id: str) -> None:
+    """Q11: After user says No once, silence auto-escalation prompts for
+    the rest of this conversation. Resets on new chat session."""
+    with _AUTOESCALATE_SILENCE_LOCK:
+        _autoescalate_silence_set.add(session_id)
+
+
+def _reset_autoescalate_silence_for_test() -> None:
+    """Test-only helper to clear in-memory silence state."""
+    with _AUTOESCALATE_SILENCE_LOCK:
+        _autoescalate_silence_set.clear()
+
+
+def _should_escalate_to_project(user_text: str, session_id: str) -> dict:
+    """2-signal gate (Step 10):
+      Signal 1: classifier verdict (is_project=True)
+      Signal 2: estimated_checkpoints >= ESCALATE_CHECKPOINTS_THRESHOLD
+
+    Plus 2 kill conditions:
+      - AGENT_AUTO_ESCALATE_ENABLED=false
+      - session_id in silence set (Q11)
+
+    Returns: {"escalate": bool, "estimated_checkpoints": int, "reason": str}
+    """
+    import os as _os
+    if _os.environ.get("AGENT_AUTO_ESCALATE_ENABLED", "true").lower() == "false":
+        return {"escalate": False, "estimated_checkpoints": 0,
+                "reason": "kill_switch_off"}
+
+    with _AUTOESCALATE_SILENCE_LOCK:
+        if session_id in _autoescalate_silence_set:
+            return {"escalate": False, "estimated_checkpoints": 0,
+                    "reason": "session_silenced", "silenced": True}
+
+    is_project, n_checkpoints, reason = _classify_chat_message(user_text)
+
+    escalate = is_project and n_checkpoints >= ESCALATE_CHECKPOINTS_THRESHOLD
+
+    return {
+        "escalate": escalate,
+        "estimated_checkpoints": n_checkpoints,
+        "reason": reason,
+        "is_project": is_project,
+    }

codec_dashboard.py

@@ -3,7 +3,6 @@
 import json
 import time
 import hmac
-import threading
 import asyncio
 from datetime import datetime, timedelta
 
@@ -862,135 +861,27 @@ async def vibe_page():
     _step_budget_for_route,
     _StepBudget,
 )
+# I1 / SR-60: auto-escalation classifier cluster (Step 10) moved to
+# codec_chat_pipeline. Re-exported here so any caller / test that imported them
+# from codec_dashboard keeps working. NOTE: the functions call each other via
+# the pipeline module namespace — tests that monkeypatch the chain must target
+# codec_chat_pipeline.*, not these re-exports (test_chat_escalation does).
+from codec_chat_pipeline import (  # noqa: E402,F401  (back-compat re-exports)
+    ESCALATE_CHECKPOINTS_THRESHOLD,
+    _AUTO_ESCALATE_SYSTEM_PROMPT,
+    _autoescalate_silence_set,
+    _AUTOESCALATE_SILENCE_LOCK,
+    _classify_chat_message,
+    _qwen_chat_classify,
+    _reset_autoescalate_silence_for_test,
+    _should_escalate_to_project,
+    silence_session_autoescalate,
+)
 
 
 # H1 / SR-59: def _try_skill → moved to routes/chat.py
 # H1 / SR-59: def _try_skill_by_name → moved to routes/chat.py
-# ── Phase 3 Step 10 — Auto-escalation classifier ──────────────────────────
-
-_AUTO_ESCALATE_SYSTEM_PROMPT = """You are CODEC's chat-input classifier. \
-Given the user's chat message, decide if it represents a "project" — \
-multi-step work that would benefit from autonomous execution by an agent \
-(file writes, browser automation, multi-checkpoint plan) — or a "quick \
-question" suitable for single-shot LLM answer.
-
-Return ONLY a JSON object:
-{
-  "is_project": <bool>,
-  "estimated_checkpoints": <int — best guess of plan size; 0 if not project>,
-  "reason": <short string explaining the verdict>
-}
-
-Rules:
-- Single-shot factual / conversational / explanatory questions → is_project=false.
-- "Build me X", "Set up Y", "Watch Z and tell me when W", "Plan launch of A" → is_project=true.
-- Be honest about checkpoint estimates; under 3 means not worth promoting.
-"""
-
-
-def _qwen_chat_classify(user_text: str, max_tokens: int = 300) -> str:
-    """Call Qwen-3.6 with the auto-escalation classifier prompt. Returns
-    raw response string. Caller handles JSON parsing + error fallback.
-
-    Hotfix: URL + model resolved from codec_config (was hardcoded to the
-    wrong dashboard port 8090; LLM lives at 8083 per ~/.codec/config.json)."""
-    try:
-        from codec_config import QWEN_BASE_URL, QWEN_MODEL as _qmodel
-        # A-12 (PR-3E-dashboard): canonical codec_llm.call (never-raises -> "").
-        # Now strips <think> + enable_thinking=False -> cleaner JSON for the
-        # downstream _classify_chat_message parse.
-        return codec_llm.call(
-            [
-                {"role": "system", "content": _AUTO_ESCALATE_SYSTEM_PROMPT},
-                {"role": "user", "content": user_text[:2000]},
-            ],
-            base_url=QWEN_BASE_URL, model=_qmodel,
-            max_tokens=max_tokens, temperature=0.1, timeout=15,
-        )
-    except Exception as e:
-        log.debug(f"_qwen_chat_classify failed: {e}")
-        return ""
-
-
-def _classify_chat_message(user_text: str) -> tuple[bool, int, str]:
-    """Returns (is_project, estimated_checkpoints, reason). Falls back to
-    (False, 0, reason) on any failure."""
-    raw = _qwen_chat_classify(user_text)
-    if not raw:
-        return (False, 0, "qwen unavailable")
-
-    raw = raw.strip()
-    if raw.startswith("```"):
-        import re as _re
-        raw = _re.sub(r"^```(?:json)?\s*", "", raw)
-        raw = _re.sub(r"\s*```\s*$", "", raw)
-
-    try:
-        d = json.loads(raw)
-    except json.JSONDecodeError:
-        return (False, 0, "qwen returned non-JSON")
-
-    return (
-        bool(d.get("is_project", False)),
-        int(d.get("estimated_checkpoints", 0)),
-        str(d.get("reason", ""))[:200],
-    )
-
-
-# ── Auto-escalation gate (in-memory session silence per Q11) ──────────────
-
-_AUTOESCALATE_SILENCE_LOCK = threading.Lock()
-_autoescalate_silence_set: set[str] = set()  # session_ids that said "no" once
-
-ESCALATE_CHECKPOINTS_THRESHOLD = 3
-
-
-def silence_session_autoescalate(session_id: str) -> None:
-    """Q11: After user says No once, silence auto-escalation prompts for
-    the rest of this conversation. Resets on new chat session."""
-    with _AUTOESCALATE_SILENCE_LOCK:
-        _autoescalate_silence_set.add(session_id)
-
-
-def _reset_autoescalate_silence_for_test() -> None:
-    """Test-only helper to clear in-memory silence state."""
-    with _AUTOESCALATE_SILENCE_LOCK:
-        _autoescalate_silence_set.clear()
-
-
-def _should_escalate_to_project(user_text: str, session_id: str) -> dict:
-    """2-signal gate (Step 10):
-      Signal 1: classifier verdict (is_project=True)
-      Signal 2: estimated_checkpoints >= ESCALATE_CHECKPOINTS_THRESHOLD
-
-    Plus 2 kill conditions:
-      - AGENT_AUTO_ESCALATE_ENABLED=false
-      - session_id in silence set (Q11)
-
-    Returns: {"escalate": bool, "estimated_checkpoints": int, "reason": str}
-    """
-    import os as _os
-    if _os.environ.get("AGENT_AUTO_ESCALATE_ENABLED", "true").lower() == "false":
-        return {"escalate": False, "estimated_checkpoints": 0,
-                "reason": "kill_switch_off"}
-
-    with _AUTOESCALATE_SILENCE_LOCK:
-        if session_id in _autoescalate_silence_set:
-            return {"escalate": False, "estimated_checkpoints": 0,
-                    "reason": "session_silenced", "silenced": True}
-
-    is_project, n_checkpoints, reason = _classify_chat_message(user_text)
-
-    escalate = is_project and n_checkpoints >= ESCALATE_CHECKPOINTS_THRESHOLD
-
-    return {
-        "escalate": escalate,
-        "estimated_checkpoints": n_checkpoints,
-        "reason": reason,
-        "is_project": is_project,
-    }
-
-
+# I1 / SR-60: Phase 3 Step 10 auto-escalation classifier cluster → moved to codec_chat_pipeline.py (re-exported below)
 # H1 / SR-59: def _chat_vision_response → moved to routes/chat.py
 # H1 / SR-59: def _build_chat_system_prompt → moved to routes/chat.py
 # H1 / SR-59:  → moved to routes/chat.py

tests/test_chat_escalation.py

@@ -2,6 +2,12 @@
 
 11 tests covering: classifier (3), 2-signal gate (3), session silence (2),
 integration with chat handler (2), kill switch (1).
+
+I1 / SR-60: the classifier cluster moved from codec_dashboard to
+codec_chat_pipeline. These tests monkeypatch the in-module call chain
+(`_qwen_chat_classify` → `_classify_chat_message` → `_should_escalate_to_project`),
+so they must patch where the functions are DEFINED (codec_chat_pipeline),
+not the codec_dashboard re-export. Hence `import codec_chat_pipeline as cd`.
 """
 from __future__ import annotations
 
@@ -17,7 +23,7 @@
 
 def test_classify_chat_message_returns_project_when_multi_step(monkeypatch):
     """LLM verdict says multi-step → returns is_project=True with checkpoints estimate."""
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     fake_response = json.dumps({
         "is_project": True,
@@ -35,7 +41,7 @@ def test_classify_chat_message_returns_project_when_multi_step(monkeypatch):
 
 
 def test_classify_chat_message_returns_not_project_for_quick_question(monkeypatch):
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     fake_response = json.dumps({
         "is_project": False, "estimated_checkpoints": 0,
@@ -50,7 +56,7 @@ def test_classify_chat_message_returns_not_project_for_quick_question(monkeypatc
 
 def test_classify_chat_message_handles_qwen_failure(monkeypatch):
     """If Qwen call fails or returns garbage, classifier returns (False, 0, reason)."""
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     monkeypatch.setattr(cd, "_qwen_chat_classify",
                         lambda text: "garbage non-json")
@@ -62,7 +68,7 @@ def test_classify_chat_message_handles_qwen_failure(monkeypatch):
 
 def test_should_escalate_when_both_signals_pass(monkeypatch):
     """LLM says project + checkpoints >= 3 → escalate."""
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     monkeypatch.setattr(cd, "_classify_chat_message",
                         lambda text: (True, 5, "multi-step"))
@@ -74,7 +80,7 @@ def test_should_escalate_when_both_signals_pass(monkeypatch):
 
 def test_should_not_escalate_when_checkpoints_below_3(monkeypatch):
     """LLM says project but estimate=2 → don't escalate."""
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     monkeypatch.setattr(cd, "_classify_chat_message",
                         lambda text: (True, 2, "small"))
@@ -85,7 +91,7 @@ def test_should_not_escalate_when_checkpoints_below_3(monkeypatch):
 
 def test_should_not_escalate_when_classifier_says_no(monkeypatch):
     """LLM says not-a-project → don't escalate even if checkpoints>=3."""
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     monkeypatch.setattr(cd, "_classify_chat_message",
                         lambda text: (False, 5, "actually quick"))
@@ -96,7 +102,7 @@ def test_should_not_escalate_when_classifier_says_no(monkeypatch):
 
 def test_session_silence_persists_across_calls(monkeypatch):
     """Q11: After silence_session(s1), subsequent _should_escalate calls return escalate=False."""
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     monkeypatch.setattr(cd, "_classify_chat_message",
                         lambda text: (True, 5, "always-project"))
@@ -117,7 +123,7 @@ def test_session_silence_persists_across_calls(monkeypatch):
 
 def test_kill_switch_disables_all_escalation(monkeypatch):
     """AGENT_AUTO_ESCALATE_ENABLED=false → never escalate."""
-    import codec_dashboard as cd
+    import codec_chat_pipeline as cd
 
     monkeypatch.setenv("AGENT_AUTO_ESCALATE_ENABLED", "false")
     monkeypatch.setattr(cd, "_classify_chat_message",