From 8fe4ad2f3839d0a6fd9b3f4c482841b08f6c0e3c Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 23:50:43 +0800 Subject: [PATCH] feat: routing policy dry-run (v0d, never applies) New module cloud/routing_dryrun.py: dry-run a merged policy document or an env mapping (file first, JSON overrides whole tier entries, CLOUD_ROUTING_MODE overrides global mode only, never explicit tier modes), reporting per-tier effective model/mode/mode-source, dropped tiers with reasons, absent and unknown tiers, and would_activate/would_shadow/would_off. Dropped or unknown tiers block (CLI exit 1); warnings alone (no version, fallback_models parsed but not applied, failed file/JSON sources that the loader silently ignores) do not block but are always printed. Model names are NOT validated against a catalog (deliberately not whitelisted) and sample counts are NOT checked here (that is the shadow report's job); the CLI never writes env, files, or worker state. Tests: test_routing_dryrun.py (8 tests). --- cloud/routing_dryrun.py | 282 +++++++++++++++++++++++++++++ cloud/tests/test_routing_dryrun.py | 169 +++++++++++++++++ 2 files changed, 451 insertions(+) create mode 100644 cloud/routing_dryrun.py create mode 100644 cloud/tests/test_routing_dryrun.py diff --git a/cloud/routing_dryrun.py b/cloud/routing_dryrun.py new file mode 100644 index 0000000..893020e --- /dev/null +++ b/cloud/routing_dryrun.py @@ -0,0 +1,282 @@ +"""Routing policy dry-run (v0d): validate before applying, never apply. + +A routing policy takes effect the moment a worker loads it from the +host env -- there is no CI gate for an env change. A typo in a tier +mode silently drops that tier, a bad model name is never checked +against a whitelist (models are not whitelisted by design), and a +policy file that fails to parse is silently ignored by the loader +(fail-closed to the worker default, but also silently). This module +surfaces all of that BEFORE the policy is applied, and it never +applies anything itself: it only reports what WOULD happen if the +given policy document were loaded. + +What it reports, per tier (plan / goal / default): +* the effective model and mode, and whether the mode came from an + explicit tier mode or from the global mode, +* tiers that were dropped, and why (invalid tier mode, no model, + unrecognised entry) -- a dropped tier is a BLOCKER, not a warning + to skim past, because applying the policy would silently not route + that tier, +* tiers absent from the document (worker default, not routed), +* unknown tier names in the document (ignored by the parser), +* whether any tier would actually activate on next dispatch. + +What it does NOT do, honestly: +* It does NOT validate model names against a provider or relay + catalog. Task models are deliberately not whitelisted in this + codebase (the relay catalog changes), so a model typo here is + reported as "unverified, not whitelisted" only if the caller + passes a known-model set; without one, model names are reported + as-is and the caller must check them against the relay catalog + themselves. This module never invents a catalog. +* It does NOT check shadow sample counts. That is the shadow + report's job (GET /routing-shadow); a dry-run that said "ready + to activate" without sample data would be fabrication. +* It does NOT change host env, worker state, or any file. The CLI + exits 1 when the policy is invalid or any tier was dropped or + unknown, so a deploy script can gate on it; warnings alone + (no version, fallback_models present) do not block, but are + always printed. + +Loading order, mirrored from load_routing_policy in models.py: +file (CLOUD_ROUTING_POLICY_FILE) first, then JSON +(CLOUD_ROUTING_POLICY_JSON) overrides whole tier entries (not +per-field: a JSON tier entry replaces the file's tier entry +entirely), then CLOUD_ROUTING_MODE overrides the GLOBAL mode +only -- it never overrides an explicit tier mode. Failed file/JSON +sources, which the loader silently ignores, are reported here as +warnings instead of being swallowed. +""" + +import argparse +import json +import os +import sys + +try: + from cloud.models import ROUTING_MODES, ROUTING_TIERS + from cloud.models import parse_routing_policy +except ImportError: # run as a script from the repo root + sys.path.insert(0, os.path.join( + os.path.dirname(os.path.abspath(__file__)), "..")) + from cloud.models import ROUTING_MODES, ROUTING_TIERS + from cloud.models import parse_routing_policy + + +def _tier_dropped_reason(raw_entry): + if isinstance(raw_entry, dict): + if "mode" in raw_entry and \ + raw_entry.get("mode") not in ROUTING_MODES: + return "invalid tier mode %r" % (raw_entry.get("mode"),) + return "no usable model" + if isinstance(raw_entry, str): + return "empty model" if not raw_entry.strip() else "unknown" + return "unrecognised tier entry (expected string or dict)" + + +def dry_run_policy_document(data, env_mode_override=None): + """Dry-run one merged policy document (pure, no IO). + + ``data`` is the merged document as parse_routing_policy would + receive it. ``env_mode_override`` is the raw CLOUD_ROUTING_MODE + value (or None): a valid value overrides the global mode only, + an invalid value is ignored with a warning. + """ + warnings = [] + tiers_report = {} + dropped = [] + unknown = [] + if not isinstance(data, dict): + return { + "valid": False, "blocked": True, "version": None, + "global_mode": None, "global_mode_source": None, + "tiers": tiers_report, "dropped_tiers": dropped, + "unknown_tiers": unknown, + "warnings": ["policy document is not an object; " + "the loader would ignore it entirely"], + "would_activate": [], "would_shadow": [], "would_off": [], + } + raw_tiers = data.get("tiers") + if isinstance(raw_tiers, dict): + for name in sorted(raw_tiers): + if name not in ROUTING_TIERS: + unknown.append(name) + warnings.append( + "unknown tier %r is ignored by the parser" % name) + policy = parse_routing_policy(data) + if policy is None: + warnings.append( + "policy is unusable (no usable tiers, or global mode " + "outside off/shadow/active); the loader would ignore it " + "entirely and workers would keep the worker default") + return { + "valid": False, "blocked": True, "version": None, + "global_mode": None, "global_mode_source": None, + "tiers": tiers_report, "dropped_tiers": dropped, + "unknown_tiers": unknown, "warnings": warnings, + "would_activate": [], "would_shadow": [], "would_off": [], + } + global_mode = policy.mode + global_source = "policy" + if env_mode_override: + if env_mode_override in ROUTING_MODES: + global_mode = env_mode_override + global_source = "env_override" + warnings.append( + "CLOUD_ROUTING_MODE overrides the global mode to %r, " + "but does NOT override explicit tier modes" % + env_mode_override) + else: + warnings.append( + "CLOUD_ROUTING_MODE %r is invalid and ignored; " + "the global mode from the policy stands" % + env_mode_override) + # Apply the env global override to the effective per-tier modes + # without mutating the parsed policy: a tier with an explicit + # mode keeps it, a tier without one follows the overridden global. + effective_policy_mode = global_mode + for tier in ROUTING_TIERS: + raw_entry = raw_tiers.get(tier) if isinstance(raw_tiers, + dict) else None + if tier not in policy.tiers: + if raw_entry is not None: + reason = _tier_dropped_reason(raw_entry) + dropped.append(tier) + warnings.append( + "tier %r was dropped: %s; applying this policy " + "would silently not route %r tasks" % + (tier, reason, tier)) + tiers_report[tier] = { + "model": None, "mode": None, + "mode_source": None, "status": "dropped", + "dropped_reason": reason, + } + else: + tiers_report[tier] = { + "model": None, "mode": None, + "mode_source": None, "status": "absent", + "dropped_reason": None, + } + continue + if tier in policy.tier_modes: + mode = policy.tier_modes[tier] + mode_source = "tier_mode" + else: + mode = effective_policy_mode + mode_source = "global_mode" + status = "active" if mode == "active" else ( + "shadow" if mode == "shadow" else "off") + tiers_report[tier] = { + "model": policy.tiers[tier], "mode": mode, + "mode_source": mode_source, "status": status, + "dropped_reason": None, + } + would_activate = [t for t in ROUTING_TIERS + if tiers_report[t]["status"] == "active"] + would_shadow = [t for t in ROUTING_TIERS + if tiers_report[t]["status"] == "shadow"] + would_off = [t for t in ROUTING_TIERS + if tiers_report[t]["status"] == "off"] + if policy.version is None: + warnings.append( + "no policy version: policy changes will be hard to " + "attribute in shadow reports (decisions record " + "policy_version)") + if data.get("fallback_models") is not None: + warnings.append( + "fallback_models in the policy are parsed but NOT " + "applied; only worker CLI --fallback-models enables a " + "real fallback retry") + if would_activate: + warnings.append( + "activation warning: tier(s) %s would actually route " + "tasks on next worker dispatch if this policy were " + "applied; this dry-run does not activate anything" % + ", ".join(would_activate)) + blocked = bool(dropped or unknown) + return { + "valid": True, "blocked": blocked, + "version": policy.version, "global_mode": global_mode, + "global_mode_source": global_source, + "tiers": tiers_report, "dropped_tiers": dropped, + "unknown_tiers": unknown, "warnings": warnings, + "would_activate": would_activate, + "would_shadow": would_shadow, "would_off": would_off, + } + + +def dry_run_from_env(env=None): + """Dry-run from an env mapping, mirroring load_routing_policy. + + Returns the report dict. File/JSON failures that the loader + silently ignores are reported as warnings here. + """ + env = env if env is not None else os.environ + warnings = [] + merged = None + path = env.get("CLOUD_ROUTING_POLICY_FILE") or "" + if path: + try: + with open(path, encoding="utf-8") as fh: + merged = json.load(fh) + except (OSError, ValueError) as exc: + warnings.append( + "policy file %r could not be read/parsed (%s); " + "the loader would ignore it" % (path, exc)) + merged = None + raw = env.get("CLOUD_ROUTING_POLICY_JSON") or "" + if raw: + try: + override = json.loads(raw) + except ValueError as exc: + warnings.append( + "CLOUD_ROUTING_POLICY_JSON could not be parsed " + "(%s); the loader would ignore it" % exc) + override = None + if isinstance(override, dict): + if not isinstance(merged, dict): + merged = override + else: + merged = dict(merged) + merged.update(override) + if isinstance(merged.get("tiers"), dict) and \ + isinstance(override.get("tiers"), dict): + tiers = dict(merged["tiers"]) + tiers.update(override["tiers"]) + merged["tiers"] = tiers + report = dry_run_policy_document( + merged, env_mode_override=env.get("CLOUD_ROUTING_MODE") or None) + # Source-failure warnings belong before the document warnings, + # and they do not change blocked (a failed source alone is not + # a dropped tier; the document report already covers validity). + report["warnings"] = warnings + report["warnings"] + return report + + +def main(argv=None): + parser = argparse.ArgumentParser( + description="Dry-run a routing policy (never applies it).") + parser.add_argument("--policy-file", default=None, + help="Policy JSON file (same as " + "CLOUD_ROUTING_POLICY_FILE)") + parser.add_argument("--policy-json", default=None, + help="Policy JSON string (same as " + "CLOUD_ROUTING_POLICY_JSON)") + parser.add_argument("--mode", default=None, + help="Global mode override (same as " + "CLOUD_ROUTING_MODE)") + args = parser.parse_args(argv) + env = {} + if args.policy_file: + env["CLOUD_ROUTING_POLICY_FILE"] = args.policy_file + if args.policy_json: + env["CLOUD_ROUTING_POLICY_JSON"] = args.policy_json + if args.mode: + env["CLOUD_ROUTING_MODE"] = args.mode + report = dry_run_from_env(env) + print(json.dumps(report, indent=2, sort_keys=True)) + return 1 if (not report["valid"] or report["blocked"]) else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cloud/tests/test_routing_dryrun.py b/cloud/tests/test_routing_dryrun.py new file mode 100644 index 0000000..7deb4ef --- /dev/null +++ b/cloud/tests/test_routing_dryrun.py @@ -0,0 +1,169 @@ +"""策略干跑校验 v0d 单测：有效档、丢档、未知档、环境覆盖、CLI。 + +运行（在仓库根目录下）： + python3 -m unittest discover -s cloud/tests -p "test_routing_dryrun.py" -v + +说明：纯函数部分用 plain dict，不碰库、不联网、不跑模型； +CLI 部分用子进程跑 `python3 -m cloud.routing_dryrun`，只读 +传入的策略，不改任何文件或环境。关键断言：干跑本身不 +激活任何东西（只报告 would_activate）；丢档会阻塞退出码； +文件/JSON 合并是整档替换、不是按字段合并；环境模式只 +覆盖全局、不覆盖显式档模式。 +""" + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.routing_dryrun import dry_run_from_env + from cloud.routing_dryrun import dry_run_policy_document +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.routing_dryrun import dry_run_from_env + from cloud.routing_dryrun import dry_run_policy_document + + +class DryRunDocumentTestCase(unittest.TestCase): + def test_plan_active_only(self): + report = dry_run_policy_document({ + "version": "v1", "mode": "shadow", + "tiers": { + "plan": {"model": "gpt-6.1-sol", "mode": "active"}, + "goal": "gpt-6.1-sol", + "default": "gpt-6.1-sol", + }, + }) + self.assertTrue(report["valid"]) + self.assertFalse(report["blocked"]) + self.assertEqual(report["tiers"]["plan"]["mode"], "active") + self.assertEqual(report["tiers"]["plan"]["mode_source"], + "tier_mode") + self.assertEqual(report["tiers"]["goal"]["mode"], "shadow") + self.assertEqual(report["tiers"]["goal"]["mode_source"], + "global_mode") + self.assertEqual(report["would_activate"], ["plan"]) + + def test_dropped_tier_blocks(self): + report = dry_run_policy_document({ + "mode": "active", + "tiers": { + "plan": {"model": "gpt-6.1-sol", "mode": "activ"}, + "default": "gpt-6.1-sol", + }, + }) + self.assertTrue(report["valid"]) + self.assertTrue(report["blocked"]) + self.assertEqual(report["dropped_tiers"], ["plan"]) + self.assertEqual(report["tiers"]["plan"]["status"], "dropped") + self.assertEqual(report["tiers"]["default"]["status"], "active") + + def test_unknown_tier_and_absent_tier(self): + report = dry_run_policy_document({ + "mode": "shadow", + "tiers": {"plan": "gpt-6.1-sol", "weird": "gpt-6.1-sol"}, + }) + self.assertEqual(report["unknown_tiers"], ["weird"]) + self.assertTrue(report["blocked"]) + self.assertEqual(report["tiers"]["goal"]["status"], "absent") + self.assertIsNone(report["tiers"]["goal"]["model"]) + + def test_invalid_global_mode_and_no_version_warning(self): + report = dry_run_policy_document({ + "mode": "bogus", "tiers": {"plan": "gpt-6.1-sol"}, + }) + self.assertFalse(report["valid"]) + self.assertTrue(report["blocked"]) + valid = dry_run_policy_document({ + "mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"}, + "fallback_models": ["gpt-5.6-terra"], + }) + self.assertTrue(any("no policy version" in w + for w in valid["warnings"])) + self.assertTrue(any("fallback_models" in w + for w in valid["warnings"])) + + def test_env_mode_overrides_global_not_tier(self): + report = dry_run_policy_document({ + "mode": "shadow", + "tiers": { + "plan": {"model": "gpt-6.1-sol", "mode": "active"}, + "default": "gpt-6.1-sol", + }, + }, env_mode_override="active") + self.assertEqual(report["global_mode"], "active") + self.assertEqual(report["global_mode_source"], "env_override") + self.assertEqual(report["tiers"]["plan"]["mode"], "active") + self.assertEqual(report["tiers"]["default"]["mode"], "active") + bad = dry_run_policy_document({ + "mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"}, + }, env_mode_override="bogus") + self.assertEqual(bad["global_mode"], "shadow") + self.assertTrue(any("invalid and ignored" in w + for w in bad["warnings"])) + + +class DryRunEnvTestCase(unittest.TestCase): + def test_file_json_merge_replaces_whole_tier(self): + with tempfile.TemporaryDirectory() as tmpdir: + policy_path = os.path.join(tmpdir, "policy.json") + with open(policy_path, "w", encoding="utf-8") as fh: + json.dump({ + "mode": "shadow", + "tiers": { + "plan": {"model": "gpt-6.1-sol", + "mode": "active"}, + "default": "gpt-6.1-sol", + }, + }, fh) + # JSON 覆盖 plan 为字符串：整档替换，档模式丢失， + # plan 回到全局 shadow，而不是字段合并保留 active。 + report = dry_run_from_env({ + "CLOUD_ROUTING_POLICY_FILE": policy_path, + "CLOUD_ROUTING_POLICY_JSON": json.dumps({ + "tiers": {"plan": "gpt-6.1-sol"}, + }), + }) + self.assertEqual(report["tiers"]["plan"]["mode"], "shadow") + self.assertEqual(report["tiers"]["plan"]["mode_source"], + "global_mode") + + def test_unreadable_file_is_warning_not_silent(self): + report = dry_run_from_env({ + "CLOUD_ROUTING_POLICY_FILE": "/nonexistent/policy.json", + "CLOUD_ROUTING_POLICY_JSON": json.dumps({ + "mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"}, + }), + }) + self.assertTrue(report["valid"]) + self.assertTrue(any("could not be read" in w + for w in report["warnings"])) + + def test_cli_exit_codes(self): + repo_root = str(Path(__file__).resolve().parents[2]) + good = subprocess.run( + [sys.executable, "-m", "cloud.routing_dryrun", + "--policy-json", json.dumps({ + "mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"}, + })], + cwd=repo_root, capture_output=True, text=True, timeout=30) + self.assertEqual(good.returncode, 0, good.stderr) + report = json.loads(good.stdout) + self.assertTrue(report["valid"]) + bad = subprocess.run( + [sys.executable, "-m", "cloud.routing_dryrun", + "--policy-json", json.dumps({ + "mode": "active", + "tiers": { + "plan": {"model": "gpt-6.1-sol", "mode": "activ"}, + "default": "gpt-6.1-sol", + }, + })], + cwd=repo_root, capture_output=True, text=True, timeout=30) + self.assertEqual(bad.returncode, 1, bad.stdout) + self.assertEqual(json.loads(bad.stdout)["dropped_tiers"], + ["plan"])