Derive verifier navigation from installed checker metadata

src/jacobian/adapters/mcp/projections.py

@@ -62,54 +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",
-        ),
-    ),
-    "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",
-        ),
-    ),
 }
 
 
@@ -128,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,
@@ -210,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]
@@ -247,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
@@ -258,6 +231,40 @@ def _discovery_operation_card(
     }
 
 
+def _compact_discovery_relationships(
+    response: dict[str, Any],
+    matches: list[dict[str, Any]],
+) -> None:
+    """Remove deterministic suffix links until the discovery response is bounded."""
+
+    while (
+        len(_mcp_text_json_bytes(response)) > CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT
+    ):
+        for match in reversed(matches):
+            related = match.get("related_capabilities")
+            if isinstance(related, list) and related:
+                related.pop()
+                response["related_capabilities_truncated"] = True
+                response["truncation_reason"] = "BYTE_LIMIT"
+                break
+        else:
+            return
+
+
+def _compact_inspection_relationships(response: dict[str, Any]) -> None:
+    """Bound exact inspection without discarding its authoritative contract."""
+
+    related = response.get("related_capabilities")
+    while (
+        len(_mcp_text_json_bytes(response)) > CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT
+        and isinstance(related, list)
+        and related
+    ):
+        related.pop()
+        response["related_capabilities_truncated"] = True
+        response["truncation_reason"] = "BYTE_LIMIT"
+
+
 def _discovery_recovery_paths(
     request: CapabilityDiscoveryRequest,
     *,
@@ -313,7 +320,7 @@ def _capability_descriptor_view(
     view: CapabilityDescriptionView,
 ) -> dict[str, Any]:
     if view == "FULL":
-        return descriptor.model_dump(mode="json")
+        return descriptor.model_dump(mode="json", exclude={"related_capabilities"})
     runtime = descriptor.provider_runtime
     if view == "SUMMARY":
         runtime_summary = (
@@ -462,8 +469,11 @@ def _capability_discovery_response(
         "recovery_paths_are_unranked": True,
         "response_byte_limit": CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT,
         "truncation_reason": None,
+        "related_capabilities_truncated": False,
     }
     matches = cast(list[dict[str, Any]], response["matches"])
+    response["match_metadata_truncated"] = False
+    _compact_discovery_relationships(response, matches)
     while (
         len(_mcp_text_json_bytes(response)) > CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT
         and len(matches) > 1
@@ -482,7 +492,6 @@ def _capability_discovery_response(
         available_domains.pop()
         response["available_domains_truncated"] = True
         response["truncation_reason"] = "BYTE_LIMIT"
-    response["match_metadata_truncated"] = False
     compact_fields = (
         "tags",
         "matched_on",

src/jacobian/adapters/mcp/tools.py

@@ -10,12 +10,17 @@
 from mcp_types import CallToolResult, TextContent
 from pydantic import BaseModel, ConfigDict, Field, RootModel, StrictInt
 
-from jacobian.adapters.mcp.constants import _CAPABILITY_SCOPE_RULE, ReasoningLogMode
+from jacobian.adapters.mcp.constants import (
+    _CAPABILITY_SCOPE_RULE,
+    CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT,
+    ReasoningLogMode,
+)
 from jacobian.adapters.mcp.context import AppState, _runtime
 from jacobian.adapters.mcp.projections import (
     _capability_descriptor_view,
     _capability_discovery_response,
     _capability_inspection_extensions,
+    _compact_inspection_relationships,
 )
 from jacobian.adapters.mcp.tooling import (
     AgentRecoveryError,
@@ -84,6 +89,7 @@ class _CapabilityDiscoveryResult(_CapabilityDiscoveryFields):
     truncation_reason: str | None = None
     available_domains_total: StrictInt
     available_domains_truncated: bool
+    related_capabilities_truncated: bool
     match_metadata_truncated: bool
 
 
@@ -92,6 +98,9 @@ class _CapabilityInspectionResult(_CapabilityDiscoveryFields):
     view: CapabilityDescriptionView
     capability: dict[str, Any]
     scope_rule: str | dict[str, Any]
+    response_byte_limit: StrictInt
+    truncation_reason: str | None = None
+    related_capabilities_truncated: bool
     invocations: list[dict[str, Any]] | None = None
     related_capabilities: list[dict[str, Any]] | None = None
     synchronous_execution: dict[str, Any] | None = None
@@ -198,6 +207,10 @@ def _find_text_projection(response: dict[str, Any]) -> dict[str, Any]:
             "routing_basis": response.get("routing_basis"),
             "total_matches": response.get("total_matches"),
             "truncated": response.get("truncated"),
+            "truncation_reason": response.get("truncation_reason"),
+            "related_capabilities_truncated": response.get(
+                "related_capabilities_truncated"
+            ),
             "next_cursor": response.get("next_cursor"),
             "available_recovery_paths": response.get("available_recovery_paths"),
         }
@@ -229,6 +242,11 @@ def _find_text_projection(response: dict[str, Any]) -> dict[str, Any]:
         projection["invocations"] = response["invocations"]
     if response.get("related_capabilities"):
         projection["related_capabilities"] = response["related_capabilities"]
+    projection["response_byte_limit"] = response.get("response_byte_limit")
+    projection["truncation_reason"] = response.get("truncation_reason")
+    projection["related_capabilities_truncated"] = response.get(
+        "related_capabilities_truncated"
+    )
     return projection
 
 
@@ -410,6 +428,9 @@ async def capability_describe(
             "policy_digest": capability_catalog.policy_digest,
             "capability": _capability_descriptor_view(descriptor, view=view),
             "scope_rule": _CAPABILITY_SCOPE_RULE,
+            "response_byte_limit": CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT,
+            "truncation_reason": None,
+            "related_capabilities_truncated": False,
         }
         if view == "SUMMARY":
             response["next_views"] = {
@@ -456,6 +477,7 @@ async def capability_describe(
             response.update(
                 _capability_inspection_extensions(capability_id, descriptors)
             )
+            _compact_inspection_relationships(response)
         if (
             view != "SUMMARY"
             and capability_id == "lean.check"

src/jacobian/capability_registry.py

@@ -8,6 +8,7 @@
 from jacobian.capability_validation import validator
 from jacobian.contracts.capabilities import (
     CapabilityCatalog,
+    CapabilityCatalogRelationshipKind,
     CapabilityProviderAvailability,
 )
 
@@ -43,6 +44,18 @@ def register(self: Any, adapter: AdapterLike) -> None:
                 f"capability {descriptor.capability_id} is unavailable: "
                 f"{descriptor.provider_runtime.diagnostic}"
             )
+        verification_relationship_kinds = {
+            CapabilityCatalogRelationshipKind.INDEPENDENT_VERIFIER,
+            CapabilityCatalogRelationshipKind.VERIFIABLE_RESULT_PRODUCER,
+        }
+        if any(
+            relationship.kind in verification_relationship_kinds
+            for relationship in descriptor.related_capabilities
+        ):
+            raise CapabilityError(
+                "verification-sensitive catalog relationships require "
+                "operator-authorized checker registration"
+            )
         validator(descriptor.input_schema)
         validator(descriptor.output_schema)
         for example in descriptor.invocation_examples:
@@ -58,16 +71,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 installer-authorized 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(