d3cb6e9045
Schemas, examples and a flag registry for the twelve HUX cards, a dependency-free validator, the governance rules (memory ledger, autonomy matrix, friendly modes mapped to real Switchyard routes, privacy defaults, suggestion gating, release state machine) and the contract doc UI work codes against. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM
325 lines
16 KiB
Python
325 lines
16 KiB
Python
"""Contract tests for the HUX schemas and the governance rules behind them.
|
|
|
|
Every schema must stay inside the keyword subset the in-repo validator
|
|
understands, every shipped example must validate, every friendly mode must
|
|
resolve to a Switchyard route that really exists, and the state machines must
|
|
refuse the transitions that would let merged/built/deployed be confused.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import tomllib
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "services" / "hermes" / "scripts"
|
|
CONTRACTS = ROOT / "services" / "hermes" / "contracts" / "hux"
|
|
|
|
|
|
def _load(name: str):
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
contracts = _load("hux_contracts")
|
|
policy = _load("hux_policy")
|
|
SCHEMAS = contracts.load_all()
|
|
EXAMPLES = {path.stem: json.loads(path.read_text()) for path in sorted((CONTRACTS / "examples").glob("*.json"))}
|
|
NOW = datetime(2026, 8, 23, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _switchyard_route_ids() -> set[str]:
|
|
manifest = next(iter(yaml.safe_load_all((ROOT / "services/hermes/switchyard-configmap.yaml").read_text())))
|
|
routes = tomllib.loads(manifest["data"]["routes.toml"])["routes"]
|
|
return {route["id"] for route in routes.values()}
|
|
|
|
|
|
# --- schema hygiene -----------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("name", contracts.SCHEMA_FILES)
|
|
def test_schema_uses_only_supported_keywords(name):
|
|
assert contracts.unsupported_keywords(SCHEMAS[name]) == []
|
|
assert SCHEMAS[name]["$id"].endswith(name)
|
|
|
|
|
|
def test_every_schema_file_is_registered():
|
|
on_disk = {path.name for path in CONTRACTS.glob("*.schema.json")}
|
|
assert on_disk == set(contracts.SCHEMA_FILES)
|
|
|
|
|
|
def test_walker_flags_unknown_keywords():
|
|
assert contracts.unsupported_keywords({"type": "object", "properties": {"x": {"format": "email"}}}) == ["#/properties/x/format"]
|
|
|
|
|
|
# --- examples -----------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("name", sorted(EXAMPLES))
|
|
def test_example_validates_by_its_schema_field(name):
|
|
assert contracts.validate_record(EXAMPLES[name], SCHEMAS) == []
|
|
|
|
|
|
def test_every_record_schema_has_an_example():
|
|
covered = {example["schema"] for example in EXAMPLES.values()}
|
|
covered |= {"hux.mode.v1", "hux.privacy_policy.v1"}
|
|
assert set(contracts.record_schema_names(SCHEMAS)) == covered
|
|
|
|
|
|
def test_validate_record_rejects_unknown_or_missing_schema():
|
|
assert contracts.validate_record({"schema": "hux.nope.v1"}, SCHEMAS) == ["$: unknown record schema 'hux.nope.v1'"]
|
|
assert contracts.validate_record(["not", "a", "record"], SCHEMAS)[0].startswith("$: record has no")
|
|
|
|
|
|
def test_default_loading_paths_work():
|
|
assert contracts.validate_record(EXAMPLES["event"]) == []
|
|
assert contracts.validate("event.schema.json", EXAMPLES["event"]) == []
|
|
assert "HUX-01" in {card["card"] for card in contracts.load_flags()["cards"]}
|
|
|
|
|
|
# --- validator keyword coverage ----------------------------------------------
|
|
|
|
def _broken(name: str, mutate) -> list[str]:
|
|
record = copy.deepcopy(EXAMPLES[name])
|
|
mutate(record)
|
|
return contracts.validate_record(record, SCHEMAS)
|
|
|
|
|
|
def test_validator_catches_each_constraint_class():
|
|
assert any("expected integer" in e for e in _broken("event", lambda r: r.update(seq=True)))
|
|
assert any("below minimum" in e for e in _broken("event", lambda r: r.update(seq=-1)))
|
|
assert any("not in enum" in e for e in _broken("event", lambda r: r.update(kind="tool.dance")))
|
|
assert any("unexpected property" in e for e in _broken("event", lambda r: r.update(extra=1)))
|
|
assert any("missing required" in e for e in _broken("event", lambda r: r.pop("summary")))
|
|
assert any("longer than" in e for e in _broken("event", lambda r: r.update(summary="x" * 281)))
|
|
assert any("shorter than" in e for e in _broken("event", lambda r: r.update(summary="")))
|
|
assert any("does not match" in e for e in _broken("event", lambda r: r.update(ts="yesterday")))
|
|
wrong_const = copy.deepcopy(EXAMPLES["event"])
|
|
wrong_const["schema"] = "hux.event.v0"
|
|
assert any("expected constant" in e for e in contracts.validate("event.schema.json", wrong_const, SCHEMAS))
|
|
assert any("more than" in e for e in _broken("event", lambda r: r.update(evidence=[{"kind": "run", "id": "r"}] * 65)))
|
|
assert any("not unique" in e for e in _broken("project", lambda r: r.update(tags=["a", "a"])))
|
|
assert any("fewer than" in e for e in _broken("memory", lambda r: r.update(audit=[])))
|
|
assert any("above maximum" in e for e in _broken("suggestion", lambda r: r.update(priority=101)))
|
|
assert any("expected object" in e for e in _broken("event", lambda r: r.update(provenance="x")))
|
|
|
|
|
|
def test_one_of_requires_exactly_one_branch():
|
|
ambiguous = {"schema": "hux.policy.v1"}
|
|
errors = contracts.validate("permission.schema.json", ambiguous, SCHEMAS)
|
|
assert any("matched 0 oneOf branches" in e for e in errors)
|
|
|
|
|
|
def test_pointer_validation_and_bad_refs():
|
|
assert contracts.validate("permission.schema.json", EXAMPLES["approval"], SCHEMAS, "/$defs/approval") == []
|
|
with pytest.raises(contracts.ContractError):
|
|
contracts.validate("permission.schema.json", {}, SCHEMAS, "/$defs/missing")
|
|
with pytest.raises(contracts.ContractError):
|
|
contracts._resolve_ref("nowhere.schema.json#/x", "event.schema.json", SCHEMAS)
|
|
with pytest.raises(contracts.ContractError):
|
|
contracts._validate({"type": "date"}, "x", "$", [], "event.schema.json", SCHEMAS)
|
|
|
|
|
|
# --- memory ledger -------------------------------------------------------------
|
|
|
|
def test_memory_state_machine_is_append_only():
|
|
allowed = {(a, b) for a, bs in policy.MEMORY_TRANSITIONS.items() for b in bs}
|
|
assert allowed == {("proposed", "active"), ("proposed", "rejected"), ("active", "expired"), ("active", "forgotten"), ("expired", "forgotten")}
|
|
assert policy.transition_allowed(policy.MEMORY_TRANSITIONS, "proposed", "active")
|
|
assert not policy.transition_allowed(policy.MEMORY_TRANSITIONS, "forgotten", "active")
|
|
assert not policy.transition_allowed(policy.MEMORY_TRANSITIONS, "unknown", "active")
|
|
|
|
|
|
def test_memory_policy_rules():
|
|
base = EXAMPLES["memory"]
|
|
assert policy.memory_policy_violations(base) == []
|
|
restricted = {**base, "sensitivity": "restricted"}
|
|
assert "restricted content may not be remembered" in policy.memory_policy_violations(restricted)
|
|
lax = {**base, "sensitivity": "sensitive", "approval_mode": "automatic", "ttl": {"policy": "never"}}
|
|
problems = policy.memory_policy_violations(lax)
|
|
assert "sensitive memory requires approval_mode=ask" in problems
|
|
assert "sensitive memory must expire or decay" in problems
|
|
creds = {**base, "topic": "credentials"}
|
|
assert "topic credentials may not be written to memory" in policy.memory_policy_violations(creds)
|
|
assert policy.memory_policy_violations({**creds, "status": "rejected"}) == []
|
|
assert "ttl.policy=expires_at requires expires_at" in policy.memory_policy_violations({**base, "ttl": {"policy": "expires_at"}})
|
|
assert "ttl.policy=decay requires decay_days" in policy.memory_policy_violations({**base, "ttl": {"policy": "decay"}})
|
|
assert "forgotten entries must drop their content" in policy.memory_policy_violations({**base, "status": "forgotten"})
|
|
assert policy.memory_policy_violations({**base, "status": "forgotten", "content": ""}) == []
|
|
|
|
|
|
# --- autonomy -------------------------------------------------------------------
|
|
|
|
def test_capability_matrix_shape():
|
|
matrix = policy.default_capability_matrix()
|
|
assert set(matrix) == {"ask_first", "safe", "autonomous"}
|
|
for level, row in matrix.items():
|
|
assert set(row) == set(policy.CAPABILITIES)
|
|
assert row["deploy"] == "ask", level
|
|
assert row["read_files"] == "allow"
|
|
assert matrix["ask_first"]["shell"] == "ask"
|
|
assert matrix["safe"]["network"] == "deny"
|
|
assert matrix["autonomous"]["shell"] == "allow"
|
|
assert matrix["autonomous"]["network"] == "allow"
|
|
assert set(policy.CAPABILITIES) == set(SCHEMAS["permission.schema.json"]["$defs"]["capability"]["enum"])
|
|
|
|
|
|
def test_effective_decision_honours_grants_expiry_and_deny():
|
|
base = copy.deepcopy(EXAMPLES["policy"])
|
|
assert policy.effective_decision(base, "network", NOW) == "ask"
|
|
assert policy.effective_decision(base, "network", datetime(2026, 9, 1, tzinfo=timezone.utc)) == "deny"
|
|
base["grants"] = [{"capability": "shell", "decision": "allow"}, {"capability": "shell", "decision": "deny"}]
|
|
assert policy.effective_decision(base, "shell", NOW) == "deny"
|
|
base["grants"] = [{"capability": "deploy", "decision": "allow"}]
|
|
assert policy.effective_decision(base, "deploy", NOW) == "ask"
|
|
base["grants"] = [{"capability": "write_files", "decision": "allow"}]
|
|
assert policy.effective_decision(base, "write_files") == "allow"
|
|
assert policy.effective_decision(base, "read_files") == "allow"
|
|
|
|
|
|
def test_approval_states_are_terminal_after_pending():
|
|
for state, nexts in policy.APPROVAL_TRANSITIONS.items():
|
|
assert bool(nexts) == (state == "pending")
|
|
|
|
|
|
# --- friendly modes ------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("mode", sorted(policy.MODE_CATALOG))
|
|
def test_mode_contract_validates_and_resolves_to_real_switchyard_route(mode):
|
|
record = policy.mode_contract(mode)
|
|
assert contracts.validate_record(record, SCHEMAS) == []
|
|
assert record["switchyard"]["route_id"] in _switchyard_route_ids()
|
|
assert "codex" not in record["intent"].lower() and "claude" not in record["intent"].lower()
|
|
|
|
|
|
def test_private_mode_never_leaves_the_cluster():
|
|
record = policy.mode_contract("private")
|
|
assert record["constraints"] == {
|
|
"providers": ["local"], "local_only": True, "effort": {"min": "low", "max": "medium"},
|
|
"tools": {"web": "denied", "shell": "denied", "artifacts": "allowed", "delegate": "denied"},
|
|
"memory": {"read": False, "write": False}, "citations_required": False, "retention": "ephemeral",
|
|
}
|
|
with pytest.raises(ValueError):
|
|
policy.mode_contract("private", "atlas/manual/claude/opus/high")
|
|
assert policy.mode_contract("private", "atlas/manual/local/qwen-14b")["switchyard"]["override_route_id"] == "atlas/manual/local/qwen-14b"
|
|
|
|
|
|
def test_hosted_modes_stay_provider_neutral_and_overridable():
|
|
for mode in ("fast", "thoughtful", "research", "create"):
|
|
assert set(policy.MODE_CATALOG[mode]["providers"]) == {"codex", "claude"}
|
|
override = policy.mode_contract("thoughtful", "atlas/manual/codex/sol/high")
|
|
assert override["switchyard"]["override_route_id"] in _switchyard_route_ids()
|
|
assert policy.MODE_CATALOG["research"]["citations_required"]
|
|
assert policy.MODE_CATALOG["research"]["tools"]["web"] == "required"
|
|
|
|
|
|
def test_effort_bands():
|
|
assert policy.effort_within("fast", "low")
|
|
assert not policy.effort_within("fast", "xhigh")
|
|
assert policy.effort_within("research", "xhigh")
|
|
assert not policy.effort_within("research", "medium")
|
|
|
|
|
|
# --- privacy -------------------------------------------------------------------
|
|
|
|
def test_privacy_policy_validates_and_is_strict_for_restricted_topics():
|
|
record = policy.privacy_policy()
|
|
assert contracts.validate_record(record, SCHEMAS) == []
|
|
topics = {row["topic"]: row for row in record["topics"]}
|
|
assert set(topics) == set(SCHEMAS["privacy.schema.json"]["$defs"]["topic"]["enum"])
|
|
for row in topics.values():
|
|
assert (row["sensitivity"] == "restricted") == (row["memory_write"] == "deny")
|
|
assert row["decay_days"] <= 30
|
|
assert record["topic_scoping"]["cross_surface_sharing"] == "never"
|
|
|
|
|
|
# --- suggestions ---------------------------------------------------------------
|
|
|
|
def test_suggestion_gating():
|
|
suggestion = EXAMPLES["suggestion"]
|
|
assert policy.suggestion_allowed(suggestion, None, NOW)
|
|
state = copy.deepcopy(EXAMPLES["suggestion_state"])
|
|
assert policy.suggestion_allowed(suggestion, state, NOW)
|
|
assert not policy.suggestion_allowed(suggestion, {**state, "never_again": True}, NOW)
|
|
assert not policy.suggestion_allowed(suggestion, {**state, "dismissed_at": "2026-08-22T11:00:00Z"}, NOW)
|
|
assert not policy.suggestion_allowed(suggestion, {**state, "shows": 2}, NOW)
|
|
assert not policy.suggestion_allowed(suggestion, {**state, "last_shown_at": "2026-08-23T11:30:00Z"}, NOW)
|
|
assert policy.suggestion_allowed(suggestion, {**state, "last_shown_at": None}, NOW)
|
|
|
|
|
|
# --- release follow-through ----------------------------------------------------
|
|
|
|
def test_release_cannot_skip_states_or_claim_deployed_without_evidence():
|
|
record = copy.deepcopy(EXAMPLES["release"])
|
|
assert policy.release_transition_problems(record, "verified") == ["verified requires evidence.harbor_digest", "harbor_digest must equal image_digest"]
|
|
assert "built -> deployed skips or reverses the release order" in policy.release_transition_problems(record, "deployed")
|
|
record["evidence"]["harbor_digest"] = record["evidence"]["image_digest"]
|
|
assert policy.release_transition_problems(record, "verified") == []
|
|
record["state"] = "deployed"
|
|
record["evidence"]["pod_digest"] = "sha256:" + "0" * 64
|
|
assert "pod_digest must equal image_digest" in policy.release_transition_problems(record, "converged")
|
|
record["evidence"]["pod_digest"] = record["evidence"]["image_digest"]
|
|
assert policy.release_transition_problems(record, "converged") == []
|
|
record["state"] = "converged"
|
|
record["evidence"]["health_check"] = {"url": "https://chat.bstein.dev/healthz", "status": "fail", "at": "2026-08-23T10:00:00Z"}
|
|
assert policy.release_transition_problems(record, "live_verified") == ["health_check must pass"]
|
|
record["evidence"]["health_check"]["status"] = "pass"
|
|
assert policy.release_transition_problems(record, "live_verified") == []
|
|
assert policy.release_transition_problems(record, "rolled_back") == ["rolled_back requires evidence.rollback_target"]
|
|
record["state"] = "rolled_back"
|
|
assert policy.release_transition_problems(record, "merged")[0].startswith("a rolled back release is terminal")
|
|
assert policy.release_transition_problems({"state": "merged", "evidence": {}}, "rolled_back")[0].startswith("nothing to roll back")
|
|
assert policy.release_transition_problems({"state": "reviewed", "evidence": {}}, "merged") == ["merged requires evidence.merge_commit"]
|
|
|
|
|
|
# --- flags ---------------------------------------------------------------------
|
|
|
|
def test_flag_registry_matches_program():
|
|
registry = policy.flag_registry()
|
|
assert sorted(registry) == [f"HUX-{n:02d}" for n in range(1, 13)]
|
|
waves = {"A": {"HUX-01", "HUX-02", "HUX-05", "HUX-10", "HUX-11", "HUX-12"}, "B": {"HUX-03", "HUX-04", "HUX-06", "HUX-08"}, "C": {"HUX-07", "HUX-09"}}
|
|
for wave, cards in waves.items():
|
|
assert {card for card, entry in registry.items() if entry["wave"] == wave} == cards
|
|
for entry in registry.values():
|
|
assert entry["default"] is False
|
|
assert entry["backend_owner"] == "claude" and entry["frontend_owner"] == "codex"
|
|
assert entry["rollback"]
|
|
for schema in entry["contracts"]:
|
|
assert schema in contracts.SCHEMA_FILES
|
|
for dep in entry["depends_on"]:
|
|
assert dep in registry
|
|
order = policy.dependency_order()
|
|
assert order.index("HUX-11") < order.index("HUX-01") < order.index("HUX-05")
|
|
|
|
|
|
def test_dependency_cycle_is_rejected(monkeypatch):
|
|
registry = policy.flag_registry()
|
|
registry["HUX-11"]["depends_on"] = ["HUX-01"]
|
|
monkeypatch.setattr(policy, "flag_registry", lambda: registry)
|
|
with pytest.raises(ValueError):
|
|
policy.dependency_order()
|
|
|
|
|
|
def test_flags_default_off_and_require_dependencies():
|
|
assert policy.enabled_flags({}) == set()
|
|
assert policy.enabled_flags({"HUX_FLAGS": "hux.activity_timeline, hux.bogus"}) == {"hux.activity_timeline"}
|
|
assert not policy.flag_enabled("hux.activity_timeline", {"HUX_FLAGS": "hux.activity_timeline"})
|
|
assert policy.flag_enabled("hux.activity_timeline", {"HUX_FLAGS": "hux.activity_timeline,hux.foundation"})
|
|
assert not policy.flag_enabled("hux.bogus", {"HUX_FLAGS": "hux.bogus"})
|
|
assert not policy.flag_enabled("hux.foundation", {"HUX_FLAGS": ""})
|
|
|
|
|
|
def test_flags_read_process_environment(monkeypatch):
|
|
monkeypatch.setenv("HUX_FLAGS", "hux.foundation")
|
|
assert policy.flag_enabled("hux.foundation")
|