Check import manifests by hash and name together (INFO-WP-0028 T01-T03)

import-review takes any partner manifest and returns, per concept, whether the
name resolves in the ownership index and to which artifact, and per entry
whether the pinned SHA-256 matches the blob at the declared source commit. Both
run in one pass so neither can be recorded without the other, which is the
failure this workplan exists to prevent. It exits non-zero on a finding, reads
JSON or YAML, needs no partner checkout, and carries its own limit: resolution
proves a name exists and names one owner, nothing more.

Accepted manifests are registered under infospace/interfaces/manifests/ as
provenance-preserving copies owned by the partner, with the partner revision and
retrieval date recorded. Editing a copy to make a check pass is forbidden in the
file itself. Validation re-resolves them and reports drift as
federation_import_drift, a warning naming the partner rather than an error,
because a stale partner pin is not this repository's file to fix.

The review kit gains an extension-boundary-review template requiring hash count,
resolution count and conflict count as three separate lines, and an operating
rule saying one is never evidence of another. Both boundary files carry the
standing-check result.

Verified live: security-canon resolves 11 of 11, interface-canon 23 of 25 with
the two known Interface and Endpoint pins. make check passes with 58 tests,
clean validation and those two warnings.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 3588@bnt-lap001
Assistant-Session: 24b80f66-e5a7-4e61-99fe-2d422e6d17da

src/info_tech_canon/cli.py

@@ -127,6 +127,11 @@ def build_parser() -> argparse.ArgumentParser:
         help="Measure declared concepts against candidates defined in artifact prose")
     concepts.add_argument("--artifact", help="Limit the report to one artifact id")
     concepts.set_defaults(handler=_concept_coverage)
+    imports_cmd = sub.add_parser(
+        "import-review",
+        help="Resolve a partner import manifest: pinned hashes and concept names")
+    imports_cmd.add_argument("manifest")
+    imports_cmd.set_defaults(handler=_import_review)
     bundle = sub.add_parser("export-emission-contract", help="Export a content-addressed contract tar")
     bundle.add_argument("destination")
     bundle.set_defaults(handler=_export_emission)
@@ -261,6 +266,13 @@ def _concept_coverage(args):
     return dict(report, ok=True)
 
 
+def _import_review(args):
+    from .federation import import_manifest_review
+    from .service import load_context
+
+    return import_manifest_review(load_context(_root(args)), args.manifest)
+
+
 def _export_emission(args):
     from .maintenance import export_emission_bundle
     from .paths import infospace_root

src/info_tech_canon/federation.py

@@ -0,0 +1,126 @@
+"""Resolve a partner import manifest: pinned blob hashes and concept names.
+
+A hash proves the reviewed file is the pinned file. It says nothing about
+whether the concept named in the manifest exists in it. Both questions are
+answered here, in one pass, so neither can be recorded without the other.
+"""
+
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+
+import yaml
+
+from . import generation
+
+RESOLVED = "resolved"
+WRONG_ARTIFACT = "wrong_artifact"
+UNOWNED = "unowned"
+
+
+def _manifest(path: Path) -> dict:
+    from .service import CanonServiceError
+
+    try:
+        text = path.read_text(encoding="utf-8")
+    except OSError as exc:
+        raise CanonServiceError("manifest_unreadable", str(exc)) from exc
+    try:
+        data = json.loads(text) if path.suffix == ".json" else yaml.safe_load(text)
+    except (json.JSONDecodeError, yaml.YAMLError) as exc:
+        raise CanonServiceError("manifest_unparsable", str(exc)) from exc
+    if not isinstance(data, dict) or not isinstance(data.get("imports"), list):
+        raise CanonServiceError("manifest_shape", "Expected a mapping with an 'imports' list")
+    return data
+
+
+def _blob_sha256(repository: Path, commit: str, path: str) -> str | None:
+    """Hash the file as the pinned revision holds it, not as it stands today."""
+    try:
+        blob = subprocess.run(["git", "show", f"{commit}:{path}"], cwd=repository,
+                              capture_output=True, check=True).stdout
+    except (OSError, subprocess.CalledProcessError):
+        return None
+    return hashlib.sha256(blob).hexdigest()
+
+
+def import_manifest_review(context, manifest_path: str | Path) -> dict:
+    data = _manifest(Path(manifest_path))
+    commit = data.get("source_commit")
+    repository = context.infospace_root.parent
+    owners: dict[str, set[str]] = {}
+    for item in generation.concept_ownership(context)["concepts"]:
+        owners.setdefault(generation._normalize_concept(item["concept"]), set()).add(item["owner"])
+    by_path = {artifact.path: artifact.id for artifact in context.infospace.artifacts}
+
+    entries, findings = [], []
+    prefix = context.infospace_root.name + "/"
+    for entry in data["imports"]:
+        path = entry.get("path", "")
+        # Manifests pin repository-relative paths; the registry keys on infospace-relative.
+        pinned = by_path.get(path) or by_path.get(path[len(prefix):] if path.startswith(prefix) else path)
+        concepts = []
+        for name in entry.get("concepts") or []:
+            found = sorted(owners.get(generation._normalize_concept(name), set()))
+            if pinned and found == [pinned]:
+                status = RESOLVED
+            elif found:
+                status = WRONG_ARTIFACT
+            else:
+                status = UNOWNED
+            concepts.append({"concept": name, "status": status,
+                             "owner": found[0] if len(found) == 1 else found or None})
+            if status != RESOLVED:
+                findings.append({"code": f"import_{status}", "concept": name,
+                                 "pinned_path": path, "pinned_artifact": pinned,
+                                 "resolves_to": found or None})
+        expected = _blob_sha256(repository, commit, path) if commit else None
+        declared = entry.get("sha256")
+        if expected is None:
+            hash_status = "unverifiable"
+        elif declared == expected:
+            hash_status = "match"
+        else:
+            hash_status = "mismatch"
+            findings.append({"code": "import_hash_mismatch", "pinned_path": path,
+                             "declared": declared, "actual": expected})
+        entries.append({"path": path, "artifact": pinned, "hash": hash_status,
+                        "declared_sha256": declared, "concepts": concepts})
+
+    resolved = sum(1 for item in entries for concept in item["concepts"]
+                   if concept["status"] == RESOLVED)
+    total = sum(len(item["concepts"]) for item in entries)
+    return {"ok": not findings, "manifest": str(manifest_path),
+            "source_commit": commit, "entries": entries, "findings": findings,
+            "resolved": resolved, "declared": total,
+            "limit": "Resolution proves the name exists and names one owner. Whether the "
+                     "partner uses the concept as its owner defines it stays with boundary review."}
+
+
+REGISTERED = "interfaces/manifests/manifests.yaml"
+
+
+def registered_manifests(context) -> list[dict]:
+    path = context.infospace_root / REGISTERED
+    if not path.exists():
+        return []
+    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+    return data.get("manifests") or []
+
+
+def registered_drift(context) -> list[dict]:
+    """Stale partner pins are the partner's to fix, so this warns and never errors."""
+    warnings = []
+    for item in registered_manifests(context):
+        manifest = context.infospace_root / REGISTERED
+        review = import_manifest_review(context, manifest.parent / item["file"])
+        for finding in review["findings"]:
+            warnings.append({"code": "federation_import_drift",
+                             "partner": item.get("partner"),
+                             "partner_revision": item.get("partner_revision"),
+                             "finding": finding["code"],
+                             "concept": finding.get("concept"),
+                             "pinned_path": finding.get("pinned_path"),
+                             "resolves_to": finding.get("resolves_to")})
+    return warnings

src/info_tech_canon/service.py

@@ -233,6 +233,8 @@ def validate_canon(root: Path | str | None = None) -> dict[str, Any]:
     declaration = concept_declaration_checks(context, ownership)
     errors.extend(declaration["errors"])
     warnings.extend(declaration["warnings"])
+    from .federation import registered_drift
+    warnings.extend(registered_drift(context))
 
     return {
         "ok": not errors,