299 lines
12 KiB
Python
299 lines
12 KiB
Python
#!/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())
|