fix(mcp): preserve produced types in text projection and decouple poset resolver from storage

Include produced_artifact_types in both the discovery-match and
descriptor allowlists of _find_text_projection so agents consuming the
agent-facing content projection (rather than structured_content) can
connect poset.finite.materialize to poset.width.compute. Covered by
test_mcp_text_projection_preserves_produced_artifact_types.

Replace MaterializedOperation.input_resolver with artifact_converter,
artifact_payload_model, and artifact_uri_field. The installer now owns
artifact retrieval, schema/semantics validation, and payload
deserialization; the domain module provides only a pure typed conversion
from FinitePosetMaterializationResult to PosetRequest. This removes the
ArtifactRepository import from posets/operations.py, keeping the
mathematical domain module decoupled from storage mechanics as the
operation-library contract requires.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: morluto

src/jacobian/adapters/mcp/tools.py

@@ -90,6 +90,7 @@ def _find_text_projection(response: dict[str, Any]) -> dict[str, Any]:
                         "modes",
                         "accepted_input_kinds",
                         "accepted_artifact_types",
+                        "produced_artifact_types",
                         "input_schema_summary",
                         "output_schema_summary",
                         "scope",
@@ -122,6 +123,7 @@ def _find_text_projection(response: dict[str, Any]) -> dict[str, Any]:
             "modes",
             "accepted_input_kinds",
             "accepted_artifact_types",
+            "produced_artifact_types",
             "input_schema_summary",
             "output_schema_summary",
         )

src/jacobian/domains/posets/operations.py

@@ -32,7 +32,6 @@
 )
 from jacobian.domains._examples import example
 from jacobian.operations import ComputedSuccess, MaterializedOperation
-from jacobian.storage.repository import ArtifactRepository
 
 
 def _presentation_graph(request: FinitePosetRequest) -> nx.DiGraph[str]:
@@ -172,24 +171,20 @@ def _width(request: PosetRequest) -> ComputedSuccess[PosetWidthResult]:
     )
 
 
-def _resolve_poset_request(
+def _convert_materialized_poset(
     request: PosetRequest,
-    store: ArtifactRepository,
-    accepted_artifact_types: tuple[str, ...],
-    semantics_uri: str,
+    payload: FinitePosetMaterializationResult,
 ) -> tuple[PosetRequest, tuple[str, ...]]:
-    if request.poset is not None:
-        return request, ()
-    if request.poset_artifact_uri is None:
-        raise ValueError("poset artifact URI is required")
-    artifact = store.get(request.poset_artifact_uri)
-    if (
-        artifact.manifest.schema_uri not in accepted_artifact_types
-        or artifact.manifest.semantics_uri != semantics_uri
-    ):
-        raise ValueError("artifact is not a compatible finite poset")
-    materialized = FinitePosetMaterializationResult.model_validate(artifact.payload)
-    return PosetRequest(poset=materialized.poset), (request.poset_artifact_uri,)
+    """Convert a materialized poset artifact into a domain request.
+
+    This is a pure typed conversion with no storage dependency; the installer
+    handles artifact retrieval, schema/semantics validation, and payload
+    deserialization before calling this converter.
+    """
+
+    return PosetRequest(poset=payload.poset), (
+        request.poset_artifact_uri,
+    ) if request.poset_artifact_uri else ()
 
 
 def _linear_extensions(
@@ -378,7 +373,9 @@ def _mobius(
         implementation=_width,
         relation_id="poset.width.dilworth.relation",
         accepted_result_capability_ids=("poset.finite.materialize",),
-        input_resolver=_resolve_poset_request,
+        artifact_converter=_convert_materialized_poset,
+        artifact_payload_model=FinitePosetMaterializationResult,
+        artifact_uri_field="poset_artifact_uri",
         tags=(
             "poset",
             "width",

src/jacobian/operation_installation.py

@@ -332,13 +332,24 @@ def invoke(self, request: CapabilityRequest) -> CapabilityResult:
                 request.input
             )
             source_artifact_uris: tuple[str, ...] = ()
-            if self.operation.input_resolver is not None:
-                validated_request, source_artifact_uris = self.operation.input_resolver(
-                    validated_request,
-                    self.resources.artifacts.store,
-                    self._descriptor.accepted_artifact_types,
-                    self.resources.semantics_uri,
-                )
+            if self.operation.artifact_converter is not None:
+                uri_field = self.operation.artifact_uri_field
+                payload_model = self.operation.artifact_payload_model
+                assert uri_field is not None and payload_model is not None
+                artifact_uri = getattr(validated_request, uri_field)
+                if artifact_uri is not None:
+                    artifact = self.resources.artifacts.store.get(artifact_uri)
+                    if (
+                        artifact.manifest.schema_uri
+                        not in self._descriptor.accepted_artifact_types
+                        or artifact.manifest.semantics_uri
+                        != self.resources.semantics_uri
+                    ):
+                        raise ValueError("artifact is not a compatible producer result")
+                    payload = payload_model.model_validate(artifact.payload)
+                    validated_request, source_artifact_uris = (
+                        self.operation.artifact_converter(validated_request, payload)
+                    )
         except (StorageError, ValidationError, ValueError) as exc:
             raise CapabilityInvocationError(
                 self.operation.invalid_request

src/jacobian/operations.py

@@ -10,7 +10,6 @@
     from jacobian.checker_operations import ExactReplayCheckerDeclaration
     from jacobian.installation.context import InstallationContext
     from jacobian.operation_installation import InstalledDomainBundle
-    from jacobian.storage.repository import ArtifactRepository
 
 from jacobian.contracts.capabilities import (
     CapabilityDiagnostic,
@@ -97,23 +96,28 @@ class MaterializedOperation[
     preview: Callable[[ArtifactT], PreviewT] | None = None
     preview_complete: bool = False
     accepted_result_capability_ids: tuple[str, ...] = ()
-    input_resolver: (
-        Callable[
-            [RequestT, ArtifactRepository, tuple[str, ...], str],
-            tuple[RequestT, tuple[str, ...]],
-        ]
-        | None
+    artifact_converter: (
+        Callable[[RequestT, ArtifactT], tuple[RequestT, tuple[str, ...]]] | None
     ) = None
+    artifact_payload_model: type[ContractModel] | None = None
+    artifact_uri_field: str | None = None
     version: str = "2"
 
     def __post_init__(self) -> None:
         if self.preview_complete and self.preview is None:
             raise ValueError("a complete materialized preview requires a preview")
-        if bool(self.accepted_result_capability_ids) != bool(self.input_resolver):
+        has_converter = self.artifact_converter is not None
+        if bool(self.accepted_result_capability_ids) != has_converter:
             raise ValueError(
-                "accepted result capabilities and an input resolver must be "
+                "accepted result capabilities and an artifact converter must be "
                 "declared together"
             )
+        if has_converter and (
+            self.artifact_payload_model is None or self.artifact_uri_field is None
+        ):
+            raise ValueError(
+                "an artifact converter requires a payload model and URI field"
+            )
 
 
 @dataclass(frozen=True, slots=True)

tests/boundary/mcp/test_mcp_catalog_and_policy.py

@@ -176,3 +176,28 @@ async def scenario() -> None:
             assert invalid["error"]["code"] == "INVALID_CURSOR"
 
     asyncio.run(scenario())
+
+
+def test_mcp_text_projection_preserves_produced_artifact_types(
+    tmp_path: Path,
+) -> None:
+    """The agent-facing text projection must include produced_artifact_types."""
+
+    async def scenario() -> None:
+        from mcp import Client
+
+        async with Client(create_server(tmp_path), raise_exceptions=True) as client:
+            listed = await client.call_tool(
+                "math.find",
+                {"query": "poset", "limit": 20},
+            )
+            text = json.loads(listed.content[0].text)
+            assert text["kind"] == "discovery"
+            producing = [m for m in text["matches"] if "produced_artifact_types" in m]
+            assert producing, (
+                "text projection must preserve produced_artifact_types for at "
+                "least one poset capability so agents can connect producers to "
+                "consumers without structured_content"
+            )
+
+    asyncio.run(scenario())