feat: 冻结 Brain 运行状态契约 v1 (#149)

2026-08-31 08:44:36 +08:00
parent 49aa79f3b9
commit e4fed702c4
16 changed files with 682 additions and 0 deletions
@@ -0,0 +1,9 @@
+# runtime-status/v1 契约测试
+
+从仓库根目录运行：
+
+```powershell
+python -m unittest discover -s contracts/tests/runtime-status-v1 -p "test_*.py" -v
+```
+
+测试只使用 Python 标准库，不安装依赖、不访问网络。它对冻结 Schema 的已用关键字执行验证，并覆盖状态迁移、时间/陈旧边界、offline/recovery、配置 revision、未知主版本、倒序消息和敏感字段拒绝。产品 adapter 还需在各自工单中运行本地模型映射测试。
@@ -0,0 +1,274 @@
+import copy
+import json
+import re
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+
+TEST_DIR = Path(__file__).resolve().parent
+CONTRACT_DIR = TEST_DIR.parents[1] / "runtime-status" / "v1"
+SCHEMA = json.loads((CONTRACT_DIR / "runtime-status.schema.json").read_text(encoding="utf-8"))
+VALID_DIR = CONTRACT_DIR / "examples" / "valid"
+INVALID_DIR = CONTRACT_DIR / "examples" / "invalid"
+
+
+def parse_datetime(value):
+    if not isinstance(value, str):
+        raise ValueError("not a string")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if parsed.tzinfo is None:
+        raise ValueError("timezone is required")
+    return parsed.astimezone(timezone.utc)
+
+
+def matches_type(value, expected):
+    if expected == "null":
+        return value is None
+    if expected == "object":
+        return isinstance(value, dict)
+    if expected == "array":
+        return isinstance(value, list)
+    if expected == "string":
+        return isinstance(value, str)
+    if expected == "integer":
+        return isinstance(value, int) and not isinstance(value, bool)
+    if expected == "number":
+        return isinstance(value, (int, float)) and not isinstance(value, bool)
+    if expected == "boolean":
+        return isinstance(value, bool)
+    raise AssertionError(f"unsupported schema type in test validator: {expected}")
+
+
+def resolve_ref(ref):
+    if not ref.startswith("#/"):
+        raise AssertionError(f"external refs are not supported: {ref}")
+    node = SCHEMA
+    for part in ref[2:].split("/"):
+        node = node[part.replace("~1", "/").replace("~0", "~")]
+    return node
+
+
+def validate(instance, schema=None, path="$", errors=None):
+    schema = SCHEMA if schema is None else schema
+    errors = [] if errors is None else errors
+    if "$ref" in schema:
+        return validate(instance, resolve_ref(schema["$ref"]), path, errors)
+
+    for subschema in schema.get("allOf", []):
+        validate(instance, subschema, path, errors)
+    if "if" in schema:
+        condition_errors = validate(instance, schema["if"], path, [])
+        branch = schema.get("then") if not condition_errors else schema.get("else")
+        if branch is not None:
+            validate(instance, branch, path, errors)
+
+    if "type" in schema:
+        allowed = schema["type"] if isinstance(schema["type"], list) else [schema["type"]]
+        if not any(matches_type(instance, expected) for expected in allowed):
+            errors.append(f"{path}: expected {allowed}")
+            return errors
+
+    if "const" in schema and instance != schema["const"]:
+        errors.append(f"{path}: expected constant {schema['const']!r}")
+    if "enum" in schema and instance not in schema["enum"]:
+        errors.append(f"{path}: value is not in enum")
+
+    if isinstance(instance, dict):
+        required = schema.get("required", [])
+        for name in required:
+            if name not in instance:
+                errors.append(f"{path}: missing required property {name}")
+        properties = schema.get("properties", {})
+        if schema.get("additionalProperties") is False:
+            for name in instance:
+                if name not in properties:
+                    errors.append(f"{path}: additional property {name}")
+        for name, value in instance.items():
+            if name in properties:
+                validate(value, properties[name], f"{path}.{name}", errors)
+
+    if isinstance(instance, list):
+        if "maxItems" in schema and len(instance) > schema["maxItems"]:
+            errors.append(f"{path}: too many items")
+        if schema.get("uniqueItems"):
+            encoded = [json.dumps(item, sort_keys=True) for item in instance]
+            if len(encoded) != len(set(encoded)):
+                errors.append(f"{path}: duplicate items")
+        if "items" in schema:
+            for index, value in enumerate(instance):
+                validate(value, schema["items"], f"{path}[{index}]", errors)
+
+    if isinstance(instance, str):
+        if "minLength" in schema and len(instance) < schema["minLength"]:
+            errors.append(f"{path}: string is too short")
+        if "maxLength" in schema and len(instance) > schema["maxLength"]:
+            errors.append(f"{path}: string is too long")
+        if "pattern" in schema and re.fullmatch(schema["pattern"], instance) is None:
+            errors.append(f"{path}: pattern mismatch")
+        if schema.get("format") == "date-time":
+            try:
+                parse_datetime(instance)
+            except (TypeError, ValueError):
+                errors.append(f"{path}: invalid date-time")
+
+    if isinstance(instance, (int, float)) and not isinstance(instance, bool):
+        if "minimum" in schema and instance < schema["minimum"]:
+            errors.append(f"{path}: below minimum")
+        if "maximum" in schema and instance > schema["maximum"]:
+            errors.append(f"{path}: above maximum")
+    return errors
+
+
+def load(path):
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def freshness(observed_at, evaluation_time):
+    age = evaluation_time - parse_datetime(observed_at)
+    if age < timedelta(seconds=-30):
+        return "future_rejected"
+    return "stale" if age > timedelta(seconds=90) else "fresh"
+
+
+ALLOWED_TRANSITIONS = {
+    "unconfigured": {"starting", "stopped"},
+    "starting": {"running", "degraded", "failed", "stopped"},
+    "running": {"degraded", "failed", "stopped"},
+    "degraded": {"running", "failed", "stopped"},
+    "failed": {"starting", "stopped"},
+    "stopped": {"starting", "unconfigured"},
+}
+
+
+def may_transition(previous, current):
+    return previous == current or current in ALLOWED_TRANSITIONS[previous]
+
+
+def may_replace(previous, candidate, evaluation_time):
+    if candidate["schema_version"] != "runtime-status/v1":
+        return False
+    if validate(candidate):
+        return False
+    if freshness(candidate["observed_at"], evaluation_time) == "future_rejected":
+        return False
+    if candidate["brain_instance_ref"] != previous["brain_instance_ref"]:
+        return False
+    if candidate["sequence"] <= previous["sequence"]:
+        return False
+    return may_transition(previous["runtime"]["state"], candidate["runtime"]["state"])
+
+
+class RuntimeStatusV1ContractTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.running = load(VALID_DIR / "running.json")
+
+    def test_schema_declares_frozen_version_and_closed_objects(self):
+        self.assertEqual(SCHEMA["$schema"], "https://json-schema.org/draft/2020-12/schema")
+        self.assertEqual(SCHEMA["properties"]["schema_version"]["const"], "runtime-status/v1")
+        self.assertFalse(SCHEMA["additionalProperties"])
+        for name in ("runtime", "model", "configuration", "health"):
+            self.assertFalse(SCHEMA["properties"][name]["additionalProperties"])
+
+    def test_all_valid_examples_satisfy_schema(self):
+        paths = sorted(VALID_DIR.glob("*.json"))
+        self.assertGreaterEqual(len(paths), 5)
+        for path in paths:
+            with self.subTest(path=path.name):
+                self.assertEqual(validate(load(path)), [])
+
+    def test_all_invalid_examples_are_rejected(self):
+        paths = sorted(INVALID_DIR.glob("*.json"))
+        self.assertGreaterEqual(len(paths), 5)
+        for path in paths:
+            with self.subTest(path=path.name):
+                self.assertNotEqual(validate(load(path)), [])
+
+    def test_every_runtime_state_is_schema_valid(self):
+        for state in ALLOWED_TRANSITIONS:
+            message = copy.deepcopy(self.running)
+            message["runtime"]["state"] = state
+            with self.subTest(state=state):
+                self.assertEqual(validate(message), [])
+
+    def test_state_transition_matrix(self):
+        self.assertTrue(may_transition("unconfigured", "starting"))
+        self.assertTrue(may_transition("starting", "running"))
+        self.assertTrue(may_transition("running", "degraded"))
+        self.assertTrue(may_transition("degraded", "running"))
+        self.assertTrue(may_transition("running", "failed"))
+        self.assertTrue(may_transition("failed", "stopped"))
+        self.assertFalse(may_transition("unconfigured", "running"))
+        self.assertFalse(may_transition("stopped", "running"))
+
+    def test_stale_and_future_boundaries(self):
+        observed = parse_datetime(self.running["observed_at"])
+        self.assertEqual(freshness(self.running["observed_at"], observed + timedelta(seconds=90)), "fresh")
+        self.assertEqual(freshness(self.running["observed_at"], observed + timedelta(seconds=91)), "stale")
+        self.assertEqual(freshness(self.running["observed_at"], observed - timedelta(seconds=30)), "fresh")
+        self.assertEqual(freshness(self.running["observed_at"], observed - timedelta(seconds=31)), "future_rejected")
+
+    def test_offline_keeps_last_known_and_recovery_replaces_it(self):
+        last_known = load(VALID_DIR / "offline-last-known.json")
+        evaluation = parse_datetime(last_known["observed_at"]) + timedelta(seconds=180)
+        self.assertEqual(freshness(last_known["observed_at"], evaluation), "stale")
+        self.assertEqual(last_known["runtime"]["state"], "degraded")
+        recovered = load(VALID_DIR / "recovered.json")
+        self.assertTrue(may_replace(last_known, recovered, parse_datetime(recovered["observed_at"])))
+
+    def test_unknown_version_and_out_of_order_do_not_replace_projection(self):
+        unknown = load(INVALID_DIR / "unknown-major.json")
+        evaluation = parse_datetime(self.running["observed_at"])
+        self.assertFalse(may_replace(self.running, unknown, evaluation))
+        older = copy.deepcopy(self.running)
+        older["sequence"] = self.running["sequence"] - 1
+        self.assertFalse(may_replace(self.running, older, evaluation))
+
+    def test_configuration_revision_mismatch_is_consumer_derived(self):
+        message = load(VALID_DIR / "config-mismatch.json")
+        desired_revision = "sense-rev-21"
+        self.assertEqual(validate(message), [])
+        self.assertNotEqual(message["configuration"]["applied_revision"], desired_revision)
+        self.assertNotIn("desired_revision", message["configuration"])
+
+    def test_configuration_apply_state_invariants(self):
+        cases = [
+            ("not_configured", "rev-forbidden", None),
+            ("applied", None, None),
+            ("rejected", "rev-old", None),
+        ]
+        for apply_state, revision, error_code in cases:
+            message = copy.deepcopy(self.running)
+            message["configuration"] = {
+                "apply_state": apply_state,
+                "applied_revision": revision,
+                "error_code": error_code,
+            }
+            with self.subTest(apply_state=apply_state):
+                self.assertNotEqual(validate(message), [])
+
+    def test_sensitive_and_business_fields_are_rejected_by_name(self):
+        for forbidden in ("access_token", "password", "credential", "internal_path", "stack", "user_session", "video", "face", "alert"):
+            message = copy.deepcopy(self.running)
+            message[forbidden] = "forbidden"
+            with self.subTest(forbidden=forbidden):
+                self.assertTrue(any("additional property" in error for error in validate(message)))
+
+    def test_logical_references_reject_paths(self):
+        for value in ("C:\\models\\private.pt", "/srv/models/private.pt", "../private.pt"):
+            message = copy.deepcopy(self.running)
+            message["model"]["model_ref"] = value
+            with self.subTest(value=value):
+                self.assertNotEqual(validate(message), [])
+
+    def test_mapper_responsibilities_are_documented(self):
+        mapping = (CONTRACT_DIR / "mapping.md").read_text(encoding="utf-8")
+        for field in ("schema_version", "status_id", "brain_instance_ref", "sequence", "observed_at", "runtime.*", "model.*", "configuration.*", "health.*", "inputs[]"):
+            self.assertIn(f"`{field}`", mapping)
+        self.assertIn("Brain", mapping)
+        self.assertIn("Sense", mapping)
+
+
+if __name__ == "__main__":
+    unittest.main()