b4145861d2
HUX-11 contract freeze: identity/capabilities/manifest/error records, event turn and idempotency and delegation/side-effect kinds, memory no-store/supersedes/ retrieval removal, revisions for optimistic concurrency, artifact access, budget scope/spend/subagents and external side-effect gating, notebook notes and dedupe keys. ADR-0001 records the wire and compatibility rules. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM
317 lines
16 KiB
Python
317 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 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"] == "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")
|