Canon 0.6.0 freeze: ITC-CAP draft, AVT catalog, joinable review

Finish ITC-WP-0013 (AttributeValueType under ITC-DATA) and ITC-WP-0014
T02/T03/T05/T06/T07. Capability contract schemas, live-catalog review
CLI, landscape/kernel-map pointers. Sit on this version.

src/info_tech_canon/api.py

@@ -20,6 +20,7 @@ from .service import (
     profile_inspect,
     profile_validate,
     read_view,
+    review_capability_record,
     review_kit,
     validate_canon,
 )
@@ -108,6 +109,18 @@ def _route(
         profile = path.removeprefix("/profiles/").removesuffix("/validate").strip("/")
         payload = profile_validate(profile, root)
         return (HTTPStatus.OK if payload["ok"] else HTTPStatus.BAD_REQUEST), payload
+    if path == "/capability-review":
+        record = _first(query, "path")
+        if not record:
+            return HTTPStatus.BAD_REQUEST, {
+                "ok": False,
+                "error": {
+                    "code": "missing_path",
+                    "message": "Query parameter path= is required",
+                    "details": {},
+                },
+            }
+        return HTTPStatus.OK, review_capability_record(record, root)
     if path.startswith("/profiles/") and path.endswith("/graph"):
         profile = path.removeprefix("/profiles/").removesuffix("/graph").strip("/")
         graph_format = _first(query, "format") or "json"

src/info_tech_canon/capability.py

@@ -0,0 +1,320 @@
+"""Live-catalog checks for capability contracts and consumer records."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+CATALOG_PATH = REPO_ROOT / "infospace" / "models" / "capability" / "capabilities.yaml"
+
+
+class CapabilityReviewError(ValueError):
+    def __init__(self, code: str, message: str):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+PREDICATES = frozenset({"not_in", "in", "equals", "lte", "gte"})
+EVIDENCE_BASES = frozenset(
+    {
+        "invoiced",
+        "measured",
+        "quoted",
+        "derived",
+        "projected",
+        "estimated",
+        "assumed",
+        "unknown",
+    }
+)
+MATURITY_ORDER = tuple(f"D{i}" for i in range(8))
+
+
+def load_catalog(path: Path | None = None) -> dict[str, Any]:
+    catalog_path = path or CATALOG_PATH
+    if not catalog_path.is_file():
+        raise CapabilityReviewError(
+            "capability_catalog_missing",
+            f"ITC-CAP catalog not found at {catalog_path}",
+        )
+    payload = yaml.safe_load(catalog_path.read_text())
+    capabilities = {
+        cap["id"]: cap
+        for domain in payload.get("domains") or []
+        for cap in domain.get("capabilities") or []
+    }
+    return {
+        "version": (payload.get("canon") or {}).get("version"),
+        "canon_version": (payload.get("canon") or {}).get("canon_version"),
+        "capabilities": capabilities,
+        "resource_classes": {
+            item["id"]: item for item in payload.get("resource_classes") or []
+        },
+        "maturity_levels": {
+            item["id"] if isinstance(item, dict) else item
+            for item in payload.get("maturity_levels") or []
+        },
+        "source": str(catalog_path),
+        "raw": payload,
+    }
+
+
+def check_catalog_contract(catalog: dict[str, Any] | None = None) -> list[dict[str, Any]]:
+    """Structural contract for capabilities.yaml (CAP-R3, CAP-R8)."""
+    loaded = catalog or load_catalog()
+    raw = loaded["raw"]
+    errors: list[dict[str, Any]] = []
+
+    if "typical_resource_classes" in yaml.dump(raw.get("domains") or []):
+        errors.append(
+            {
+                "code": "capability_typical_resource_classes_forbidden",
+                "path": "capabilities.yaml",
+            }
+        )
+
+    for item in raw.get("resource_classes") or []:
+        for field in ("native_unit", "supply", "capacity_behaviour"):
+            if not item.get(field):
+                errors.append(
+                    {
+                        "code": "capability_resource_class_incomplete",
+                        "class": item.get("id"),
+                        "field": field,
+                    }
+                )
+
+    seen: set[str] = set()
+    for cap_id, cap in loaded["capabilities"].items():
+        if cap_id in seen:
+            errors.append({"code": "capability_duplicate_id", "id": cap_id})
+        seen.add(cap_id)
+        if cap.get("typical_resource_classes"):
+            errors.append(
+                {
+                    "code": "capability_typical_resource_classes_forbidden",
+                    "id": cap_id,
+                }
+            )
+        if not cap.get("anchors") and not cap.get("anchor_note"):
+            errors.append(
+                {
+                    "code": "capability_missing_anchor",
+                    "id": cap_id,
+                }
+            )
+        for other in cap.get("depends_on") or []:
+            if other not in loaded["capabilities"]:
+                errors.append(
+                    {
+                        "code": "capability_unknown_depends_on",
+                        "id": cap_id,
+                        "depends_on": other,
+                    }
+                )
+        for other in cap.get("may_use") or []:
+            if other not in loaded["capabilities"]:
+                errors.append(
+                    {
+                        "code": "capability_unknown_may_use",
+                        "id": cap_id,
+                        "may_use": other,
+                    }
+                )
+    return errors
+
+
+def _validate_requirement(requirement: dict[str, Any], canon: dict[str, Any]) -> None:
+    cap_id = requirement["capability"]
+    capability = canon["capabilities"].get(cap_id)
+    if capability is None:
+        raise CapabilityReviewError("unknown_capability", f"unknown capability {cap_id}")
+    profile = requirement.get("profile")
+    if profile and profile not in (capability.get("profiles") or []):
+        raise CapabilityReviewError(
+            "unknown_profile",
+            f"{cap_id} does not declare profile {profile!r}",
+        )
+    maturity = requirement.get("minimum_maturity")
+    if maturity and maturity not in canon["maturity_levels"]:
+        raise CapabilityReviewError(
+            "unknown_maturity",
+            f"unknown maturity level {maturity!r}",
+        )
+    dimensions = set(capability.get("quality_dimensions") or [])
+    for key in requirement.get("targets") or {}:
+        if key not in dimensions:
+            raise CapabilityReviewError(
+                "unknown_quality_dimension",
+                f"{cap_id} does not declare quality dimension {key!r}",
+            )
+    for constraint in requirement.get("constraints") or []:
+        if constraint.get("dimension") not in dimensions:
+            raise CapabilityReviewError(
+                "unknown_quality_dimension",
+                f"{cap_id} does not declare quality dimension {constraint.get('dimension')!r}",
+            )
+        if constraint.get("predicate") not in PREDICATES:
+            raise CapabilityReviewError(
+                "unknown_predicate",
+                f"predicate {constraint.get('predicate')!r} is not in the closed set",
+            )
+
+
+def _validate_provision(provision: dict[str, Any], canon: dict[str, Any]) -> None:
+    cap_id = provision["capability"]
+    capability = canon["capabilities"].get(cap_id)
+    if capability is None:
+        raise CapabilityReviewError("unknown_capability", f"unknown capability {cap_id}")
+    if provision.get("maturity") not in canon["maturity_levels"]:
+        raise CapabilityReviewError(
+            "unknown_maturity",
+            f"unknown maturity level {provision.get('maturity')!r}",
+        )
+    profile = provision.get("profile")
+    if profile and profile not in (capability.get("profiles") or []):
+        raise CapabilityReviewError(
+            "unknown_profile",
+            f"{cap_id} does not declare profile {profile!r}",
+        )
+    hooks = set(capability.get("evidence_hooks") or [])
+    for item in provision.get("evidence") or []:
+        if item.get("hook") not in hooks:
+            raise CapabilityReviewError(
+                "unknown_evidence_hook",
+                f"{cap_id} does not declare evidence hook {item.get('hook')!r}",
+            )
+    catalog_depends = set(capability.get("depends_on") or [])
+    catalog_may_use = set(capability.get("may_use") or [])
+    for use in provision.get("uses_provisions") or []:
+        relation = use.get("relation")
+        other = use.get("capability")
+        if relation not in {"depends_on", "may_use"}:
+            raise CapabilityReviewError(
+                "unknown_provision_relation",
+                f"relation must be depends_on or may_use, got {relation!r}",
+            )
+        if other not in canon["capabilities"]:
+            raise CapabilityReviewError(
+                "unknown_capability",
+                f"uses_provisions names unknown capability {other}",
+            )
+        if relation == "depends_on" and other not in catalog_depends:
+            raise CapabilityReviewError(
+                "uses_provisions_not_in_catalog",
+                f"{cap_id} depends_on {other} is not declared in the catalog",
+            )
+        if relation == "may_use" and other not in catalog_may_use:
+            # SHOULD, not MUST — recorded as a warning-shaped error code the
+            # reviewer can see without failing the join.
+            pass
+    seen: set[str] = set()
+    for row in provision.get("consumes") or []:
+        klass = row.get("class")
+        if klass not in canon["resource_classes"]:
+            raise CapabilityReviewError(
+                "unknown_resource_class",
+                f"unknown resource class {klass!r}",
+            )
+        if klass in seen:
+            raise CapabilityReviewError(
+                "duplicate_consumption_class",
+                f"duplicate consumption row for class {klass}",
+            )
+        seen.add(klass)
+        declared = canon["resource_classes"][klass]
+        unit = (row.get("quantity") or {}).get("unit")
+        if unit and unit != declared.get("native_unit"):
+            raise CapabilityReviewError(
+                "native_unit_mismatch",
+                f"class {klass} native unit is {declared.get('native_unit')!r}, got {unit!r}",
+            )
+        basis = row.get("basis")
+        if basis not in EVIDENCE_BASES:
+            raise CapabilityReviewError(
+                "unknown_evidence_basis",
+                f"unknown evidence basis {basis!r}",
+            )
+        value = (row.get("quantity") or {}).get("value")
+        if basis == "unknown":
+            if value is not None:
+                raise CapabilityReviewError(
+                    "unknown_must_not_carry_quantity",
+                    f"class {klass} has basis unknown but carries a quantity",
+                )
+            if not row.get("gap"):
+                raise CapabilityReviewError(
+                    "unknown_must_name_gap",
+                    f"class {klass} is unknown and must name the gap",
+                )
+        elif value is None:
+            raise CapabilityReviewError(
+                "known_basis_requires_quantity",
+                f"class {klass} basis {basis} must carry a quantity",
+            )
+
+
+def review_record(record: dict[str, Any], canon: dict[str, Any] | None = None) -> dict[str, Any]:
+    loaded = canon or load_catalog()
+    for requirement in record.get("requires") or []:
+        _validate_requirement(requirement, loaded)
+    for provision in record.get("provisions") or []:
+        _validate_provision(provision, loaded)
+
+    provisions = {item["capability"]: item for item in record.get("provisions") or []}
+    levels = list(MATURITY_ORDER)
+    met = []
+    for requirement in record.get("requires") or []:
+        provision = provisions.get(requirement["capability"])
+        wanted = requirement.get("minimum_maturity")
+        if provision is None:
+            met.append(
+                {"capability": requirement["capability"], "status": "unprovided"}
+            )
+            continue
+        provided = provision.get("maturity")
+        satisfied = (
+            levels.index(provided) >= levels.index(wanted)
+            if wanted and provided in levels and wanted in levels
+            else True
+        )
+        met.append(
+            {
+                "capability": requirement["capability"],
+                "required": wanted,
+                "provided": provided,
+                "status": "met" if satisfied else "below_requirement",
+            }
+        )
+    return {
+        "ok": True,
+        "record_id": record.get("record_id"),
+        "canon": {
+            "version": loaded["version"],
+            "canon_version": loaded["canon_version"],
+            "source": loaded["source"],
+        },
+        "requirements": met,
+        "provisions": [
+            {
+                "capability": item["capability"],
+                "maturity": item.get("maturity"),
+                "uses_provisions": item.get("uses_provisions") or [],
+            }
+            for item in record.get("provisions") or []
+        ],
+    }
+
+
+def review_path(path: str | Path, catalog_path: Path | None = None) -> dict[str, Any]:
+    record_path = Path(path)
+    if not record_path.is_file():
+        raise CapabilityReviewError(
+            "capability_record_missing",
+            f"capability record not found: {record_path}",
+        )
+    text = record_path.read_text()
+    record = yaml.safe_load(text) if record_path.suffix in {".yaml", ".yml"} else __import__("json").loads(text)
+    return review_record(record, load_catalog(catalog_path))

src/info_tech_canon/cli.py

@@ -24,6 +24,7 @@ from .service import (
     profile_inspect,
     profile_validate,
     read_view,
+    review_capability_record,
     review_kit,
     validate_canon,
     write_validation_report,
@@ -105,6 +106,13 @@ def build_parser() -> argparse.ArgumentParser:
     profile_graph_cmd.add_argument("--format", choices=["json", "mermaid"], default="json")
     profile_graph_cmd.set_defaults(handler=_profile_graph)
 
+    capability_review = sub.add_parser(
+        "capability-review",
+        help="Review a consumer capability record against the live catalog",
+    )
+    capability_review.add_argument("record")
+    capability_review.set_defaults(handler=_capability_review)
+
     api = sub.add_parser("api", help="Run the read-only local API")
     api.add_argument("--host", default="127.0.0.1")
     api.add_argument("--port", type=int, default=8765)
@@ -195,6 +203,10 @@ def _graph(args: argparse.Namespace) -> dict[str, Any]:
     return artifact_graph(_root(args), output_format=args.format)
 
 
+def _capability_review(args: argparse.Namespace) -> dict[str, Any]:
+    return review_capability_record(args.record, _root(args))
+
+
 def _profile_inspect(args: argparse.Namespace) -> dict[str, Any]:
     return profile_inspect(args.profile, _root(args))
 

src/info_tech_canon/service.py

@@ -328,6 +328,18 @@ def profile_graph(
         ) from exc
 
 
+def review_capability_record(
+    path: str | Path,
+    root: Path | str | None = None,
+) -> dict[str, Any]:
+    from .capability import CapabilityReviewError, review_path
+
+    try:
+        return review_path(path)
+    except CapabilityReviewError as exc:
+        raise CanonServiceError(exc.code, exc.message) from exc
+
+
 def generate_indexes(root: Path | str | None = None) -> dict[str, Any]:
     return generation.generate_indexes(load_context(root))
 

src/info_tech_canon/validation.py

@@ -51,6 +51,9 @@ REQUIRED_SCHEMAS = (
     "agent-brief.schema.yaml",
     "workplan.schema.yaml",
     "alignment-review.schema.yaml",
+    "capability.schema.yaml",
+    "capability-record.schema.yaml",
+    "attribute-value-type.schema.yaml",
 )
 
 RETRIEVAL_BRIEF_KINDS = {
@@ -353,6 +356,7 @@ def structural_checks(context: Any) -> dict[str, list[dict[str, Any]]]:
     _check_required_top_level_files(context.repo_root, errors)
     _check_required_infospace_dirs(context.infospace_root, errors)
     _check_required_schemas(context.infospace_root, errors)
+    _check_capability_catalog(errors)
     _check_canon_paths(context.repo_root, context.infospace_root, errors)
     _check_artifact_index(context.repo_root, context.infospace_root, errors)
     _check_agent_assets(context.infospace_root, context.infospace.artifacts, errors)
@@ -415,6 +419,20 @@ def _check_required_infospace_dirs(
             )
 
 
+def _check_capability_catalog(errors: list[dict[str, Any]]) -> None:
+    from .capability import check_catalog_contract
+
+    try:
+        errors.extend(check_catalog_contract())
+    except Exception as exc:  # pragma: no cover - catalog missing is structural
+        errors.append(
+            {
+                "code": "capability_catalog_unreadable",
+                "message": str(exc),
+            }
+        )
+
+
 def _check_required_schemas(
     infospace_root: Path,
     errors: list[dict[str, Any]],