S1: declaration-owned MCP pipeline result boundary and renderers

docs/refactor/01-INDEX.md

@@ -99,3 +99,7 @@ NRA's complete scan of the package is running and will be added to the evidence
 Written just in time: [`R0-stop-the-inflow.md`](R0-stop-the-inflow.md),
 and the authoritative [L0 pattern-resolver checkpoint](L0-pattern-resolver.rst).
 The latter closes one confirmed dead module, not the whole L0 surface.
+
+Current [S1 MCP result-owner surface receipt](S1-mcp-result-owners.rst) records
+the coherent pipeline checkpoint and named remaining renderer families; it does
+not close the full S1 surface or installed acceptance.

docs/refactor/S1-main-f24-integration-20261001.rst

@@ -0,0 +1,60 @@
+S1 pipeline checkpoint: normal main integration
+==============================================
+
+At owner's explicit request, normally merged main
+f24a828da16691b83284622a01e1765b1d147e66 into own334 ba3b1a4862e81ac44beec6661392d8a685843673.
+Merge commit: 4be5af14a8eaa77af8a8deb469b2a82b1fc9d653, ordinary ort merge,
+no reset/rebase/force push. All eleven S1 production modules have zero diff
+against ba3b1a. The full pipeline source checkpoint and named S1 remainder
+remain exactly as described in S1-mcp-result-owners.rst.
+
+Focused source sanity: 37 passed (pipeline results, transport failures, config
+renderer), 11.78s/258356 KiB RSS, existing interpreter/dependencies, one CPU60s,
+plugin autoload disabled. Actual production ingress/generated rendering and
+both diamond/new-subclass controls stay intact. No existing assertion edited.
+Source extensions rebuilt in this worktree only, no package installation.
+Original bounded R0/R1 evidence remains historical with its actual limits:
+R1 incomplete is not a global proof. No live/native/GUI/scientific launch.
+
+Parent's separate installed candidate e16d74 has the same source relation.
+Parent owns exact live-session close, installed qualification and merge decision;
+the public partial receipt s1-installed-20261001/CHECKPOINT.rst claims no broad
+native/step/job completeness. This source merge/sanity is ready for that review,
+not a claim that installed/native acceptance is complete.
+
+Generated catalog-preparation connection arguments are a separately owned
+follow-up, not an expansion of334: source --help exposes no --port/--host and
+--port5993 is rejected by argument parsing, before any MCP launch. Existing
+capability declares ExecutionConnectionSpec, while generated input projection
+recognizes scalar/AgentCliRequest only. Arendt owns the follow-up defect;
+parent owns live acceptance. No overlapping implementation was made here.
+
+Evidence: receipts/S1-main-f24-integration-20261001.tar.gz includes complete
+source-sanity XML/log, build log and CLI help/rejection. Owned disposable output
+was tracked under /home/ts/.cache/agent-scratch/openhcs-imagexpress-readonly-
+inventory-337-20261001 (compiler-334/build-lib-334/334-merge-fixtures). Source
+and persistent worktrees stay; archived logs survive scratch cleanup.
+
+Current main356 integration (subsequent owner request)
+-----------------------------------------------------
+
+Normally merged currentmain3569105574cb707d75922dc165a313365e1bd33c:
+14e66ad42c6c84142f87cb1c0202d72520187e70, ordinary conflict-free ort merge.
+No production edit or competing runtime patch. All eleven S1 production modules
+remain byte-unchanged from reviewed ba3b1a/bdfd; merged runtime326 belongs to its
+existing owner. Existing source extensions rebuilt locally, 3.49s/143728 KiB,
+no installation or native application launch.
+
+Same focused source sanity: 37 PASS, 6.46s/259196 KiB RSS, CPU0/60s.
+Current build/log/XML added to the existing archive; earlier evidence retained.
+Source integration is ready for parent shipment without optional CI.
+
+Parent reports completed installed e16/f24 native and human/generated/MCP
+qualification, recorded publicly at PR334 comment5925763208. Actual source
+execution returned job1, native COMPLETE with four addressed/calibrated planes
+and exact samples; separate compile-only job2 COMPLETE. Parent reports original
+owned native/MCP processes closed and lock released. That is scoped historical
+acceptance of the eleven unchanged S1 modules, not installed356 qualification,
+biology/UI acceptance or a global R1 pass. Parent will rebuild exact merged main
+for its next phase. Source owner has not opened scientific inputs or launched
+a live endpoint. Issue339 separately owns omitted generated connection arguments.

openhcs/agent/capabilities.py

@@ -11,11 +11,14 @@
 from importlib.metadata import distributions
 from inspect import getdoc
 from math import isfinite
-from typing import ClassVar, Generic, Self, TypeAlias, TypeVar
+from types import UnionType
+from typing import ClassVar, Generic, Self, TypeAlias, TypeVar, get_args, get_type_hints
 
 from metaclass_registry import AutoRegisterMeta
 from zmqruntime.client import EndpointShutdownResult
 
+from openhcs.agent.services.execution_session_service import ExecutionSessionService
+
 from openhcs.agent.dto.architecture import (
     ArchitectureTopic,
     ArchitectureTopicPage,
@@ -611,7 +614,32 @@ def schema_name(self) -> str:
         return self.field_name
 
 
-AgentContract: TypeAlias = type | AgentScalarInputContract
+@dataclass(frozen=True, slots=True)
+class AgentResultFamilyContract:
+    """Actual producer alternatives and their existing external nominal identity.
+
+    MCP's advertised owner is an external metadata fact, not the entire result
+    family. Keep those questions distinct while deriving decode membership from
+    the producer's declared union. No presentation roster owns that membership.
+    """
+
+    advertised_contract: type
+    producer: Callable[..., object]
+
+    @property
+    def result_type(self) -> UnionType:
+        return get_type_hints(self.producer, include_extras=True)["return"]
+
+    @property
+    def result_types(self) -> tuple[type, ...]:
+        return get_args(self.result_type)
+
+    @property
+    def schema_name(self) -> str:
+        return self.advertised_contract.__name__
+
+
+AgentContract: TypeAlias = type | AgentScalarInputContract | AgentResultFamilyContract
 AgentContextT = TypeVar("AgentContextT")
 AgentServiceT = TypeVar("AgentServiceT")
 AgentRequestT = TypeVar("AgentRequestT")
@@ -635,12 +663,14 @@ def _enum_member_title(value: Enum) -> str:
 def _contract_schema_name(contract: AgentContract | None) -> str | None:
     if contract is None:
         return None
-    if isinstance(contract, AgentScalarInputContract):
+    if isinstance(contract, (AgentScalarInputContract, AgentResultFamilyContract)):
         return contract.schema_name
     return contract.__name__
 
 
 def require_agent_type_contract(contract: AgentContract | None) -> type:
+    if isinstance(contract, AgentResultFamilyContract):
+        return contract.advertised_contract
     if not isinstance(contract, type):
         raise TypeError(f"Expected agent type contract, got {contract!r}.")
     return contract
@@ -958,6 +988,17 @@ class AgentCapabilitySpec:
     progress_worker_thread_safe: bool = True
     input_contract: AgentContract | None = None
     output_contract: AgentContract | None = None
+
+    @property
+    def output_contract_types(self) -> tuple[type, ...]:
+        if isinstance(self.output_contract, AgentResultFamilyContract):
+            return self.output_contract.result_types
+        return (
+            ()
+            if self.output_contract is None
+            else (require_agent_type_contract(self.output_contract),)
+        )
+
     exposition: AgentCapabilityExposition | None = None
 
     def __post_init__(self) -> None:
@@ -2709,7 +2750,9 @@ class SubmitCompileCapability(HeadlessExecutionCapability):
     mutating = True
     side_effects = ("submits_zmq_compile_job",)
     input_contract = CompileSubmissionRequest
-    output_contract = ExecutionJobRef
+    output_contract = AgentResultFamilyContract(
+        ExecutionJobRef, ExecutionSessionService._submit_job
+    )
     request_invocation = AgentDataclassRequestServiceInvocation(
         service=lambda context: context.execution_service,
         method=lambda service, request: service.submit_compile(
@@ -2740,7 +2783,9 @@ class SubmitPipelineExecutionCapability(HeadlessExecutionCapability):
     mutating = True
     side_effects = ("submits_zmq_execution_job",)
     input_contract = PipelineExecutionSubmissionRequest
-    output_contract = ExecutionJobRef
+    output_contract = AgentResultFamilyContract(
+        ExecutionJobRef, ExecutionSessionService._submit_job
+    )
     request_invocation = AgentDataclassRequestServiceInvocation(
         service=lambda context: context.execution_service,
         method=lambda service, request: service.submit_execution(
@@ -2833,7 +2878,9 @@ class ObserveOwnedRuntimeCapability(RuntimeServerCliConnectionCapability):
     service = "runtime_server"
     input_contract = RuntimeBootstrapObserveRequest
     output_contract = RuntimeBootstrapState
-    exposition = StartOwnedRuntimeCapability.exposition.refine(workflow_stage=CapabilityWorkflowStage.STATUS)
+    exposition = StartOwnedRuntimeCapability.exposition.refine(
+        workflow_stage=CapabilityWorkflowStage.STATUS
+    )
     request_invocation = AgentDataclassRequestServiceInvocation(
         service=lambda context: context.runtime_server_service,
         method=lambda service, request: service.observe_bootstrap(request),

openhcs/agent/dto/pipeline.py

@@ -5,7 +5,11 @@
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 
-from openhcs.agent.dto.common import AgentResultEnvelope, JsonObject
+from openhcs.agent.dto.common import (
+    AgentResultEnvelope,
+    JsonObject,
+    JsonValue as JsonValue,
+)
 from openhcs.agent.dto.config import ConfigPatch
 from openhcs.agent.dto.functions import FunctionIdentity
 

openhcs/mcp/dev_client.py

@@ -203,11 +203,11 @@ def execute(
                 server_stderr_tail=captured_server_stderr_tail(self._server_stderr),
             )
         payload = require_json_object_payload(to_jsonable(response))
-        returncode = 0 if args.allow_error_payloads else int(_command_failed(payload))
+        returncode = 0 if args.allow_error_payloads else int(response.has_errors())
         return McpDevCommandExecution(
             argv=normalized_argv,
             payload=payload,
-            rendered_output=command_spec.render_response(payload, args),
+            rendered_output=command_spec.render_result(response, args),
             returncode=returncode,
             server_stderr_tail=captured_server_stderr_tail(self._server_stderr),
         )
@@ -482,16 +482,14 @@ def main(argv: Sequence[str] | None = None) -> int:
             return _run_persistent_shell(args)
         command_spec = McpDevCommandSpec.for_name(args.command)
         try:
-            payload = require_json_object_payload(
-                to_jsonable(asyncio.run(_run_async(args)))
-            )
+            response = asyncio.run(_run_async(args))
         except McpDevCliUsageError as exc:
             parser.error(str(exc))
-        if not write_stdout(command_spec.render_response(payload, args)):
+        if not write_stdout(command_spec.render_result(response, args)):
             return 0
         if args.allow_error_payloads:
             return 0
-        return 1 if _command_failed(payload) else 0
+        return int(response.has_errors())
     finally:
         logging.disable(disabled_level)
         registry_logger.setLevel(registry_level)

openhcs/mcp/dev_client_commanding.py

@@ -209,6 +209,21 @@ def render_response(
         del args
         return json.dumps(payload, indent=2, sort_keys=True)
 
+    def render_result(self, response, args: argparse.Namespace) -> str:
+        """Render a framed production result without requiring a JSON round trip."""
+        from openhcs.serialization.json import to_jsonable
+
+        return self.render_response(to_jsonable(response), args)
+
+
+class TypedCompositeCommandSpec(McpDevCommandSpec):
+    """Composite commands consume the same decoded batch as their wire ingress."""
+
+    def render_result(self, response, args: argparse.Namespace) -> str:
+        if args.json:
+            return super().render_result(response, args)
+        return self.render_response(response, args)
+
 
 class CapabilityBackedCommandSpec(McpDevCommandSpec):
     """Command whose primary MCP tool capability is declared on the command."""
@@ -260,7 +275,9 @@ def parser_aliases(self) -> tuple[str, ...]:
     def output_renderer_binding(self):
         output_contract = self.capability.output_contract
         return McpDevOutputRenderer.for_output_contract(
-            output_contract if isinstance(output_contract, type) else None
+            None
+            if output_contract is None
+            else require_agent_type_contract(output_contract)
         )
 
     def configure_reflected_parser(self, parser: argparse.ArgumentParser) -> None:
@@ -278,6 +295,12 @@ def render_call_response(
             self.call_render_args(tool_arguments),
         )
 
+    def render_call_result(
+        self, response, tool_arguments: Mapping[str, JsonValue]
+    ) -> str:
+        """Generic calls use the same nominal command contract as named calls."""
+        return self.render_result(response, self.call_render_args(tool_arguments))
+
     def renderer_options(
         self,
         args: argparse.Namespace,
@@ -301,6 +324,12 @@ def render_response(
             payload, self.renderer_options(args)
         )
 
+    def render_result(self, response, args: argparse.Namespace) -> str:
+        binding = self.output_renderer_binding()
+        if args.json or binding is None:
+            return super().render_result(response, args)
+        return binding.render_result(response, self.renderer_options(args))
+
 
 class ToolsCommandSpec(McpDevCommandSpec):
     command = "tools"