"""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 json import sys import tomllib from datetime import datetime, timezone from pathlib import Path import pytest import yaml ROOT = Path(__file__).resolve().parents[2] FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" CONTRACTS = ROOT / "services" / "hermes" / "contracts" / "hux" if str(FOUNDATION) not in sys.path: sys.path.insert(0, str(FOUNDATION)) from hux import contracts, rules as policy # noqa: E402 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["external_side_effect"] == "ask", level assert row["read_files"] == "allow" assert matrix["ask_first"]["shell"] == "ask" assert matrix["safe"]["network"] == "ask", "browsing under safe parks a visible prompt, never a silent deny" assert matrix["safe"]["web_search"] == "ask" assert matrix["ask_first"]["network"] == "ask" and matrix["ask_first"]["web_search"] == "ask" assert matrix["autonomous"]["shell"] == "allow" assert matrix["autonomous"]["network"] == "allow" assert matrix["autonomous"]["web_search"] == "allow" assert "deny" not in {value for row in matrix.values() for value in row.values()}, "no default silently denies" 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" base["grants"][0]["decision"] = "allow" assert policy.effective_decision(base, "network", NOW) == "allow" assert policy.effective_decision(base, "network", datetime(2026, 9, 1, tzinfo=timezone.utc)) == "ask", ( "an expired allow grant falls back to the safe matrix, which asks for network" ) 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")