From bac2a39377160e59ad3988acc5edbea5ab7bb8d5 Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 23:05:27 +0800 Subject: [PATCH] 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 | 211 +++++++++++++++++++++++++++++ cloud/server.py | 77 +++++++++++ cloud/tests/test_routing_shadow.py | 184 +++++++++++++++++++++++++ cloud/worker.py | 11 +- 4 files changed, 481 insertions(+), 2 deletions(-) create mode 100644 cloud/routing_shadow.py create mode 100644 cloud/tests/test_routing_shadow.py diff --git a/cloud/routing_shadow.py b/cloud/routing_shadow.py new file mode 100644 index 0000000..961e7a2 --- /dev/null +++ b/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 `` tier= source= applied=``; + v0b extends it with ``suggested= chosen=`` + (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\S+)\s+tier=(?P\S+)\s+source=(?P\S+)" + r"\s+applied=(?P\S+)" + r"(?:\s+suggested=(?P\S+))?" + r"(?:\s+chosen=(?P\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, + } diff --git a/cloud/server.py b/cloud/server.py index c1e9cd4..c3b1a70 100644 --- a/cloud/server.py +++ b/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: diff --git a/cloud/tests/test_routing_shadow.py b/cloud/tests/test_routing_shadow.py new file mode 100644 index 0000000..da7da32 --- /dev/null +++ b/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) diff --git a/cloud/worker.py b/cloud/worker.py index e2ee95a..31f7529 100644 --- a/cloud/worker.py +++ b/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 保持干净；通用