feat: routing shadow evaluation v0b (read-only, never auto-activates)

New pure module cloud/routing_shadow.py: parse worker_routing
audit details (old format without suggested model parses with
suggested unknown, never guessed), evaluate per-tier routed
samples with one-task-one-sample dedup, explicit tasks counted
separately, status under the chosen model, token/USD with
unpriced tokens kept apart, and a min-20 gate per tier. The
report explicitly marks the counterfactual not_measured:
shadow suggested models never ran, so no suggested-vs-chosen
cost/quality comparison is possible. Worker audit detail now
includes suggested/chosen for future rows. GET /routing-shadow
serves the report (project read permission, truncated flag
over 10000 audit rows, cost table ensured so a fresh DB does
not 500).

Tests: test_routing_shadow.py (5 tests).

cloud/routing_shadow.py

@@ -0,0 +1,211 @@
+"""Routing v0b: shadow evaluation (read-only, never auto-activates).
+
+This module answers the one question routing v0 left unanswered:
+"is there enough shadow evidence to review a tier for activation?"
+It does NOT decide activation, and it does NOT claim a counterfactual.
+
+Why the counterfactual is explicitly not measured
+--------------------------------------------------
+In shadow mode the suggested model never runs. The task's actual
+outcome (status, tokens, cost) is the outcome under the chosen
+worker-default model, not under the suggested model. So this report
+can measure, per tier:
+
+* how many distinct tasks were routed (one task = one sample,
+  retries do not double-count),
+* their status breakdown under the chosen model,
+* the distribution of suggested models,
+* total tokens and priced USD, with unpriced tokens kept separate,
+
+but it cannot say the suggested model "would have been cheaper or
+better". Any report that compared shadow suggested cost to actual
+cost as if both had run would be fabrication. The field
+``counterfactual`` in every report says ``not_measured`` for exactly
+this reason, and the top-level recommendation never activates a tier:
+activation stays a manual decision, plan tier first, per the v0 design.
+
+Data sources, and their limits
+-------------------------------
+* ``audit_log`` rows with action ``worker_routing``. The v0 format
+  was ``<task_id> tier=<tier> source=<source> applied=<bool>``;
+  v0b extends it with ``suggested=<model|-> chosen=<model|->``
+  (see worker.py). Rows in the old format parse fine, with
+  ``suggested_model`` reported as unknown and counted separately
+  (``suggested_unknown_n``), never guessed.
+* ``tasks`` rows for status and explicit-model separation. A task
+  with a creation-time ``model`` is explicit: routing never audits
+  it (decide_routing returns source ``explicit`` and the worker
+  does not audit), so explicit tasks are counted separately and
+  never inflate a tier's sample count.
+* ``cost_records`` aggregated per task. USD here is whatever the
+  cost store recorded: with the current proxy-price overlay that is
+  the gpt-6.1-sol proxy price, NOT the relay's actual bill, and
+  unpriced rows contribute tokens but no USD. The report keeps
+  ``usd_micros`` (priced only) and ``unpriced_tokens`` apart and
+  labels USD as proxy-priced when any unpriced tokens exist.
+
+Pure functions only: this module never touches a DB, a provider,
+or an environment variable, so it is unit-testable with plain dicts.
+The server endpoint (GET /routing-shadow) is the only DB reader,
+and it passes already-aggregated rows in.
+"""
+
+import re
+
+MIN_SAMPLES = 20
+TIERS = ("plan", "goal", "default")
+ACTIVATION_ORDER = ("plan", "goal", "default")
+
+_DETAIL_RE = re.compile(
+    r"^(?P<task_id>\S+)\s+tier=(?P<tier>\S+)\s+source=(?P<source>\S+)"
+    r"\s+applied=(?P<applied>\S+)"
+    r"(?:\s+suggested=(?P<suggested>\S+))?"
+    r"(?:\s+chosen=(?P<chosen>\S+))?\s*$")
+
+
+def parse_worker_routing_detail(detail):
+    """Parse one worker_routing audit detail, or None if not parseable.
+
+    Old format (no suggested/chosen) parses with both models None.
+    A ``-`` placeholder also means None, never a model named "-".
+    """
+    if not isinstance(detail, str):
+        return None
+    match = _DETAIL_RE.match(detail.strip())
+    if not match:
+        return None
+    applied_raw = match.group("applied").lower()
+    applied = applied_raw in ("true", "1", "yes")
+    suggested = match.group("suggested")
+    chosen = match.group("chosen")
+    return {
+        "task_id": match.group("task_id"),
+        "tier": match.group("tier"),
+        "source": match.group("source"),
+        "applied": applied,
+        "suggested_model": None if suggested in (None, "-") else suggested,
+        "chosen_model": None if chosen in (None, "-") else chosen,
+    }
+
+
+def _empty_tier():
+    return {
+        "routed_n": 0,
+        "shadow_n": 0,
+        "policy_n": 0,
+        "completed": 0,
+        "failed": 0,
+        "other_status": 0,
+        "unknown_status": 0,
+        "total_tokens": 0,
+        "usd_micros": 0,
+        "unpriced_tokens": 0,
+        "suggested_models": {},
+        "suggested_unknown_n": 0,
+        "gate": "not_ready",
+        "sample_shortfall": MIN_SAMPLES,
+    }
+
+
+def evaluate_shadow(audit_rows, tasks_by_id, costs_by_task,
+                    explicit_n=0, min_samples=MIN_SAMPLES):
+    """Evaluate shadow evidence from already-fetched rows.
+
+    ``audit_rows``: dicts with at least ``detail`` (and optionally
+    ``created_at`` for earliest-first dedup; without it, first-seen
+    wins). ``tasks_by_id``: task_id -> {"status": str}.
+    ``costs_by_task``: task_id -> {"total_tokens": int,
+    "usd_micros": int|None, "unpriced_tokens": int}.
+    ``explicit_n``: number of tasks with a creation-time model in
+    the same project (counted separately, never as routed samples).
+    """
+    tiers = {tier: _empty_tier() for tier in TIERS}
+    seen = set()
+    routed_task_ids = set()
+    for row in audit_rows or []:
+        parsed = parse_worker_routing_detail(
+            (row or {}).get("detail"))
+        if parsed is None:
+            continue
+        task_id = parsed["task_id"]
+        if task_id in seen:
+            continue
+        seen.add(task_id)
+        tier = parsed["tier"]
+        if tier not in tiers:
+            # Unknown tier: do not invent a bucket, do not count it
+            # as a sample for any known tier.
+            continue
+        bucket = tiers[tier]
+        routed_task_ids.add(task_id)
+        bucket["routed_n"] += 1
+        if parsed["source"] == "shadow":
+            bucket["shadow_n"] += 1
+        elif parsed["source"] == "policy":
+            bucket["policy_n"] += 1
+        else:
+            # Other sources (e.g. a future source) are routed
+            # samples, but neither shadow nor applied-policy samples.
+            pass
+        task = tasks_by_id.get(task_id) if tasks_by_id else None
+        status = (task or {}).get("status") if task else None
+        if status == "completed":
+            bucket["completed"] += 1
+        elif status == "failed":
+            bucket["failed"] += 1
+        elif status is None:
+            bucket["unknown_status"] += 1
+        else:
+            bucket["other_status"] += 1
+        cost = costs_by_task.get(task_id) if costs_by_task else None
+        if cost:
+            bucket["total_tokens"] += int(cost.get("total_tokens") or 0)
+            if cost.get("usd_micros") is not None:
+                bucket["usd_micros"] += int(cost["usd_micros"])
+            bucket["unpriced_tokens"] += int(
+                cost.get("unpriced_tokens") or 0)
+        if parsed["suggested_model"]:
+            models = bucket["suggested_models"]
+            models[parsed["suggested_model"]] = \
+                models.get(parsed["suggested_model"], 0) + 1
+        else:
+            bucket["suggested_unknown_n"] += 1
+    for tier, bucket in tiers.items():
+        bucket["gate"] = (
+            "ready_for_review" if bucket["routed_n"] >= min_samples
+            else "not_ready")
+        bucket["sample_shortfall"] = max(
+            0, min_samples - bucket["routed_n"])
+    ready = [t for t in ACTIVATION_ORDER
+             if tiers[t]["gate"] == "ready_for_review"]
+    if ready:
+        recommendation = (
+            "Ready for manual review (not activation): %s. "
+            "Review plan tier first, then goal, then default; "
+            "activation stays a manual decision and is never "
+            "automatic from this report." % ", ".join(ready))
+    else:
+        recommendation = (
+            "Not ready for review: no tier has %d routed samples. "
+            "Keep shadow mode running; do not activate any tier."
+            % min_samples)
+    return {
+        "min_samples": min_samples,
+        "tiers": tiers,
+        "explicit_n": int(explicit_n or 0),
+        "routed_task_n": len(routed_task_ids),
+        "counterfactual": "not_measured",
+        "counterfactual_note": (
+            "Shadow suggested models never ran, so this report cannot "
+            "compare suggested vs chosen cost or quality; it measures "
+            "sample size, status under the chosen model, and the "
+            "suggested-model distribution only."),
+        "usd_note": (
+            "usd_micros sums priced cost records only; unpriced "
+            "tokens are kept separate. With the current proxy-price "
+            "overlay, USD is proxy-priced (gpt-6.1-sol), not the "
+            "relay's actual bill."),
+        "activation_order": list(ACTIVATION_ORDER),
+        "ready_for_review": ready,
+        "recommendation": recommendation,
+    }

cloud/server.py

@@ -66,6 +66,7 @@ from cloud.auth import AuthStore
 from cloud.coordinator import parse_parts as _coord_parse_parts
 from cloud.coordinator import validate_parts as _coord_validate_parts
 from cloud.cost import CostStore, unpriced_token_cap
+from cloud.routing_shadow import evaluate_shadow as _evaluate_shadow
 from cloud.db import connect as _db_connect
 from cloud.db import create_database as _create_database
 from cloud.db import init_postgres_schema as _init_postgres_schema
@@ -2341,6 +2342,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._create_upload(user_id)
         elif path == "/tasks" and method == "GET":
             self._list_tasks(user_id, query)
+        elif path == "/routing-shadow" and method == "GET":
+            self._routing_shadow(user_id, query)
         elif path == "/admin/sweep" and method == "POST":
             self._admin_sweep(user_id)
         elif path == "/admin/backup" and method == "POST":
@@ -4129,6 +4132,80 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                               "daily": daily})
 
     # -- 任务成本与模型总览 ----------------------------------------------
+    def _routing_shadow(self, user_id, query):
+        # GET /routing-shadow?project_id=...：路由 v0b 影子评估，
+        # 只读，不改任何模式、不激活任何档。权限同 _list_tasks：
+        # 项目 read 权限，没有则 403。审计行按 created_at 升序
+        # 取（评估里按任务去重，第一次出现即样本），上限 10000
+        # 行——超过时报告里仍按取到的行算，调用方须知样本被
+        # 截断（见 response 的 truncated 字段），不假装全量。
+        project_ids = query.get("project_id")
+        project_id = project_ids[0] if project_ids else None
+        if not project_id:
+            self._send_json(400, {"error": "project_id required"})
+            return
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(user_id, project_id, "read"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        # cost_records 表只在 CostStore 首次初始化时建：新鲜
+        # 库上直接查会报 no such table，报告端点不应因此 500。
+        # 先建表（幂等），再读；建表失败时成本按空算、报告
+        # 里 token/usd 为 0，不假装成本已计。
+        try:
+            _cost_store = CostStore(self.server.db_path)
+            try:
+                _cost_store._conn.close()
+            except Exception:
+                pass
+        except Exception:
+            pass
+        conn = self._db()
+        try:
+            audit_rows = conn.execute(
+                "SELECT detail FROM audit_log"
+                " WHERE project_id = ? AND action = 'worker_routing'"
+                " ORDER BY created_at, id LIMIT 10001",
+                (project_id,),
+            ).fetchall()
+            truncated = len(audit_rows) > 10000
+            audit_rows = audit_rows[:10000]
+            task_rows = conn.execute(
+                "SELECT id, status, model FROM tasks"
+                " WHERE project_id = ?",
+                (project_id,),
+            ).fetchall()
+            cost_rows = conn.execute(
+                "SELECT task_id, total_tokens, usd_micros"
+                " FROM cost_records WHERE project_id = ?",
+                (project_id,),
+            ).fetchall()
+        finally:
+            conn.close()
+        tasks_by_id = {}
+        explicit_n = 0
+        for row in task_rows:
+            tasks_by_id[row["id"]] = {"status": row["status"]}
+            if row["model"]:
+                explicit_n += 1
+        costs_by_task = {}
+        for row in cost_rows:
+            entry = costs_by_task.setdefault(
+                row["task_id"],
+                {"total_tokens": 0, "usd_micros": 0,
+                 "unpriced_tokens": 0})
+            entry["total_tokens"] += int(row["total_tokens"] or 0)
+            if row["usd_micros"] is None:
+                entry["unpriced_tokens"] += int(row["total_tokens"] or 0)
+            else:
+                entry["usd_micros"] += int(row["usd_micros"])
+        report = _evaluate_shadow(
+            [{"detail": row["detail"]} for row in audit_rows],
+            tasks_by_id, costs_by_task, explicit_n=explicit_n)
+        report["project_id"] = project_id
+        report["truncated"] = truncated
+        self._send_json(200, report)
+
     def _task_cost(self, user_id, task_id):
         row = self._fetch_task(task_id)
         if row is None:

cloud/tests/test_routing_shadow.py

@@ -0,0 +1,184 @@
+"""路由影子评估 v0b 单测：解析、按档统计、门禁、服务端点。
+
+运行（在仓库根目录下）：
+    python3 -m unittest discover -s cloud/tests -p "test_routing_shadow.py" -v
+
+说明：纯函数部分用 plain dict，不碰库；服务端点部分用
+create_server 本地 HTTP（test_task_options 同款），审计行
+直接写进测试库（审计表由 AuthStore 自建），不跑真实模型、
+不联网。关键断言：同一任务两次审计只算一个样本；显式
+任务不算路由样本；影子从不声称反事实；未满 20 不 ready。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.routing_shadow import MIN_SAMPLES
+    from cloud.routing_shadow import evaluate_shadow
+    from cloud.routing_shadow import parse_worker_routing_detail
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.routing_shadow import MIN_SAMPLES
+    from cloud.routing_shadow import evaluate_shadow
+    from cloud.routing_shadow import parse_worker_routing_detail
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+
+def _detail(task_id, tier, source, applied, suggested=None, chosen=None):
+    detail = "%s tier=%s source=%s applied=%s" % (
+        task_id, tier, source, applied)
+    if suggested is not None:
+        detail += " suggested=%s" % suggested
+    if chosen is not None:
+        detail += " chosen=%s" % chosen
+    return detail
+
+
+class ParseTestCase(unittest.TestCase):
+    def test_new_and_old_format(self):
+        parsed = parse_worker_routing_detail(
+            _detail("t1", "plan", "shadow", "False",
+                    suggested="gpt-6.1-sol", chosen="gpt-5.6-terra"))
+        self.assertEqual(parsed["suggested_model"], "gpt-6.1-sol")
+        self.assertEqual(parsed["chosen_model"], "gpt-5.6-terra")
+        self.assertFalse(parsed["applied"])
+        old = parse_worker_routing_detail(
+            _detail("t2", "goal", "shadow", "False"))
+        self.assertIsNone(old["suggested_model"])
+        self.assertIsNone(old["chosen_model"])
+
+    def test_unparseable_returns_none(self):
+        self.assertIsNone(parse_worker_routing_detail("garbage"))
+        self.assertIsNone(parse_worker_routing_detail(None))
+
+
+class EvaluateTestCase(unittest.TestCase):
+    def test_dedup_explicit_and_gate(self):
+        audit_rows = [
+            {"detail": _detail("t1", "plan", "shadow", "False",
+                               suggested="gpt-6.1-sol")},
+            # 同一任务重试再记一次：仍只算一个样本。
+            {"detail": _detail("t1", "plan", "shadow", "False",
+                               suggested="gpt-6.1-sol")},
+            {"detail": _detail("t2", "plan", "shadow", "False")},
+            {"detail": _detail("t3", "default", "policy", "True",
+                               suggested="-", chosen="gpt-6.1-sol")},
+        ]
+        tasks = {"t1": {"status": "completed"},
+                 "t2": {"status": "failed"},
+                 "t3": {"status": "completed"}}
+        costs = {"t1": {"total_tokens": 100, "usd_micros": 200,
+                        "unpriced_tokens": 0},
+                 "t2": {"total_tokens": 50, "usd_micros": None,
+                        "unpriced_tokens": 50}}
+        report = evaluate_shadow(audit_rows, tasks, costs, explicit_n=5)
+        plan = report["tiers"]["plan"]
+        self.assertEqual(plan["routed_n"], 2)
+        self.assertEqual(plan["shadow_n"], 2)
+        self.assertEqual(plan["completed"], 1)
+        self.assertEqual(plan["failed"], 1)
+        self.assertEqual(plan["total_tokens"], 150)
+        self.assertEqual(plan["usd_micros"], 200)
+        self.assertEqual(plan["unpriced_tokens"], 50)
+        self.assertEqual(plan["suggested_models"], {"gpt-6.1-sol": 1})
+        self.assertEqual(plan["suggested_unknown_n"], 1)
+        self.assertEqual(plan["gate"], "not_ready")
+        self.assertEqual(plan["sample_shortfall"], MIN_SAMPLES - 2)
+        self.assertEqual(report["explicit_n"], 5)
+        self.assertEqual(report["routed_task_n"], 3)
+        self.assertEqual(report["counterfactual"], "not_measured")
+        self.assertEqual(report["ready_for_review"], [])
+        self.assertIn("Not ready", report["recommendation"])
+
+    def test_gate_ready_at_threshold(self):
+        audit_rows = [
+            {"detail": _detail("t%d" % i, "plan", "shadow", "False",
+                               suggested="gpt-6.1-sol")}
+            for i in range(MIN_SAMPLES)]
+        tasks = {"t%d" % i: {"status": "completed"}
+                 for i in range(MIN_SAMPLES)}
+        report = evaluate_shadow(audit_rows, tasks, {})
+        self.assertEqual(report["tiers"]["plan"]["gate"],
+                         "ready_for_review")
+        self.assertEqual(report["ready_for_review"], ["plan"])
+        self.assertIn("manual review", report["recommendation"])
+        self.assertNotIn("activate plan", report["recommendation"])
+
+
+class ServerShadowTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+        # 审计行与任务行直接写库（审计表由 AuthStore 自建）。
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO audit_log(user_id, project_id, action,"
+                " detail, created_at) VALUES(?, ?, ?, ?, ?)",
+                ("system", self.project["id"], "worker_routing",
+                 _detail("t1", "plan", "shadow", "False",
+                         suggested="gpt-6.1-sol"), 1.0))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        headers = {"Content-Type": "application/json"}
+        if key:
+            headers["X-API-Key"] = key
+        conn.request(method, path,
+                     json.dumps(body) if body is not None else None, headers)
+        resp = conn.getresponse()
+        raw = resp.read()
+        conn.close()
+        try:
+            data = json.loads(raw)
+        except ValueError:
+            data = None
+        return resp.status, data
+
+    def test_endpoint_and_permission(self):
+        status, report = self._req(
+            "GET", "/routing-shadow?project_id=" + self.project["id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200, report)
+        self.assertEqual(report["tiers"]["plan"]["routed_n"], 1)
+        self.assertEqual(report["counterfactual"], "not_measured")
+        self.assertFalse(report["truncated"])
+        # 无 project_id 400；无 key 时 _dispatch 的鉴权在前，
+        # 这里只断言缺 project_id 的 400 已由端点处理。
+        status, _data = self._req(
+            "GET", "/routing-shadow", key=self.owner_key)
+        self.assertEqual(status, 400)

cloud/worker.py

@@ -5920,10 +5920,17 @@ class Worker:
             # 不需要再用一条事件重复说明。
             if decision.source in ("policy", "shadow"):
                 record_routing({"id": task_id, "events": events}, decision)
+                # 影子评估（v0b）需要建议模型：旧格式只有
+                # tier/source/applied，评估时建议模型只能记
+                # unknown；新格式补 suggested/chosen，旧解析
+                # 器仍能读（见 cloud/routing_shadow.py）。
                 self._audit(task["project_id"], "worker_routing",
-                            "%s tier=%s source=%s applied=%s" % (
+                            "%s tier=%s source=%s applied=%s"
+                            " suggested=%s chosen=%s" % (
                                 task_id, decision.tier, decision.source,
-                                decision.applied))
+                                decision.applied,
+                                decision.suggested_model or "-",
+                                decision.chosen_model or "-"))
             task["model"] = decision.chosen_model or self.model
         # agent 指令通道（v2）：local/docker 由 sandbox 的
         # --developer-instructions 派发，prompt 保持干净；通用