.gitignore

@@ -3,6 +3,7 @@ db/*.db
 *.db-shm
 *.db-wal
 .env
+.watchdog.lock
 __pycache__/
 *.pyc
 router.log

admin/frontend/controls.html

@@ -382,6 +382,49 @@ header.navbar{padding-top:2px!important;padding-bottom:2px!important}
             </div>
           </div>
 
+          <!-- Watchdog -->
+          <div class="col-12">
+            <div class="card" id="watchdog-card">
+              <div class="card-header">
+                <h3 class="card-title"><span class="me-2" data-icon="alert"></span>Watchdog</h3>
+                <div class="card-actions d-flex align-items-center gap-2">
+                  <span id="watchdog-last-tick" class="badge bg-secondary">no tick</span>
+                </div>
+              </div>
+              <div class="card-body py-3">
+                <div class="row g-3">
+                  <div class="col-md-4">
+                    <div class="text-muted mb-1" style="font-size:.78rem">Last tick</div>
+                    <div id="wd-tick-time" class="small">-</div>
+                  </div>
+                  <div class="col-md-4">
+                    <div class="text-muted mb-1" style="font-size:.78rem">Sessions seen</div>
+                    <div id="wd-sessions" class="small">-</div>
+                  </div>
+                  <div class="col-md-4">
+                    <div class="text-muted mb-1" style="font-size:.78rem">Flagged verdicts</div>
+                    <div id="wd-flagged" class="small">-</div>
+                  </div>
+                </div>
+                <div class="mt-3">
+                  <div class="text-muted mb-1" style="font-size:.78rem">Open alerts</div>
+                  <div id="wd-alerts" class="small">-</div>
+                </div>
+                <div class="mt-3">
+                  <div class="text-muted mb-2" style="font-size:.78rem">Notification channels</div>
+                  <div id="wd-channels" class="d-flex flex-wrap gap-2">-</div>
+                </div>
+                <div class="d-flex align-items-center justify-content-between mt-3">
+                  <span id="watchdog-status" class="text-muted" style="font-size:.72rem"></span>
+                  <div class="d-flex gap-2">
+                    <button class="btn btn-outline-secondary btn-sm" onclick="loadWatchdog()">Refresh</button>
+                    <button class="btn btn-success btn-sm" onclick="testWatchdogAlert()">Test Alert</button>
+                  </div>
+                </div>
+              </div>
+            </div>
+          </div>
+
           <!-- One knob table: runtime and persisted state pair by config key,
                so drift (live value the next restart will undo) reads on the
                same row as the controls that cause and fix it. -->
@@ -1389,6 +1432,61 @@ function clearCloudFallback() {
 
 
 
+/* ═══════════════════════════════════════
+   WATCHDOG
+   ═══════════════════════════════════════ */
+
+async function loadWatchdog() {
+  const status = await apiFetch(`${API}api/watchdog/status`);
+  if (status) {
+    const tick = status.last_tick;
+    const tickEl = document.getElementById('wd-tick-time');
+    const sessionsEl = document.getElementById('wd-sessions');
+    const flaggedEl = document.getElementById('wd-flagged');
+    const alertsEl = document.getElementById('wd-alerts');
+    const badge = document.getElementById('watchdog-last-tick');
+    if (tick) {
+      tickEl.textContent = tick.ticked_at || '-';
+      sessionsEl.textContent = tick.sessions_seen != null ? tick.sessions_seen : '-';
+      badge.textContent = tick.outcome || 'ok';
+      badge.className = tick.outcome === 'ok' ? 'badge bg-success' : 'badge bg-warning';
+    }
+    if (status.verdicts) {
+      flaggedEl.textContent = `${status.verdicts.flagged || 0} / ${status.verdicts.total || 0}`;
+    }
+    alertsEl.textContent = status.open_alerts || '0';
+  }
+  const channels = await apiFetch(`${API}api/watchdog/channels`);
+  const chEl = document.getElementById('wd-channels');
+  if (channels && channels.length) {
+    chEl.innerHTML = channels.map(c =>
+      `<span class="badge ${c.enabled ? 'bg-success' : 'bg-secondary'}">${escapeHtml(c.channel_name)} (${escapeHtml(c.min_severity)})</span>`
+    ).join('');
+  } else {
+    chEl.innerHTML = '<span class="text-muted small">no channels configured</span>';
+  }
+}
+
+async function testWatchdogAlert() {
+  const statusEl = document.getElementById('watchdog-status');
+  statusEl.textContent = 'sending test alert...';
+  statusEl.style.color = 'var(--tblr-secondary)';
+  const resp = await apiFetch(`${API}api/watchdog/test-alert`, {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ severity: 'warning' }),
+  });
+  if (resp) {
+    statusEl.textContent = `sent to ${resp.sent} channel(s)`;
+    statusEl.style.color = 'var(--tblr-success)';
+    toast('Test alert sent', 'success');
+  } else {
+    statusEl.textContent = 'failed';
+    statusEl.style.color = 'var(--tblr-danger)';
+    toast('Test alert failed', 'error');
+  }
+}
+
 /* ═══════════════════════════════════════
    DATA FETCHING
    ═══════════════════════════════════════ */
@@ -1409,6 +1507,7 @@ async function loadControls() {
   if (config) renderConfig(config);
   await loadClassifierConfig();
   await loadCloudFallback();
+  loadWatchdog();
 }
 
 /* ═══════════════════════════════════════

admin/frontend/index.html

@@ -417,6 +417,17 @@ header.navbar{padding-top:2px!important;padding-bottom:2px!important}
           </div>
         </div>
 
+        <!-- Loops: open watchdog alerts. -->
+        <div class="card mt-3" id="loops">
+          <div class="card-header" style="padding:.55rem .8rem">
+            <h3 class="card-title mb-0" style="font-size:.9rem">Loops</h3>
+            <span class="card-subtitle text-muted ms-2" style="font-size:.72rem" id="loops-count">0 open</span>
+          </div>
+          <div class="card-body" style="padding:.6rem .8rem">
+            <div id="loops-body"><div class="text-muted small">Loading&hellip;</div></div>
+          </div>
+        </div>
+
         <!-- LIVE: the router working, as the page's subject. -->
         <div id="view-live" hidden>
           <div class="live-wrap">
@@ -1321,6 +1332,32 @@ function hashCode(str) {
   return hash;
 }
 
+/* ═══════════════════════════════════════
+   LOOPS (open watchdog alerts)
+   ═══════════════════════════════════════ */
+
+async function loadLoops() {
+  const rows = await apiFetch(`${API}api/watchdog/loops`);
+  const body = document.getElementById('loops-body');
+  const countEl = document.getElementById('loops-count');
+  if (!rows || !rows.length) {
+    body.innerHTML = '<div class="text-muted small">No open alerts</div>';
+    countEl.textContent = '0 open';
+    return;
+  }
+  countEl.textContent = `${rows.length} open`;
+  body.innerHTML = rows.map(r => {
+    const sev = r.severity === 'critical' ? 'bg-danger'
+      : r.severity === 'warning' ? 'bg-warning'
+      : 'bg-info';
+    return `<div class="d-flex align-items-center gap-2 py-1" style="border-bottom:1px solid rgba(255,255,255,.05)">
+      <span class="badge ${sev}" style="font-size:.62rem">${escapeHtml(r.severity)}</span>
+      <span class="small" style="flex:1">${escapeHtml(r.dedup_key)}</span>
+      <span class="text-muted" style="font-size:.66rem">${escapeHtml(r.opened_at || '')}</span>
+    </div>`;
+  }).join('');
+}
+
 /* ═══════════════════════════════════════
    INIT
    ═══════════════════════════════════════ */
@@ -1334,6 +1371,7 @@ function init() {
   loadHistory();
   loadActivity();
   loadLiveHistory();
+  loadLoops();
   connectSSE();
   // Auto-refresh
   setInterval(() => {
@@ -1341,6 +1379,7 @@ function init() {
     loadHistory();
     loadActivity();
     loadLiveHistory();
+    loadLoops();
   }, REFRESH_MS);
 }
 

admin/frontend/models.html

@@ -229,7 +229,7 @@ header.navbar{padding-top:2px!important;padding-bottom:2px!important}
                   <svg xmlns="http://www.w3.org/2000/svg" width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><ellipse cx="12" cy="5" rx="9" ry="3"/><path d="M21 12c0 1.66-4 3-9 3s-9-1.34-9-3"/><path d="M3 5v14c0 1.66 4 3 9 3s9-1.34 9-3V5"/></svg>
                   <h3 class="card-title mb-0">Model Availability</h3>
                 </div>
-                <div class="card-subtitle text-muted ms-2">Change the override dropdown to mark a model active, deprecated, or stale</div>
+                <div class="card-subtitle text-muted ms-2">Change the override dropdown to mark a model active, blocked, deprecated, or stale</div>
                 <div class="ms-auto d-flex align-items-center gap-2 flex-wrap">
                   <span class="text-muted small" id="model-count">&mdash;</span>
                 </div>
@@ -259,6 +259,7 @@ header.navbar{padding-top:2px!important;padding-bottom:2px!important}
                     <option value="routable">Routable only</option>
                     <option value="">All statuses</option>
                     <option value="active">active</option>
+                    <option value="blocked">blocked</option>
                     <option value="deprecated">deprecated</option>
                     <option value="stale">stale</option>
                   </select>
@@ -596,8 +597,8 @@ function renderModels(models) {
   }
 
   const html = visible.map(m => {
-    const availClass = { active: 'var(--tblr-success)', deprecated: 'var(--tblr-danger)', stale: 'var(--tblr-warning)' }[m.effective_availability] || 'var(--tblr-secondary)';
+    const availClass = { active: 'var(--tblr-success)', blocked: 'var(--tblr-danger)', deprecated: 'var(--tblr-danger)', stale: 'var(--tblr-warning)' }[m.effective_availability] || 'var(--tblr-secondary)';
-    const selectOpts = ['active','deprecated','stale'].map(a =>
+    const selectOpts = ['active','blocked','deprecated','stale'].map(a =>
       `<option value="${a}"${m.effective_availability===a?' selected':''}>${a}</option>`
     ).join('');
     const tierClass = `tier-${m.tier || '?'}`;

config/config.yaml

@@ -1113,3 +1113,29 @@ logging:
   #   systemctl --user restart llm-router
   level: info
 
+watchdog:
+  enabled: true
+  local_llm_enabled: true
+  model: null                # defaults to verification.model
+  read_only_agents:
+    - explore
+    - librarian
+    - oracle
+  detector:
+    window: 60
+    dup_min: 0.25
+    top_min: 12
+    top_min_ro: 8
+    cum_min: 15
+    cover_min: 4.0
+    min_calls: 40
+  dashboard_base_url: "http://127.0.0.1:8080/admin"  # base URL for alert links
+  # tick interval is 5 min (set in deploy/llm-router-watchdog.timer)
+
+notifications:
+  channels:
+    - name: default
+      type: desktop
+      enabled: true
+      min_severity: warning
+

config/schema.sql

@@ -435,3 +435,51 @@ CREATE TABLE IF NOT EXISTS provider_model_allowlist (
     PRIMARY KEY (provider, model_id)
 );
 CREATE INDEX IF NOT EXISTS idx_allowlist_provider ON provider_model_allowlist (provider);
+
+-- Watchdog monitoring tables. See src/watchdog_store.py for the code-side
+-- inline-create that mirrors this schema for live databases.
+CREATE TABLE IF NOT EXISTS watchdog_ticks (
+    id               INTEGER PRIMARY KEY AUTOINCREMENT,
+    ticked_at        TEXT NOT NULL,
+    sessions_seen    INTEGER NOT NULL DEFAULT 0,
+    outcome          TEXT
+);
+CREATE TABLE IF NOT EXISTS watchdog_verdicts (
+    id               INTEGER PRIMARY KEY AUTOINCREMENT,
+    tick_id          INTEGER NOT NULL REFERENCES watchdog_ticks(id),
+    session_id       TEXT NOT NULL,
+    session_root     TEXT,
+    agent            TEXT,
+    model_id         TEXT,
+    provider         TEXT,
+    flagged          INTEGER NOT NULL DEFAULT 0,
+    dup              REAL,
+    top              INTEGER,
+    top_what         TEXT,
+    landed           INTEGER,
+    slow             INTEGER,
+    coverage         REAL,
+    calls_since_landed    INTEGER,
+    cost_since_landed_usd REAL,
+    llm_second_opinion TEXT,
+    created_at       TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS watchdog_alerts (
+    dedup_key        TEXT PRIMARY KEY,
+    state            TEXT NOT NULL,
+    severity         TEXT NOT NULL,
+    flagged_ticks    INTEGER NOT NULL DEFAULT 0,
+    opened_at        TEXT NOT NULL,
+    last_fired_at    TEXT NOT NULL,
+    resolved_at      TEXT
+);
+CREATE TABLE IF NOT EXISTS watchdog_channel_settings (
+    id               INTEGER PRIMARY KEY AUTOINCREMENT,
+    channel_name     TEXT NOT NULL UNIQUE,
+    enabled          INTEGER NOT NULL DEFAULT 1,
+    min_severity     TEXT NOT NULL DEFAULT 'warning'
+);
+CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_model_id ON watchdog_verdicts(model_id);
+CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_created_at ON watchdog_verdicts(created_at);
+CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_flagged ON watchdog_verdicts(flagged);
+CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_session_root ON watchdog_verdicts(session_root);

deploy/llm-router-watchdog.service

@@ -0,0 +1,12 @@
+[Unit]
+Description=LLM Router Watchdog — opencode loop detection
+
+[Service]
+Type=oneshot
+WorkingDirectory=%h/llm-router
+Environment=PYTHONPATH=%h/llm-router/src
+ExecStart=%h/llm-router/.venv/bin/python -m watchdog --once
+ProtectHome=read-only
+ReadWritePaths=%h/llm-router
+
+# NO EnvironmentFile: watchdog needs no API keys (local Ollama only)

deploy/llm-router-watchdog.timer

@@ -0,0 +1,10 @@
+[Unit]
+Description=LLM Router Watchdog — 5-minute interval
+
+[Timer]
+OnBootSec=5min
+OnUnitActiveSec=5min
+Persistent=true
+
+[Install]
+WantedBy=timers.target

deploy/opencode-plugin/router-link.js

@@ -218,4 +218,7 @@ export const RouterLink = async ({ client, directory }) => {
   };
 };
 
-export { parentCache };
+// Attach parentCache for test introspection.  opencode's plugin loader
+// rejects any non-function export (see loader-contract test), so the map
+// lives as a property on the factory function rather than a named export.
+RouterLink.parentCache = parentCache;

deploy/opencode-plugin/router-link.test.mjs

@@ -12,7 +12,9 @@
 import { describe, it, mock, before, after, beforeEach } from "node:test";
 import assert from "node:assert/strict";
 
-import { RouterLink, parentCache } from "./router-link.js";
+import { RouterLink } from "./router-link.js";
+
+const parentCache = RouterLink.parentCache;
 
 const SESSION_ID = "ses_test123";
 const AGENT = "build";
@@ -698,4 +700,13 @@ describe("parent cache", () => {
     // An entry in the middle still present
     assert.equal(parentCache.has("ses_evict_0500"), true);
   });
+});
+
+describe("loader contract", () => {
+  it("every named export from the module is a function (opencode rejects non-function exports)", async () => {
+    const mod = await import("./router-link.js");
+    for (const [name, value] of Object.entries(mod)) {
+      assert.equal(typeof value, "function", `export "${name}" must be a function`);
+    }
+  });
 });

scripts/export_progress_fixture.py

@@ -0,0 +1,327 @@
+#!/usr/bin/env python3
+"""Export labelled calibration sessions from opencode SQLite DB into fixture JSON.
+
+Usage:
+    python scripts/export_progress_fixture.py
+
+Output: tests/fixtures/progress/fixture.json
+
+Fixture format: array of {session_id, label, root_id, title, agent, file_lines, calls}
+
+Content scrubbing: sha1 everything except filePath/path/offset/limit/command/pattern.
+These keys are the detector's target and must pass through verbatim.
+"""
+
+import hashlib
+import json
+import sqlite3
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+DB_PATH = Path.home() / ".local" / "share" / "opencode" / "opencode.db"
+OUTPUT_PATH = Path(__file__).resolve().parent.parent / "tests" / "fixtures" / "progress" / "fixture.json"
+
+GIT_REPO = "/home/alee/Sources/6krrt"
+WORKTREE_PREFIX = "/home/alee/Sources/6krrt-cockpit-quick-wins/"
+MAIN_REPO_PREFIX = "/home/alee/Sources/6krrt/"
+WORKTREE_COMMIT = "827c408"
+
+# Regex to find likely file paths in bash command text
+import re
+
+_PATH_RE = re.compile(r"/home/alee/Sources/6krrt(?:-cockpit-quick-wins)?/[\w./-]+\.(?:py|js|html|css|json|yaml|yml|md|toml|sh|sql|mjs|ts|tsx)")
+
+LABELS: dict[str, str] = {
+    "ses_f24cc0258ffepbJPoOzoiLZi9s": "must_flag",
+    "ses_f24fcf651ffeOm9O4E14QSZeJ4": "must_flag",
+    "ses_f24890640ffewKnIPqjjAfFvOk": "must_flag",
+    "ses_f245607eaffeGxPH8F1tvvdvkG": "must_flag",
+    "ses_f237e7e31ffeOzz4kUKu4CWzMu": "must_flag",
+    "ses_f235d755bffeo0uWtD1g2HwQUg": "must_flag",
+    "ses_f2506ac70ffebLUxBsq0aauOn9": "must_flag",
+    "ses_f2380dd4cffeVwcZn3Kv3nQN7v": "must_flag",
+    "ses_f246e699fffeCHUPTLr5ajBhWC": "must_not_flag",
+    "ses_f245f78f8ffeGMn3PgArpOwhFy": "must_not_flag",
+    "ses_f243ea08effemMwXdMRE1gpnhK": "must_not_flag",
+    "ses_f242e5e28ffehF97ZxxdbTeylQ": "must_not_flag",
+    "ses_f236bbdaaffezwnvH10X2rgrxc": "must_not_flag",
+    "ses_f236bc7baffeG5LQsr2K1c46xO": "must_not_flag",
+    "ses_f2e81929fffeaVbWui0X9QjmRQ": "must_not_flag",
+}
+
+SAFE_KEYS = {"filePath", "path", "offset", "limit", "command", "pattern", "output_mode", "include", "tmux_command"}
+
+# ── line count cache ────────────────────────────────────────────────────────
+
+_LEN_CACHE: dict[str, int | None] = {}
+
+
+def _count_lines_worktree(relpath: str) -> int | None:
+    """Count lines of a file in the deleted worktree via git show."""
+    try:
+        r = subprocess.run(
+            ["git", "-C", GIT_REPO, "show", f"{WORKTREE_COMMIT}:{relpath}"],
+            capture_output=True, text=True, timeout=15, check=False,
+        )
+        if r.returncode == 0:
+            return max(1, r.stdout.count("\n"))
+        return None
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+
+
+def _count_lines_disk(abspath: str) -> int | None:
+    """Count lines of a file that exists on disk."""
+    try:
+        with open(abspath, "rb") as fh:
+            return max(1, fh.read().count(b"\n"))
+    except OSError:
+        return None
+
+
+def count_lines(path: str) -> int | None:
+    """Count lines of a file. Cached."""
+    if path not in _LEN_CACHE:
+        if path.startswith(WORKTREE_PREFIX):
+            rel = path[len(WORKTREE_PREFIX):]
+            _LEN_CACHE[path] = _count_lines_worktree(rel)
+        elif Path(path).exists():
+            _LEN_CACHE[path] = _count_lines_disk(path)
+        else:
+            _LEN_CACHE[path] = None
+    return _LEN_CACHE[path]
+
+
+# ── helpers ─────────────────────────────────────────────────────────────────
+
+def sha1(s: str) -> str:
+    return hashlib.sha1(s.encode("utf-8")).hexdigest()
+
+
+_sha1_prefix: str | None = None
+
+
+def _hash_short(value: str) -> str:
+    """8-hex-char prefix of sha1."""
+    s = sha1(value)
+    return s[:8]
+
+
+def scrub_args(inp: dict[str, Any]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for k, v in inp.items():
+        if k in SAFE_KEYS:
+            result[k] = v
+        elif isinstance(v, str):
+            result[k] = _hash_short(v) if len(v) > 0 else v
+        elif isinstance(v, (list, dict)):
+            result[k] = _hash_short(json.dumps(v, sort_keys=True))
+        else:
+            result[k] = v
+    return result
+
+
+def extract_root_id(db: sqlite3.Connection, session_id: str) -> str:
+    current = session_id
+    visited = set()
+    while current and current not in visited:
+        visited.add(current)
+        row = db.execute("SELECT parent_id FROM session WHERE id = ?", (current,)).fetchone()
+        if row and row["parent_id"]:
+            current = row["parent_id"]
+        else:
+            break
+    return current
+
+
+def extract_title_and_agent(db: sqlite3.Connection, session_id: str) -> tuple:
+    """Return (title, agent) for a session. Agent from the first user message."""
+    row = db.execute("SELECT title FROM session WHERE id = ?", (session_id,)).fetchone()
+    title = row["title"] if row and row["title"] else ""
+
+    agent = ""
+    msg_rows = db.execute(
+        """SELECT data FROM message
+           WHERE session_id = ?
+           ORDER BY time_created""",
+        (session_id,),
+    ).fetchall()
+
+    for mr in msg_rows:
+        mdata = json.loads(mr["data"])
+        if mdata.get("role") == "user":
+            agent = mdata.get("agent", "")
+            break
+
+    return title, agent
+
+
+# ── file_lines (MAP {path: line_count or null}) ──────────────────────────────
+
+def extract_file_lines(db: sqlite3.Connection, session_id: str) -> dict[str, int | None]:
+    """Collect all file paths targeted by read/bash calls and resolve line counts."""
+    rows = db.execute(
+        """SELECT p.data FROM part p
+           WHERE p.session_id = ? AND json_extract(p.data, '$.type') = 'tool'
+           ORDER BY p.time_created""",
+        (session_id,),
+    ).fetchall()
+
+    paths: dict[str, int] = {}  # path -> order seen (to preserve insertion order)
+    order = 0
+
+    for r in rows:
+        data = json.loads(r["data"])
+        tool = data.get("tool", "")
+        state = data.get("state", {})
+        if not isinstance(state, dict):
+            continue
+        inp = state.get("input", {})
+        if not isinstance(inp, dict):
+            continue
+
+        fpath = None
+        if tool in ("read", "edit", "write"):
+            fpath = inp.get("filePath")
+        elif tool in ("bash", "interactive_bash"):
+            cmd = inp.get("command") or inp.get("tmux_command") or ""
+            for m in _PATH_RE.findall(cmd):
+                fpath = m
+                break
+
+        if fpath and isinstance(fpath, str) and fpath not in paths:
+            paths[fpath] = order
+            order += 1
+
+    result: dict[str, int | None] = {}
+    for p in sorted(paths, key=paths.get):
+        result[p] = count_lines(p)
+
+    return result
+
+
+# ── calls (landed is BOOLEAN per prototype) ─────────────────────────────────
+
+def extract_calls(db: sqlite3.Connection, session_id: str) -> list[dict[str, Any]]:
+    rows = db.execute(
+        """SELECT p.data FROM part p
+           WHERE p.session_id = ? AND json_extract(p.data, '$.type') = 'tool'
+           ORDER BY p.time_created""",
+        (session_id,),
+    ).fetchall()
+
+    calls: list[dict[str, Any]] = []
+
+    for r in rows:
+        data = json.loads(r["data"])
+        tool = data.get("tool", "")
+        state = data.get("state", {})
+        if not isinstance(state, dict):
+            continue
+
+        inp = state.get("input", {})
+        if not isinstance(inp, dict):
+            inp = {}
+
+        md = state.get("metadata", {})
+        if not isinstance(md, dict):
+            md = {}
+
+        ts_raw = state.get("time", {})
+        ts = ts_raw.get("start", 0) if isinstance(ts_raw, dict) else 0
+
+        # prototype:44-45
+        landed = (
+            tool in ("edit", "write", "patch") and bool(md.get("diff"))
+        ) or (
+            tool == "bash"
+            and "git commit" in (inp.get("command") or "")
+            and md.get("exit") == 0
+        )
+
+        calls.append({
+            "t": ts,
+            "tool": tool,
+            "args": scrub_args(inp),
+            "landed": landed,
+        })
+
+    return calls
+
+
+# ── assembly ────────────────────────────────────────────────────────────────
+
+def build_fixture() -> list[dict[str, Any]]:
+    if not DB_PATH.exists():
+        print(f"Error: DB not found at {DB_PATH}", file=sys.stderr)
+        sys.exit(1)
+
+    db = sqlite3.connect(str(DB_PATH))
+    db.row_factory = sqlite3.Row
+
+    fixture: list[dict[str, Any]] = []
+    errors: list[str] = []
+
+    for session_id, label in LABELS.items():
+        row = db.execute("SELECT id, parent_id FROM session WHERE id = ?", (session_id,)).fetchone()
+        if not row:
+            errors.append(f"Session not found in DB: {session_id}")
+            continue
+
+        title, agent = extract_title_and_agent(db, session_id)
+        root_id = extract_root_id(db, session_id)
+        file_lines = extract_file_lines(db, session_id)
+        calls = extract_calls(db, session_id)
+
+        fixture.append({
+            "session_id": session_id,
+            "label": label,
+            "root_id": root_id,
+            "title": title,
+            "agent": agent,
+            "file_lines": file_lines,
+            "calls": calls,
+        })
+
+    db.close()
+
+    if errors:
+        for err in errors:
+            print(err, file=sys.stderr)
+        if not fixture:
+            sys.exit(1)
+
+    return fixture
+
+
+def main():
+    fixture = build_fixture()
+    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with open(OUTPUT_PATH, "w") as f:
+        json.dump(fixture, f, indent=None, separators=(",", ":"), ensure_ascii=False)
+
+    total_calls = sum(len(e["calls"]) for e in fixture)
+    total_fl = sum(len(e["file_lines"]) for e in fixture)
+    size_kb = OUTPUT_PATH.stat().st_size / 1024
+    mf = sum(1 for e in fixture if e["label"] == "must_flag")
+    mn = sum(1 for e in fixture if e["label"] == "must_not_flag")
+
+    print(f"Exported {len(fixture)} sessions ({mf} must_flag, {mn} must_not_flag)")
+    print(f"  Total tool calls: {total_calls}")
+    print(f"  Total file_lines entries: {total_fl}")
+    print(f"  File size: {size_kb:.1f} KB")
+    print(f"  Output: {OUTPUT_PATH}")
+
+    print(f"\n{'session':36s} {'label':15s} {'title':60s} {'agent':15s} {'calls':>6s} {'landed':>7s} fl_entries")
+    print("-" * 150)
+    for e in fixture:
+        sid = e["session_id"][:35]
+        landed_true = sum(1 for c in e["calls"] if c["landed"])
+        nonnull_fl = sum(1 for v in e["file_lines"].values() if v is not None)
+        print(f"{sid:36s} {e['label']:15s} {e['title'][:60]:60s} {e['agent']:15s} {len(e['calls']):6d} {landed_true:7d} {len(e['file_lines']):3d} ({nonnull_fl} with count)")
+
+
+if __name__ == "__main__":
+    main()

scripts/progress_backtest.py

@@ -0,0 +1,205 @@
+#!/usr/bin/env python
+"""Progress-detection backtest: replay sessions through the detector.
+
+Reads sessions from a fixture JSON file (default ~/.local/share/opencode/rc-
+servers.json for live mode, or --fixture <path>) and runs
+progress_detect.evaluate() over a sliding window with step 5.
+
+Output per session:
+    FLAG|ok  calls=NNNN first_flag_at=N dup=0.00 top=N landed=0 ro=N | session_title
+
+Usage:
+    # Live mode — reads sessions from the opencode rc-servers.json
+    PYTHONPATH=src python -m scripts.progress_backtest
+
+    # Fixture mode
+    PYTHONPATH=src python -m scripts.progress_backtest --fixture /path/to/fixture.json
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+
+from progress_detect import DetectConfig, evaluate
+
+_DEFAULT_CFG = DetectConfig()
+_FIXTURE_PATH = os.path.join(
+    os.path.dirname(__file__), "..", "tests", "fixtures", "progress", "fixture.json"
+)
+_RC_SERVERS = os.path.expanduser("~/.local/share/opencode/rc-servers.json")
+
+
+def _calls_from_fixture(session):
+    """Convert a fixture session dict to call tuples for evaluate()."""
+    return [
+        (c["t"], c["tool"], json.dumps(c["args"]), c["landed"])
+        for c in session["calls"]
+    ]
+
+
+def _file_lines_fn(file_lines_dict):
+    """Return a callable for coverage() from a file_path→line_count dict."""
+    def fn(fp):
+        if fp is None:
+            return 0
+        return file_lines_dict.get(fp, 0) or 0
+    return fn
+
+
+def _backtest_session(session, cfg):
+    """Run sliding-window backtest on one session. Returns (first_flag_at, reason_dict)."""
+    calls = _calls_from_fixture(session)
+    fl = session.get("file_lines", {})
+    fl_fn = _file_lines_fn(fl)
+    title = session.get("title", "")
+
+    if len(calls) < cfg.min_calls:
+        return (None, None)
+
+    for end_idx in range(cfg.min_calls, len(calls) + 1, 5):
+        flag, reason = evaluate(calls, cfg, fl_fn, title, end_idx=end_idx)
+        if flag and reason:
+            return (end_idx, reason)
+    return (None, None)
+
+
+def _format_line(session, first_flag_at, reason):
+    """Format one backtest output line.
+
+    Pattern: FLAG|ok  calls=NNNN first_flag_at=N dup=0.00 top=N landed=ro=N | title
+    """
+    calls_n = len(session["calls"])
+    title = session.get("title", "")
+
+    if reason is not None:
+        dup = reason.get("dup", 0.0)
+        top = reason.get("top", 0)
+        landed = 1 if reason.get("landed") else 0
+        ro = 1 if reason.get("ro") else 0
+        first_flag = first_flag_at
+        flags = "FLAG  "
+    else:
+        dup = 0.0
+        top = 0
+        landed = 0
+        ro = 0
+        first_flag = "N/A"
+        flags = "ok    "
+
+    return (
+        f"{flags}calls={calls_n:4d} first_flag_at={first_flag}"
+        f" dup={dup:.2f} top={top}"
+        f" landed={landed} ro={ro}"
+        f" | {title}"
+    )
+
+
+def _load_fixture(path):
+    """Load a fixture JSON file and return list of session dicts."""
+    with open(path) as f:
+        return json.load(f)
+
+
+def _load_live_sessions():
+    """Load sessions from the rc-servers.json opencode API file."""
+    if not os.path.isfile(_RC_SERVERS):
+        print(f"Error: {_RC_SERVERS} not found. Use --fixture instead.", file=sys.stderr)
+        sys.exit(1)
+
+    with open(_RC_SERVERS) as f:
+        data = json.load(f)
+
+    sessions = []
+    # rc-servers.json structure: {<session_id>: {<agent>: {calls: [...]}}}
+    for session_id, agents in data.items():
+        for agent_name, agent_data in agents.items():
+            calls = agent_data.get("calls", [])
+            sessions.append({
+                "session_id": session_id,
+                "title": agent_data.get("title", f"{agent_name} session"),
+                "agent": agent_name,
+                "calls": calls,
+                "file_lines": {},
+                "label": "unknown",
+            })
+    return sessions
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Progress-detection backtest: replay sessions through the detector."
+    )
+    parser.add_argument(
+        "--fixture",
+        default=None,
+        help="Path to fixture JSON file (default: live mode from rc-servers.json)",
+    )
+    parser.add_argument(
+        "--window",
+        type=int,
+        default=60,
+        help="Sliding window size in calls (default: 60)",
+    )
+    parser.add_argument(
+        "--min-calls",
+        type=int,
+        default=40,
+        help="Minimum calls before evaluation (default: 40)",
+    )
+    parser.add_argument(
+        "--step",
+        type=int,
+        default=5,
+        help="Sliding window step (default: 5)",
+    )
+    parser.add_argument(
+        "--output",
+        type=str,
+        default=None,
+        help="Write output to file instead of stdout",
+    )
+    args = parser.parse_args()
+
+    cfg = DetectConfig(
+        window=args.window,
+        min_calls=args.min_calls,
+        read_only_agent_keywords=("explore", "librarian", "oracle"),
+    )
+
+    if args.fixture:
+        fixture_path = args.fixture
+    else:
+        fixture_path = _FIXTURE_PATH
+
+    if not os.path.isfile(fixture_path):
+        print(f"Error: fixture not found at {fixture_path}", file=sys.stderr)
+        sys.exit(1)
+
+    sessions = _load_fixture(fixture_path)
+
+    results = []
+    for session in sessions:
+        first_flag_at, reason = _backtest_session(session, cfg)
+        line = _format_line(session, first_flag_at, reason)
+        results.append(line)
+
+    output = "\n".join(results) + "\n"
+
+    if args.output:
+        os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True)
+        with open(args.output, "w") as f:
+            f.write(output)
+    else:
+        sys.stdout.write(output)
+
+    # Print summary
+    flagged = sum(1 for r in results if r.startswith("FLAG"))
+    total = len(results)
+    print(f"\n# backtest complete: {flagged}/{total} flagged", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

src/admin.py

@@ -518,6 +518,16 @@ _CONFIG_ALLOWLIST: dict[str, tuple[str, ...]] = {
     "pinch.relevance.enabled": ("pinch", "relevance", "enabled"),
     "routing.default_flex_preference": ("routing", "default_flex_preference"),
     "routing.default_profile": ("routing", "default_profile"),
+    # Watchdog — persisted config knobs for the model-response watchdog.
+    "watchdog.enabled": ("watchdog", "enabled"),
+    "watchdog.local_llm_enabled": ("watchdog", "local_llm_enabled"),
+    "watchdog.detector.window": ("watchdog", "detector", "window"),
+    "watchdog.detector.dup_min": ("watchdog", "detector", "dup_min"),
+    "watchdog.detector.top_min": ("watchdog", "detector", "top_min"),
+    "watchdog.detector.top_min_ro": ("watchdog", "detector", "top_min_ro"),
+    "watchdog.detector.cum_min": ("watchdog", "detector", "cum_min"),
+    "watchdog.detector.cover_min": ("watchdog", "detector", "cover_min"),
+    "watchdog.detector.min_calls": ("watchdog", "detector", "min_calls"),
 }
 
 # Order preserves config.yaml layout for the GET response.
@@ -538,6 +548,15 @@ _CONFIG_GET_ORDER: list[str] = [
     "pinch.relevance.enabled",
     "routing.default_flex_preference",
     "routing.default_profile",
+    "watchdog.enabled",
+    "watchdog.local_llm_enabled",
+    "watchdog.detector.window",
+    "watchdog.detector.dup_min",
+    "watchdog.detector.top_min",
+    "watchdog.detector.top_min_ro",
+    "watchdog.detector.cum_min",
+    "watchdog.detector.cover_min",
+    "watchdog.detector.min_calls",
 ]
 
 
@@ -2006,7 +2025,7 @@ def build_router(
         """
         # Validate availability value (Pydantic only typed as str; keep explicit
         # allow-list so unknown values get a clear 422).
-        valid_avail = {"active", "deprecated", "stale"}
+        valid_avail = {"active", "blocked", "deprecated", "stale"}
         avail_val = body.availability
         reason_val = body.reason
         if avail_val not in valid_avail:
@@ -2050,14 +2069,16 @@ def build_router(
             for r in model_rows:
                 if r["model_id"] == model_id and r["provider"] == provider:
                     r["availability"] = avail_val
+                    if avail_val == "blocked":
+                        r["blocked"] = 1
                     if avail_val == "deprecated":
                         r["deprecated"] = 1
                     break
 
             exclude_set = metrics._admin_excluded_models(conn)
-            # "stale" excludes as well as "deprecated"; see
+            # "stale" and "blocked" exclude as well as "deprecated"; see
             # dispatcher._admin_excluded_models for why.
-            if avail_val in ("deprecated", "stale"):
+            if avail_val in ("blocked", "deprecated", "stale"):
                 exclude_set = exclude_set | {model_id}
             ctx = metrics.context_ceilings_with_rows(
                 conn, cfg, model_rows, exclude_models=exclude_set
@@ -2784,4 +2805,125 @@ def build_router(
             "message": "A restart is required for this change to take effect",
         }
 
+    # ── watchdog API ──────────────────────────────────────────────────────
+
+    @router.get("/api/watchdog/status")
+    def _watchdog_status():
+        conn = _db_callable()
+        try:
+            tick = conn.execute(
+                "SELECT * FROM watchdog_ticks ORDER BY id DESC LIMIT 1"
+            ).fetchone()
+            verdicts = conn.execute(
+                "SELECT COUNT(*) as total, SUM(flagged) as flagged "
+                "FROM watchdog_verdicts WHERE tick_id = ?",
+                (tick["id"],) if tick else (),
+            ).fetchone() if tick else None
+            alerts = conn.execute(
+                "SELECT COUNT(*) as open_count "
+                "FROM watchdog_alerts WHERE resolved_at IS NULL"
+            ).fetchone()
+            return {
+                "last_tick": dict(tick) if tick else None,
+                "verdicts": dict(verdicts) if verdicts else None,
+                "open_alerts": alerts["open_count"] if alerts else 0,
+            }
+        finally:
+            conn.close()
+
+    @router.get("/api/watchdog/loops")
+    def _watchdog_loops():
+        conn = _db_callable()
+        try:
+            rows = conn.execute(
+                "SELECT * FROM watchdog_alerts WHERE resolved_at IS NULL "
+                "ORDER BY opened_at DESC"
+            ).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    @router.get("/api/watchdog/channels")
+    def _watchdog_channels():
+        conn = _db_callable()
+        try:
+            rows = conn.execute(
+                "SELECT * FROM watchdog_channel_settings ORDER BY channel_name"
+            ).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    @router.post("/api/watchdog/channels")
+    def _watchdog_channels_update(body: dict):
+        conn = _db_callable()
+        try:
+            channel_name = body.get("channel_name")
+            if not channel_name:
+                raise HTTPException(status_code=422, detail="channel_name required")
+            enabled = body.get("enabled")
+            min_severity = body.get("min_severity")
+            if min_severity not in (None, "info", "warning", "critical"):
+                raise HTTPException(status_code=422, detail="invalid min_severity")
+            conn.execute(
+                "INSERT INTO watchdog_channel_settings (channel_name, enabled, min_severity) "
+                "VALUES (?, ?, ?) "
+                "ON CONFLICT(channel_name) DO UPDATE SET "
+                "enabled = COALESCE(?, enabled), "
+                "min_severity = COALESCE(?, min_severity)",
+                (
+                    channel_name,
+                    1 if enabled is None else (1 if enabled else 0),
+                    min_severity or "warning",
+                    1 if enabled is None else (1 if enabled else 0),
+                    min_severity,
+                ),
+            )
+            conn.commit()
+            return {"ok": True}
+        finally:
+            conn.close()
+
+    @router.post("/api/watchdog/test-alert")
+    def _watchdog_test_alert(body: dict):
+        from notifier import AlertEvent, Notifier
+        severity = body.get("severity", "warning")
+        channel_name = body.get("channel_name")
+        if severity not in ("info", "warning", "critical"):
+            raise HTTPException(status_code=422, detail="invalid severity")
+        event = AlertEvent(
+            dedup_key=f"test-{uuid.uuid4().hex[:8]}",
+            severity=severity,
+            state="trigger",
+            title="Watchdog test alert",
+            summary=f"Test alert at severity {severity}",
+        )
+        conn = _db_callable()
+        try:
+            channels = conn.execute(
+                "SELECT * FROM watchdog_channel_settings WHERE enabled = 1"
+            ).fetchall()
+            if channel_name:
+                channels = [c for c in channels if c["channel_name"] == channel_name]
+            if not channels:
+                return {"sent": 0, "message": "no enabled channels"}
+            from types import SimpleNamespace as _NS
+
+            notifier = Notifier.__new__(Notifier)
+            notifier._config = type("Obj", (), {"channels": channels})()
+            notifier._last_fired = lambda _: None
+            notifier._channels = [
+                _NS(
+                    name=c["channel_name"],
+                    type="desktop",
+                    enabled=bool(c["enabled"]),
+                    min_severity=c["min_severity"],
+                )
+                for c in channels
+            ]
+            notifier.deliver(event)
+            return {"sent": len(channels), "severity": severity}
+        finally:
+            conn.close()
+
     return router

src/config.py

@@ -1354,6 +1354,61 @@ class LoggingConfig(StrictModel):
         return v.strip().lower()
 
 
+class DetectorConfig(StrictModel):
+    """Heuristic thresholds for the watchdog's model-response anomaly detector.
+
+    Each threshold is a rough boundary tuned on the reference deployment's
+    traffic; you should expect to adjust them for your own workload.
+    """
+
+    window: int = 60
+    dup_min: float = 0.25
+    top_min: int = 12
+    top_min_ro: int = 8
+    cum_min: int = 15
+    cover_min: float = 4.0
+    min_calls: int = 40
+
+
+class ChannelConfig(StrictModel):
+    """A single notification channel that the watchdog can alert through.
+
+    The ``type`` field is deliberately restricted — unknown types fail at
+    load rather than being silently ignored.
+    """
+
+    name: str
+    type: Literal["desktop"]
+    enabled: bool = True
+    min_severity: Literal["info", "warning", "critical"] = "warning"
+
+
+class NotificationsConfig(StrictModel):
+    """Watchdog alert routing: which channels receive which severities."""
+
+    channels: list[ChannelConfig] = [
+        {"name": "default", "type": "desktop", "enabled": True, "min_severity": "warning"}
+    ]
+
+
+class WatchdogConfig(StrictModel):
+    """Runtime model-response watchdog that monitors for anomalous behavior.
+
+    Watches the live decision stream for duplicate/truncated/repetitive
+    output patterns and sends desktop notifications when thresholds are
+    crossed.
+    """
+
+    enabled: bool = True
+    local_llm_enabled: bool = True
+    # Defaults to verification.model at runtime when None.
+    model: str | None = None
+    read_only_agents: list[str] = ["explore", "librarian", "oracle"]
+    detector: DetectorConfig = DetectorConfig()
+    # Base URL for alert links back to the admin dashboard.
+    dashboard_base_url: str = "http://127.0.0.1:8080/admin"
+
+
 def _is_loopback_host(raw_url: str) -> bool:
     """Return True if the URL's hostname is a loopback address.
 
@@ -1398,6 +1453,8 @@ class RouterConfig(StrictModel):
     local_energy: LocalEnergyConfig = LocalEnergyConfig()
     local_dispatch_models: list[LocalDispatchModel] = []
     profiles: dict[str, RoutingProfile] = {}
+    watchdog: WatchdogConfig = WatchdogConfig()
+    notifications: NotificationsConfig = NotificationsConfig()
 
     @model_validator(mode="after")
     def local_energy_needs_tariff_when_enabled(self) -> "RouterConfig":

src/dispatcher.py

@@ -69,6 +69,7 @@ import session_cache
 import textcut
 from capabilities import detect_capabilities, iter_image_url_values
 from config import BUILTIN_PROFILES, FlexPreference, RouterConfig, RoutingProfile, load_config
+from watchdog_store import ensure_watchdog_tables
 from conversation_identity import parse_conversation_id, resolve_identity
 from context_prune import (
     _text_only,
@@ -550,6 +551,7 @@ def _ensure_tables() -> None:
             _ensure_energy_observations_table(conn)
             local_energy.ensure_local_energy_table(conn)
             admin.ensure_admin_tables(conn)
+            ensure_watchdog_tables(conn)
         finally:
             conn.close()
     except Exception as exc:  # noqa: BLE001
@@ -3406,7 +3408,7 @@ def _admin_excluded_models(conn: sqlite3.Connection) -> set[str]:
             row["model_id"]
             for row in conn.execute(
                 "SELECT model_id FROM admin_model_overrides "
-                "WHERE availability IN ('deprecated', 'stale')"
+                "WHERE availability IN ('deprecated', 'stale', 'blocked')"
             )
         }
     except sqlite3.OperationalError:

src/metrics.py

@@ -1496,8 +1496,8 @@ def _admin_excluded_models(conn: sqlite3.Connection) -> set[str]:
         return {
             row["model_id"]
             for row in conn.execute(
-                "SELECT model_id FROM admin_model_overrides "
+            "SELECT model_id FROM admin_model_overrides "
-                "WHERE availability IN ('deprecated', 'stale')"
+            "WHERE availability IN ('deprecated', 'stale', 'blocked')"
             )
         }
     except sqlite3.OperationalError:

src/notifier.py

@@ -0,0 +1,169 @@
+"""Desktop notification channel with PagerDuty Events v2 shape.
+
+Sends desktop alerts via ``notify-send`` for watchdog-detected anomalies.
+Each channel type in the config gets its own dispatch path; currently only
+``desktop`` is implemented. The module is designed to be importable without
+``DISPLAY`` or ``notify-send`` installed — delivery errors are logged, never
+raised.
+"""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Final
+
+from config import ChannelConfig, NotificationsConfig
+
+logger = logging.getLogger(__name__)
+
+# ── constants ──────────────────────────────────────────────────────────
+
+RATE_LIMIT_SECONDS: Final[int] = 300
+
+# ── data classes ───────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class AlertEvent:
+    """A single alert matching the PagerDuty Events v2 shape.
+
+    Fields mirror the PagerDuty v2 envelope: *dedup_key* for deduplication,
+    *severity* (info / warning / critical), *state* (trigger / escalate /
+    resolve), *title*, *summary*, optional *details*, and a *source*
+    identifier defaulting to ``6krrt-watchdog``.
+    """
+
+    dedup_key: str
+    severity: str  # "info" | "warning" | "critical"
+    state: str  # "trigger" | "escalate" | "resolve"
+    title: str
+    summary: str
+    details: str | None = None
+    source: str = "6krrt-watchdog"
+
+
+# ── notifier ────────────────────────────────────────────────────────────
+
+_SEVERITY_ORDER: dict[str, int] = {
+    "info": 0,
+    "warning": 1,
+    "critical": 2,
+}
+
+_URGENCY_MAP: dict[str, str] = {
+    "critical": "critical",
+    "warning": "normal",
+    "info": "low",
+}
+
+
+class Notifier:
+    """Routes alerts to configured notification channels.
+
+    Each channel is checked against the event's severity via the channel's
+    ``min_severity`` threshold. Rate limiting uses an injected
+    ``last_fired`` callable so the caller (e.g. the watchdog) can back it
+    with persistent state.
+    """
+
+    def __init__(
+        self,
+        config: NotificationsConfig,
+        last_fired: Callable[[str], float | None] = lambda _: None,
+    ) -> None:
+        self._config = config
+        self._last_fired = last_fired
+        # Normalise channels: Pydantic may not coerce dict defaults into
+        # ChannelConfig objects when ``from __future__ import annotations``
+        # is active and ``model_rebuild()`` has not been called.
+        self._channels = [
+            ChannelConfig(**ch) if isinstance(ch, dict) else ch
+            for ch in config.channels
+        ]
+
+    # ── public API ─────────────────────────────────────────────────────
+
+    def deliver(self, event: AlertEvent) -> None:
+        """Deliver *event* through every matching channel."""
+        for channel in self._channels:
+            if not channel.enabled:
+                continue
+            if not self._passes_severity_gate(channel, event):
+                continue
+            if not self._passes_rate_limit(channel, event):
+                continue
+            self._dispatch(channel, event)
+
+    # ── helpers ────────────────────────────────────────────────────────
+
+    @staticmethod
+    def _passes_severity_gate(
+        channel: ChannelConfig, event: AlertEvent
+    ) -> bool:
+        """Return True when the event severity meets the channel threshold."""
+        required = _SEVERITY_ORDER.get(channel.min_severity, 0)
+        actual = _SEVERITY_ORDER.get(event.severity, 0)
+        return actual >= required
+
+    def _passes_rate_limit(
+        self, channel: ChannelConfig, event: AlertEvent
+    ) -> bool:
+        """Return True when the event may fire without hitting the rate
+        limit.
+
+        Resolve events always bypass the rate window — they are the
+        mechanism for clearing a previous alert and must not be suppressed.
+        """
+        if event.state == "resolve":
+            return True
+        last = self._last_fired(channel.name)
+        if last is None:
+            return True
+        return (time.monotonic() - last) >= RATE_LIMIT_SECONDS
+
+    def _dispatch(self, channel: ChannelConfig, event: AlertEvent) -> None:
+        """Send *event* through *channel*."""
+        if channel.type == "desktop":
+            self._desktop_notify(event)
+        else:
+            logger.warning(
+                "notifier: unknown channel type %r for channel %r — skipped",
+                channel.type,
+                channel.name,
+            )
+
+    def _desktop_notify(self, event: AlertEvent) -> None:
+        """Send a desktop notification via ``notify-send``."""
+        urgency = _URGENCY_MAP.get(event.severity, "normal")
+        try:
+            subprocess.run(
+                ["notify-send", "-u", urgency, event.title, event.summary],
+                check=False,
+                timeout=5,
+            )
+        except FileNotFoundError:
+            logger.warning(
+                "notifier: notify-send not installed — desktop alert "
+                "dropped (dedup_key=%r, title=%r)",
+                event.dedup_key,
+                event.title,
+            )
+        except OSError as exc:
+            logger.warning(
+                "notifier: OS error sending desktop alert "
+                "(dedup_key=%r, title=%r): %s",
+                event.dedup_key,
+                event.title,
+                exc,
+            )
+        except subprocess.TimeoutExpired:
+            logger.warning(
+                "notifier: notify-send timed out after 5s "
+                "(dedup_key=%r, title=%r)",
+                event.dedup_key,
+                event.title,
+            )

src/progress_detect.py

@@ -0,0 +1,183 @@
+"""Progress detection"""
+
+from __future__ import annotations
+
+import collections
+import json
+import os
+import re
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass
+
+_PATHISH = re.compile(r"[\w./-]+\.(?:json|md|py|js|html|yaml|yml|toml|txt|sql|sh|mjs|css)\b")
+
+_CD_PREFIX = re.compile(r"^\s*cd\s+\S+\s*(?:&&|;)\s*")
+_ENV_PREFIX = re.compile(r"^\s*(?:[A-Z_][A-Z0-9_]*=\S+\s+)+")
+
+
+def _normalize_bash(cmd: str) -> str:
+    lines = []
+    for ln in (cmd or "").splitlines():
+        stripped = ln.lstrip()
+        if stripped.startswith("#"):
+            # Strip leading '#' from comments so distinct comments produce
+            # distinct normalization keys instead of all collapsing to "#".
+            lines.append(stripped[1:])
+        else:
+            lines.append(ln)
+    c = " ".join(lines).strip()
+    for _ in range(3):
+        c2 = _ENV_PREFIX.sub("", _CD_PREFIX.sub("", c))
+        if c2 == c:
+            break
+        c = c2
+    return c
+
+
+@dataclass(frozen=True)
+class DetectConfig:
+    """Tunable thresholds for progress-detection signals.
+
+    window : int
+        Sliding window in number of calls (not seconds). Last N tool calls.
+    min_calls : int
+        Minimum total calls before evaluation produces a signal.
+    """
+    window: int = 60
+    dup_min: float = 0.25
+    top_min: int = 12
+    top_min_ro: int = 8
+    cum_min: int = 15
+    cover_min: float = 4.0
+    min_calls: int = 40
+    read_only_agent_keywords: tuple[str, ...] = ()
+
+
+def is_ro(title: str, keywords: tuple[str, ...]) -> bool:
+    t = (title or "").lower()
+    return any(k.lower() in t for k in keywords)
+
+
+def target_of(tool: str, args_json: str) -> tuple:
+    try:
+        a = json.loads(args_json)
+    except Exception:  # noqa: BLE001
+        return (tool, args_json[:80])
+    if a.get("filePath"):
+        if tool == "read":
+            return ("file", os.path.basename(a["filePath"]), a.get("offset"), a.get("limit"))
+        return ("file", os.path.basename(a["filePath"]))
+    if tool == "bash":
+        norm_cmd = _normalize_bash(a.get("command") or "")
+        files = sorted({os.path.basename(f) for f in _PATHISH.findall(norm_cmd)})
+        if files:
+            nums = tuple(sorted(set(re.findall(r"\b\d+\b", a.get("command") or ""))))
+            return ("bash-files", ",".join(files[:3]), nums)
+        return ("bash", " ".join(norm_cmd.split()[:2]))
+    if tool in ("grep", "glob"):
+        return (tool, str(a.get("pattern"))[:60])
+    return (tool, args_json[:80])
+
+
+def _canonical_args(args_json: str) -> str:
+    """Parse args_json as JSON and re-dump with sort_keys for canonical comparison.
+
+    Falls back to raw args_json if JSON parsing fails.
+    """
+    try:
+        a = json.loads(args_json)
+        return json.dumps(a, sort_keys=True)
+    except Exception:  # noqa: BLE001
+        return args_json
+
+
+def window_stats(calls, n: int) -> tuple[float, int, int]:
+    win = calls[-n:] if n > 0 else calls
+    fp = collections.Counter((c[1], _canonical_args(c[2])) for c in win)
+    dup = sum(v - 1 for v in fp.values() if v > 1) / max(1, len(win))
+    tg = collections.Counter(target_of(c[1], c[2]) for c in win)
+    top_n = tg.most_common(1)[0][1] if tg else 0
+    all_fp = collections.Counter((c[1], _canonical_args(c[2])) for c in calls)
+    slow_count = all_fp.most_common(1)[0][1] if all_fp else 0
+    return dup, top_n, slow_count
+
+
+def coverage(calls, n: int, file_lines: Callable[[str], int]) -> tuple[float, str]:
+    win = calls[-n:] if n > 0 else calls
+    req = collections.Counter()
+    for c in win:
+        if c[1] != "read":
+            continue
+        try:
+            a = json.loads(c[2])
+        except json.JSONDecodeError:
+            continue
+        fp = a.get("filePath")
+        nlines = file_lines(fp) if fp else None
+        if not nlines:
+            continue
+        req[fp] += min(a.get("limit") or nlines, nlines)
+    best = max(((v / file_lines(k), k) for k, v in req.items()), default=(0.0, ""))
+    return round(best[0], 1), os.path.basename(best[1])
+
+
+def tree_landed(calls, n: int) -> bool:
+    win = calls[-n:] if n > 0 else calls
+    return any(c[3] for c in win)
+
+
+def evaluate(
+    calls,
+    config: DetectConfig,
+    file_lines: Callable[[str], int],
+    title: str,
+    end_idx: int | None = None,
+    extra_landed_times: Iterable[float] = (),
+) -> tuple[bool, dict | None]:
+    """Evaluate session for no-progress pattern.
+
+    Call-count-based window: last config.window calls from calls[:end_idx].
+    landed is time-based (session-wide check within window's time span) plus
+    any extra_landed_times inside [t0, t1] (e.g. descendant landed calls).
+    """
+    subset = calls if end_idx is None else calls[:end_idx]
+    if len(subset) < config.min_calls:
+        return (False, None)
+    win = subset[-config.window:]
+    if len(win) < config.min_calls:
+        return (False, None)
+    t0, t1 = win[0][0], win[-1][0]
+    landed = any(t0 <= c[0] <= t1 and c[3] for c in calls) or any(
+        t0 <= t <= t1 for t in extra_landed_times
+    )
+    dup, top_n, slow_count = window_stats(subset, config.window)
+    ro = is_ro(title, config.read_only_agent_keywords)
+    cov_val, cov_file = coverage(subset, config.window, file_lines)
+    slow = (not landed) and slow_count >= config.cum_min
+    reread = (not landed) and cov_val >= config.cover_min
+    tg = collections.Counter(target_of(c[1], c[2]) for c in win)
+    top_k = tg.most_common(1)[0][0] if tg else ("", "")
+    if reread and not (top_n >= config.top_min):
+        top_n = int(cov_val)
+        top_k = ("coverage", f"{cov_file} read {cov_val}x its length")
+    if ro:
+        flag = top_n >= config.top_min_ro or slow or reread
+    else:
+        flag = ((not landed) and (dup >= config.dup_min or top_n >= config.top_min)) or slow or reread
+    if slow and not (top_n >= (config.top_min_ro if ro else config.top_min)):
+        top_n = slow_count
+        top_k = ("exact-total",)
+    if flag:
+        reason = {
+            "flag": True,
+            "dup": round(dup, 2),
+            "top": top_n,
+            "top_what": " ".join(str(x) for x in top_k)[:120],
+            "landed": landed,
+            "ro": ro,
+            "n": len(subset),
+            "t1": win[-1][0] if win else 0.0,
+            "coverage": cov_val,
+        }
+        return (True, reason)
+    return (False, None)

src/watchdog.py

@@ -0,0 +1,846 @@
+"""Watchdog orchestrator — polls opencode sessions, detects stalls, alerts."""
+from __future__ import annotations
+
+import argparse
+import collections
+import fcntl
+import json
+import logging
+import os
+import sqlite3
+import sys
+import time
+import urllib.error
+import urllib.request
+from collections.abc import Callable
+from datetime import datetime, timezone
+from typing import Any
+
+from notifier import AlertEvent, Notifier
+from progress_detect import DetectConfig, evaluate
+from watchdog_store import ensure_watchdog_tables
+
+logger = logging.getLogger(__name__)
+
+RC_SERVERS_PATH = os.path.expanduser(
+    "~/.local/share/opencode/rc-servers.json"
+)
+_LOCK_NAME = ".watchdog.lock"
+_OPCODE_TIMEOUT = 10
+_OLLAMA_TIMEOUT = 15
+_MAX_LLM = 3
+_FIRST_RUN_BACKOFF = 600
+# After this many flagged ticks (each tick is 5 min) without resolve,
+# escalate from warning to critical.  3 ticks = 15 min.
+_ESCALATE_THRESHOLD = 3
+
+
+def detect_config_from_pydantic(pc: Any) -> DetectConfig:
+    return DetectConfig(
+        window=pc.window,
+        dup_min=pc.dup_min,
+        top_min=pc.top_min,
+        top_min_ro=pc.top_min_ro,
+        cum_min=pc.cum_min,
+        cover_min=pc.cover_min,
+        min_calls=pc.min_calls,
+        read_only_agent_keywords=tuple(
+            getattr(pc, "_ro_agents", ()) or ()
+        ),
+    )
+
+
+def _http_get(url: str, timeout: float = _OPCODE_TIMEOUT) -> Any:
+    resp = urllib.request.urlopen(
+        urllib.request.Request(url), timeout=timeout
+    )
+    return json.loads(resp.read().decode())
+
+
+def calls_of(
+    server_url: str, session_id: str, http_get=_http_get,
+) -> list[tuple]:
+    """Fetch ``(t, tool, args_json, landed)`` tuples for one session."""
+    url = f"{server_url.rstrip('/')}/session/{session_id}/message"
+    try:
+        messages = http_get(url)
+    except (urllib.error.HTTPError, urllib.error.URLError, OSError):
+        logger.warning("calls_of: fetch failed for %s", session_id)
+        return []
+    except json.JSONDecodeError:
+        return []
+
+    out: list[tuple] = []
+    for m in messages:
+        for p in m.get("parts") or []:
+            if p.get("type") != "tool":
+                continue
+            st = p.get("state") or {}
+            md = st.get("metadata") or {}
+            inp = st.get("input") or {}
+            t = (
+                (st.get("time") or {}).get("start")
+                or m.get("info", {}).get("time", {}).get("created", 0)
+            )
+            tool = p.get("tool")
+            if tool in ("edit", "write", "patch") and md.get("diff"):
+                landed = True
+            elif tool == "bash" and isinstance(inp, dict):
+                cmd = inp.get("command", "")
+                landed = md.get("exit") == 0 and isinstance(cmd, str) and "git commit" in cmd
+            else:
+                landed = False
+            try:
+                args_json = json.dumps(inp, sort_keys=True)
+            except (TypeError, ValueError):
+                args_json = json.dumps(str(inp), sort_keys=True)
+            out.append((t, tool, args_json, landed))
+    return out
+
+
+def resolve_roots(
+    sessions: list[dict],
+) -> tuple[dict[str, str], dict[str, list[str]]]:
+    """Map each session to its root and build reverse child index."""
+    id_to_s: dict[str, dict] = {}
+    kids: dict[str, list[str]] = collections.defaultdict(list)
+    for s in sessions:
+        sid = s.get("id", "")
+        pid = s.get("parentID")
+        if sid:
+            id_to_s[sid] = s
+            if pid:
+                kids[pid].append(sid)
+
+    root_cache: dict[str, str] = {}
+
+    def _resolve(sid: str) -> str:
+        if sid in root_cache:
+            return root_cache[sid]
+        s = id_to_s.get(sid)
+        if not s:
+            root_cache[sid] = sid
+            return sid
+        pid = s.get("parentID")
+        if not pid:
+            root_cache[sid] = sid
+            return sid
+        root = _resolve(pid)
+        root_cache[sid] = root
+        return root
+
+    session_root: dict[str, str] = {}
+    for sid in id_to_s:
+        session_root[sid] = _resolve(sid)
+
+    children: dict[str, list[str]] = collections.defaultdict(list)
+    for sid, root in session_root.items():
+        children[root].append(sid)
+    for v in children.values():
+        v.sort()
+    return session_root, dict(children)
+
+
+def _fetch_sessions(
+    server_url: str, http_get: Callable[[str], Any],
+) -> list[dict] | None:
+    """Probe one opencode server; returns session list or None."""
+    base = server_url.rstrip("/")
+    try:
+        raw = http_get(f"{base}/session")
+        if isinstance(raw, list):
+            return raw
+        if isinstance(raw, dict) and "sessions" in raw:
+            return raw["sessions"]
+    except (urllib.error.HTTPError, urllib.error.URLError, OSError,
+            json.JSONDecodeError, KeyError):
+        logger.warning("probe failed for %s", base)
+    return None
+
+
+def _read_rc_servers(
+    http_get: Callable[[str], Any] = _http_get,
+) -> list[str]:
+    """Read rc-servers.json; returns list of serverUrl values."""
+    if not os.path.exists(RC_SERVERS_PATH):
+        return []
+    try:
+        with open(RC_SERVERS_PATH) as f:
+            data = json.loads(f.read())
+    except (json.JSONDecodeError, OSError):
+        return []
+
+    urls: list[str] = []
+    # Handles both nested {"servers": [...]} and flat path-keyed format.
+    if "servers" in data and isinstance(data["servers"], list):
+        for entry in data["servers"]:
+            if isinstance(entry, dict):
+                u = entry.get("serverUrl", "")
+                if u:
+                    urls.append(u.rstrip("/"))
+    else:
+        for entry in data.values():
+            if isinstance(entry, dict):
+                u = entry.get("serverUrl", "")
+                if u:
+                    urls.append(u.rstrip("/"))
+    # Deduplicate preserving order.
+    seen: set = set()
+    out: list[str] = []
+    for u in urls:
+        if u not in seen:
+            seen.add(u)
+            out.append(u)
+    return out
+
+
+def _db_dir(conn: sqlite3.Connection) -> str:
+    try:
+        path = conn.execute("PRAGMA database_list").fetchone()[2]
+        return os.path.dirname(path) if path else "."
+    except Exception:  # noqa: BLE001
+        return "."
+
+
+def _acquire_lock(conn: sqlite3.Connection,
+                  db_dir: str | None = None) -> int | None:
+    if db_dir is None:
+        db_dir = _db_dir(conn)
+    lock = os.path.join(db_dir, _LOCK_NAME)
+    try:
+        fd = os.open(lock, os.O_CREAT | os.O_RDWR)
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        return fd
+    except (OSError, BlockingIOError):
+        return None
+
+
+def _last_tick_ts(conn: sqlite3.Connection) -> float:
+    row = conn.execute(
+        "SELECT ticked_at FROM watchdog_ticks ORDER BY id DESC LIMIT 1"
+    ).fetchone()
+    if row:
+        try:
+            return datetime.fromisoformat(row[0]).timestamp()
+        except (ValueError, TypeError):
+            pass
+    return time.time() - _FIRST_RUN_BACKOFF
+
+
+def _attribution(
+    conn: sqlite3.Connection,
+    fingerprints: list[str],
+    since: float,
+) -> tuple[str | None, str | None, float, int]:
+    """Look up route_decisions by session fingerprint. Returns
+    ``(model_id, provider, cost_usd, call_count)``."""
+    if not fingerprints:
+        return (None, None, 0.0, 0)
+    since_iso = datetime.fromtimestamp(since, tz=timezone.utc).isoformat()
+    rows = conn.execute(
+        "SELECT selected_model, selected_provider, est_cost_usd "
+        "FROM route_decisions WHERE session_key IN ("
+        + ",".join("?" for _ in fingerprints) + ") AND observed_at > ?",
+        fingerprints + [since_iso],
+    ).fetchall()
+    mc: dict[str, int] = collections.Counter()
+    cost_usd = 0.0
+    cnt = 0
+    for r in rows:
+        m = r[0]
+        if m:
+            mc[m] += 1
+        cost_usd += r[2] or 0
+        cnt += 1
+    if mc:
+        w = mc.most_common(1)[0][0]
+        prov = conn.execute(
+            "SELECT provider FROM models WHERE model_id = ?", (w,)
+        ).fetchone()
+        return (w, prov[0] if prov else None, cost_usd, cnt)
+    return (None, None, 0.0, 0)
+
+
+def _local_llm_second_opinion(
+    cfg: Any, server_url: str, agent: str,
+    reason_str: str, http_get: Callable[[str], Any],
+) -> str | None:
+    """Ask local Ollama for a yes/no opinion. Returns yes/no/None."""
+    model = cfg.verification.model or "qwen2.5-coder-router:14b"
+    base = cfg.verification.base_url.rstrip("/")
+    prompt = (
+        f"A coding agent session is looping without progress. "
+        f"Agent: {agent}. Evidence: {reason_str}. "
+        "Respond with only 'yes' or 'no'. No explanation."
+    )
+    url = f"{base}/api/chat"
+    payload = json.dumps({
+        "model": model,
+        "messages": [{"role": "user", "content": prompt}],
+        "stream": False,
+    }).encode()
+    req = urllib.request.Request(
+        url, data=payload,
+        headers={"Content-Type": "application/json"},
+    )
+    try:
+        resp = urllib.request.urlopen(req, timeout=_OLLAMA_TIMEOUT)
+        body = json.loads(resp.read().decode())
+        reply = (body.get("message") or {}).get("content", "")
+        return "yes" if "yes" in reply.lower() else "no"
+    except Exception:  # noqa: BLE001
+        return None
+
+
+def _make_key(session_root: str) -> str:
+    return f"opencode-loop:{session_root}"
+
+
+def _make_last_fired(
+    conn: sqlite3.Connection,
+) -> Callable[[str], float | None]:
+    """Return a callable that reads last_fired_at from watchdog_alerts."""
+
+    def _last_fired(channel_name: str) -> float | None:
+        row = conn.execute(
+            "SELECT last_fired_at FROM watchdog_alerts "
+            "ORDER BY opened_at DESC LIMIT 1",
+        ).fetchone()
+        if row is None:
+            return None
+        try:
+            dt = datetime.fromisoformat(row[0])
+            return dt.timestamp()
+        except (ValueError, TypeError):
+            return None
+
+    return _last_fired
+
+
+def _fire_alert(
+    conn: sqlite3.Connection, dedup_key: str,
+    state: str, severity: str, title: str,
+) -> None:
+    """Record an alert transition in the watchdog_alerts table.
+
+    Trigger: INSERT a new alert row with flagged_ticks = 1
+             (no-op if already open — that path uses escalate).
+    Escalate: UPDATE flagged_ticks += 1, severity = 'critical'.
+    Resolve:  UPDATE resolved_at, state = 'resolved'.
+    """
+    now = datetime.now(timezone.utc).isoformat()
+    if state == "resolve":
+        conn.execute(
+            "UPDATE watchdog_alerts SET resolved_at = ?, "
+            "state = 'resolved', severity = 'info' "
+            "WHERE dedup_key = ? AND resolved_at IS NULL",
+            (now, dedup_key),
+        )
+    elif state == "escalate":
+        conn.execute(
+            "UPDATE watchdog_alerts SET state = 'open', "
+            "severity = 'critical', flagged_ticks = flagged_ticks + 1, "
+            "last_fired_at = ? WHERE dedup_key = ?",
+            (now, dedup_key),
+        )
+    elif state == "trigger":
+        # Insert only if no open alert exists (Fix 4: trigger once).
+        existing = conn.execute(
+            "SELECT 1 FROM watchdog_alerts "
+            "WHERE dedup_key = ? AND resolved_at IS NULL",
+            (dedup_key,),
+        ).fetchone()
+        if existing is None:
+            conn.execute(
+                "INSERT INTO watchdog_alerts "
+                "(dedup_key, state, severity, flagged_ticks, "
+                "opened_at, last_fired_at, resolved_at) "
+                "VALUES (?, 'open', ?, 1, ?, ?, NULL)",
+                (dedup_key, severity, now, now),
+            )
+    logger.info(
+        "watchdog=%s dedup=%r state=%s severity=%s title=%s",
+        now, dedup_key, state, severity, title,
+    )
+
+
+def tick(
+    cfg: Any,
+    conn: sqlite3.Connection,
+    notifier: Notifier | None = None,
+    http_get: Callable[[str], Any] | None = None,
+) -> int:
+    """Run one watchdog tick.  Returns number of alerts fired."""
+    if http_get is None:
+        http_get = _http_get
+
+    now = datetime.now(timezone.utc)
+    now_iso = now.isoformat()
+
+    # 1. Acquire lock.
+    db_dir = _db_dir(conn)
+    fd = _acquire_lock(conn, db_dir)
+    if fd is None:
+        return 0
+    alerts_fired = 0
+    try:
+        # 2. Read rc-servers.json.
+        urls = _read_rc_servers(http_get)
+        if not urls:
+            conn.execute(
+                "INSERT INTO watchdog_ticks (ticked_at, sessions_seen, "
+                "outcome) VALUES (?, 0, 'no_opencode')", (now_iso,),
+            )
+            conn.commit()
+            return 0
+
+        # 3. Probe opencode servers.
+        answering: list[str] = []
+        for url in urls:
+            if _fetch_sessions(url, http_get) is not None:
+                answering.append(url)
+        if not answering:
+            conn.execute(
+                "INSERT INTO watchdog_ticks (ticked_at, sessions_seen, "
+                "outcome) VALUES (?, 0, 'no_opencode')", (now_iso,),
+            )
+            conn.commit()
+            return 0
+
+        server_url = answering[0].rstrip("/")
+        sessions = _fetch_sessions(server_url, http_get) or []
+        session_root, children = resolve_roots(sessions)
+
+        # 4. Last tick → activity window.  last_tick is epoch SECONDS; opencode
+        #    call times (c[0]) are epoch MILLISECONDS, so compare against
+        #    last_tick_ms.  Comparing seconds to ms made every call "recent" and
+        #    re-judged the whole session history on every tick.
+        last_tick = _last_tick_ts(conn)
+        last_tick_ms = last_tick * 1000
+
+        # 5. Fetch calls for ALL sessions.  Keep FULL history for evaluation;
+        #    filter to recent only for the "has this session been active" test.
+        all_calls: dict[str, list[tuple]] = {}
+        sessions_with_recent: dict[str, bool] = {}
+        for sid, root in session_root.items():
+            calls = calls_of(server_url, sid, http_get=http_get)
+            if not calls:
+                continue
+            all_calls[sid] = calls
+            recent = [c for c in calls if c[0] >= last_tick_ms]
+            sessions_with_recent[sid] = len(recent) > 0
+
+        # Build session-info lookup.
+        sess_by_id: dict[str, dict] = {}
+        for s in sessions:
+            sid = s.get("id", "")
+            if sid:
+                sess_by_id[sid] = s
+
+        # Build direct-children index from parentID (not from session_root).
+        parent_to_kids: dict[str, list[str]] = collections.defaultdict(list)
+        for sid, s in sess_by_id.items():
+            pid = s.get("parentID")
+            if pid:
+                parent_to_kids[pid].append(sid)
+
+        def _transitive_descendants(sid: str) -> list[str]:
+            result: list[str] = []
+            for kid in parent_to_kids.get(sid, []):
+                result.append(kid)
+                result.extend(_transitive_descendants(kid))
+            return result
+
+        # 6. Evaluate EACH active session on its own calls with its own title.
+        #    (Fix 2: not merging descendants into root)
+        dc = detect_config_from_pydantic(cfg.watchdog.detector)
+
+        def _file_lines(path: str) -> int | None:
+            try:
+                with open(path, "rb") as fh:
+                    return max(1, fh.read().count(b"\n"))
+            except OSError:
+                return None
+
+        per_session_verdict: dict[str, dict | None] = {}
+        session_agent: dict[str, str] = {}
+
+        for sid, has_recent in sessions_with_recent.items():
+            if not has_recent:
+                continue
+            cs = all_calls[sid]
+            sess_obj = sess_by_id.get(sid)
+            sess_title = sess_obj.get("title", "") if isinstance(sess_obj, dict) else ""
+            agent = sess_title or sid
+
+            # Compute descendant landed times: collect c[0] for calls
+            # where c[3] is True from the session's OWN transitive
+            # descendants (not siblings, not the session itself).
+            descendant_landed_times: list[float] = []
+            for did in _transitive_descendants(sid):
+                for c in all_calls.get(did, []):
+                    if c[3]:
+                        descendant_landed_times.append(c[0])
+
+            flagged_result, reason = evaluate(
+                cs, dc, _file_lines, agent,
+                extra_landed_times=descendant_landed_times,
+            )
+            per_session_verdict[sid] = reason if flagged_result else None
+            session_agent[sid] = agent
+
+        # 6b. Build root-level flagged set: flagged if ANY session in tree.
+        root_flagged: dict[str, dict | None] = {}
+        for sid, verdict in per_session_verdict.items():
+            root = session_root.get(sid, sid)
+            if verdict is not None:
+                root_flagged[root] = verdict
+
+        flagged: list[str] = list(root_flagged.keys())
+
+        # 7. Second opinion — ask for EACH flagged session up to max cap.
+        #    (Additional fix: remove break on first "no")
+        llm_opins: dict[str, str | None] = {}
+        if (hasattr(cfg, "local_compute")
+                and cfg.local_compute.enabled
+                and hasattr(cfg.watchdog, "local_llm_enabled")
+                and cfg.watchdog.local_llm_enabled):
+            for root in flagged:
+                if len(llm_opins) >= _MAX_LLM:
+                    break
+                # Find agent from any session in this root's tree.
+                tree_sessions = children.get(root, [root])
+                agent = None
+                for tsid in tree_sessions:
+                    agent = session_agent.get(tsid, tsid)
+                    if agent:
+                        break
+                if not agent:
+                    agent = root
+                why = json.dumps(root_flagged[root]) if root_flagged[root] else ""
+                opinion = _local_llm_second_opinion(
+                    cfg, server_url, agent, why, http_get)
+                llm_opins[root] = opinion
+
+        # 8. Model attribution — per session, using "c:" prefix keys only
+        #    (Fix 7: delete "ses_"+s hack, use c: prefix).
+        session_attr: dict[str, tuple] = {}
+        for sid in all_calls:
+            fps = ["c:" + sid]
+            session_attr[sid] = (
+                _attribution(conn, fps, last_tick)
+                if fps else (None, None, 0.0, 0)
+            )
+
+        # 9. Compute calls_since_landed (tool calls after tree's latest landed
+        #    time) and cost_since_landed_usd (route_decisions after that time).
+        tree_last_landed: dict[str, float] = {}
+        for sid in all_calls:
+            root = session_root.get(sid, sid)
+            tree_sessions = children.get(root, [root])
+            latest = 0.0
+            for tsid in tree_sessions:
+                for c in all_calls.get(tsid, []):
+                    if c[3]:
+                        latest = max(latest, c[0] or 0)
+            tree_last_landed[root] = latest
+
+        calls_since_landed_map: dict[str, int] = {}
+        cost_since_landed_map: dict[str, float] = {}
+        for sid, cs in all_calls.items():
+            root = session_root.get(sid, sid)
+            last_landed = tree_last_landed.get(root, 0.0)
+            calls_since_landed_map[sid] = sum(
+                1 for c in cs if (c[0] or 0) > last_landed
+            )
+
+            if last_landed > 0:
+                try:
+                    since_iso = datetime.fromtimestamp(last_landed / 1000, tz=timezone.utc).isoformat()
+                except (ValueError, OSError):
+                    since_iso = datetime.now(timezone.utc).isoformat()
+                fps = ["c:" + sid]
+                row = conn.execute(
+                    "SELECT SUM(est_cost_usd) FROM route_decisions "
+                    "WHERE session_key IN ("
+                    + ",".join("?" for _ in fps) + ") AND observed_at > ?",
+                    fps + [since_iso],
+                ).fetchone()
+                cost_since_landed_map[sid] = row[0] or 0.0 if row else 0.0
+            else:
+                cost_since_landed_map[sid] = 0.0
+
+        # 10. Resolve open alerts for idle/disappeared sessions FIRST
+        #     (Fix 6: iterate all OPEN alerts before checking flagged).
+        all_session_ids = {
+            s.get("id", "") for s in sessions if s.get("id")
+        }
+        open_alerts = conn.execute(
+            "SELECT dedup_key FROM watchdog_alerts "
+            "WHERE state = 'open' AND resolved_at IS NULL"
+        ).fetchall()
+        for alert_row in open_alerts:
+            dedup = alert_row[0]
+            if not dedup.startswith("opencode-loop:"):
+                continue
+            root_id = dedup[len("opencode-loop:"):]
+
+            # Resolve if the root session is no longer listed.
+            if root_id not in all_session_ids:
+                agent = root_id
+                _fire_alert(conn, dedup, "resolve", "info",
+                            f"opencode loop resolved: {agent}")
+                if notifier:
+                    event = AlertEvent(
+                        dedup_key=dedup, severity="info", state="resolve",
+                        title=f"opencode loop resolved: {agent}",
+                        summary="Session no longer active",
+                    )
+                    notifier.deliver(event)
+                alerts_fired += 1
+                continue
+
+            # Resolve if all sessions in the tree pass detection now.
+            tree_sids = children.get(root_id, [root_id])
+            any_flagged = any(
+                per_session_verdict.get(tsid) is not None
+                for tsid in tree_sids
+                if tsid in all_calls
+            )
+            if not any_flagged:
+                # Find agent name.
+                agent = root_id
+                for tsid in tree_sids:
+                    a = (session_agent.get(tsid)
+                         or (sess_by_id.get(tsid, {}) or {}).get("title", ""))
+                    if a:
+                        agent = a
+                        break
+                _fire_alert(conn, dedup, "resolve", "info",
+                            f"opencode loop resolved: {agent}")
+                if notifier:
+                    event = AlertEvent(
+                        dedup_key=dedup, severity="info", state="resolve",
+                        title=f"opencode loop resolved: {agent}",
+                        summary="Session no longer flagged",
+                    )
+                    notifier.deliver(event)
+                alerts_fired += 1
+
+        conn.commit()
+
+        # 11. Alert state machine per-root (Fix 4: trigger once, escalate once,
+        #     Fix 5: only create alert rows for flagged sessions).
+        for sid, verdict in per_session_verdict.items():
+            if verdict is None:
+                continue
+            root = session_root.get(sid, sid)
+            dedup = _make_key(root)
+            agent = session_agent.get(sid, sid)
+
+            alert_row = conn.execute(
+                "SELECT state, severity, flagged_ticks "
+                "FROM watchdog_alerts "
+                "WHERE dedup_key = ? AND resolved_at IS NULL",
+                (dedup,),
+            ).fetchone()
+
+            if alert_row is None:
+                # No open alert → trigger (Fix 4: trigger once).
+                severity = "critical" if llm_opins.get(root) == "yes" else "warning"
+                _fire_alert(conn, dedup, "trigger", severity,
+                            f"opencode loop: {agent}")
+                if notifier:
+                    summary = (
+                        f"Session {sid} flagged for looping. "
+                        f"See {cfg.watchdog.dashboard_base_url}#loops"
+                    )
+                    event = AlertEvent(
+                        dedup_key=dedup, severity=severity, state="trigger",
+                        title=f"opencode loop: {agent}",
+                        summary=summary,
+                        details=json.dumps(verdict),
+                    )
+                    notifier.deliver(event)
+                alerts_fired += 1
+            else:
+                # Open alert exists → increment flagged_ticks.
+                # Fire escalate ONCE when LLM confirms OR 3 ticks (15 min).
+                flagged_ticks = alert_row[2] + 1
+                conn.execute(
+                    "UPDATE watchdog_alerts SET flagged_ticks = ? "
+                    "WHERE dedup_key = ?",
+                    (flagged_ticks, dedup),
+                )
+                should_escalate = (
+                    flagged_ticks >= _ESCALATE_THRESHOLD
+                    or llm_opins.get(root) == "yes"
+                )
+                if should_escalate and alert_row[1] != "critical":
+                    _fire_alert(conn, dedup, "escalate", "critical",
+                                f"opencode loop: {agent}")
+                    if notifier:
+                        summary = (
+                            f"Session {sid} persisting after {flagged_ticks} "
+                            f"ticks ({flagged_ticks * 5} min). "
+                            f"See {cfg.watchdog.dashboard_base_url}#loops"
+                        )
+                        event = AlertEvent(
+                            dedup_key=dedup, severity="critical",
+                            state="escalate",
+                            title=f"opencode loop: {agent}",
+                            summary=summary,
+                            details=json.dumps(verdict),
+                        )
+                        notifier.deliver(event)
+                    alerts_fired += 1
+
+        # 12. Write tick record (Fix 8: outcome is "flagged" or "ok").
+        outcome = "flagged" if flagged else "ok"
+        conn.execute(
+            "INSERT INTO watchdog_ticks (ticked_at, sessions_seen, "
+            "outcome) VALUES (?, ?, ?)",
+            (now_iso, len(sessions), outcome),
+        )
+        tick_id = conn.execute(
+            "SELECT last_insert_rowid()"
+        ).fetchone()[0]
+
+        # 13. Write PER-SESSION verdict rows (not per-root).
+        for sid, verdict in per_session_verdict.items():
+            root = session_root.get(sid, sid)
+            agent = session_agent.get(sid, sid)
+            m_id, prov, _cost, _calls_count = session_attr.get(
+                sid, (None, None, 0.0, 0)
+            )
+            is_flagged = verdict is not None
+
+            conn.execute(
+                "INSERT INTO watchdog_verdicts ("
+                "tick_id, session_id, session_root, agent, "
+                "model_id, provider, flagged, dup, top, top_what, "
+                "landed, slow, coverage, calls_since_landed, "
+                "cost_since_landed_usd, llm_second_opinion, "
+                "created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, "
+                "?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    tick_id, sid, root, agent,
+                    m_id, prov,
+                    1 if is_flagged else 0,
+                    verdict.get("dup", 0) if verdict else 0,
+                    verdict.get("top", 0) if verdict else 0,
+                    verdict.get("top_what", "") if verdict else "",
+                    1 if (verdict and verdict.get("landed")) else 0,
+                    1 if (verdict and verdict.get("slow", False)) else 0,
+                    round(verdict.get("coverage", 0), 1) if verdict else 0.0,
+                    calls_since_landed_map.get(sid, 0),
+                    cost_since_landed_map.get(sid, 0.0),
+                    llm_opins.get(root),
+                    now_iso,
+                ),
+            )
+
+        conn.commit()
+
+        # 14. Blind warning for unexpected session structure.
+        if sessions:
+            sample = sessions[0]
+            needed = {"id", "title", "time"}
+            if not needed.issubset(
+                {str(k) for k in sample}
+            ):
+                dedup = "_blind"
+                _fire_alert(conn, dedup, "trigger", "warning",
+                            "watchdog-blind: unexpected opencode structure")
+                if notifier:
+                    event = AlertEvent(
+                        dedup_key=dedup, severity="warning", state="trigger",
+                        title="watchdog is blind",
+                        summary="Session structure does not match expected schema",
+                    )
+                    notifier.deliver(event)
+                alerts_fired += 1
+
+    except Exception:  # logger.exception below documents the catch
+        logger.exception("watchdog: tick error")
+    finally:
+        if fd is not None:
+            try:
+                fcntl.flock(fd, fcntl.LOCK_UN)
+                os.close(fd)
+            except OSError:
+                pass
+
+    return alerts_fired
+
+
+def load_config(path: str = "config/config.yaml") -> Any:
+    from config import load_config as _load
+    return _load(path)
+
+
+def _get_db_path(cfg: Any) -> str:
+    try:
+        if hasattr(cfg, "database") and hasattr(
+            cfg.database, "path"
+        ):
+            return cfg.database.path
+    except AttributeError:
+        pass
+    return "router.db"
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser(
+        description="Watchdog orchestrator for opencode loop detection",
+    )
+    ap.add_argument(
+        "--once", action="store_true", help="Run one tick and exit",
+    )
+    ap.add_argument(
+        "--config", default="config/config.yaml", help="config.yaml path",
+    )
+    args = ap.parse_args()
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    cfg = load_config(args.config)
+    db_path = _get_db_path(cfg)
+    conn = sqlite3.connect(db_path)
+    conn.execute("PRAGMA busy_timeout=5000")
+    conn.execute("PRAGMA foreign_keys=ON")
+    ensure_watchdog_tables(conn)
+
+    # Construct Notifier from config + DB overrides (Fix 3: __main__).
+    notifier: Notifier | None = None
+    if hasattr(cfg, "notifications") and cfg.notifications.channels:
+        from config import NotificationsConfig
+
+        base_channels: list[dict] = [
+            {
+                "name": ch.name, "type": ch.type,
+                "enabled": ch.enabled, "min_severity": ch.min_severity,
+            }
+            for ch in cfg.notifications.channels
+        ]
+
+        # Apply DB overrides from watchdog_channel_settings.
+        for row in conn.execute(
+            "SELECT channel_name, enabled, min_severity "
+            "FROM watchdog_channel_settings"
+        ).fetchall():
+            for ch in base_channels:
+                if ch["name"] == row[0]:
+                    ch["enabled"] = bool(row[1])
+                    ch["min_severity"] = row[2]
+                    break
+
+        notifier_cfg = NotificationsConfig(channels=base_channels)
+        last_fired_cb = _make_last_fired(conn)
+        notifier = Notifier(notifier_cfg, last_fired=last_fired_cb)
+
+    rc = tick(cfg, conn, notifier=notifier)
+    conn.close()
+    sys.exit(rc or 0)

src/watchdog_store.py

@@ -0,0 +1,73 @@
+"""Watchdog monitoring store — idempotent table creation for the watchdog subsystem.
+
+Mirrors ``ensure_route_decisions`` in dispatcher.py: the schema file
+(``config/schema.sql``) uses ``CREATE TABLE IF NOT EXISTS``, so it defines a
+NEW database and silently does nothing to an existing one. A live router.db
+predating the watchdog tables never gets them from re-running schema.sql, so
+they are created here, from code, with a guard.
+
+Safe to call any number of times against an existing connection: the CREATE
+TABLE, CREATE INDEX, and ALTER ADD COLUMN operations are all guarded by IF NOT
+EXISTS / PRAGMA table_info, so a table that is already present is left fully
+intact (rows included) and a second call no-ops.
+"""
+
+import sqlite3
+
+
+def ensure_watchdog_tables(conn: sqlite3.Connection) -> None:
+    """Idempotently create the watchdog monitoring tables."""
+    conn.executescript(
+        """
+        CREATE TABLE IF NOT EXISTS watchdog_ticks (
+            id               INTEGER PRIMARY KEY AUTOINCREMENT,
+            ticked_at        TEXT NOT NULL,
+            sessions_seen    INTEGER NOT NULL DEFAULT 0,
+            outcome          TEXT
+        );
+        CREATE TABLE IF NOT EXISTS watchdog_verdicts (
+            id               INTEGER PRIMARY KEY AUTOINCREMENT,
+            tick_id          INTEGER NOT NULL REFERENCES watchdog_ticks(id),
+            session_id       TEXT NOT NULL,
+            session_root     TEXT,
+            agent            TEXT,
+            model_id         TEXT,
+            provider         TEXT,
+            flagged          INTEGER NOT NULL DEFAULT 0,
+            dup              REAL,
+            top              INTEGER,
+            top_what         TEXT,
+            landed           INTEGER,
+            slow             INTEGER,
+            coverage         REAL,
+            calls_since_landed    INTEGER,
+            cost_since_landed_usd REAL,
+            llm_second_opinion TEXT,
+            created_at       TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS watchdog_alerts (
+            dedup_key        TEXT PRIMARY KEY,
+            state            TEXT NOT NULL,
+            severity         TEXT NOT NULL,
+            flagged_ticks    INTEGER NOT NULL DEFAULT 0,
+            opened_at        TEXT NOT NULL,
+            last_fired_at    TEXT NOT NULL,
+            resolved_at      TEXT
+        );
+        CREATE TABLE IF NOT EXISTS watchdog_channel_settings (
+            id               INTEGER PRIMARY KEY AUTOINCREMENT,
+            channel_name     TEXT NOT NULL UNIQUE,
+            enabled          INTEGER NOT NULL DEFAULT 1,
+            min_severity     TEXT NOT NULL DEFAULT 'warning'
+        );
+        CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_model_id
+            ON watchdog_verdicts(model_id);
+        CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_created_at
+            ON watchdog_verdicts(created_at);
+        CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_flagged
+            ON watchdog_verdicts(flagged);
+        CREATE INDEX IF NOT EXISTS idx_watchdog_verdicts_session_root
+            ON watchdog_verdicts(session_root);
+        """
+    )
+    conn.commit()

tests/test_admin_knob_coverage.py

@@ -328,6 +328,10 @@ DELIBERATELY_NOT_IN_ADMIN: dict[str, str] = {
         "lookback for the conversation adoption counter in /metrics; a "
         "read-only window that shapes a report, not a routing dial."
     ),
+    # --- deployment wiring: set in config, never in admin --------------------
+    "watchdog.dashboard_base_url": (
+        "deployment wiring, set in config, not admin"
+    ),
 }
 
 

tests/test_admin_watchdog.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+import pytest
+from starlette.testclient import TestClient
+
+import dispatcher
+from watchdog_store import ensure_watchdog_tables
+
+ROOT = Path(__file__).resolve().parent.parent
+SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    db_path = tmp_path / "wd.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(SCHEMA_SQL)
+    ensure_watchdog_tables(conn)
+    conn.close()
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+    with TestClient(dispatcher.app) as c:
+        yield c
+
+
+def test_watchdog_status_empty_db(client):
+    resp = client.get("/admin/api/watchdog/status")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["last_tick"] is None
+    assert data["open_alerts"] == 0
+
+
+def test_watchdog_loops_empty_db(client):
+    resp = client.get("/admin/api/watchdog/loops")
+    assert resp.status_code == 200
+    assert resp.json() == []
+
+
+def test_watchdog_channels_empty_db(client):
+    resp = client.get("/admin/api/watchdog/channels")
+    assert resp.status_code == 200
+    assert resp.json() == []
+
+
+def test_watchdog_channels_create(client):
+    resp = client.post(
+        "/admin/api/watchdog/channels",
+        json={"channel_name": "desktop", "enabled": True, "min_severity": "warning"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["ok"] is True
+    listing = client.get("/admin/api/watchdog/channels").json()
+    assert len(listing) == 1
+    assert listing[0]["channel_name"] == "desktop"
+    assert listing[0]["enabled"] == 1
+
+
+def test_watchdog_channels_update_min_severity(client):
+    client.post(
+        "/admin/api/watchdog/channels",
+        json={"channel_name": "desktop", "enabled": True, "min_severity": "warning"},
+    )
+    client.post(
+        "/admin/api/watchdog/channels",
+        json={"channel_name": "desktop", "min_severity": "critical"},
+    )
+    listing = client.get("/admin/api/watchdog/channels").json()
+    assert listing[0]["min_severity"] == "critical"
+    assert listing[0]["enabled"] == 1
+
+
+def test_watchdog_channels_requires_name(client):
+    resp = client.post("/admin/api/watchdog/channels", json={})
+    assert resp.status_code == 422
+
+
+def test_watchdog_channels_invalid_severity(client):
+    resp = client.post(
+        "/admin/api/watchdog/channels",
+        json={"channel_name": "x", "min_severity": "bogus"},
+    )
+    assert resp.status_code == 422
+
+
+def test_watchdog_test_alert_no_channels(client):
+    resp = client.post(
+        "/admin/api/watchdog/test-alert",
+        json={"severity": "warning"},
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["sent"] == 0
+
+
+def test_watchdog_test_alert_invalid_severity(client):
+    resp = client.post(
+        "/admin/api/watchdog/test-alert",
+        json={"severity": "bogus"},
+    )
+    assert resp.status_code == 422
+
+
+def test_watchdog_status_after_tick(tmp_path, monkeypatch):
+    import sqlite3 as _sqlite3
+    db_path = tmp_path / "wd_ticks.db"
+    conn = _sqlite3.connect(db_path)
+    conn.row_factory = _sqlite3.Row
+    conn.executescript(SCHEMA_SQL)
+    ensure_watchdog_tables(conn)
+    conn.execute(
+        "INSERT INTO watchdog_ticks (ticked_at, sessions_seen, outcome) VALUES (?, ?, ?)",
+        ("2026-09-25T10:00:00", 5, "ok"),
+    )
+    conn.commit()
+    conn.close()
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+    with TestClient(dispatcher.app) as c:
+        resp = c.get("/admin/api/watchdog/status")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["last_tick"] is not None
+    assert data["last_tick"]["sessions_seen"] == 5

tests/test_blocked_routing.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+import pytest
+from starlette.testclient import TestClient
+
+import dispatcher
+from dispatcher import Classification, app
+
+ROOT = Path(__file__).resolve().parent.parent
+SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
+
+KEEP = "keep-model"
+DROP = "drop-model"
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    db_path = tmp_path / "blk.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(SCHEMA_SQL)
+    import admin
+    admin.ensure_admin_tables(conn)
+    for model_id in (KEEP, DROP):
+        conn.execute(
+            """
+            INSERT INTO models (
+                model_id, provider, base_model_id, tier, context_window,
+                effective_context_window, max_output_tokens,
+                cost_per_1m_prompt, cost_per_1m_completion,
+                supports_vision, supports_json_mode,
+                latency_class, reasoning_mode, context_variant,
+                access_level, availability, last_updated
+            ) VALUES (?, 'neuralwatt', ?, 2, 262128, 192500, 16384, 1.0, 2.0,
+                      0, 1, 'standard', 'default', 'full', 'public', 'active',
+                      '2026-08-22T00:00:00+00:00')
+            """,
+            (model_id, model_id),
+        )
+    conn.commit()
+    conn.close()
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.session_cache, "enabled", False)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+    monkeypatch.setattr(
+        dispatcher, "classify",
+        lambda task, context: Classification(
+            task_category="coding_general", task_tier=2,
+            required_context_tokens=100, confidence=0.9,
+        ),
+    )
+    return TestClient(app)
+
+
+def _block(client, model_id):
+    resp = client.post(
+        f"/admin/api/models/{model_id}/neuralwatt/availability",
+        json={"availability": "blocked"},
+    )
+    assert resp.status_code == 200, resp.text
+
+
+def _listed(client):
+    return {m["id"] for m in client.get("/v1/models").json()["data"]}
+
+
+def test_blocked_removes_model_from_v1_models(client):
+    assert DROP in _listed(client)
+    _block(client, DROP)
+    assert DROP not in _listed(client)
+    assert KEEP in _listed(client), "only the blocked model should go"
+
+
+def test_blocked_refuses_a_pinned_request(client):
+    _block(client, DROP)
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": DROP, "messages": [{"role": "user", "content": "hi"}]},
+    )
+    assert resp.status_code == 503, resp.text
+    assert "override" in resp.text
+
+
+def test_blocked_drops_model_from_routing(client):
+    client.post("/route", json={"task": "write a function"}).json()
+    _block(client, DROP)
+    after = client.post("/route", json={"task": "write a function"}).json()
+    assert after.get("selected_model") != DROP
+
+
+def test_clearing_blocked_restores_model(client):
+    _block(client, DROP)
+    assert DROP not in _listed(client)
+    resp = client.delete(f"/admin/api/models/{DROP}/neuralwatt/availability")
+    assert resp.status_code == 200, resp.text
+    assert DROP in _listed(client)
+
+
+def test_unblocked_model_is_still_pinnable(client):
+    _block(client, DROP)
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": KEEP, "messages": [{"role": "user", "content": "hi"}]},
+    )
+    assert resp.status_code != 503 or "override" not in resp.text

tests/test_config_watchdog.py

@@ -0,0 +1,114 @@
+"""Config validation tests for the watchdog / notification / detector sections.
+
+Each section has defaults that load correctly when absent, and each has
+knobs that must be validated at config-load time rather than silently
+accepted.
+"""
+
+from __future__ import annotations
+
+import copy
+from pathlib import Path
+
+import pytest
+import yaml
+
+from config import RouterConfig
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+@pytest.fixture
+def raw() -> dict:
+    with open(ROOT / "config" / "config.yaml") as fh:
+        return yaml.safe_load(fh)
+
+
+def test_watchdog_config_defaults(raw):
+    """When the watchdog section is absent, all defaults are filled in."""
+    cfg = copy.deepcopy(raw)
+    cfg.pop("watchdog")
+    loaded = RouterConfig(**cfg)
+
+    assert loaded.watchdog.enabled is True
+    assert loaded.watchdog.local_llm_enabled is True
+    assert loaded.watchdog.model is None
+    assert loaded.watchdog.read_only_agents == ["explore", "librarian", "oracle"]
+    assert loaded.watchdog.dashboard_base_url == "http://127.0.0.1:8080/admin"
+
+    # Detector sub-config defaults
+    det = loaded.watchdog.detector
+    assert det.window == 60
+    assert det.dup_min == 0.25
+    assert det.top_min == 12
+    assert det.top_min_ro == 8
+    assert det.cum_min == 15
+    assert det.cover_min == 4.0
+    assert det.min_calls == 40
+
+
+def test_notifications_channels_default(raw):
+    """When the notifications section is absent, one desktop channel is the default."""
+    cfg = copy.deepcopy(raw)
+    cfg.pop("notifications")
+    loaded = RouterConfig(**cfg)
+
+    assert len(loaded.notifications.channels) == 1
+    ch = loaded.notifications.channels[0]
+    assert ch["name"] == "default"
+    assert ch["type"] == "desktop"
+    assert ch["enabled"] is True
+    assert ch["min_severity"] == "warning"
+
+
+def test_notifications_unknown_type_fails(raw):
+    """An unknown notification channel type is rejected at load time."""
+    cfg = copy.deepcopy(raw)
+    cfg["notifications"]["channels"].append(
+        {"name": "pager", "type": "pagerduty"}
+    )
+    with pytest.raises(ValueError, match="pagerduty|Input should be"):
+        RouterConfig(**cfg)
+
+
+def test_detector_thresholds(raw):
+    """Override every detector threshold and confirm parsing."""
+    cfg = copy.deepcopy(raw)
+    cfg["watchdog"]["detector"] = {
+        "window": 120,
+        "dup_min": 0.5,
+        "top_min": 24,
+        "top_min_ro": 16,
+        "cum_min": 30,
+        "cover_min": 8.0,
+        "min_calls": 80,
+    }
+    loaded = RouterConfig(**cfg)
+    det = loaded.watchdog.detector
+    assert det.window == 120
+    assert det.dup_min == 0.5
+    assert det.top_min == 24
+    assert det.top_min_ro == 16
+    assert det.cum_min == 30
+    assert det.cover_min == 8.0
+    assert det.min_calls == 80
+
+
+def test_watchdog_model_defaults_to_verification(raw):
+    """When watchdog.model is None, the config stores None and falls through
+    to verification.model at runtime."""
+    cfg = copy.deepcopy(raw)
+    # Shipped config has watchdog.model: null and verification.model set.
+    assert cfg["watchdog"]["model"] is None
+    assert cfg["verification"]["model"] is not None
+
+    loaded = RouterConfig(**cfg)
+    assert loaded.watchdog.model is None
+    # The fallback logic lives at runtime, not in the model, so we confirm
+    # the config-level contract: watchdog.model None + verification.model set.
+    assert loaded.verification.model is not None
+
+    # Also verify that when watchdog.model IS set, it takes that value.
+    cfg["watchdog"]["model"] = "custom-verifier:latest"
+    loaded2 = RouterConfig(**cfg)
+    assert loaded2.watchdog.model == "custom-verifier:latest"

tests/test_notifier.py

@@ -0,0 +1,173 @@
+"""Unit tests for the desktop notification channel (notifier.py).
+
+All tests stub ``subprocess.run`` — no real ``notify-send`` calls.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+
+from config import NotificationsConfig
+from notifier import AlertEvent, Notifier
+
+# ── fixtures ──────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def notifier() -> Notifier:
+    """Default notifier with one desktop channel (min_severity=warning)."""
+    cfg = NotificationsConfig()
+    return Notifier(cfg)
+
+
+@pytest.fixture
+def event() -> AlertEvent:
+    """A sample critical trigger event."""
+    return AlertEvent(
+        dedup_key="test-session-1",
+        severity="critical",
+        state="trigger",
+        title="Agent stuck",
+        summary="No progress for 30 minutes",
+    )
+
+
+# ── tests ─────────────────────────────────────────────────────────────
+
+
+class TestDesktopNotify:
+    """Tests for desktop notification delivery via notify-send."""
+
+    def test_deliver_critical(self, notifier: Notifier, event: AlertEvent) -> None:
+        """A critical event fires notify-send with -u critical."""
+        with mock.patch("notifier.subprocess.run") as mock_run:
+            notifier.deliver(event)
+
+        mock_run.assert_called_once_with(
+            ["notify-send", "-u", "critical", "Agent stuck",
+             "No progress for 30 minutes"],
+            check=False,
+            timeout=5,
+        )
+
+    def test_deliver_skips_below_min_severity(
+        self, notifier: Notifier
+    ) -> None:
+        """An info event is skipped when channel min_severity is warning."""
+        info_event = AlertEvent(
+            dedup_key="info-test",
+            severity="info",
+            state="trigger",
+            title="Trivial",
+            summary="Nothing to see",
+        )
+        with mock.patch("notifier.subprocess.run") as mock_run:
+            notifier.deliver(info_event)
+
+        mock_run.assert_not_called()
+
+    def test_deliver_rate_limited(self, notifier: Notifier) -> None:
+        """A second delivery within the rate window is suppressed.
+
+        We deliver once, record the timestamp as if the caller did, then
+        deliver again — the second should be rate-limited.
+        """
+        store: dict[str, float] = {}
+        # Notifier that reads from our store but does not write to it
+        rate_notifier = Notifier(
+            NotificationsConfig(),
+            last_fired=lambda key: store.get(key),
+        )
+
+        ev = AlertEvent(
+            dedup_key="dup",
+            severity="warning",
+            state="trigger",
+            title="Dup",
+            summary="Duplicate check",
+        )
+
+        with mock.patch("notifier.subprocess.run") as mock_run:
+            # First delivery — no last_fired entry, should fire
+            rate_notifier.deliver(ev)
+            # Simulate the caller recording the fired timestamp
+            store["default"] = time.monotonic()
+            # Second delivery — within rate window, should be suppressed
+            rate_notifier.deliver(ev)
+
+        assert mock_run.call_count == 1
+
+    def test_deliver_resolve_not_rate_limited(
+        self, notifier: Notifier
+    ) -> None:
+        """A resolve event bypasses the rate limit.
+
+        Deliver trigger, record timestamp, deliver resolve — both fire.
+        """
+        store: dict[str, float] = {}
+        rate_notifier = Notifier(
+            NotificationsConfig(),
+            last_fired=lambda key: store.get(key),
+        )
+
+        trigger_event = AlertEvent(
+            dedup_key="test-alert",
+            severity="critical",
+            state="trigger",
+            title="Something wrong",
+            summary="Critical issue detected",
+        )
+        resolve_event = AlertEvent(
+            dedup_key="test-alert",
+            severity="critical",
+            state="resolve",
+            title="Something wrong",
+            summary="Issue resolved",
+        )
+
+        with mock.patch("notifier.subprocess.run") as mock_run:
+            rate_notifier.deliver(trigger_event)
+            store["default"] = time.monotonic()  # simulate caller recording
+            rate_notifier.deliver(resolve_event)
+
+        assert mock_run.call_count == 2
+
+    def test_deliver_failure_logged(
+        self, caplog: pytest.LogCaptureFixture, notifier: Notifier,
+        event: AlertEvent,
+    ) -> None:
+        """FileNotFoundError from notify-send is caught and logged."""
+        caplog.set_level(logging.WARNING, logger="notifier")
+
+        with mock.patch(
+            "notifier.subprocess.run", side_effect=FileNotFoundError
+        ):
+            notifier.deliver(event)
+
+        assert len(caplog.records) >= 1
+        assert "notify-send not installed" in caplog.text
+        assert "test-session-1" in caplog.text
+
+
+class TestUnknownChannel:
+    """Tests for handling of unknown channel types."""
+
+    def test_unknown_channel_type_skips(
+        self, caplog: pytest.LogCaptureFixture, notifier: Notifier,
+        event: AlertEvent,
+    ) -> None:
+        """An unknown channel type (not 'desktop') logs a warning and skips."""
+        caplog.set_level(logging.WARNING, logger="notifier")
+
+        fake_channel = SimpleNamespace(type="unknown", name="test-channel")
+        notifier._dispatch(fake_channel, event)
+
+        assert any(
+            "unknown channel type" in r.getMessage() for r in caplog.records
+        )
+        assert "test-channel" in caplog.text

tests/test_progress_backtest.py

@@ -0,0 +1,316 @@
+"""Integration tests for scripts/progress_backtest.py.
+
+Tests the fixture-based backtest output format, label correctness,
+boundary conditions, and ensures no real opencode calls are made.
+"""
+
+import json
+import os
+import re
+import sys
+
+from progress_detect import DetectConfig, evaluate
+
+_SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts")
+_PROGRESS_BACKTEST = os.path.join(_SCRIPTS_DIR, "progress_backtest.py")
+_FIXTURE_PATH = os.path.join(
+    os.path.dirname(__file__), "fixtures", "progress", "fixture.json"
+)
+
+_ATLAS_ID = "ses_f2506ac70ffebLUxBsq0aauOn9"
+_FLAG_RE = re.compile(
+    r"^(FLAG\s+|ok\s+  )"
+    r"calls=(\s*\d+)"
+    r" first_flag_at=(\d+|N/A)"
+    r" dup=(\d+\.\d+)"
+    r" top=(\d+)"
+    r" landed=(\d+)"
+    r" ro=(\d+)"
+    r" \| (.+)$"
+)
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+def _calls_from_fixture(session):
+    """Convert a fixture session dict to call tuples for evaluate()."""
+    return [
+        (c["t"], c["tool"], json.dumps(c["args"]), c["landed"])
+        for c in session["calls"]
+    ]
+
+
+def _file_lines_fn(file_lines_dict):
+    """Return a callable for coverage() from a file_path→line_count dict."""
+    def fn(fp):
+        if fp is None:
+            return 0
+        return file_lines_dict.get(fp, 0) or 0
+    return fn
+
+
+def _run_backtest(fixture_path):
+    """Run the backtest script with --fixture and return parsed lines + stderr."""
+    import subprocess
+
+    worktree_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    result = subprocess.run(
+        [sys.executable, _PROGRESS_BACKTEST, "--fixture", fixture_path],
+        capture_output=True,
+        text=True,
+        check=False,
+        cwd=worktree_root,
+        env={**os.environ, "PYTHONPATH": os.path.join(worktree_root, "src")},
+    )
+    lines = [l for l in result.stdout.strip().split("\n") if l.strip() and not l.startswith("#")]
+    return result, lines
+
+def _load_fixture():
+    """Load fixture JSON."""
+    with open(_FIXTURE_PATH) as f:
+        return json.load(f)
+
+
+def _parse_line(line):
+    """Parse a backtest output line into a dict, or None if invalid."""
+    m = _FLAG_RE.match(line)
+    if not m:
+        return None
+    return {
+        "status": "flag" if m.group(1).startswith("FLAG") else "ok",
+        "calls": int(m.group(2)),
+        "first_flag_at": None if m.group(3) == "N/A" else int(m.group(3)),
+        "dup": float(m.group(4)),
+        "top": int(m.group(5)),
+        "landed": int(m.group(6)),
+        "ro": int(m.group(7)),
+        "title": m.group(8),
+    }
+
+
+# ---------------------------------------------------------------------------
+# fixture label correctness
+# ---------------------------------------------------------------------------
+
+
+class TestFixtureLabels:
+    """Verify backtest results match fixture labels using the detector directly."""
+
+    @staticmethod
+    def _check_session(session):
+        """Run sliding-window backtest returns (first_flag_at, reason)."""
+        cfg = DetectConfig()
+        calls = _calls_from_fixture(session)
+        fl = session.get("file_lines", {})
+        fl_fn = _file_lines_fn(fl)
+        title = session.get("title", "")
+        if len(calls) < cfg.min_calls:
+            return (False, None)
+        for end_idx in range(cfg.min_calls, len(calls) + 1, 5):
+            flag, reason = evaluate(calls, cfg, fl_fn, title, end_idx=end_idx)
+            if flag and reason:
+                return (True, reason)
+        return (False, None)
+
+    def test_must_flag_sessions_all_flag(self):
+        """Every must_flag session has at least one window where flag=True."""
+        sessions = _load_fixture()
+        must_flag = [s for s in sessions if s["label"] == "must_flag"]
+        for s in must_flag:
+            flagged, _ = self._check_session(s)
+            assert flagged, (
+                f"Session {s['session_id']} (must_flag) should flag but did not"
+            )
+
+    def test_must_not_flag_sessions_never_flag(self):
+        """Every must_not_flag session has NO window where flag=True."""
+        sessions = _load_fixture()
+        must_not = [s for s in sessions if s["label"] == "must_not_flag"]
+        for s in must_not:
+            sid = s["session_id"]
+            flagged, _ = self._check_session(s)
+            assert not flagged, (
+                f"Session {sid} (must_not_flag) should not flag but did"
+            )
+
+
+# ---------------------------------------------------------------------------
+# Atlas 02:07 boundary
+# ---------------------------------------------------------------------------
+
+
+class TestAtlasBoundary:
+    """Atlas session: no flag before 02:07; flag exists post-02:07."""
+
+    def test_atlas_no_flag_before_cutoff(self):
+        """No flagged window before 02:07 wall-clock time."""
+        sessions = _load_fixture()
+        atlas = next(s for s in sessions if s["session_id"] == _ATLAS_ID)
+        calls = _calls_from_fixture(atlas)
+        fl_fn = _file_lines_fn(atlas.get("file_lines", {}))
+        cfg = DetectConfig()
+        cutoff = 1790402856356  # 02:07:36 UTC wall-clock
+
+        pre_flagged = []
+        for i in range(cfg.min_calls, len(calls) + 1, 5):
+            subset = calls[:i]
+            if len(subset) < cfg.min_calls:
+                continue
+            win = subset[-cfg.window:]
+            if len(win) < cfg.min_calls:
+                continue
+            win_end = win[-1][0]
+            flag, _ = evaluate(calls, cfg, fl_fn, atlas["title"], end_idx=i)
+            if flag and win_end < cutoff:
+                pre_flagged.append(i)
+
+        assert len(pre_flagged) == 0, (
+            f"Expected no flags before 02:07, got {len(pre_flagged)} "
+            f"at indices {pre_flagged}"
+        )
+
+    def test_atlas_flag_after_cutoff(self):
+        """At least one flag after 02:07 wall-clock time."""
+        sessions = _load_fixture()
+        atlas = next(s for s in sessions if s["session_id"] == _ATLAS_ID)
+        calls = _calls_from_fixture(atlas)
+        fl_fn = _file_lines_fn(atlas.get("file_lines", {}))
+        cfg = DetectConfig()
+        cutoff = 1790402856356
+
+        post_flagged = []
+        for i in range(cfg.min_calls, len(calls) + 1, 5):
+            subset = calls[:i]
+            if len(subset) < cfg.min_calls:
+                continue
+            win = subset[-cfg.window:]
+            if len(win) < cfg.min_calls:
+                continue
+            win_end = win[-1][0]
+            flag, _ = evaluate(calls, cfg, fl_fn, atlas["title"], end_idx=i)
+            if flag and win_end >= cutoff:
+                post_flagged.append(i)
+
+        assert len(post_flagged) >= 1, (
+            "Expected at least one flag after 02:07, got none"
+        )
+
+
+# ---------------------------------------------------------------------------
+# backtest output format
+# ---------------------------------------------------------------------------
+
+
+class TestOutputFormat:
+    """Test that the backtest script produces correctly formatted output."""
+
+    def test_backtest_output_parseable(self):
+        """All non-comment, non-empty output lines parse correctly."""
+        result, lines = _run_backtest(_FIXTURE_PATH)
+        assert result.returncode == 0, f"Backtest failed: {result.stderr}"
+        assert len(lines) == 15, f"Expected 15 session lines, got {len(lines)}"
+        parsed = [_parse_line(l) for l in lines]
+        assert all(p is not None for p in parsed), (
+            f"Some lines failed parsing. Sample: {lines[0] if lines else '(empty)'}"
+        )
+
+    def test_flag_count_matches_fixture(self):
+        """FLAG count equals must_flag count."""
+        sessions = _load_fixture()
+        must_flag = len([s for s in sessions if s["label"] == "must_flag"])
+        expected_flags = must_flag
+        _, lines = _run_backtest(_FIXTURE_PATH)
+        parsed = [p for p in [_parse_line(l) for l in lines] if p]
+        actual_flags = sum(1 for p in parsed if p["status"] == "flag")
+        assert actual_flags == expected_flags, (
+            f"Expected {expected_flags} flagged, got {actual_flags}"
+        )
+
+    def test_flagged_session_has_first_flag_at(self):
+        """FLAG lines have a numeric first_flag_at; ok lines have None."""
+        _, lines = _run_backtest(_FIXTURE_PATH)
+        parsed = [_parse_line(l) for l in lines if _parse_line(l)]
+        for p in parsed:
+            if p["status"] == "flag":
+                assert p["first_flag_at"] is not None, (
+                    f"FLAG line should have numeric first_flag_at: {p}"
+                )
+            else:
+                assert p["first_flag_at"] is None, (
+                    f"ok line should have None first_flag_at: {p}"
+                )
+
+    def test_call_counts_match_fixture(self):
+        """Each line's calls= field matches the fixture call count."""
+        sessions = _load_fixture()
+        sid_to_calls = {s["session_id"]: len(s["calls"]) for s in sessions}
+        # We need to match lines to sessions — titles are unique
+        title_to_sid = {}
+        for s in sessions:
+            title_to_sid[s["title"]] = s["session_id"]
+        _, lines = _run_backtest(_FIXTURE_PATH)
+        parsed = [_parse_line(l) for l in lines if _parse_line(l)]
+        assert len(parsed) == len(sessions)
+        for p in parsed:
+            sid = title_to_sid.get(p["title"])
+            if sid:
+                assert p["calls"] == sid_to_calls[sid], (
+                    f"Call count mismatch for {sid}: expected {sid_to_calls[sid]}, got {p['calls']}"
+                )
+
+    def test_output_includes_summary_in_stderr(self):
+        """Stderr includes a summary line like '# backtest complete: N/M flagged'."""
+        result, _lines = _run_backtest(_FIXTURE_PATH)
+        summary_re = re.compile(r"# backtest complete: \d+/\d+ flagged")
+        assert summary_re.search(result.stderr), (
+            f"Expected summary in stderr, got: {result.stderr}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# empty fixture edge case
+# ---------------------------------------------------------------------------
+
+
+class TestEmptyFixture:
+    """Test edge cases with empty or minimal fixture data."""
+
+    def test_empty_fixture_file(self, tmp_path):
+        """Empty fixture produces zero session lines."""
+        fixture_file = tmp_path / "empty_fixture.json"
+        fixture_file.write_text("[]")
+        result, lines = _run_backtest(str(fixture_file))
+        assert result.returncode == 0
+        # Only the summary line should remain after filtering
+        assert len(lines) == 0, f"Expected 0 lines for empty fixture, got: {lines}"
+
+    def test_fixture_with_too_few_calls(self, tmp_path):
+        """Fixture with sessions under min_calls produces all ok."""
+        fixture_file = tmp_path / "few_calls.json"
+        # Build a minimal fixture with 20 calls (below min_calls=40)
+        calls = []
+        for i in range(20):
+            calls.append({
+                "t": 1790000000000 + i,
+                "tool": "read",
+                "args": {"filePath": f"/tmp/file_{i}.py"},
+                "landed": False,
+            })
+        fixture_data = [{
+            "session_id": "ses_empty_test",
+            "title": "Tiny session",
+            "agent": "test",
+            "calls": calls,
+            "file_lines": {},
+            "label": "must_not_flag",
+        }]
+        fixture_file.write_text(json.dumps(fixture_data))
+        result, lines = _run_backtest(str(fixture_file))
+        assert result.returncode == 0
+        assert len(lines) == 1
+        parsed = _parse_line(lines[0])
+        assert parsed is not None
+        assert parsed["status"] == "ok"
+        assert parsed["first_flag_at"] is None

tests/test_progress_detect.py

@@ -0,0 +1,412 @@
+"""Tests for progress_detect.py — progress detection signals.
+
+Covers target_of extraction, window stats, coverage, evaluate
+logic (normal, read-only, landed suppression, slow repeat),
+and the full fixture-driven label verification.
+"""
+
+import json
+import os
+
+from progress_detect import (
+    DetectConfig,
+    _normalize_bash,
+    coverage,
+    evaluate,
+    target_of,
+    window_stats,
+)
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+def _call(t, tool, args, landed=False):
+    """Build a call tuple as progress_detect.evaluate expects."""
+    return (t, tool, json.dumps(args), landed)
+
+
+def _calls_from_fixture(session):
+    """Convert a fixture session dict to call tuples."""
+    return [
+        (c["t"], c["tool"], json.dumps(c["args"]), c["landed"])
+        for c in session["calls"]
+    ]
+
+
+def _file_lines_fn(file_lines_dict):
+    """Return a callable for coverage() from a file-path→line-count dict."""
+    def fn(fp):
+        if fp is None:
+            return 0
+        return file_lines_dict.get(fp, 0) or 0
+    return fn
+
+
+_DEFAULT_CFG = DetectConfig()
+_FIXTURE_PATH = os.path.join(
+    os.path.dirname(__file__), "fixtures", "progress", "fixture.json"
+)
+
+# ---------------------------------------------------------------------------
+# target_of
+# ---------------------------------------------------------------------------
+
+
+class TestTargetOf:
+    """target_of(tool, args_json) -> tuple key for counting."""
+
+    def test_read_with_offset_limit(self):
+        """read with filePath, offset, limit -> file key with offset/limit."""
+        args = json.dumps({"filePath": "/home/user/proj/main.py", "offset": 10, "limit": 50})
+        result = target_of("read", args)
+        assert result[0] == "file"
+        assert result[1] == "main.py"
+        assert result[2] == 10
+        assert result[3] == 50
+
+    def test_read_without_offset_limit(self):
+        """read with filePath only -> file key, offset/limit remain None."""
+        args = json.dumps({"filePath": "/home/user/proj/data.json"})
+        result = target_of("read", args)
+        assert result[0] == "file"
+        assert result[1] == "data.json"
+        assert result[2] is None
+        assert result[3] is None
+
+    def test_bash_with_git_files(self):
+        """bash with git commands referencing files -> bash-files key."""
+        args = json.dumps({"command": "git add main.py && git commit -m 'fix' && git push origin main"})
+        result = target_of("bash", args)
+        assert result[0] == "bash-files"
+        assert "main.py" in result[1]
+
+    def test_bash_no_files(self):
+        """bash with no file-ish args -> bash key with first words."""
+        args = json.dumps({"command": "echo hello world"})
+        result = target_of("bash", args)
+        assert result[0] == "bash"
+        assert "echo" in result[1]
+
+    def test_edit_with_filepath(self):
+        """edit with filePath -> file key only."""
+        args = json.dumps({"filePath": "/home/user/proj/app.py", "oldString": "foo", "newString": "bar"})
+        result = target_of("edit", args)
+        assert result[0] == "file"
+        assert result[1] == "app.py"
+
+    def test_grep_with_pattern(self):
+        """grep -> (grep, pattern[:60])."""
+        args = json.dumps({"pattern": "def test_"})
+        result = target_of("grep", args)
+        assert result[0] == "grep"
+        assert "def test_" in result[1]
+
+    def test_glob_with_pattern(self):
+        """glob -> (glob, pattern[:60])."""
+        args = json.dumps({"pattern": "**/*.py"})
+        result = target_of("glob", args)
+        assert result[0] == "glob"
+        assert "**/*.py" in result[1]
+
+    def test_invalid_json_fallback(self):
+        """Invalid JSON -> returns (tool, truncated args)."""
+        result = target_of("read", "{{{bad json}}}")
+        assert result[0] == "read"
+        assert len(result[1]) <= 80
+
+    def test_unknown_tool_fallback(self):
+        """Unknown tool with valid JSON -> (tool, truncated JSON)."""
+        result = target_of("some_tool", json.dumps({"k": "v"}))
+        assert result[0] == "some_tool"
+
+    def test_normalize_bash_comment_lines(self):
+        """20 distinct bash-comment commands produce 20 distinct keys."""
+        tokens = set()
+        for i in range(20):
+            cmd = f"# Check file{i}: do something{i}"
+            tokens.add(_normalize_bash(cmd))
+        assert len(tokens) == 20
+
+    def test_normalize_bash_cd_prefix(self):
+        """cd <dir> && ... prefix is stripped so different commands differ."""
+        tokens = set()
+        for i in range(20):
+            cmd = f"cd /x && echo file{i}"
+            tokens.add(_normalize_bash(cmd))
+        assert len(tokens) == 20
+
+    def test_normalize_bash_env_prefix(self):
+        """VAR=val... prefix is stripped so different commands differ."""
+        r1 = _normalize_bash("PYTHONPATH=src python -m a")
+        r2 = _normalize_bash("PYTHONPATH=src python -m b")
+        assert r1 != r2
+        assert "python -m a" == r1
+        assert "python -m b" == r2
+
+
+# ---------------------------------------------------------------------------
+# window_stats
+# ---------------------------------------------------------------------------
+
+
+class TestWindowStats:
+    """window_stats(calls, n) -> (dup, top_n, slow_count)."""
+
+    def test_duplicate_calls_have_positive_dup(self):
+        """Repeated (tool, args) pairs -> dup > 0."""
+        calls = [_call(0, "read", {"filePath": "/x.py"})] * 10 + \
+                [_call(1, "read", {"filePath": "/y.py"})] * 10
+        dup, _top_n, _slow_count = window_stats(calls, 20)
+        assert dup > 0
+
+    def test_all_unique_calls_have_zero_dup(self):
+        """Every call has distinct (tool, args) -> dup = 0."""
+        calls = [_call(i, "read", {"filePath": f"/file_{i}.py"}) for i in range(20)]
+        dup, _top_n, _slow_count = window_stats(calls, 20)
+        assert dup == 0.0
+
+    def test_window_shorter_than_calls(self):
+        """Only last n calls contribute to window stats."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)] + \
+                [_call(i, "bash", {"command": f"echo {i}"}) for i in range(100, 200)]
+        dup, _top_n, _slow_count = window_stats(calls, 10)
+        # Last 10 are bash with distinct commands -> all unique -> dup = 0
+        assert dup == 0.0
+
+
+# ---------------------------------------------------------------------------
+# coverage
+# ---------------------------------------------------------------------------
+
+
+class TestCoverage:
+    """coverage(calls, n, file_lines) -> (ratio, filename)."""
+
+    def test_reads_from_single_file_produce_coverage(self):
+        """Repeated reads of one file produce coverage > 0."""
+        calls = [_call(i, "read", {"filePath": "/project/main.py", "limit": 100}) for i in range(10)]
+        fl = {"/project/main.py": 200}
+        cov, fname = coverage(calls, 20, _file_lines_fn(fl))
+        assert cov > 0
+        assert fname == "main.py"
+
+    def test_no_read_returns_zero(self):
+        """No 'read' calls -> (0.0, '')."""
+        calls = [_call(i, "bash", {"command": "ls"}) for i in range(10)]
+        cov, fname = coverage(calls, 20, _file_lines_fn({}))
+        assert cov == 0.0
+        assert fname == ""
+
+    def test_empty_window_returns_zero(self):
+        """Empty call list -> (0.0, '')."""
+        cov, fname = coverage([], 0, _file_lines_fn({}))
+        assert cov == 0.0
+        assert fname == ""
+
+
+# ---------------------------------------------------------------------------
+# evaluate — pure logic
+# ---------------------------------------------------------------------------
+
+
+class TestEvaluate:
+    """evaluate() — signal detection."""
+
+    def test_not_enough_calls(self):
+        """< 40 calls -> (False, None)."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(30)]
+        flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
+        assert flag is False
+        assert reason is None
+
+    def test_flags_normal_high_dup(self):
+        """Normal agent, no landed, high dup -> flagged."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
+        flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "normal agent")
+        assert flag is True
+        assert reason["dup"] >= _DEFAULT_CFG.dup_min
+        assert reason["landed"] is False
+        assert reason["ro"] is False
+
+    def test_flags_ro_high_top(self):
+        """Read-only agent, high top -> flagged (lower threshold)."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
+        ro_cfg = DetectConfig(read_only_agent_keywords=("explore", "librarian", "oracle"))
+        flag, reason = evaluate(calls, ro_cfg, _file_lines_fn({}), "explore subagent")
+        assert flag is True
+        assert reason["ro"] is True
+
+    def test_skips_with_landed(self):
+        """Landed call within window's time span -> not flagged."""
+        # 60 repeat reads; position 55 has landed=True within the time span
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(60)]
+        calls[55] = _call(55, "read", {"filePath": "/x.py"}, landed=True)
+        flag, _reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
+        assert flag is False
+
+    def test_slow_signal(self):
+        """Single exact (tool, args) pair repeated 15+ times across all calls -> slow flagged."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
+        # window must be >= min_calls to pass the early-exit guard
+        cfg = DetectConfig(window=50, min_calls=40, cum_min=15)
+        flag, reason = evaluate(calls, cfg, _file_lines_fn({}), "test")
+        assert flag is True
+        # The slow signal fires when slow_count >= cum_min; the exact-total
+        # cosmetic relabel only happens when top_n is lower than top_min,
+        # which is not the case when every call targets the same file.
+        assert reason["top"] >= cfg.cum_min
+
+    def test_evaluate_flag_reason_has_expected_keys(self):
+        """Flagged evaluation returns a reason dict with all expected keys."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(100)]
+        flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
+        assert flag is True
+        for key in ("flag", "dup", "top", "top_what", "landed", "ro", "n", "t1"):
+            assert key in reason
+
+    def test_evaluate_exact_min_calls_quiet(self):
+        """Exactly min_calls (40) with diverse commands -> not flagged."""
+        calls = [_call(i, "bash", {"command": f"echo {i}"}) for i in range(40)]
+        flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}), "test")
+        assert flag is False
+        assert reason is None
+
+    def test_evaluate_extra_landed_times(self):
+        """extra_landed_times inside window gates dup/top signals -> not flagged."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(80)]
+        # No own landed, but extra_landed_times has a time inside [t0, t1]
+        extra = [40.0]  # well within the 80-call window's time span
+        flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}),
+                                "test", extra_landed_times=extra)
+        assert flag is False, "Should not flag when extra landed times are inside window"
+        if reason:
+            assert reason["landed"] is True
+
+    def test_extra_landed_times_outside_window(self):
+        """extra_landed_times outside window -> still flagged (not landed)."""
+        calls = [_call(i, "read", {"filePath": "/x.py"}) for i in range(80)]
+        # extra time far outside window's time span
+        extra = [999999.0]
+        flag, reason = evaluate(calls, _DEFAULT_CFG, _file_lines_fn({}),
+                                "test", extra_landed_times=extra)
+        assert flag is True, "Should still flag when extra landed times are outside window"
+        assert reason["landed"] is False
+
+
+# ---------------------------------------------------------------------------
+# fixture labels — end-to-end validation
+# ---------------------------------------------------------------------------
+
+
+class TestFixtureLabels:
+    """Load the fixture and verify every session's label against the detector.
+
+    Uses a sliding window (step 5) from min_calls to len(calls).
+    """
+
+    @staticmethod
+    def _load_fixture():
+        with open(_FIXTURE_PATH) as f:
+            return json.load(f)
+
+    def test_all(self):
+        sessions = self._load_fixture()
+        cfg = _DEFAULT_CFG
+
+        must_flag = [s for s in sessions if s["label"] == "must_flag"]
+        must_not = [s for s in sessions if s["label"] == "must_not_flag"]
+
+        # --- must_not_flag sessions: never flag ----------------------------
+        for s in must_not:
+            sid = s["session_id"]
+            calls = _calls_from_fixture(s)
+            fl = s.get("file_lines", {})
+            fl_fn = _file_lines_fn(fl)
+            flagged = False
+            for i in range(cfg.min_calls, len(calls) + 1, 5):
+                r = evaluate(calls, cfg, fl_fn, s["title"], end_idx=i)
+                if r and r[0]:
+                    flagged = True
+                    break
+            assert not flagged, (
+                f"Session {sid} (must_not_flag) should not flag but did at end_idx={i}"
+            )
+
+        # --- must_flag sessions: at least one window flags ----------------
+        for s in must_flag:
+            sid = s["session_id"]
+            calls = _calls_from_fixture(s)
+            fl = s.get("file_lines", {})
+            fl_fn = _file_lines_fn(fl)
+            flagged = False
+            for i in range(cfg.min_calls, len(calls) + 1, 5):
+                r = evaluate(calls, cfg, fl_fn, s["title"], end_idx=i)
+                if r and r[0]:
+                    flagged = True
+                    break
+            assert flagged, (
+                f"Session {sid} (must_flag) should flag but did not"
+            )
+
+    def test_atlas_207_boundary(self):
+        """Atlas session: no flag before 02:07; at least one flag after.
+
+        02:07 (02:07:36 UTC on Sep 26) is the wall-clock time of the first
+        flagged window's last call.  Before that timestamp there are fewer
+        than min_calls (40), so evaluation cannot produce a signal.
+        """
+        sessions = self._load_fixture()
+        atlas = next(s for s in sessions if s["session_id"] == "ses_f2506ac70ffebLUxBsq0aauOn9")
+        calls = _calls_from_fixture(atlas)
+        fl_fn = _file_lines_fn(atlas.get("file_lines", {}))
+        cfg = _DEFAULT_CFG
+
+        # 02:07 wall-clock cutoff — first flagged window's last-call timestamp
+        cutoff = 1790402856356
+        pre_flagged = []
+        post_flagged = []
+        for i in range(cfg.min_calls, len(calls) + 1, 5):
+            subset = calls[:i]
+            if len(subset) < cfg.min_calls:
+                continue
+            win = subset[-cfg.window:]
+            if len(win) < cfg.min_calls:
+                continue
+            win_end = win[-1][0]
+            r = evaluate(calls, cfg, fl_fn, atlas["title"], end_idx=i)
+            if r and r[0]:
+                if win_end < cutoff:
+                    pre_flagged.append(i)
+                else:
+                    post_flagged.append(i)
+
+        assert len(pre_flagged) == 0, (
+            f"Expected no flags before 02:07, got {len(pre_flagged)} at indices {pre_flagged}"
+        )
+        assert len(post_flagged) >= 1, (
+            "Expected at least one flag after 02:07, got none"
+        )
+
+    def test_specific_sessions_flag(self):
+        """Two explicitly required sessions must flag."""
+        sessions = self._load_fixture()
+        cfg = _DEFAULT_CFG
+
+        for sid in (
+            "ses_f24fcf651ffeOm9O4E14QSZeJ4",   # item 2 worker 1st attempt, 307 calls
+            "ses_f235d755bffeo0uWtD1g2HwQUg",   # helper printed spec 12x, 81 calls
+        ):
+            s = next(x for x in sessions if x["session_id"] == sid)
+            calls = _calls_from_fixture(s)
+            fl_fn = _file_lines_fn(s.get("file_lines", {}))
+            flagged = False
+            for i in range(cfg.min_calls, len(calls) + 1, 5):
+                r = evaluate(calls, cfg, fl_fn, s["title"], end_idx=i)
+                if r and r[0]:
+                    flagged = True
+                    break
+            assert flagged, (
+                f"Session {sid} must flag but did not"
+            )