tegwick
d103955217
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
"""Resolve exactly one authoritative owner for a State Hub record type."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
AUTHORITY_SCHEMA = "repo-manager.hub-record-authority.v1"
|
|
VALID_CLASSES = frozenset(
|
|
{"file-derived", "hub-native-append", "hub-native-control", "derived-cache", "retired"}
|
|
)
|
|
DEFAULT_CONTRACT = Path(__file__).resolve().parents[2] / "config" / "hub-record-authority.yaml"
|
|
|
|
|
|
class AuthorityError(ValueError):
|
|
"""The requested route is missing or conflicts with the authority contract."""
|
|
|
|
|
|
def load_authority_contract(path: Path = DEFAULT_CONTRACT) -> dict[str, Any]:
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
if not isinstance(data, dict) or data.get("schema") != AUTHORITY_SCHEMA:
|
|
raise AuthorityError(f"unsupported authority contract: {path}")
|
|
records = data.get("records")
|
|
if not isinstance(records, dict) or not records:
|
|
raise AuthorityError("authority contract has no records")
|
|
for record_type, rule in records.items():
|
|
if not isinstance(rule, dict):
|
|
raise AuthorityError(f"invalid rule for {record_type}")
|
|
if rule.get("class") not in VALID_CLASSES or not rule.get("owner"):
|
|
raise AuthorityError(f"incomplete authority rule for {record_type}")
|
|
return data
|
|
|
|
|
|
def resolve_record_authority(
|
|
record_type: str,
|
|
*,
|
|
repo_slug: str | None = None,
|
|
domain_slug: str | None = None,
|
|
claimed_owner: str | None = None,
|
|
contract_path: Path = DEFAULT_CONTRACT,
|
|
) -> dict[str, str]:
|
|
"""Return the one authority route or reject an ambiguous/conflicting write."""
|
|
contract = load_authority_contract(contract_path)
|
|
rule = contract["records"].get(record_type)
|
|
if not isinstance(rule, dict):
|
|
raise AuthorityError(f"unknown record type: {record_type}")
|
|
|
|
record_class = str(rule["class"])
|
|
owner = str(rule["owner"])
|
|
if claimed_owner is not None and claimed_owner != owner:
|
|
raise AuthorityError(
|
|
f"authority mismatch for {record_type}: contract={owner}, claimed={claimed_owner}"
|
|
)
|
|
|
|
if record_class == "file-derived":
|
|
if not repo_slug or not domain_slug:
|
|
raise AuthorityError(
|
|
f"file-derived {record_type} requires repo_slug and domain_slug"
|
|
)
|
|
authority_key = f"repository:{domain_slug}/{repo_slug}"
|
|
elif record_class == "retired":
|
|
authority_key = f"archive:{record_type}"
|
|
else:
|
|
authority_key = f"hub:{owner}"
|
|
|
|
return {
|
|
"schema": AUTHORITY_SCHEMA,
|
|
"record_type": record_type,
|
|
"record_class": record_class,
|
|
"owner": owner,
|
|
"authority_key": authority_key,
|
|
}
|