#!/usr/bin/env python3 """Compile the artifact runtime registry from reviewed source contracts. Method contracts may reference an artifact kind, but they may not create one. Every method output must first be explicitly admitted by workflow-contracts.yaml or artifact-type-vocabulary.yaml. Runtime code reads only the generated registry; ``--check`` fails on source drift or any contract invariant violation. """ from __future__ import annotations import argparse import hashlib import os import sys from copy import deepcopy import yaml HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.environ.get("CLAUDE_PROJECT_DIR") or os.path.dirname(os.path.dirname(HERE)) REGISTRY_DIR = os.path.join(ROOT, "org-os", "00-role-registry") WORKFLOW_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "workflow-contracts.yaml") VOCABULARY_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "artifact-type-vocabulary.yaml") ROLES_PATH = os.path.join(REGISTRY_DIR, "roles.yaml") METHOD_INDEX = os.path.join(REGISTRY_DIR, "role-working-methods", "index.yaml") OUTPUT_PATH = os.path.join( ROOT, "org-os", "06-agent-work", "generated", "artifact-registry.yaml") def _load(path): with open(path, encoding="utf-8") as fh: return yaml.safe_load(fh) or {} def _sha(path): digest = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _rel(path): return os.path.relpath(path, ROOT).replace(os.sep, "/") def _method_sources(): index = _load(METHOD_INDEX).get("role-method-contracts", {}) or {} base = os.path.dirname(METHOD_INDEX) return [METHOD_INDEX] + [os.path.join(base, value) for value in index.get("includes", []) or []] def _methods(): merged = {} for path in _method_sources()[1:]: for role, entry in (_load(path).get("role-working-methods", {}) or {}).items(): if role in merged: raise ValueError(f"duplicate method role: {role}") merged[role] = entry return merged def _roles(): registry = _load(ROLES_PATH).get("role-registry", {}) or {} result = {item.get("role-id") for item in registry.get("roles", []) or [] if isinstance(item, dict) and item.get("role-id")} human = (registry.get("human-user") or {}).get("role-id") if human: result.add(human) return result def _method_outputs(methods): outputs = {} for role, entry in methods.items(): for method in entry.get("methods", []) or []: kinds = list(method.get("output-artifacts") or []) kinds += [step.get("required-output") for step in method.get("workflow", []) or []] for kind in kinds: if kind: outputs.setdefault(kind, set()).add(role) return outputs def _method_profiles(methods): """Return the explicit (role, method-id) contract map used by binding checks.""" profiles = {} for role, entry in methods.items(): for method in entry.get("methods", []) or []: method_id = method.get("method-id") if method_id: profiles[(role, method_id)] = method return profiles def _binding_errors(kind, definition, methods): """Validate workflow-artifact -> craft-method linkage at compile time. An aggregate is one immutable workflow envelope containing the outputs of every craft step through a named checkpoint. The field map prevents a step-results trace from standing in for the actual typed content. """ binding = definition.get("method-binding") if binding is None: return [] if not isinstance(binding, dict): return [f"artifact {kind}: method-binding must be an object"] mode = binding.get("mode") allowed_modes = {"workflow-control", "aggregate", "stage-synthesis", "independent-review", "lens-contribution"} if mode not in allowed_modes: return [f"artifact {kind}: method-binding.mode {mode!r} not in {sorted(allowed_modes)}"] role_methods = binding.get("role-methods") if mode != "aggregate": return ([f"artifact {kind}: only aggregate binding may declare role-methods"] if role_methods is not None else []) if not isinstance(role_methods, dict) or not role_methods: return [f"artifact {kind}: aggregate binding requires non-empty role-methods"] errors = [] producers = set(definition.get("producer-roles") or []) bound_roles = set(role_methods) if bound_roles != producers: errors.append( f"artifact {kind}: aggregate role-methods must exactly cover producer-roles " f"(bound={sorted(bound_roles)}, producers={sorted(producers)})") profiles = _method_profiles(methods) required_fields = set(definition.get("required-payload-fields") or []) for role, config in role_methods.items(): if not isinstance(config, dict): errors.append(f"artifact {kind}/{role}: aggregate config must be an object") continue method_id = config.get("method-id") profile = profiles.get((role, method_id)) if not profile: errors.append(f"artifact {kind}/{role}: unknown method-id {method_id!r}") continue workflow = profile.get("workflow", []) or [] checkpoint = config.get("checkpoint-step-id") indexes = [index for index, step in enumerate(workflow) if step.get("step-id") == checkpoint] if len(indexes) != 1: errors.append( f"artifact {kind}/{role}/{method_id}: checkpoint-step-id {checkpoint!r} " "must identify exactly one workflow step") continue embedded = config.get("embedded-outputs") if not isinstance(embedded, dict): errors.append(f"artifact {kind}/{role}/{method_id}: embedded-outputs object required") continue for step in workflow[:indexes[0] + 1]: output = step.get("required-output") fields = embedded.get(output) if isinstance(fields, str): fields = [fields] if not isinstance(fields, list) or not fields or not all( isinstance(field, str) and field for field in fields): errors.append( f"artifact {kind}/{role}/{method_id}: required-output {output!r} " "needs a non-empty embedded field list") continue undeclared = sorted(set(fields) - required_fields) if undeclared: errors.append( f"artifact {kind}/{role}/{method_id}: embedded fields are not required " f"payload fields: {undeclared}") return errors def compile_registry(): workflow_doc = _load(WORKFLOW_PATH) contract = workflow_doc.get("workflow-contracts", {}) or {} workflow_kinds = contract.get("artifact-kinds", {}) or {} vocabulary = _load(VOCABULARY_PATH).get("artifact-types", {}) or {} methods = _methods() method_outputs = _method_outputs(methods) known_roles = _roles() errors = [] role_caps = contract.get("role-capabilities", {}) or {} for capability, role_ids in role_caps.items(): unknown = sorted(set(role_ids or []) - known_roles) if unknown: errors.append(f"role-capability {capability}: unknown roles {unknown}") admitted = set(workflow_kinds) | set(vocabulary) unknown_outputs = sorted(set(method_outputs) - admitted) if unknown_outputs: errors.append( "method outputs are not explicitly admitted by workflow/vocabulary: " + ", ".join(unknown_outputs)) definitions = {} default_schema = contract.get("default-payload-schema-ref") for kind in sorted(admitted): vocab = vocabulary.get(kind) or {} direct = workflow_kinds.get(kind) or {} definition = { "producer-roles": sorted(set(vocab.get("producer-roles") or [])), "reviewer-capability": "artifact-reviewer", "required-payload-fields": list(vocab.get("required-fields") or []), "registry-sources": (["artifact-type-vocabulary"] if kind in vocabulary else []), } if vocab.get("schema-ref"): definition["method-schema-ref"] = vocab.get("schema-ref") if kind in workflow_kinds: definition.update(deepcopy(direct)) definition["registry-sources"] = definition.get("registry-sources", []) + ["workflow-contracts"] definition["producer-roles"] = sorted(set(direct.get("producer-roles") or [])) declared_producers = set(definition.get("producer-roles") or []) method_producers = method_outputs.get(kind, set()) missing_producers = sorted(method_producers - declared_producers) if missing_producers: errors.append( f"artifact {kind}: method producer roles not admitted {missing_producers}") unknown_producers = sorted(declared_producers - known_roles) if unknown_producers: errors.append(f"artifact {kind}: unknown producer roles {unknown_producers}") capability = definition.get("reviewer-capability") if capability not in role_caps: errors.append(f"artifact {kind}: unknown reviewer-capability {capability!r}") schema_ref = definition.get("payload-schema-ref") or default_schema if not schema_ref or not os.path.isfile(os.path.join(ROOT, ".claude", "schemas", schema_ref)): errors.append(f"artifact {kind}: missing payload schema {schema_ref!r}") errors.extend(_binding_errors(kind, definition, methods)) definitions[kind] = definition referenced = set() for bundle in (contract.get("artifact-bundles", {}) or {}).values(): referenced.update(bundle.get("always") or []) for conditional in bundle.get("conditional", []) or []: referenced.update(conditional.get("require") or []) for workflow in (contract.get("workflows", {}) or {}).values(): for stage in (workflow.get("stages", {}) or {}).values(): outputs = stage.get("outputs") or {} referenced.update(outputs.get("bundle") or []) missing_references = sorted(referenced - set(definitions)) if missing_references: errors.append("workflow/bundle references unknown artifact kinds: " + ", ".join(missing_references)) schema_refs = {default_schema} schema_refs.update(definition.get("payload-schema-ref") for definition in definitions.values()) schema_paths = [os.path.join(ROOT, ".claude", "schemas", ref) for ref in sorted(value for value in schema_refs if value)] source_paths = [WORKFLOW_PATH, VOCABULARY_PATH, ROLES_PATH] + _method_sources() + schema_paths source_hashes = {_rel(path): _sha(path) for path in source_paths} result = { "artifact-registry": { "version": 1, "generated-by": ".claude/hooks/compile_artifact_registry.py", "source-sha256": source_hashes, "artifact-kind-count": len(definitions), "artifact-kinds": definitions, } } return result, errors def _serialized(document): return yaml.safe_dump(document, allow_unicode=True, sort_keys=False, width=120) def main(argv=None): parser = argparse.ArgumentParser() parser.add_argument("--check", action="store_true", help="validate invariants and generated-file drift") parser.add_argument("--output", default=OUTPUT_PATH) args = parser.parse_args(argv) try: document, errors = compile_registry() except Exception as exc: errors = [str(exc)] document = None if errors: for error in errors: print(f"[artifact-registry] ERROR: {error}", file=sys.stderr) return 2 expected = _serialized(document) if args.check: try: with open(args.output, encoding="utf-8") as fh: actual = fh.read() except OSError: actual = "" if actual != expected: print("[artifact-registry] ERROR: generated registry drift; run compiler without --check", file=sys.stderr) return 2 print(f"[artifact-registry] OK: {document['artifact-registry']['artifact-kind-count']} kinds") return 0 os.makedirs(os.path.dirname(args.output), exist_ok=True) with open(args.output, "w", encoding="utf-8") as fh: fh.write(expected) print(f"[artifact-registry] wrote {args.output}") return 0 if __name__ == "__main__": raise SystemExit(main())