host
06e4e16bbb
- I-3: new models.evaluate_capability returning a CapabilityVerdict (ok/missing/unknown from check, known/unknown_missing from split, status/verb derivations); decide_routing, worker gateway check, worker persisted recheck, dry-run tier+gateway checks and server shadow endpoint all call it. Blocking rules stay at the call sites (gateway has no worker-default fallback, recheck blocks only on known, dry-run/shadow never block); all user-visible audit details, dry-run notes and shadow notes are byte-identical. - I-4: new models.merge_policy_documents shared by load_routing_policy and dry_run_from_env; the old tiers-update block is kept verbatim (it was a no-op after update(): JSON tiers replace the whole map, file tiers absent from JSON are dropped -- preserved, not fixed, as that would be a behaviour change). dropped_tiers/partial_failures remain load-time records, not merge products. - Tests: test_audit_batch3.py, 13 tests (verdict matrix + parity with check/split, decide/dry-run wording, merge boundaries + 300 seeded random pairs + loader/dry-run agreement). Full suite: 1694 OK (skipped=11), baseline 1681 + 13, 0 failures.
327 lines
14 KiB
Python
327 lines
14 KiB
Python
"""Audit batch 3 (I-3, I-4): capability verdict unification and policy
|
|
merge unification. Pure refactor: no behaviour change.
|
|
|
|
Run: python3 -m unittest discover -s cloud/tests -p "test_audit_batch3.py" -v
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import random
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
|
|
from cloud.models import (
|
|
check_model_capabilities,
|
|
decide_routing,
|
|
evaluate_capability,
|
|
load_routing_policy,
|
|
merge_policy_documents,
|
|
parse_routing_policy,
|
|
split_model_capability_missing,
|
|
)
|
|
from cloud.routing_dryrun import dry_run_from_env, dry_run_policy_document
|
|
|
|
REQUIRES = {"tool_calling": True, "vision": True,
|
|
"min_context_tokens": 1000}
|
|
|
|
CAPS = {
|
|
"good": {"tool_calling": True, "vision": True,
|
|
"context_tokens": 200000},
|
|
"no-tool": {"tool_calling": False, "vision": True,
|
|
"context_tokens": 200000},
|
|
"no-vision-field": {"tool_calling": True,
|
|
"context_tokens": 200000},
|
|
"small-ctx": {"tool_calling": True, "vision": True,
|
|
"context_tokens": 100},
|
|
"no-ctx": {"tool_calling": True, "vision": True},
|
|
"mixed": {"tool_calling": False, "context_tokens": 200000},
|
|
}
|
|
|
|
|
|
def _policy(mode, tier_model, requires, caps, worker_model=None):
|
|
doc = {
|
|
"version": "batch3",
|
|
"mode": mode,
|
|
"tiers": {"default": {"model": tier_model,
|
|
"requires": requires}},
|
|
"model_capabilities": caps,
|
|
}
|
|
return parse_routing_policy(doc)
|
|
|
|
|
|
class EvaluateCapabilityTestCase(unittest.TestCase):
|
|
def test_matrix_known_unknown_mixed_absent_empty(self):
|
|
cases = [
|
|
# (model, ok, missing, unknown, known, unknown_missing,
|
|
# status, verb)
|
|
("good", True, [], False, [], [], "ok", None),
|
|
("no-tool", False, ["tool_calling"], False,
|
|
["tool_calling"], [], "fail", "lacks"),
|
|
("no-vision-field", False, ["vision"], True,
|
|
[], ["vision"], "unknown", "cannot verify"),
|
|
("small-ctx", False, ["min_context_tokens"], False,
|
|
["min_context_tokens"], [], "fail", "lacks"),
|
|
("no-ctx", False, ["min_context_tokens"], True,
|
|
[], ["min_context_tokens"], "unknown", "cannot verify"),
|
|
("mixed", False, ["tool_calling", "vision"], True,
|
|
["tool_calling"], ["vision"], "unknown", "cannot verify"),
|
|
("absent-model", False,
|
|
["min_context_tokens", "tool_calling", "vision"], True,
|
|
[], ["min_context_tokens", "tool_calling", "vision"],
|
|
"unknown", "cannot verify"),
|
|
]
|
|
for (model, ok, missing, unknown, known, unknown_missing,
|
|
status, verb) in cases:
|
|
with self.subTest(model=model):
|
|
verdict = evaluate_capability(CAPS, model, REQUIRES)
|
|
self.assertEqual(verdict.ok, ok)
|
|
self.assertEqual(verdict.missing, missing)
|
|
self.assertEqual(verdict.unknown, unknown)
|
|
self.assertEqual(verdict.known_missing, known)
|
|
self.assertEqual(verdict.unknown_missing, unknown_missing)
|
|
self.assertEqual(verdict.status, status)
|
|
if verb is not None:
|
|
self.assertEqual(verdict.verb, verb)
|
|
|
|
def test_empty_requires_always_ok_even_for_absent_model(self):
|
|
verdict = evaluate_capability(CAPS, "absent-model", {})
|
|
self.assertTrue(verdict.ok)
|
|
self.assertEqual(verdict.missing, [])
|
|
self.assertFalse(verdict.unknown)
|
|
self.assertEqual(verdict.status, "ok")
|
|
|
|
def test_parity_with_old_check_and_split(self):
|
|
# The refactor contract: every verdict field equals what the
|
|
# two old functions returned for the same inputs, across a
|
|
# matrix of directories, models and requires — including the
|
|
# preserved edge where an absent model plus an uncleaned
|
|
# requires key makes the total check list the invalid key
|
|
# while the split lists only recognised keys. That edge is
|
|
# documented here as preserved parity, not as desired
|
|
# semantics; changing it would be a behaviour change.
|
|
requires_sets = [
|
|
REQUIRES,
|
|
{"tool_calling": True},
|
|
{"min_context_tokens": 500},
|
|
{},
|
|
{"tool_calling": True, "bogus_key": True},
|
|
]
|
|
models = list(CAPS) + ["absent-model", "", None]
|
|
for requires in requires_sets:
|
|
for model in models:
|
|
with self.subTest(model=model, requires=requires):
|
|
verdict = evaluate_capability(
|
|
CAPS, model, requires)
|
|
ok, missing, unknown = check_model_capabilities(
|
|
CAPS, model, requires)
|
|
known, unknown_missing = \
|
|
split_model_capability_missing(
|
|
CAPS, model, requires)
|
|
self.assertEqual(verdict.ok, ok)
|
|
self.assertEqual(verdict.missing, missing)
|
|
self.assertEqual(verdict.unknown, unknown)
|
|
self.assertEqual(verdict.known_missing, known)
|
|
self.assertEqual(
|
|
verdict.unknown_missing, unknown_missing)
|
|
|
|
|
|
class DecideRoutingWordingTestCase(unittest.TestCase):
|
|
def test_active_known_missing_falls_back_to_worker_default(self):
|
|
policy = _policy("active", "no-tool", REQUIRES, CAPS)
|
|
decision = decide_routing({}, "good", policy=policy)
|
|
self.assertEqual(decision.source, "worker_default")
|
|
self.assertTrue(decision.capability_fallback)
|
|
self.assertEqual(decision.chosen_model, "good")
|
|
self.assertIn("capability lacks: tool_calling",
|
|
decision.reason)
|
|
|
|
def test_active_both_fail_blocks_with_exact_reason_shape(self):
|
|
policy = _policy("active", "no-tool", REQUIRES, CAPS)
|
|
decision = decide_routing({}, "no-tool", policy=policy)
|
|
self.assertEqual(decision.source, "capability_blocked")
|
|
self.assertIsNone(decision.chosen_model)
|
|
self.assertIn("refusing to dispatch an incapable or"
|
|
" unverifiable model", decision.reason)
|
|
|
|
def test_shadow_unknown_and_known_notes(self):
|
|
policy = _policy("shadow", "no-vision-field", REQUIRES, CAPS)
|
|
decision = decide_routing({}, "good", policy=policy)
|
|
self.assertEqual(decision.source, "shadow")
|
|
self.assertFalse(decision.applied)
|
|
self.assertIn("capability unknown: vision", decision.reason)
|
|
policy = _policy("shadow", "no-tool", REQUIRES, CAPS)
|
|
decision = decide_routing({}, "good", policy=policy)
|
|
self.assertIn("capability lacks: tool_calling",
|
|
decision.reason)
|
|
|
|
|
|
class DryRunWordingTestCase(unittest.TestCase):
|
|
def _doc(self, mode, tier_model):
|
|
return {
|
|
"version": "batch3",
|
|
"mode": mode,
|
|
"tiers": {"default": {"model": tier_model,
|
|
"requires": REQUIRES}},
|
|
"model_capabilities": CAPS,
|
|
}
|
|
|
|
def test_tier_check_notes_and_block_flags(self):
|
|
report = dry_run_policy_document(
|
|
self._doc("active", "no-tool"))
|
|
tier = report["tiers"]["default"]
|
|
self.assertEqual(tier["capabilities"]["status"], "fail")
|
|
self.assertTrue(report["blocked"])
|
|
self.assertTrue(any(
|
|
"lacks capability: tool_calling" in w
|
|
and "would block 'default' tasks at dispatch" in w
|
|
for w in report["warnings"]))
|
|
report = dry_run_policy_document(
|
|
self._doc("shadow", "no-vision-field"))
|
|
tier = report["tiers"]["default"]
|
|
self.assertEqual(tier["capabilities"]["status"], "unknown")
|
|
self.assertFalse(report["blocked"])
|
|
self.assertTrue(any(
|
|
"cannot verify capability: vision" in w
|
|
and "shadow mode does not block dispatch" in w
|
|
for w in report["warnings"]))
|
|
|
|
def test_gateway_check_notes_and_block_flags(self):
|
|
report = dry_run_policy_document(
|
|
self._doc("active", "good"), gateway_model="no-tool")
|
|
gw = report["gateway"]["tiers"]["default"]
|
|
self.assertEqual(gw["status"], "fail")
|
|
self.assertTrue(report["blocked"])
|
|
self.assertTrue(any(
|
|
"gateway model 'no-tool' lacks capability: tool_calling"
|
|
in w
|
|
and "would block 'default' tasks at the gateway gate" in w
|
|
for w in report["warnings"]))
|
|
report = dry_run_policy_document(
|
|
self._doc("shadow", "good"), gateway_model=None)
|
|
gw = report["gateway"]["tiers"]["default"]
|
|
self.assertEqual(gw["status"], "unknown")
|
|
self.assertTrue(gw["unknown"])
|
|
self.assertTrue(any(
|
|
"gateway model is not set in CLOUD_SUB2API_DEFAULT_MODEL"
|
|
in w for w in report["warnings"]))
|
|
|
|
|
|
class MergePolicyDocumentsTestCase(unittest.TestCase):
|
|
def test_boundary_cases(self):
|
|
self.assertIsNone(merge_policy_documents(None, None))
|
|
self.assertEqual(merge_policy_documents(None, {"a": 1}),
|
|
{"a": 1})
|
|
self.assertEqual(merge_policy_documents({"a": 1}, None),
|
|
{"a": 1})
|
|
self.assertEqual(merge_policy_documents({"a": 1}, ["x"]),
|
|
{"a": 1})
|
|
self.assertEqual(merge_policy_documents(["x"], {"a": 1}),
|
|
{"a": 1})
|
|
# No tiers anywhere: plain shallow update.
|
|
self.assertEqual(
|
|
merge_policy_documents({"mode": "shadow", "version": "v1"},
|
|
{"mode": "active"}),
|
|
{"mode": "active", "version": "v1"})
|
|
|
|
def test_json_replaces_whole_tiers_map_not_per_tier(self):
|
|
# Preserved (not fixed) semantics, documented in
|
|
# merge_policy_documents: JSON tiers replace the whole map.
|
|
# File tiers absent from JSON are dropped, not preserved.
|
|
base = {"mode": "shadow",
|
|
"tiers": {"plan": "m-file-plan",
|
|
"default": "m-file-default"}}
|
|
override = {"tiers": {"plan": "m-json-plan"}}
|
|
self.assertEqual(merge_policy_documents(base, override),
|
|
{"mode": "shadow",
|
|
"tiers": {"plan": "m-json-plan"}})
|
|
|
|
def test_inputs_not_mutated(self):
|
|
base = {"mode": "shadow",
|
|
"tiers": {"plan": "m-file-plan",
|
|
"default": "m-file-default"}}
|
|
override = {"tiers": {"plan": "m-json-plan"}}
|
|
base_before, override_before = copy.deepcopy(base), \
|
|
copy.deepcopy(override)
|
|
merge_policy_documents(base, override)
|
|
self.assertEqual(base, base_before)
|
|
self.assertEqual(override, override_before)
|
|
|
|
def test_property_random_pairs_match_shallow_update(self):
|
|
rng = random.Random(20261102)
|
|
tier_values = ["m1", "m2", {"model": "m3"},
|
|
{"model": "m4", "mode": "shadow"},
|
|
{"model": "m5", "mode": "bogus"},
|
|
{"model": ""}, " ", 7, None]
|
|
top_values = ["shadow", "active", "v9", 3, None,
|
|
{"x": 1}, ["a"]]
|
|
|
|
def _rand_doc():
|
|
doc = {}
|
|
if rng.random() < 0.8:
|
|
doc["mode"] = rng.choice(["shadow", "active", "off"])
|
|
if rng.random() < 0.5:
|
|
doc["version"] = rng.choice(top_values)
|
|
if rng.random() < 0.8:
|
|
tiers = {}
|
|
for tier in rng.sample(
|
|
["plan", "goal", "default", "weird"],
|
|
rng.randint(0, 3)):
|
|
tiers[tier] = rng.choice(tier_values)
|
|
if rng.random() < 0.1:
|
|
doc["tiers"] = rng.choice(["x", 5, None])
|
|
else:
|
|
doc["tiers"] = tiers
|
|
return doc
|
|
|
|
for _ in range(300):
|
|
base = _rand_doc()
|
|
override = _rand_doc()
|
|
got = merge_policy_documents(base, override)
|
|
expected = dict(base)
|
|
expected.update(override)
|
|
self.assertEqual(got, expected)
|
|
|
|
def test_loader_and_dryrun_agree_on_same_env(self):
|
|
# Loader and dry-run now share merge_policy_documents: for
|
|
# the same file+JSON env, the parsed policy tiers and the
|
|
# dry-run report tiers (and fingerprints) must agree.
|
|
cases = [
|
|
({"mode": "shadow",
|
|
"tiers": {"plan": "m-file-plan",
|
|
"default": "m-file-default"}},
|
|
{"tiers": {"plan": "m-json-plan"}}),
|
|
({"mode": "active",
|
|
"tiers": {"default": {"model": "m1",
|
|
"requires": {"tool_calling": True}}},
|
|
"model_capabilities": {"m1": {"tool_calling": True}}},
|
|
{"mode": "shadow"}),
|
|
]
|
|
for base_doc, override_doc in cases:
|
|
with tempfile.NamedTemporaryFile(
|
|
"w", suffix=".json", delete=False) as fh:
|
|
json.dump(base_doc, fh)
|
|
path = fh.name
|
|
try:
|
|
env = {"CLOUD_ROUTING_POLICY_FILE": path,
|
|
"CLOUD_ROUTING_POLICY_JSON":
|
|
json.dumps(override_doc)}
|
|
policy = load_routing_policy(env)
|
|
report = dry_run_from_env(env)
|
|
self.assertIsNotNone(policy)
|
|
self.assertTrue(report["valid"])
|
|
for tier, model in policy.tiers.items():
|
|
self.assertEqual(
|
|
report["tiers"][tier]["model"], model)
|
|
self.assertEqual(report["content_fingerprint"],
|
|
policy.content_fingerprint())
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|