Clarify inline verifier invocation envelopes

src/jacobian/adapters/mcp/projections.py

@@ -16,6 +16,7 @@
 from jacobian.canonical import canonicalize_json
 from jacobian.capability_service import CapabilityDiscoveryCursorError
 from jacobian.contracts.capabilities import (
+    CapabilityCatalogRelationshipKind,
     CapabilityDescriptor,
     CapabilityDiscoveryBrowseRecoveryPath,
     CapabilityDiscoveryInspectCatalogRecoveryPath,
@@ -103,6 +104,11 @@ def _capability_inspection_extensions(
         extensions["related_capabilities"] = [
             related[related_id] for related_id in sorted(related)
         ]
+    invocation_protocol = _inline_verifier_invocation_protocol(
+        capability_id, descriptors[capability_id]
+    )
+    if invocation_protocol is not None:
+        extensions["invocation_protocol"] = invocation_protocol
     if capability_id.startswith(("sat.", "smt.")):
         extensions["synchronous_execution"] = {
             "remote_safe_wall_seconds_max": 150,
@@ -117,6 +123,41 @@ def _capability_inspection_extensions(
     return extensions
 
 
+def _inline_verifier_invocation_protocol(
+    capability_id: str,
+    descriptor: CapabilityDescriptor,
+) -> dict[str, Any] | None:
+    """Expose the generic ``math.run`` envelope for inline replay verifiers."""
+
+    schema = descriptor.input_schema
+    has_producer_relationship = any(
+        relationship.kind
+        is CapabilityCatalogRelationshipKind.VERIFIABLE_RESULT_PRODUCER
+        for relationship in descriptor.related_capabilities
+    )
+    if (
+        CapabilityMode.VERIFY not in descriptor.modes
+        or not has_producer_relationship
+        or set(schema.get("required", [])) != {"input", "candidate"}
+    ):
+        return None
+    return {
+        "tool": "math.run",
+        "arguments_shape": {
+            "capability_id": capability_id,
+            "mode": "VERIFY",
+            "payload": {
+                "input": "<exact producer input object>",
+                "candidate": "<producer response.output.result>",
+            },
+        },
+        "payload_rule": (
+            "input and candidate are sibling fields inside payload; neither field "
+            "belongs beside payload or nested inside the other"
+        ),
+    }
+
+
 def _compact_json_schema(value: Any) -> Any:
     """Drop annotation-only prose while preserving validation semantics."""
 
@@ -206,6 +247,9 @@ def _discovery_operation_card(
             <= _DISCOVERY_INVOCATION_EXAMPLE_BYTE_LIMIT
         ):
             invocation_example = candidate
+    invocation_protocol = _inline_verifier_invocation_protocol(
+        descriptor.capability_id, descriptor
+    )
     return {
         **match,
         "accepted_input_kinds": [
@@ -226,6 +270,11 @@ def _discovery_operation_card(
                 "input_schema_summary": _input_schema_summary(descriptor.input_schema)
             }
         ),
+        **(
+            {"invocation_protocol": invocation_protocol}
+            if invocation_protocol is not None
+            else {}
+        ),
     }
 
 

src/jacobian/adapters/mcp/tools.py

@@ -77,6 +77,29 @@ class _DiscoveryInvocationExample(_MCPOutputModel):
     payload: dict[str, Any]
 
 
+class _InlineVerifierPayload(_MCPOutputModel):
+    """Producer values required by an inline verifier replay."""
+
+    input: Literal["<exact producer input object>"]
+    candidate: Literal["<producer response.output.result>"]
+
+
+class _InlineVerifierArguments(_MCPOutputModel):
+    """Complete ``math.run`` argument envelope for inline verification."""
+
+    capability_id: CapabilityId
+    mode: Literal[CapabilityMode.VERIFY]
+    payload: _InlineVerifierPayload
+
+
+class _InlineVerifierInvocationProtocol(_MCPOutputModel):
+    """Closed discovery contract for invoking an inline verifier."""
+
+    tool: Literal["math.run"]
+    arguments_shape: _InlineVerifierArguments
+    payload_rule: str
+
+
 class _CapabilityDiscoveryOperationCard(CapabilityDiscoveryMatch):
     accepted_input_kinds: tuple[CapabilityInputKind, ...]
     accepted_artifact_types: tuple[ArtifactUri, ...]
@@ -87,6 +110,7 @@ class _CapabilityDiscoveryOperationCard(CapabilityDiscoveryMatch):
     provider_availability: CapabilityProviderAvailability | Literal["UNKNOWN"]
     related_capabilities: tuple[_RelatedCapability, ...]
     invocation_example: _DiscoveryInvocationExample | None = None
+    invocation_protocol: _InlineVerifierInvocationProtocol | None = None
 
 
 class _ProviderRuntimeProjection(_MCPOutputModel):
@@ -226,6 +250,7 @@ class _CapabilityInspectionResult(_CapabilityDiscoveryFields):
     truncation_reason: str | None = None
     related_capabilities_truncated: bool
     invocations: tuple[_CapabilityInvocation, ...] | None = None
+    invocation_protocol: _InlineVerifierInvocationProtocol | None = None
     related_capabilities: tuple[_RelatedCapability, ...] | None = None
     synchronous_execution: _SynchronousExecution | None = None
     next_views: _NextCapabilityViews | None = None
@@ -323,6 +348,7 @@ def _find_text_projection(response: dict[str, Any]) -> dict[str, Any]:
                         "provider_availability",
                         "related_capabilities",
                         "invocation_example",
+                        "invocation_protocol",
                         "lexical_fit",
                     )
                     if key in match
@@ -378,6 +404,8 @@ def _find_text_projection(response: dict[str, Any]) -> dict[str, Any]:
     projection["related_capabilities_truncated"] = response.get(
         "related_capabilities_truncated"
     )
+    if response.get("invocation_protocol"):
+        projection["invocation_protocol"] = response["invocation_protocol"]
     return projection
 
 

tests/boundary/mcp/test_mcp_discovery_projection.py

@@ -68,6 +68,21 @@ async def scenario() -> None:
                 if match["capability_id"] == "matrix.determinant.verify"
             )
             assert "invocation_example" not in verifier
+            assert verifier["invocation_protocol"] == {
+                "tool": "math.run",
+                "arguments_shape": {
+                    "capability_id": "matrix.determinant.verify",
+                    "mode": "VERIFY",
+                    "payload": {
+                        "input": "<exact producer input object>",
+                        "candidate": "<producer response.output.result>",
+                    },
+                },
+                "payload_rule": (
+                    "input and candidate are sibling fields inside payload; neither "
+                    "field belongs beside payload or nested inside the other"
+                ),
+            }
             assert verifier["input_schema_summary"] == {
                 "type": "object",
                 "required": ["input", "candidate"],
@@ -82,6 +97,32 @@ async def scenario() -> None:
                 verifier_text["input_schema_summary"]
                 == verifier["input_schema_summary"]
             )
+            assert (
+                verifier_text["invocation_protocol"]
+                == verifier["invocation_protocol"]
+            )
+
+            verifier_contract_result = await client.call_tool(
+                "math.find",
+                {
+                    "capability_id": "matrix.determinant.verify",
+                    "view": "CONTRACT",
+                },
+            )
+            assert isinstance(verifier_contract_result.structured_content, dict)
+            verifier_contract = verifier_contract_result.structured_content
+            assert (
+                verifier_contract["invocation_protocol"]
+                == verifier["invocation_protocol"]
+            )
+            assert verifier_contract["invocations"] == []
+            verifier_contract_text = json.loads(
+                verifier_contract_result.content[0].text
+            )
+            assert (
+                verifier_contract_text["invocation_protocol"]
+                == verifier["invocation_protocol"]
+            )
 
             compute_contract = await client.call_tool(
                 "math.find",