Derive verifier navigation from installed checker metadata (#899)

* Fix graph reliability verifier discovery

* Derive verifier navigation from checker metadata

* fix: compact discovery relationships within byte limit

* fix: bound capability relationship inspection

* fix: avoid duplicate full-view relationships

* fix: reserve verifier relationships for checker authority

* fix: keep verifier relationship authority installer-private

* fix: fail closed on oversized capability inspection

* fix: scope inspection bounds to relationships

* test: avoid fixed checker inventory count

* test: cover oversized exact inspections

* fix: account for discovery metadata before compaction

* fix: snapshot validated capability descriptors

---------

Co-authored-by: morluto <76467478+morluto@users.noreply.github.com>
Co-authored-by: Grace Lee Rui Yue <graceleeruiyue@Graces-MacBook-Air-2.local>

Author: 3 people

src/jacobian/adapters/mcp/constants.py

@@ -8,6 +8,7 @@
 
 _LOGGER = logging.getLogger(__name__)
 CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT = 16_384
+CAPABILITY_INSPECTION_RELATIONSHIPS_BYTE_LIMIT = 16_384
 CapabilityDescriptionView = Literal["SUMMARY", "CONTRACT", "FULL"]
 
 

src/jacobian/adapters/mcp/projections.py

@@ -8,7 +8,10 @@
 import threading
 from typing import TYPE_CHECKING, Any, Literal, cast
 
-from jacobian.adapters.mcp.constants import CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT
+from jacobian.adapters.mcp.constants import (
+    CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT,
+    CAPABILITY_INSPECTION_RELATIONSHIPS_BYTE_LIMIT,
+)
 from jacobian.bounded_process import bounded_process_cancellation
 from jacobian.canonical import canonicalize_json
 from jacobian.capability_service import CapabilityDiscoveryCursorError
@@ -62,54 +65,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 +83,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 +176,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 +223,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 +234,41 @@ 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 relationships without truncating the descriptor."""
+
+    related = response.get("related_capabilities")
+    while (
+        len(_mcp_text_json_bytes(related))
+        > CAPABILITY_INSPECTION_RELATIONSHIPS_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 +324,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 +473,13 @@ def _capability_discovery_response(
         "recovery_paths_are_unranked": True,
         "response_byte_limit": CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT,
         "truncation_reason": None,
+        "related_capabilities_truncated": False,
+        "available_domains_total": len(discovered_payload["available_domains"]),
+        "available_domains_truncated": False,
+        "match_metadata_truncated": False,
     }
     matches = cast(list[dict[str, Any]], response["matches"])
+    _compact_discovery_relationships(response, matches)
     while (
         len(_mcp_text_json_bytes(response)) > CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT
         and len(matches) > 1
@@ -473,16 +489,13 @@ def _capability_discovery_response(
         response["next_cursor"] = matches[-1]["capability_id"]
         response["truncation_reason"] = "BYTE_LIMIT"
     available_domains = cast(list[str], response["available_domains"])
-    response["available_domains_total"] = len(available_domains)
-    response["available_domains_truncated"] = False
     while (
         len(_mcp_text_json_bytes(response)) > CAPABILITY_DISCOVERY_RESPONSE_BYTE_LIMIT
         and available_domains
     ):
         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_INSPECTION_RELATIONSHIPS_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]
+    related_capabilities_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,13 @@ 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["related_capabilities_byte_limit"] = response.get(
+        "related_capabilities_byte_limit"
+    )
+    projection["truncation_reason"] = response.get("truncation_reason")
+    projection["related_capabilities_truncated"] = response.get(
+        "related_capabilities_truncated"
+    )
     return projection
 
 
@@ -410,6 +430,11 @@ async def capability_describe(
             "policy_digest": capability_catalog.policy_digest,
             "capability": _capability_descriptor_view(descriptor, view=view),
             "scope_rule": _CAPABILITY_SCOPE_RULE,
+            "related_capabilities_byte_limit": (
+                CAPABILITY_INSPECTION_RELATIONSHIPS_BYTE_LIMIT
+            ),
+            "truncation_reason": None,
+            "related_capabilities_truncated": False,
         }
         if view == "SUMMARY":
             response["next_views"] = {
@@ -471,6 +496,7 @@ async def capability_describe(
                     else {"status": "UNAVAILABLE", "detail": None}
                 ),
             }
+        _compact_inspection_relationships(response)
         return _find_result(response)
 
 

src/jacobian/capability_dispatch.py

@@ -47,7 +47,7 @@ def invoke(self: Any, request: CapabilityRequest) -> CapabilityResult:
             result = _unknown_capability_failure(self, request)
             log_invocation(result, started)
             return result
-        descriptor = adapter.descriptor
+        descriptor = self._descriptors[request.capability_id]
         resolution = _capability_resolution_failure(self, request, descriptor)
         if resolution is not None:
             log_invocation(resolution, started)