Derive verifier navigation from checker metadata

Author: yuelgrace1810-ops

src/jacobian/adapters/mcp/projections.py

@@ -62,66 +62,6 @@ def _mcp_text_json_bytes(value: object) -> bytes:
             "materialize named Boolean CNF for finite colorings and forbidden patterns",
         ),
     ),
-    "graph.invariant.maximum_matching.compute": (
-        (
-            "graph.invariant.maximum_matching.verify",
-            "independently replay the stored Tutte-Berge certificate",
-        ),
-    ),
-    "graph.invariant.maximum_matching.verify": (
-        (
-            "graph.invariant.maximum_matching.compute",
-            "produce a matching witness and Tutte-Berge certificate",
-        ),
-    ),
-    "probability.graph_reliability.connection_probability.compute": (
-        (
-            "probability.graph_reliability.connection_probability.verify",
-            "independently replay every edge state and terminal connection",
-        ),
-    ),
-    "probability.graph_reliability.connection_probability.verify": (
-        (
-            "probability.graph_reliability.connection_probability.compute",
-            "compute the complete exact edge-state ledger",
-        ),
-    ),
-    "graph.hamiltonian_path.decide": (
-        (
-            "graph.hamiltonian_path.verify",
-            "independently verify the stored positive or negative decision",
-        ),
-    ),
-    "graph.hamiltonian_path.verify": (
-        (
-            "graph.hamiltonian_path.decide",
-            "produce a complete bounded decision and optional path witness",
-        ),
-    ),
-    "polynomial.jacobian_syzygy.minimum_degree.compute": (
-        (
-            "polynomial.jacobian_syzygy.minimum_degree.verify",
-            "independently rebuild the graded maps, ranks, minors, and first kernel",
-        ),
-    ),
-    "polynomial.jacobian_syzygy.minimum_degree.verify": (
-        (
-            "polynomial.jacobian_syzygy.minimum_degree.compute",
-            "produce the provenance-bound graded rank ledger and kernel witness",
-        ),
-    ),
-    "geometry.projective_line_arrangement.flats.materialize": (
-        (
-            "geometry.projective_line_arrangement.flats.verify",
-            "independently rebuild all projective flats and pair accounting",
-        ),
-    ),
-    "geometry.projective_line_arrangement.flats.verify": (
-        (
-            "geometry.projective_line_arrangement.flats.materialize",
-            "materialize normalized lines, exact flats, incidences and multiplicities",
-        ),
-    ),
 }
 
 
@@ -140,16 +80,27 @@ def _capability_inspection_extensions(
     descriptors: dict[str, CapabilityDescriptor],
 ) -> dict[str, Any]:
     extensions: dict[str, Any] = {}
-    related = [
+    related = {
+        item.capability_id: item.model_dump(mode="json")
+        for item in descriptors[capability_id].related_capabilities
+        if item.capability_id in descriptors and item.capability_id != capability_id
+    }
+    related.update(
         {
-            "capability_id": related_id,
-            "relationship": relationship,
+            related_id: {
+                "capability_id": related_id,
+                "relationship": relationship,
+            }
+            for related_id, relationship in _RELATED_CAPABILITIES.get(capability_id, ())
+            if related_id in descriptors
+            and related_id != capability_id
+            and related_id not in related
         }
-        for related_id, relationship in _RELATED_CAPABILITIES.get(capability_id, ())
-        if related_id in descriptors
-    ]
+    )
     if related:
-        extensions["related_capabilities"] = related
+        extensions["related_capabilities"] = [
+            related[related_id] for related_id in sorted(related)
+        ]
     if capability_id.startswith(("sat.", "smt.")):
         extensions["synchronous_execution"] = {
             "remote_safe_wall_seconds_max": 150,
@@ -222,16 +173,26 @@ def _discovery_operation_card(
     """Add compact decision facts without recommending a research action."""
 
     runtime = descriptor.provider_runtime
-    related = [
+    related = {
+        item.capability_id: item.model_dump(mode="json")
+        for item in descriptor.related_capabilities
+        if item.capability_id in descriptors
+        and item.capability_id != descriptor.capability_id
+    }
+    related.update(
         {
-            "capability_id": related_id,
-            "relationship": relationship,
+            related_id: {
+                "capability_id": related_id,
+                "relationship": relationship,
+            }
+            for related_id, relationship in _RELATED_CAPABILITIES.get(
+                descriptor.capability_id, ()
+            )
+            if related_id in descriptors
+            and related_id != descriptor.capability_id
+            and related_id not in related
         }
-        for related_id, relationship in _RELATED_CAPABILITIES.get(
-            descriptor.capability_id, ()
-        )
-        if related_id in descriptors
-    ]
+    )
     invocation_example = None
     if descriptor.invocation_examples:
         example = descriptor.invocation_examples[0]
@@ -259,7 +220,7 @@ def _discovery_operation_card(
         "provider_availability": (
             runtime.availability.value if runtime is not None else "UNKNOWN"
         ),
-        "related_capabilities": related,
+        "related_capabilities": [related[item] for item in sorted(related)],
         **(
             {"invocation_example": invocation_example}
             if invocation_example is not None

src/jacobian/capability_registry.py

@@ -58,16 +58,44 @@ def register(self: Any, adapter: AdapterLike) -> None:
         self._adapters[descriptor.capability_id] = adapter
 
     def catalog(self: Any) -> CapabilityCatalog:
-        visible = tuple(
+        projected = tuple(
             projected
             for name in sorted(self._adapters)
             if (projected := self.policy.project(self._adapters[name].descriptor))
             is not None
         )
+        visible_ids = {descriptor.capability_id for descriptor in projected}
+        visible = []
+        for descriptor in projected:
+            relationships = {
+                item.capability_id: item for item in descriptor.related_capabilities
+            }
+            for related_id, relationship in self._catalog_relationships.get(
+                descriptor.capability_id, {}
+            ).items():
+                previous = relationships.get(related_id)
+                if previous is not None and previous != relationship:
+                    raise CapabilityError(
+                        "conflicting projected catalog relationship: "
+                        f"{descriptor.capability_id} -> {related_id}"
+                    )
+                relationships[related_id] = relationship
+            visible.append(
+                descriptor.model_copy(
+                    update={
+                        "related_capabilities": tuple(
+                            relationships[related_id]
+                            for related_id in sorted(relationships)
+                            if related_id in visible_ids
+                            and related_id != descriptor.capability_id
+                        )
+                    }
+                )
+            )
         return CapabilityCatalog(
             policy_profile=self.policy.profile,
             policy_digest=self.policy.digest,
-            capabilities=visible,
+            capabilities=tuple(visible),
         )
 
 

src/jacobian/capability_service.py

@@ -32,6 +32,7 @@
 from jacobian.capability_registry import CapabilityRegistryMixin
 from jacobian.capability_verification import CapabilityVerificationMixin
 from jacobian.contracts.capabilities import (
+    CapabilityCatalogRelationship,
     CapabilityDescriptor,
     CapabilityMode,
     CapabilityRequest,
@@ -195,6 +196,27 @@ def __init__(
         self.store = store
         self.policy = policy or CapabilityPolicy()
         self._adapters: dict[str, CapabilityAdapter] = {}
+        self._catalog_relationships: dict[
+            str, dict[str, CapabilityCatalogRelationship]
+        ] = {}
+
+    def register_catalog_relationship(
+        self,
+        source_capability_id: str,
+        relationship: CapabilityCatalogRelationship,
+    ) -> None:
+        """Register one authoritative directed catalog relationship."""
+
+        if source_capability_id == relationship.capability_id:
+            raise CapabilityError("a capability cannot relate to itself")
+        related = self._catalog_relationships.setdefault(source_capability_id, {})
+        previous = related.get(relationship.capability_id)
+        if previous is not None and previous != relationship:
+            raise CapabilityError(
+                "conflicting catalog relationship: "
+                f"{source_capability_id} -> {relationship.capability_id}"
+            )
+        related[relationship.capability_id] = relationship
 
 
 def load_capability_adapter(

src/jacobian/contracts/capabilities.py

@@ -412,6 +412,34 @@ class CapabilityCompletenessStatus(StrEnum):
     COMPLETE = "COMPLETE"
 
 
+class CapabilityCatalogRelationshipKind(StrEnum):
+    """Factual installed-capability relationship exposed by the catalog."""
+
+    INDEPENDENT_VERIFIER = "INDEPENDENT_VERIFIER"
+    VERIFIABLE_RESULT_PRODUCER = "VERIFIABLE_RESULT_PRODUCER"
+
+
+class CapabilityCatalogRelationship(ContractModel):
+    """One typed navigation edge to another installed capability."""
+
+    capability_id: CapabilityId
+    kind: CapabilityCatalogRelationshipKind
+    relationship: str = Field(min_length=1, max_length=256)
+
+
+class CapabilityCatalogRelationshipRegistration(ContractModel):
+    """One authoritative directed relationship before catalog projection."""
+
+    source_capability_id: CapabilityId
+    related_capability: CapabilityCatalogRelationship
+
+    @model_validator(mode="after")
+    def reject_self_relationship(self) -> Self:
+        if self.source_capability_id == self.related_capability.capability_id:
+            raise ValueError("a capability cannot relate to itself")
+        return self
+
+
 class CapabilityDescriptor(ContractModel):
     """One installed operation advertised by an operator-installed adapter."""
 
@@ -432,6 +460,7 @@ class CapabilityDescriptor(ContractModel):
     )
     accepted_artifact_types: tuple[ArtifactUri, ...] = ()
     produced_artifact_types: tuple[ArtifactUri, ...] = ()
+    related_capabilities: tuple[CapabilityCatalogRelationship, ...] = ()
     discovery_visible: bool = True
     invocation_examples: tuple[CapabilityInvocationExample, ...] = ()
 
@@ -447,6 +476,11 @@ def require_modes_and_canonical_schemas(self) -> Self:
         )
         if len(set(self.produced_artifact_types)) != len(self.produced_artifact_types):
             raise ValueError("produced artifact types must be unique")
+        related_ids = [item.capability_id for item in self.related_capabilities]
+        if self.capability_id in related_ids:
+            raise ValueError("a capability cannot relate to itself")
+        if len(set(related_ids)) != len(related_ids):
+            raise ValueError("related capability IDs must be unique")
         if len({example.name for example in self.invocation_examples}) != len(
             self.invocation_examples
         ):

src/jacobian/exact_domain_checkers.py

@@ -4,7 +4,7 @@
 
 import logging
 from collections.abc import Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import Any, Literal
 
 from pydantic import ValidationError
@@ -17,6 +17,9 @@
 from jacobian.contracts.capabilities import (
     CapabilityAssurance,
     CapabilityAssuranceLevel,
+    CapabilityCatalogRelationship,
+    CapabilityCatalogRelationshipKind,
+    CapabilityCatalogRelationshipRegistration,
     CapabilityCompleteness,
     CapabilityCompletenessStatus,
     CapabilityDescriptor,
@@ -97,6 +100,7 @@ class ExactDomainCheckerInstallation:
     provider_runtimes: dict[str, CapabilityProviderRuntime]
     witness_schema_uri: str | None = None
     diagnostics: tuple[CapabilityDiagnostic, ...] = ()
+    catalog_relationships: tuple[CapabilityCatalogRelationshipRegistration, ...] = ()
 
 
 @dataclass(frozen=True, slots=True)
@@ -368,6 +372,7 @@ def install_exact_domain_verification(
     ):
         return (), installation
     adapters: list[CapabilityAdapter] = []
+    catalog_relationships: list[CapabilityCatalogRelationshipRegistration] = []
     result_models = {
         operation.capability_id: operation.result_model
         for bundle, _installed_bundle in bundles.values()
@@ -404,6 +409,38 @@ def install_exact_domain_verification(
                 stored_result_input=declaration.capability_id in stored_producers,
             )
         )
+        verifier_id = declaration.verification_capability_id
+        if verifier_id is None:
+            raise ValueError("exact replay declaration has no verifier capability ID")
+        catalog_relationships.extend(
+            (
+                CapabilityCatalogRelationshipRegistration(
+                    source_capability_id=declaration.capability_id,
+                    related_capability=CapabilityCatalogRelationship(
+                        capability_id=verifier_id,
+                        kind=(CapabilityCatalogRelationshipKind.INDEPENDENT_VERIFIER),
+                        relationship=(
+                            "independently verify this exact producer result"
+                        ),
+                    ),
+                ),
+                CapabilityCatalogRelationshipRegistration(
+                    source_capability_id=verifier_id,
+                    related_capability=CapabilityCatalogRelationship(
+                        capability_id=declaration.capability_id,
+                        kind=(
+                            CapabilityCatalogRelationshipKind.VERIFIABLE_RESULT_PRODUCER
+                        ),
+                        relationship=(
+                            "produce the exact result accepted by this verifier"
+                        ),
+                    ),
+                ),
+            )
+        )
+    installation = replace(
+        installation, catalog_relationships=tuple(catalog_relationships)
+    )
     return tuple(adapters), installation
 
 

src/jacobian/portfolio/core_installation.py

@@ -206,6 +206,11 @@ def install_domain_verification(
         )
         for adapter in adapters:
             self.context.register_capability(adapter)
+        for relationship in result.exact_domain_checkers.catalog_relationships:
+            ctx.capabilities.register_catalog_relationship(
+                relationship.source_capability_id,
+                relationship.related_capability,
+            )
 
 
 def _conjecture_ingestion_installation(