diff --git a/docs/refactor/01-INDEX.md b/docs/refactor/01-INDEX.md index 33caba08b8..a13794b3cc 100644 --- a/docs/refactor/01-INDEX.md +++ b/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. diff --git a/docs/refactor/S1-main-f24-integration-20261001.rst b/docs/refactor/S1-main-f24-integration-20261001.rst new file mode 100644 index 0000000000..257f3a317e --- /dev/null +++ b/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. diff --git a/docs/refactor/S1-mcp-result-owners.rst b/docs/refactor/S1-mcp-result-owners.rst new file mode 100644 index 0000000000..499b651d1c --- /dev/null +++ b/docs/refactor/S1-mcp-result-owners.rst @@ -0,0 +1,343 @@ +S1: MCP development-client result ownership +=========================================== + +Source owner: Codex S1. Integration owner: parent. Audited canonical main: +e690c3bfc0f2042dcc2c75e6205d03fffe8aa603. Branch: +refactor/mcp-result-owners-s1-20261001. Worktree: +/home/ts/wt/openhcs-mcp-result-owners-s1-20261001. + +Current tested source pin: c4ef5f34735e3337a78801aad1dfc70f4dee737b. +Normally integrated canonical main: 2cbfc4a0771bf24d153b076bff4a5e321ac039f4. +Visible draft: https://github.com/OpenHCSDev/openhcs/pull/334. The original +audited head above remains the before-source census, not today's integration base. + +Binding instructions read completely before decisions: S1-DISPATCH-20261001.rst, +GOAL-SCOPE-REFRACTOR-20260930.md, canonical 00-RULES.md and 01-INDEX.md, +OWNER-OVERRIDES.md, both NRA/refactor-audit SKILL.md files, the authoritative +NRA skills/refactor-audit.skill archive SKILL.md, patterns README, complete +boundaries, implementation, membership, identity and over-time patterns, +surface-receipt instructions and NRA batching instructions. + +Before-source measurement and antipattern review +------------------------------------------------ + +Original debt_census.py, existing Python 3.14.7, snapshot at the audited head: +8 renderer modules, 6068 code lines; literal_key_get 166.61/1000 (1011 reads), +isinstance_call 20.60/1000 (125), none_identity 27.03/1000 (164). +pipeline.py has 502 code lines and weighted debt 100. Snapshot, not a zero diff. + +Explicit antipattern review confirmed from current source: + +* BOUND-2: pipeline.py:207 declares ArtifactPlanInspection, while :216-241 + read its fields as raw keys and :247-289 read SourceWorkspaceSummary and + SourceWorkspaceFileRecord as raw records. execution.py DTO declarations own + these fields, including nested materialization and streaming owners. +* BOUND-1/IMPL-12/IMPL-13: draft :27-43 and execution :534-543 repeatedly + project tool envelopes through raw helpers; field validation differs among + nested_mapping, optional_bool/int and sequence_of_mappings. The existing + python_introspect/dataclass_projection.py codec already performs recursive + annotation-owned reconstruction and strict rejection. Reuse it, not copies. +* MEMB-1/2/4: dev_client_rendering.py:129-209 and capability declarations own + renderer/output membership through AutoRegisterMeta and render_bindings. + Extend that owner; do not introduce a decoder catalog or name/type switches. +* MEMB-5/TIME-9: McpDevToolResult and McpDevToolBatchResponse already own framing + at dev_client_core.py:393 and :670. No replacement envelope or DTO facade. + Their single payload sequence must carry decoded values, not a parallel store. +* IDEN-1/3/5: absent tool, missing payload, malformed contract and agent failure + are distinct observations. Keep transport/agent diagnostics and original + malformed receipt, not an empty successful record. No duplicated payload cache. +* IMPL-14: strict field validation belongs in the existing generic codec, not + new anonymous field/key conditions in renderers. Errors identify their path. +* BOUND-4: draft repair guidance currently parses a human missing-kwargs message + (:156-192). Preserve current user guidance; do not claim it is structured proof. + +Producer/consumer trace and required relation +--------------------------------------------- + +Stdio/socket JSON RPC request -> _content_payloads (structuredContent or MCP +text content) -> McpDevToolResult.from_payload -> command run_session -> batch -> +compact command rendering / external JSON serialization. Native source producer +DTOs -> to_jsonable -> MCP content are unchanged. Existing capability output +declarations select the renderer binding; that binding owns payload reconstruction +through the existing codec. Typed renderer members read actual DTO fields. + +ExecutionJobRecord.status at execution_session_service.py:537-558 returns the +existing richer ExecutionJobStatus. _submit_job at :1210-1265 returns +ExecutionJobRef or ExecutionJobStatus (wait and submission failure), despite +capabilities.py:2712/:2743 advertising only ExecutionJobRef. S1 owns a reproducer +and full-receipt decoding for this source presentation mismatch. Current +AgentResultFamilyContract derives its members directly from the existing actual +return owner ExecutionSessionService._submit_job using get_type_hints with +include_extras. Its Ref|Status annotation is the decode family authority; the +extra ExecutionSubmissionResult alias was deleted following parent's review. +No service-file edit, shared-owner collision, duplicate union or presentation +alternative roster. Original advertised ExecutionJobRef metadata remains exact. +Native producer +bytes and advertised metadata remain unchanged. Both real siblings render through +ExecutionJobRenderer's ExecutionJobIdentity-owned common procedure. + +Initial coherent checkpoint +---------------------------- + +Migrate pipeline draft, validation, rendered source, artifact plan and execution +as one command family. Extend existing renderer binding and framing owners; +preserve one JSON serialization path. Delete this family's old raw projections, +per-field checks and nested record readers in place. Real command consumers use +the same decoded values as ingress; serialized command input is decoded at its +boundary, not via an old reader. Diagnostics retain full receipts; compact +display limits do not modify the stored result. Genuine JSON extension metadata +and native execution response bags remain mappings. + +No stored format changes, migrations, numeric processing or scientific artifacts. +MCP text/structured content and existing native DTO serialization remain external +and unchanged. No package, configured-skill or managed-blind instruction changes. + +New-case decisions +------------------ + +Before: a new nested source/plan fact requires a DTO declaration plus another +raw reader and validation at each consumer. After: fields decode from their true +owner; a renderer adds presentation only. A new capability returning a registered +output contract needs no decoder or caller change. Test that extension through +AutoRegisterMeta. Compose MI only for demonstrated overlapping capabilities; +there is no evidence requiring new MI in the initial pipeline family. + +Current source checkpoint +------------------------- + +Parent's three concrete pre-checkpoint ownership risks were reviewed against +source, not treated as a completed parent review. The presentation-only Status +retry binding was deleted; producer union membership is now the capability fact. +Status no longer inherits the Ref renderer. Both share the actual identity owner. + +Fourth parent review risk: the first union alias still paralleled three existing +producer annotations (:1018/:1039/:1219). Resolved by deleting the new alias and +deriving the contract from the existing common _submit_job return declaration, +not editing shared service source. Both submission capabilities point to that +same callable. A source behavioral test checks actual callable identity and +resolved return annotation together with unchanged advertised external metadata. +Source import of this declaration relation passes, 4.32s/242684 KiB. This is +source-only import, no submission/runtime/server invocation. +has_errors/agent_error_codes no longer convert typed results to JSON and rescan +raw records: one declaration-driven diagnostic traversal handles typed AgentError +identity, typed dataclass fields and genuine external JSON extension metadata. +Malformed ingress retains one original receipt with typed boundary diagnoses. + +Deleted mechanisms: pipeline family's literal-key field roster and type checks; +its repeated nested workspace/artifact/materialization readers; draft/session +consumers' raw extraction; duplicate renderer-membership tuple and registration +store; copied raw diagnostic scans; cross-family source truncation procedure. +Shared source clipping now belongs to CodeDocumentRenderOptions. Composite +command rendering belongs to TypedCompositeCommandSpec; per-case presentation +belongs to actual declared output renderers. Existing generic dataclass codec and +existing AutoRegisterMeta registry are the only reconstruction/membership owners. +The registry key inheritance defect is fixed at its existing nominal owner: +subclass registration projects only its declaration, not its parent's assigned +key. MRO lookup reuses ancestor behavior without pretending sibling DTO subtyping. + +New-case evidence: ExtendedInspection adds one DTO declaration and inherits +ArtifactPlanInspection behavior with its extra fact retained. Its single capability +declaration selects it at actual ingress and generated-command rendering. Diamond +composes two independent presentation capabilities over their real value-rendering +ancestor. Both Left/Right and Right/Left orders traverse cooperative super(): each +capability and the ancestor execute once, including through actual transport +framing and generated-command rendering. Ancestor MRO and derived hierarchy views +contain each identity once; existing AutoRegisterMeta contains the declared key. +No consumer, dispatcher or field roster edits. Existing parser implementation +and controls are not replaced; this checkpoint extends only renderer controls. + +Generic call closure: CallCommandSpec now delegates render_call_result to the +nominal capability command. Migrated pipeline compact calls no longer serialize +and re-decode their batch; the actual CLI main test forbids serialization while +rendering a nested artifact plan. The pending selected-workflow leaf retains its +existing separate action presentation; it is not claimed typed or closed here. +Its typed-batch poll presentation is checked equal to the existing source view. +Existing generic plate/widget-tree/field-help/window-snapshot cases also pass. + +Named remaining defect, owner Codex S1: pending selected-workflow compact call +currently passes a nested action_result receipt to a raw UiActionInvokeRenderer +that expects top-level action fields. The new exploratory compact assertion +failed with missing status/selection; preserved in pipeline-final-integrated.log +and XML. This is existing pending UI BOUND-2 debt, not pipeline closure. The +checkpoint's native-receipt test uses the actual declared DTOs and --json, keeps +all run_plate/accepted/scope assertions and adds exact full external receipt +equality. It does not assert UI compact acceptance. Reproducer and acceptance +for the remaining UI family: actual UiSelectedPlateWorkflowResult through call +must expose action identity/status/receipt/targets without raw fallback. No UI +family adapter or schema-shape mirror was introduced to conceal that debt. + +Validation and resource ownership +--------------------------------- + +Run original packaged R0 and unchanged scripts/check_refactor_r1.py, no exceptions +or changed guard inputs. Provider-free family tests must drive production command +rendering and transport-content reconstruction with real nested DTOs, diagnostics, +progress, malformed/missing/not-run and a declaration extension. Preserve original +failed shard results before splitting or repairing incomplete internal fixtures. +Existing interpreters/dependencies only; serial shards <=512 MiB, one CPU, 60s. + +Initial new production-path shard: 14 passed, 5.28s wall, 250284 KiB RSS. +Combined coherent pipeline checkpoint: 25 passed (18 new, seven existing), +11.14s wall, 312636 KiB RSS, one CPU. New coverage drives real run_session, +call_mcp_tool framing and command rendering; text/structured content, nested +diagnostics/hints, source limits, native Ref/Status branches, progress, extension +metadata, malformed nested/missing/undeclared/bool-as-int shapes, absent/not-run, +transport/MCP failures, exact malformed receipt retention and actual CLI main's +compact/JSON entrypoints with a controlled wire response, no runtime launch. +Native server metadata remains exactly openhcs/outputContract=ExecutionJobRef. +Typed error interpretation is tested with serialization forbidden. + +After normal integration of parent main b31807d9de067e18d212bdf9353d05099757fc74: +29 passed (22 new, seven existing), 17.17s wall, 310812 KiB RSS. Single-declaration +extension now also exercises real capability ingress and generated command +rendering. Four nullable-fact cases prove false, zero and empty strings are not +silently omitted as absence. Original R0 found eight repeated foreign optional +probes; replaced all nine nullable composition sites with the typed ancestor's +shared optional_lines policy, not local variable rebinding or guard exceptions. +Leaf owners format present facts; the native None absence fact is unchanged. + +Original R0 at agent-comms 3b03785f45df2ef5dc62ba6aed99294192ecbb01, +head 018caf90c, base b31807d9: openhcs PASS, 26.53s/87592 KiB. No increased +measure. Original failed R0 remains archived. Syntax/correctness Ruff subset +E9/F63/F7/F82 passes; broad inherited lint debt was not mass-edited. +Whole renderer snapshot after pipeline: 5962 code lines, literal_key_get +157.33/1000 (938 reads), isinstance_call 19.96/1000 (119), None 26.00/1000 +(155). Pipeline weighted debt 100 -> 11; 502 -> 400 code lines. This measures +remaining family debt honestly and does not count shuffling as closure. + +Unchanged R1, authoritative NRA source 0844525ecaba93e090a064a4ae4914466b2dae60, +base b31807d9, current source head c47b1f6a9: INCOMPLETE, not PASS. First bounded +attempt could not mmap Git's pack under a 512 MiB address-space limit. Retried +the same roots/revisions/tool with bounded Git mapping windows (8m/128m), not +changed detector inputs: materialized recorded source/dependency context, then +ScanDeadlineExceeded during parse_python_module at 55.000s. Wall 56.20s, +263856 KiB RSS. No comparison or descent/proof result returned. Roots remain +openhcs, scripts, benchmark plus all recorded external submodule Python context; +report scope remains changed surviving production files. No exception/waiver, +no substituted scoped result, no full-package/global NRA proof claim. Continuing +source progress does not repair or claim completion of this independent tool. +Owned persistent source snapshot scratch: worktree/.s1-r1-scratch; original tool +TemporaryDirectory cleaned its snapshots on the caught deadline exit. + +R0 scripts and benchmark also PASS unchanged (empty changed Python scope). +Shared-owner regression shard: 29 existing knowledge/architecture/function/ +authoring/tool-list/code-document command cases PASS, 28.04s/284140 KiB, +provider-free, no runtime. These are behavior checks, not full remaining-family +descent closure. Evidence archive: receipts/S1-pipeline-20261001.tar.gz contains +original failures, subsequent passing logs/XML, full guard outputs and census. +It excludes compiler/build products. Regenerate it after further owned checks; +owned terminal scratch is disposable once the archive is verified. + +Final integrated checkpoint evidence +------------------------------------ + +Production source at d374482c0700a15e6677281fa0c3485a1405f43a is identical to +the current tested source pin; c4ef5f347 adds only stronger diamond behavior +controls. Relative to integrated main, production deletes 525 lines first and +adds 768; the pipeline renderer deletes 417 and adds 331. This is factoring and +test coverage, not a performance claim or a global debt certificate. + +Final shard: 39 PASS, 286 deselected, 14.66s pytest/15.76s wall, 317360 KiB RSS. +27 owned family/extension/boundary/CLI/client cases plus 12 existing pipeline, +generic-call and selected-workflow poll cases. Continuous McpDevClient.execute +journeys replace only the wire peer and retain real run_session, framing, decode, +rendering and exit-status behavior, with nested contracts, typed diagnostics, +malformed input and full native JSON receipts. Separate actual main rendering +checks return a controlled typed batch from _run_async to avoid runtime startup; +they do not prove process/transport startup. Renderers, declaration selection and +the existing codec remain production code, not substitutes. +Both diamond orders run through unchanged generated-command consumers. Earlier +38-case results and the pending-UI compact failure remain archived, not erased. + +Recipe, from this worktree, using the existing interpreter:: + + env PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 PYTHONDONTWRITEBYTECODE=1 \ + OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 MKL_NUM_THREADS=1 NUMEXPR_NUM_THREADS=1 \ + /usr/bin/time -v taskset -c 0 timeout 60 \ + /home/ts/code/projects/openhcs/.venv/bin/python -c \ + 'import openhcs, pytest; raise SystemExit(pytest.main(["--noconftest", "-o", "addopts=", "tests/unit/agent/test_mcp_dev_client_pipeline_results.py", "tests/unit/agent/test_mcp_server.py", "-k", "pipeline_results or mcp_dev_client_draft_pipeline_step or mcp_dev_client_artifact_plan or artifact_plan_exposes_source_workspace_count or mcp_dev_client_execute_source_composes or mcp_dev_client_call_renders or mcp_dev_client_selected_workflow_poll_renders_compact_summary"]))' + +Pre-importing OpenHCS selects this checkout's recorded dependency objects before +pytest imports external modules. Unknown asyncio config warnings reflect disabled +pytest plugin autoload, not a failed assertion. No packages were installed. + +Unchanged original R0, base 2cbfc4a0771bf24d153b076bff4a5e321ac039f4, production +head d374482c0700a15e6677281fa0c3485a1405f43a: PASS, 23.84s/87664 KiB, zero +exceptions and no increased measures. Archived r0-pinned-source.log. Scripts and +benchmark have no S1 diff against this base; their earlier empty-scope passes are +retained. Correctness Ruff E9/F63/F7/F82 and git diff --check PASS on current source. + +Unchanged full-context R1 at these same revisions: INCOMPLETE, exit 1, +ScanDeadlineExceeded during parse_python_module at 55.001s/55.000s; wall 56.55s, +263808 KiB RSS. One CPU and 512 MiB address-space ceiling, original production +roots and all recorded dependency Python context, bounded Git mapping windows +8m/128m. No root, detector, descent guard, assertion or comparison input changed. +Materialized the committed source context; parsing did not complete, so there is +no before/after certificate or global proof. Archived r1-pinned-full-context.log. +The owned R1 snapshot directory is empty following the tool's own cleanup. + +Actual source-only command ``python -m openhcs.mcp.dev_client artifact-plan --help`` +PASS, exit 0, 4.79s/248692 KiB, one CPU/60s; archived source-command-help.log. +This checks the generated parser/entrypoint, not a live MCP/server/installed run. + +Authoritative archive SHA256 reverified at the checkpoint: +100fbe8ef89664b866777e87b2c8640a3432e8a10e9188dff81c97942d551bf6. +Actual antipattern review above influenced implementation: shared decode/optional +composition on McpDevTypedOutputRenderer, source clipping on render options, +common execution presentation on the real ExecutionJobIdentity owner, typed +diagnostic descent without JSON rescans, and producer-annotation-derived family +membership. The removed alias and replaced readers/stores are absent from the +combined source diff; no shared execution service file was changed. + +Original failures remain in the owned output archive: source-first import order +collection failure; JsonValue annotation namespace failure; scoped registry/MRO +failures; five existing fixture failures. Existing fixture updates supply missing +real server/schema/config/ref/source-identity/output-dir fields, remove nonexistent +errors on PipelineSpec/RenderedSource and replace nonexistent base_path with +shared_output_stem. No behavior assertion was removed or weakened. Two actual +presentation regressions (Validate errors label and None group text) were fixed. + +Source imports needed the checkout's declared C++ extensions. Built only this +worktree's extensions with existing setup.py, compiler and dependency objects, +one build worker: 3.29s/143740 KiB. No interpreter/package/install changes. +Recorded submodules initialized only here at existing gitlinks; no gitlink edits. +Own compiler/build-lib scratch and two local abi3 binaries are disposable and +must be cleaned after evidence is archived. No application native launch. + +Resource helper before work reports warning/exit 2: initially swap used 14.2 GiB, +available RAM 20.0 GiB, /home free 23.1 GiB; latest swap 13.3 GiB, available RAM +18.5 GiB, /home free 20.6 GiB. No extra agents or large tests authorized while +this persists; use only bounded focused checks. Owned disposable output: +/home/ts/.cache/agent-scratch/openhcs-mcp-result-owners-s1-20261001, purpose source +census/guard/test logs, compiler output and archive-check fresh extraction, +owner Codex S1. Archive useful receipts in this worktree, +then remove that exact owned run directory after termination. No volatile source. + +Cleanup at checkpoint: evidence archive refreshed, freshly extracted to the owned +archive-check child and compared byte-for-byte with every original log/XML/census. +Moved the exact owned output/build/verification directory (14 MiB including +the verification extraction) and only these reproducible local source-build +binaries to recoverable desktop trash; removed the empty .s1-r1-scratch directory: +openhcs/core/_tabular_native.abi3.so and +openhcs/processing/backends/cellprofiler/_granularity_native.abi3.so. Persistent +source/guard worktrees remain. Rebuild these checkout-only extensions with the +existing setup.py/interpreter/compiler before reproducing imports; do not install +or change active packages. Original receipts remain recoverable in the archive. + +Source-context scan must include OpenHCS production and recorded dependency +sources where safe. Any budget/tool limitation is incomplete coverage, not global +proof or a blocker to independent implementation. No held-out/reference/science +data is read. No native/MCP/JVM/viewer launches. + +Remaining S1 and installed acceptance +------------------------------------- + +Codex S1 retains plate, viewer/runtime, UI/code-document/state-surface, +object-state, knowledge/function/authoring and config families after the initial +pipeline checkpoint. XML329, runtime/BaSiCPy217/326 and canonical skill references +remain with their existing owners. Current PR331 core admission is also disjoint. +No parent/foreign/main worktree edits. Parent merges useful locally verified draft +checkpoints with CI deferred. Parent owns affected installed CLI/MCP user journeys +after the scientific lock release. A source import/help/render check is not +installed or live acceptance. Neither this checkpoint nor a partial NRA scan +closes S1, the full ZIP or the parent's scientific goal. diff --git a/docs/refactor/receipts/S1-main-f24-integration-20261001.tar.gz b/docs/refactor/receipts/S1-main-f24-integration-20261001.tar.gz new file mode 100644 index 0000000000..bb9af005a6 Binary files /dev/null and b/docs/refactor/receipts/S1-main-f24-integration-20261001.tar.gz differ diff --git a/docs/refactor/receipts/S1-pipeline-20261001.tar.gz b/docs/refactor/receipts/S1-pipeline-20261001.tar.gz new file mode 100644 index 0000000000..1dc1e8e46a Binary files /dev/null and b/docs/refactor/receipts/S1-pipeline-20261001.tar.gz differ diff --git a/openhcs/agent/capabilities.py b/openhcs/agent/capabilities.py index 8ef1b2c7c4..961943fae1 100644 --- a/openhcs/agent/capabilities.py +++ b/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), diff --git a/openhcs/agent/dto/pipeline.py b/openhcs/agent/dto/pipeline.py index 78f3a4811e..c9e989db03 100644 --- a/openhcs/agent/dto/pipeline.py +++ b/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 diff --git a/openhcs/mcp/dev_client.py b/openhcs/mcp/dev_client.py index e041a7cd15..61d146b27d 100644 --- a/openhcs/mcp/dev_client.py +++ b/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) diff --git a/openhcs/mcp/dev_client_commanding.py b/openhcs/mcp/dev_client_commanding.py index 9f0a9cca9d..96716b011c 100644 --- a/openhcs/mcp/dev_client_commanding.py +++ b/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" diff --git a/openhcs/mcp/dev_client_commands/knowledge_pipeline.py b/openhcs/mcp/dev_client_commands/knowledge_pipeline.py index 4c9a92af61..2e07a6e2dc 100644 --- a/openhcs/mcp/dev_client_commands/knowledge_pipeline.py +++ b/openhcs/mcp/dev_client_commands/knowledge_pipeline.py @@ -24,7 +24,10 @@ PipelineSourceRenderRequest, PipelineValidationRequest, ) -from openhcs.mcp.dev_client_commanding import McpDevCommandSpec, SingleToolCommandSpec +from openhcs.mcp.dev_client_commanding import ( + SingleToolCommandSpec, + TypedCompositeCommandSpec, +) from openhcs.mcp.dev_client_core import ( DEFAULT_REGISTRY_DISCOVERY_TIMEOUT_SECONDS, McpDevCliUsageError, @@ -38,9 +41,7 @@ execute_source_session_tool_arguments, execute_source_submit_timeout_seconds, execute_source_submit_tool_arguments, - first_mapping_payload, optional_int, - optional_str, parse_optional_json_object, parse_required_axis_labels, pipeline_source_from_args, @@ -216,13 +217,17 @@ def configure_parser(self, parser: argparse.ArgumentParser) -> None: ) for connection_field in fields(ExecutionConnectionSpec): add_request_factory_option( - parser, CustomFunctionRegistrationRequest.from_fields, - connection_field.name, f"--{connection_field.name.replace('_', '-')}", + parser, + CustomFunctionRegistrationRequest.from_fields, + connection_field.name, + f"--{connection_field.name.replace('_', '-')}", ) for field_name in ("function_name", "storage_dir"): add_request_factory_option( - parser, CustomFunctionRegistrationRequest.from_fields, - field_name, f"--{field_name.replace('_', '-')}", + parser, + CustomFunctionRegistrationRequest.from_fields, + field_name, + f"--{field_name.replace('_', '-')}", ) parser.add_argument( "--json", @@ -237,10 +242,12 @@ def tool_arguments( source_code = args.source_code if source_code is None: source_code = Path(args.source_file).read_text(encoding="utf-8") - connection = ExecutionConnectionSpec(**{ - connection_field.name: vars(args)[connection_field.name] - for connection_field in fields(ExecutionConnectionSpec) - }) + connection = ExecutionConnectionSpec( + **{ + connection_field.name: vars(args)[connection_field.name] + for connection_field in fields(ExecutionConnectionSpec) + } + ) connection.require_port("Custom function registration") return McpToolArgumentAuthority.from_payload( { @@ -321,7 +328,7 @@ def call_render_args( ) -class DraftPipelineStepCommandSpec(McpDevCommandSpec): +class DraftPipelineStepCommandSpec(TypedCompositeCommandSpec): command = "draft-pipeline-step" help = ( "Create, add one FunctionStep, validate, and render a draft in one MCP session." @@ -380,11 +387,9 @@ async def run_session( timeout_seconds, ) results = [create_result] - create_payload = first_mapping_payload(create_result) - pipeline_id = ( - None if create_payload is None else create_payload.get("pipeline_id") - ) - if isinstance(pipeline_id, str): + create_payload = create_result.first_decoded_payload() + pipeline_id = None if create_payload is None else create_payload.pipeline_id + if pipeline_id is not None: add_arguments: dict[str, JsonValue] = { "pipeline_id": pipeline_id, "function_id": args.function_id, @@ -499,7 +504,7 @@ def tool_arguments( ) -class ExecuteSourceCommandSpec(McpDevCommandSpec): +class ExecuteSourceCommandSpec(TypedCompositeCommandSpec): command = "execute-source" help = "Create and submit a source-backed headless execution session." default_timeout_seconds: ClassVar[float] = 120.0 @@ -574,12 +579,8 @@ async def run_session( timeout_seconds, ) results = [create_result] - create_payload = first_mapping_payload(create_result) - session_id = ( - optional_str(create_payload.get("session_id")) - if create_payload is not None - else None - ) + create_payload = create_result.first_decoded_payload() + session_id = create_payload.session_id if create_payload is not None else None if session_id is not None: results.append( await call_mcp_tool( diff --git a/openhcs/mcp/dev_client_commands/ui.py b/openhcs/mcp/dev_client_commands/ui.py index 081d57242d..d5cb811cee 100644 --- a/openhcs/mcp/dev_client_commands/ui.py +++ b/openhcs/mcp/dev_client_commands/ui.py @@ -172,6 +172,12 @@ def render_response( args.arguments, ) + def render_result(self, response, args: argparse.Namespace) -> str: + command_spec = CapabilityBackedCommandSpec.for_capability_name(args.tool_name) + if args.json or command_spec is None: + return super().render_result(response, args) + return command_spec.render_call_result(response, args.arguments) + class SelectedWorkflowCommandSpec(CapabilityBackedCommandSpec): capability = agent_capabilities.ui_selected_plate_workflow @@ -479,6 +485,13 @@ def render_call_response( action_id=workflow, ) + def render_call_result(self, response, tool_arguments: Mapping[str, JsonValue]) -> str: + # This distinct workflow presentation is still in the named pending + # UI/state family. Preserve its existing action view at that boundary. + from openhcs.serialization.json import to_jsonable + + return self.render_call_response(to_jsonable(response), tool_arguments) + @staticmethod def _poll_summary_payload(payload: JsonObject) -> Mapping[str, JsonValue] | None: for result in SelectedWorkflowCommandSpec._result_mappings(payload): diff --git a/openhcs/mcp/dev_client_core.py b/openhcs/mcp/dev_client_core.py index 633db24d26..992efe91c5 100644 --- a/openhcs/mcp/dev_client_core.py +++ b/openhcs/mcp/dev_client_core.py @@ -13,8 +13,9 @@ from abc import ABC, abstractmethod from collections.abc import Callable, Mapping, Sequence from contextlib import AsyncExitStack, asynccontextmanager -from dataclasses import dataclass, field +from dataclasses import dataclass, field, fields, is_dataclass, replace from enum import Enum +from functools import singledispatch from pathlib import Path from typing import ClassVar, Self, TextIO, TypeVar, cast, get_type_hints @@ -389,13 +390,83 @@ def from_exception( ) +@dataclass(frozen=True, slots=True) +class McpDevPayloadFailure: + """Rejected response contract, retaining the complete external receipt.""" + + receipt: JsonValue + errors: tuple[AgentError, ...] + + +@to_jsonable.register(McpDevPayloadFailure) +def _jsonable_payload_failure(value: McpDevPayloadFailure) -> JsonValue: + return value.receipt + + @dataclass(frozen=True, slots=True) class McpDevToolResult: """JSON-facing result for one MCP tool call.""" tool: str mcp_error: bool - payloads: tuple[JsonValue, ...] + payloads: tuple[object, ...] + + def decoded_for_rendering(self) -> "McpDevToolResult": + """Descend through the capability's existing renderer/output binding.""" + from openhcs.mcp.dev_client_rendering import McpDevOutputRenderer + + try: + capability = get_agent_capability(self.tool) + except KeyError: + return self + bindings = tuple( + McpDevOutputRenderer.for_output_contract(contract) + for contract in capability.output_contract_types + ) + if not bindings or any(binding is None for binding in bindings): + return self + return replace( + self, + payloads=tuple( + self._decode_payload(payload, bindings) for payload in self.payloads + ), + ) + + @staticmethod + def _decode_payload(payload, bindings): + if isinstance(payload, McpDevPayloadFailure): + return payload + rejections: list[AgentError] = [] + for binding in bindings: + try: + return binding.decode_payload(payload) + except (TypeError, ValueError) as error: + rejections.append( + AgentError.from_exception("mcp_payload_invalid", error) + ) + return McpDevPayloadFailure(payload, (*_agent_errors(payload), *rejections)) + + def first_decoded_payload(self): + """A missing or rejected payload is not a successful empty record.""" + if not self.payloads: + return None + payload = self.payloads[0] + return None if isinstance(payload, McpDevPayloadFailure) else payload + + def diagnostic_errors(self) -> tuple[AgentError, ...]: + errors = tuple( + error for payload in self.payloads for error in _agent_errors(payload) + ) + if self.mcp_error and not errors: + return (AgentError(code="mcp_tool_error", message=f"{self.tool} failed."),) + if not self.payloads and not errors: + return ( + AgentError( + code="mcp_payload_missing", + message=f"{self.tool} returned no payload.", + ), + ) + return errors @classmethod def from_payload( @@ -407,20 +478,16 @@ def from_payload( tool=tool_name, mcp_error=result.get("isError") is True, payloads=_content_payloads(result), - ) + ).decoded_for_rendering() def has_errors(self) -> bool: """Return whether the tool or any structured agent payload failed.""" - return self.mcp_error or any( - _contains_agent_error(payload) for payload in self.payloads - ) + return bool(self.diagnostic_errors()) def agent_error_codes(self) -> tuple[str, ...]: """Project structured agent error codes without interpreting their domain.""" - return tuple( - code for payload in self.payloads for code in _agent_error_codes(payload) - ) + return tuple(error.code for error in self.diagnostic_errors()) def has_only_agent_error_code(self, code: str) -> bool: """Return whether every structured failure carries one declared code.""" @@ -645,6 +712,9 @@ class McpDevResponse(ABC): server: McpDevServerIdentity errors: tuple[McpDevTransportFailure, ...] = () + def has_errors(self) -> bool: + return bool(self.errors) + @classmethod def from_transport_failure( cls, @@ -672,6 +742,39 @@ class McpDevToolBatchResponse(McpDevResponse): results: tuple[McpDevToolResult, ...] = () + def has_errors(self) -> bool: + return super(McpDevToolBatchResponse, self).has_errors() or any( + result.has_errors() for result in self.results + ) + + @classmethod + def for_rendering( + cls, + response: "McpDevToolBatchResponse | JsonObject", + ) -> "McpDevToolBatchResponse": + """Decode the external framing and selected DTOs once at view ingress.""" + framed = ( + response + if isinstance(response, cls) + else dataclass_from_mapping(cls, response) + ) + return replace( + framed, + results=tuple(result.decoded_for_rendering() for result in framed.results), + ) + + def payload_for(self, capability: AgentCapabilitySpec): + result = next( + (result for result in self.results if result.tool == capability.name), None + ) + return None if result is None else result.first_decoded_payload() + + def diagnostic_errors(self) -> tuple[AgentError | McpDevTransportFailure, ...]: + return ( + *self.errors, + *(error for result in self.results for error in result.diagnostic_errors()), + ) + @classmethod def from_results( cls, @@ -1039,37 +1142,61 @@ def require_json_object_payload(value: JsonValue) -> JsonObject: return cast(JsonObject, value) -def _contains_agent_error(value: JsonValue) -> bool: - if isinstance(value, Mapping): - errors = AgentResultEnvelope.error_items_from_serialized_mapping(value) - if errors: - return True - return any(_contains_agent_error(child) for child in value.values()) - if isinstance(value, list): - return any(_contains_agent_error(child) for child in value) - return False +@singledispatch +def _agent_errors(value: object) -> tuple[AgentError, ...]: + """One diagnostic descent mechanism over actual declared and JSON values.""" + if is_dataclass(value) and not isinstance(value, type): + return tuple( + error + for declaration in fields(value) + for error in _agent_errors(getattr(value, declaration.name)) + ) + return () -def _agent_error_codes(value: JsonValue) -> tuple[str, ...]: - """Recursively project declared error codes from one JSON-facing payload.""" - - if isinstance(value, Mapping): - projected: list[str] = [] - errors = AgentResultEnvelope.error_items_from_serialized_mapping(value) - if errors is not None: - for error in errors: - if not isinstance(error, Mapping): - continue - code = AgentError.code_from_serialized_mapping(error) - if code is not None: - projected.append(code) - for child in value.values(): - if child is not errors: - projected.extend(_agent_error_codes(child)) - return tuple(projected) - if isinstance(value, list): - return tuple(code for child in value for code in _agent_error_codes(child)) - return () +@_agent_errors.register(AgentError) +def _declared_agent_error(value: AgentError) -> tuple[AgentError, ...]: + return (value,) + + +@_agent_errors.register(McpDevPayloadFailure) +def _failed_payload_errors(value: McpDevPayloadFailure) -> tuple[AgentError, ...]: + return value.errors + + +@_agent_errors.register(Mapping) +def _boundary_agent_errors(value: Mapping) -> tuple[AgentError, ...]: + """JSON extension bags and unmigrated wire payloads remain real mappings.""" + errors = AgentResultEnvelope.error_items_from_serialized_mapping(value) + projected: list[AgentError] = [] + for error in errors or (): + try: + projected.append(dataclass_from_mapping(AgentError, error)) + except (TypeError, ValueError) as exception: + projected.append( + AgentError.from_exception("mcp_diagnostic_invalid", exception) + ) + projected.extend( + error + for child in value.values() + if child is not errors + for error in _agent_errors(child) + ) + return tuple(projected) + + +@_agent_errors.register(list) +@_agent_errors.register(tuple) +def _sequence_agent_errors(value) -> tuple[AgentError, ...]: + return tuple(error for child in value for error in _agent_errors(child)) + + +def _contains_agent_error(value: object) -> bool: + return bool(_agent_errors(value)) + + +def _agent_error_codes(value: object) -> tuple[str, ...]: + return tuple(error.code for error in _agent_errors(value)) def _command_failed(payload: JsonObject) -> bool: diff --git a/openhcs/mcp/dev_client_renderers/knowledge.py b/openhcs/mcp/dev_client_renderers/knowledge.py index 1d5cbb9947..aa47ecbefe 100644 --- a/openhcs/mcp/dev_client_renderers/knowledge.py +++ b/openhcs/mcp/dev_client_renderers/knowledge.py @@ -26,11 +26,11 @@ from openhcs.mcp.dev_client_rendering import ( AuthoringContextRenderOptions, CatalogRenderOptions, + CodeDocumentRenderOptions, McpDevOutputRenderer, McpDevPayloadProjection, ) from openhcs.mcp.dev_client_renderers.object_state import ObjectStateScopeRenderer -from openhcs.mcp.dev_client_renderers.ui_bridge import CodeDocumentRenderer from openhcs.mcp.dev_client_renderers.viewer import ( RuntimeServerRenderer, ViewerValidationRenderer, @@ -286,7 +286,7 @@ def render( bounded_limit = max(limit, 0) visible_topics = topics[:bounded_limit] lines = [ - "Architecture topics: " f"matched={len(topics)} shown={len(visible_topics)}" + f"Architecture topics: matched={len(topics)} shown={len(visible_topics)}" ] if contains: lines.append(f"Filter: contains={contains}") @@ -660,6 +660,8 @@ def render(cls, response: JsonObject, *, max_chars: int = 2_000) -> str: if isinstance(content, str): lines.append("Content:") lines.append( - CodeDocumentRenderer._source_text(content, max_source_chars=max_chars) + CodeDocumentRenderOptions(max_source_chars=max_chars).source_text( + content + ) ) return "\n".join(lines) diff --git a/openhcs/mcp/dev_client_renderers/pipeline.py b/openhcs/mcp/dev_client_renderers/pipeline.py index 8c0933352e..c31fc530d1 100644 --- a/openhcs/mcp/dev_client_renderers/pipeline.py +++ b/openhcs/mcp/dev_client_renderers/pipeline.py @@ -1,174 +1,199 @@ -"""Pipeline authoring and execution renderers for the MCP dev client.""" +"""Typed pipeline authoring, planning and execution presentation owners.""" from __future__ import annotations -import json -from collections.abc import Mapping from inspect import getdoc from openhcs.agent.capabilities import agent_capabilities -from openhcs.agent.dto.common import JsonObject, JsonValue -from openhcs.agent.dto.execution import ArtifactPlanInspection, SourceWorkspaceSummary +from openhcs.agent.dto.common import AgentError, AgentWarning, RenderedSource +from openhcs.agent.dto.execution import ( + ArtifactInputPlanSummary, + ArtifactMaterializationPlanSummary, + ArtifactPlanInspection, + CompiledStepPlanSummary, + ExecutionJobRef, + ExecutionJobIdentity, + ExecutionJobStatus, + OrchestratorSessionRef, + SourceWorkspaceSummary, +) +from openhcs.agent.dto.pipeline import ( + FunctionStepSpec, + PipelineRef, + PipelineSpec, + PipelineValidationResult, +) +from openhcs.mcp.dev_client_core import McpDevToolBatchResponse from openhcs.mcp.dev_client_rendering import ( - McpDevOutputRenderer, + CodeDocumentRenderOptions, + McpDevOutputRenderOptions, McpDevPayloadProjection, + McpDevTypedOutputRenderer, + McpDiagnosticRenderer, ) -from openhcs.mcp.dev_client_core import optional_bool, optional_int -from openhcs.mcp.dev_client_renderers.object_state import ObjectStateScopeRenderer -from openhcs.mcp.dev_client_renderers.ui_bridge import CodeDocumentRenderer -from openhcs.mcp.dev_client_renderers.viewer import ViewerValidationRenderer -class PipelineDraftStepRenderer: - """Compact renderer for one-step pipeline draft smoke checks.""" +def _sequence_text(values) -> str: + return ",".join(str(value) for value in values) + + +def _mapping_text(values) -> str: + return ", ".join( + f"{key}={McpDevPayloadProjection.text(value)}" for key, value in values.items() + ) + + +def _append_messages( + lines: list[str], + errors: tuple[AgentError, ...], + warnings: tuple[AgentWarning, ...], + label: str = "", +) -> None: + prefix = f"{label} " if label else "" + if errors: + lines.append(f"{prefix}errors:") + lines.extend(McpDiagnosticRenderer.typed_error_lines(errors)) + if warnings: + lines.append(f"{prefix}warnings:") + lines.extend(McpDiagnosticRenderer.typed_error_lines(warnings)) + + +class PipelineReferenceRenderer(McpDevTypedOutputRenderer): + output_contract = PipelineRef @classmethod - def render(cls, response: JsonObject, *, max_source_chars: int = 2_000) -> str: - create_payload = cls._payload_for_tool( - response, agent_capabilities.create_pipeline.name - ) - add_payload = cls._payload_for_tool( - response, agent_capabilities.add_function_step.name - ) - validate_payload = cls._payload_for_tool( - response, agent_capabilities.validate_pipeline.name - ) - render_payload = cls._payload_for_tool( - response, agent_capabilities.render_pipeline_source.name - ) - pipeline_id = ( - create_payload.get("pipeline_id") if create_payload is not None else None - ) - pipeline_steps = ( - McpDevPayloadProjection.sequence_of_mappings(add_payload.get("steps")) - if add_payload is not None - else () - ) - lines = [ + def render_payload( + cls, payload: PipelineRef, options: McpDevOutputRenderOptions + ) -> str: + return f"Pipeline: id={payload.pipeline_id} uri={payload.uri}" + + +class PipelineSpecRenderer(McpDevTypedOutputRenderer): + output_contract = PipelineSpec + + @classmethod + def render_payload( + cls, payload: PipelineSpec, options: McpDevOutputRenderOptions + ) -> str: + return "\n".join( ( - "Pipeline draft: " - f"id={McpDevPayloadProjection.text(pipeline_id)} " - f"valid={cls._valid_text(validate_payload)} " - f"steps={len(pipeline_steps)}" + f"Pipeline: id={payload.pipeline_id} steps={len(payload.steps)}", + *cls.step_lines(payload.steps), ) + ) + + @staticmethod + def step_lines(steps: tuple[FunctionStepSpec, ...]) -> list[str]: + return [ + f"- {step.step_id}: name={McpDevPayloadProjection.quoted_text(step.name)} " + f"enabled={step.enabled} functions={','.join(function.function_id for function in step.functions) or ''}" + for step in steps ] - if create_payload is not None: - lines.append( - f"Ref: uri={McpDevPayloadProjection.text(create_payload.get('uri'))}" - ) - cls._append_payload_messages(lines, create_payload, "Create") - cls._append_payload_messages(lines, add_payload, "Add step") - cls._append_payload_messages(lines, validate_payload, "Validate") - cls._append_repair_hints(lines, validate_payload, pipeline_steps) - if pipeline_steps: - lines.append("Steps:") - lines.extend(cls._step_lines(pipeline_steps)) - if render_payload is not None: - source = render_payload.get("source") - if isinstance(source, str): - lines.append( - "Source: " - f"title={McpDevPayloadProjection.quoted_text(render_payload.get('title'))} " - f"bytes={len(source)}" - ) - lines.append( - CodeDocumentRenderer._source_text( - source, - max_source_chars=max_source_chars, - ) - ) - cls._append_payload_messages(lines, render_payload, "Render") + + +class PipelineValidationRenderer(McpDevTypedOutputRenderer): + output_contract = PipelineValidationResult + + @classmethod + def render_payload( + cls, payload: PipelineValidationResult, options: McpDevOutputRenderOptions + ) -> str: + lines = [ + f"Pipeline validation: id={payload.pipeline_ref.pipeline_id} valid={payload.valid}" + ] + _append_messages(lines, (), payload.warnings) return "\n".join(lines) - @staticmethod - def _payload_for_tool( - response: JsonObject, - tool_name: str, - ) -> Mapping[str, JsonValue] | None: - return McpDevPayloadProjection.tool_payload(response, tool_name) - @staticmethod - def _valid_text(payload: Mapping[str, JsonValue] | None) -> str: - if payload is None: - return "" - return McpDevPayloadProjection.text(payload.get("valid")) +class PipelineSourceRenderer(McpDevTypedOutputRenderer): + output_contract = RenderedSource + render_options_type = CodeDocumentRenderOptions @classmethod - def _step_lines(cls, steps: tuple[Mapping[str, JsonValue], ...]) -> list[str]: - lines: list[str] = [] - for step in steps: - functions = McpDevPayloadProjection.sequence_of_mappings( - step.get("functions") - ) - function_ids = tuple( - McpDevPayloadProjection.text(function.get("function_id")) - for function in functions - ) + def render_payload( + cls, payload: RenderedSource, options: CodeDocumentRenderOptions + ) -> str: + lines = [ + f"Source: title={McpDevPayloadProjection.quoted_text(payload.title)} bytes={len(payload.source)}" + ] + if options.include_source: + lines.append(options.source_text(payload.source)) + return "\n".join(lines) + + +class PipelineDraftStepRenderer: + """Composite presentation of the same decoded production authoring results.""" + + @classmethod + def render(cls, response, *, max_source_chars: int = 2_000) -> str: + decoded = McpDevToolBatchResponse.for_rendering(response) + created: PipelineRef | None = decoded.payload_for( + agent_capabilities.create_pipeline + ) + added: PipelineSpec | None = decoded.payload_for( + agent_capabilities.add_function_step + ) + validated: PipelineValidationResult | None = decoded.payload_for( + agent_capabilities.validate_pipeline + ) + source: RenderedSource | None = decoded.payload_for( + agent_capabilities.render_pipeline_source + ) + steps = () if added is None else added.steps + lines = [ + f"Pipeline draft: id={McpDevPayloadProjection.text(None if created is None else created.pipeline_id)} " + f"valid={'' if validated is None else validated.valid} steps={len(steps)}" + ] + if created is not None: + lines.append(f"Ref: uri={created.uri}") + if validated is not None and validated.errors: + lines.append("Validate errors:") + lines.extend( + McpDiagnosticRenderer.typed_error_lines(decoded.diagnostic_errors()) + ) + if validated is not None: + _append_messages(lines, (), validated.warnings, "Validate") + cls._append_repair_hints(lines, validated.errors, steps) + if steps: + lines.append("Steps:") + lines.extend(PipelineSpecRenderer.step_lines(steps)) + if source is not None: lines.append( - "- " - f"{McpDevPayloadProjection.text(step.get('step_id'))}: " - f"name={McpDevPayloadProjection.quoted_text(step.get('name'))} " - f"enabled={McpDevPayloadProjection.text(step.get('enabled'))} " - f"functions={','.join(function_ids) if function_ids else ''}" + PipelineSourceRenderer.render_payload( + source, CodeDocumentRenderOptions(max_source_chars=max_source_chars) + ) ) - return lines - - @staticmethod - def _append_payload_messages( - lines: list[str], - payload: Mapping[str, JsonValue] | None, - label: str, - ) -> None: - if payload is None: - return - errors = McpDevPayloadProjection.sequence_of_mappings(payload.get("errors")) - if errors: - lines.append(f"{label} errors:") - lines.extend(ViewerValidationRenderer._error_lines(errors)) - warnings = McpDevPayloadProjection.sequence_of_mappings(payload.get("warnings")) - if warnings: - lines.append(f"{label} warnings:") - lines.extend(ViewerValidationRenderer._error_lines(warnings)) + return "\n".join(lines) @classmethod def _append_repair_hints( cls, lines: list[str], - validate_payload: Mapping[str, JsonValue] | None, - steps: tuple[Mapping[str, JsonValue], ...], + errors: tuple[AgentError, ...], + steps: tuple[FunctionStepSpec, ...], ) -> None: - if validate_payload is None or not steps: - return - errors = McpDevPayloadProjection.sequence_of_mappings( - validate_payload.get("errors") - ) missing_kwargs = cls._missing_function_kwargs(errors) - if not missing_kwargs: - return - function_id = cls._first_function_id(steps) - if function_id is None: + functions = tuple(function for step in steps for function in step.functions) + if not missing_kwargs or not functions: return + function_id = functions[0].function_id kwargs_shape = ", ".join(f'"{name}": ' for name in missing_kwargs) lines.append(f"Next: function {function_id}") lines.append( - "Retry shape: " - f"draft-pipeline-step {function_id} --kwargs '{{{kwargs_shape}}}'" + f"Retry shape: draft-pipeline-step {function_id} --kwargs '{{{kwargs_shape}}}'" ) - @staticmethod + @classmethod def _missing_function_kwargs( - errors: tuple[Mapping[str, JsonValue], ...], + cls, errors: tuple[AgentError, ...] ) -> tuple[str, ...]: for error in errors: - if error.get("code") != "missing_function_kwargs": - continue - for key in ("hint", "message"): - value = error.get(key) - if not isinstance(value, str): - continue - names = PipelineDraftStepRenderer._parse_missing_kwargs(value) - if names: - return names + if error.code == "missing_function_kwargs": + for text in (error.hint, error.message): + if text is not None: + names = cls._parse_missing_kwargs(text) + if names: + return names return () @staticmethod @@ -181,355 +206,244 @@ def _parse_missing_kwargs(text: str) -> tuple[str, ...]: else: return () tail = tail.strip().rstrip(".") - names = tuple( + return tuple( name.strip().strip("`'") for name in tail.split(",") if name.strip() ) - return names - - @staticmethod - def _first_function_id( - steps: tuple[Mapping[str, JsonValue], ...], - ) -> str | None: - for step in steps: - functions = McpDevPayloadProjection.sequence_of_mappings( - step.get("functions") - ) - for function in functions: - function_id = function.get("function_id") - if isinstance(function_id, str) and function_id: - return function_id - return None - -class PipelineArtifactPlanRenderer(McpDevOutputRenderer): - """Compact renderer for pycodified pipeline artifact-plan inspection.""" +class PipelineArtifactPlanRenderer(McpDevTypedOutputRenderer): output_contract = ArtifactPlanInspection @classmethod - def render(cls, response: JsonObject) -> str: - payload = McpDevPayloadProjection.first_tool_payload(response) - if payload is None: - return json.dumps(response, indent=2, sort_keys=True) + def render_payload( + cls, payload: ArtifactPlanInspection, options: McpDevOutputRenderOptions + ) -> str: lines = [ - ( - "Artifact plan: " - f"plate={McpDevPayloadProjection.text(payload.get('plate_path'))} " - f"axes={McpDevPayloadProjection.text(payload.get('axis_count'))} " - f"steps={McpDevPayloadProjection.text(payload.get('step_count'))} " - f"progress_events={McpDevPayloadProjection.text(payload.get('progress_event_count'))}" - ) + f"Artifact plan: plate={payload.plate_path} axes={payload.axis_count} steps={payload.step_count} progress_events={payload.progress_event_count}" ] - ObjectStateScopeRenderer._append_messages(lines, payload) - axes = payload.get("axes") - if axes: - lines.append(f"Axes: {ViewerValidationRenderer._sequence_text(axes)}") - axis_filter = payload.get("axis_filter") - if axis_filter: + _append_messages(lines, (), payload.warnings) + if payload.axes: + lines.append(f"Axes: {_sequence_text(payload.axes)}") + if payload.axis_filter: + lines.append(f"Axis filter: {_sequence_text(payload.axis_filter)}") + lines.extend(cls._source_workspace_lines(payload.source_workspace)) + if payload.worker_assignments: lines.append( - f"Axis filter: {ViewerValidationRenderer._sequence_text(axis_filter)}" + "Workers: " + + ", ".join( + f"{worker}=[{_sequence_text(axes)}]" + for worker, axes in payload.worker_assignments.items() + ) ) - source_workspace = McpDevPayloadProjection.nested_mapping( - payload, - "source_workspace", - ) - if source_workspace: - lines.extend(cls._source_workspace_lines(source_workspace)) - worker_assignments = McpDevPayloadProjection.nested_mapping( - payload, - "worker_assignments", - ) - if worker_assignments: - lines.append(f"Workers: {cls._worker_text(worker_assignments)}") - steps = McpDevPayloadProjection.sequence_of_mappings(payload.get("steps")) - if steps: + if payload.steps: lines.append("Steps:") - lines.extend(cls._step_lines(steps)) + lines.extend(cls._step_lines(payload.steps)) return "\n".join(lines) - @classmethod - def _source_workspace_lines( - cls, - source_workspace: Mapping[str, JsonValue], - ) -> list[str]: + @staticmethod + def _source_workspace_lines(source_workspace: SourceWorkspaceSummary) -> list[str]: lines = [ - ( - "Source workspace (source-bound files): " - f"files={McpDevPayloadProjection.text(source_workspace.get('file_count'))} " - f"truncated={McpDevPayloadProjection.text(source_workspace.get('truncated_file_count'))}" - ) + f"Source workspace (source-bound files): files={source_workspace.file_count} truncated={source_workspace.truncated_file_count}" ] - axis_file_counts = McpDevPayloadProjection.nested_mapping( - source_workspace, - "axis_file_counts", - ) - if axis_file_counts: - lines.append(f" axis files: {cls._mapping_text(axis_file_counts)}") - if optional_int(source_workspace.get("file_count")) == 0: - lines.append(f" note: {getdoc(SourceWorkspaceSummary)}") - files = McpDevPayloadProjection.sequence_of_mappings( - source_workspace.get("files") - ) - for file_record in files[:5]: - virtual_path = McpDevPayloadProjection.text(file_record.get("virtual_path")) - source_path = McpDevPayloadProjection.text(file_record.get("source_path")) - metadata = McpDevPayloadProjection.nested_mapping( - file_record, - "source_metadata", + if source_workspace.axis_file_counts: + lines.append( + f" axis files: {_mapping_text(source_workspace.axis_file_counts)}" ) - line = f" - {virtual_path}" - if source_path and source_path != "" and source_path != virtual_path: - line += f" -> {source_path}" - if metadata: - line += f" components={cls._mapping_text(metadata)}" + if source_workspace.file_count == 0: + lines.append(f" note: {getdoc(SourceWorkspaceSummary)}") + for record in source_workspace.files[:5]: + line = f" - {record.virtual_path}" + if record.source_path and record.source_path != record.virtual_path: + line += f" -> {record.source_path}" + if record.source_metadata: + line += f" components={_mapping_text(record.source_metadata)}" lines.append(line) - if len(files) > 5: + if len(source_workspace.files) > 5: lines.append(" - ...") return lines @classmethod - def _step_lines( - cls, - steps: tuple[Mapping[str, JsonValue], ...], - ) -> list[str]: + def _step_lines(cls, steps: tuple[CompiledStepPlanSummary, ...]) -> list[str]: lines: list[str] = [] for step in steps: lines.append( - "- " - f"{McpDevPayloadProjection.text(step.get('step_index'))}: " - f"{McpDevPayloadProjection.text(step.get('step_name'))} " - f"axis={McpDevPayloadProjection.text(step.get('axis_id'))} " - f"groups={ViewerValidationRenderer._sequence_text(step.get('execution_groups'))}" - ) - axis_persistence = optional_bool( - step.get("main_flow_axis_persistence_enabled") + f"- {step.step_index}: {step.step_name} axis={step.axis_id} groups={_sequence_text(step.execution_groups)}" ) - if axis_persistence is not None: - lines.append( - " main-flow axis persistence: " - f"{'enabled' if axis_persistence else 'runtime-only'}" + lines.extend( + cls.optional_lines( + step.main_flow_axis_persistence_enabled, + lambda enabled: ( + " main-flow axis persistence: " + + ("enabled" if enabled else "runtime-only"), + ), ) - main_flow_materialization = McpDevPayloadProjection.nested_mapping( - step, - "main_flow_materialization", ) - if main_flow_materialization: - lines.append( - " main-flow checkpoint: " - f"backend={McpDevPayloadProjection.text(main_flow_materialization.get('backend'))} " - f"output_dir={McpDevPayloadProjection.text(main_flow_materialization.get('output_dir'))} " - f"sub_dir={McpDevPayloadProjection.text(main_flow_materialization.get('sub_dir'))}" + lines.extend( + cls.optional_lines( + step.main_flow_materialization, + lambda checkpoint: ( + f" main-flow checkpoint: backend={checkpoint.backend} output_dir={checkpoint.output_dir} sub_dir={checkpoint.sub_dir}", + ), ) - viewer_streaming = McpDevPayloadProjection.sequence_of_mappings( - step.get("viewer_streaming") ) - for viewer in viewer_streaming: - effective_config = McpDevPayloadProjection.nested_mapping( - viewer, - "effective_config", - ) - line = ( - " viewer stream: " - f"viewer={McpDevPayloadProjection.text(viewer.get('viewer_type'))} " - f"config={McpDevPayloadProjection.text(viewer.get('config_key'))} " - f"backend={McpDevPayloadProjection.text(viewer.get('backend'))}" - ) - if effective_config: - line += f" effective={cls._mapping_text(effective_config)}" + for viewer in step.viewer_streaming: + line = f" viewer stream: viewer={viewer.viewer_type.value} config={viewer.config_key} backend={viewer.backend}" + if viewer.effective_config: + line += f" effective={_mapping_text(viewer.effective_config)}" lines.append(line) - artifact_inputs = McpDevPayloadProjection.sequence_of_mappings( - step.get("artifact_inputs") - ) - for artifact in artifact_inputs: + for artifact in step.artifact_inputs: lines.append(cls._artifact_input_line(artifact)) - artifacts = McpDevPayloadProjection.sequence_of_mappings( - step.get("artifact_outputs") - ) - for artifact in artifacts: + for artifact in step.artifact_outputs: + lines.append( + f" artifact {artifact.name}: kind={artifact.kind} path={artifact.path} groups={_sequence_text(artifact.group_keys)}" + ) + lines.extend( + cls.optional_lines( + artifact.materialization, cls._materialization_lines + ) + ) + if step.truncated_artifact_input_count > 0: lines.append( - " artifact " - f"{McpDevPayloadProjection.text(artifact.get('name'))}: " - f"kind={McpDevPayloadProjection.text(artifact.get('kind'))} " - f"path={McpDevPayloadProjection.text(artifact.get('path'))} " - f"groups={ViewerValidationRenderer._sequence_text(artifact.get('group_keys'))}" + f" artifact input ... truncated={step.truncated_artifact_input_count}" ) - materialization = McpDevPayloadProjection.nested_mapping( - artifact, - "materialization", + if step.truncated_artifact_output_count > 0: + lines.append( + f" artifact ... truncated={step.truncated_artifact_output_count}" ) - if materialization: - lines.extend(cls._materialization_lines(materialization)) - truncated_input_count = step.get("truncated_artifact_input_count") - if isinstance(truncated_input_count, int) and truncated_input_count > 0: - lines.append(f" artifact input ... truncated={truncated_input_count}") - truncated_count = step.get("truncated_artifact_output_count") - if isinstance(truncated_count, int) and truncated_count > 0: - lines.append(f" artifact ... truncated={truncated_count}") return lines - @staticmethod - def _artifact_input_line(artifact: Mapping[str, JsonValue]) -> str: + @classmethod + def _artifact_input_line(cls, artifact: ArtifactInputPlanSummary) -> str: source_parts: list[str] = [] - source_step_id = artifact.get("source_step_id") - if source_step_id is not None: - source_parts.append( - f"source_step={McpDevPayloadProjection.text(source_step_id)}" + source_parts.extend( + cls.optional_lines( + artifact.source_step_id, lambda value: (f"source_step={value}",) ) - source_step_scope_id = artifact.get("source_step_scope_id") - if source_step_scope_id is not None: - source_parts.append( - f"source_scope={McpDevPayloadProjection.text(source_step_scope_id)}" + ) + source_parts.extend( + cls.optional_lines( + artifact.source_step_scope_id, lambda value: (f"source_scope={value}",) ) - suffix = f" {' '.join(source_parts)}" if source_parts else "" - return ( - " artifact input " - f"{McpDevPayloadProjection.text(artifact.get('name'))}: " - f"kind={McpDevPayloadProjection.text(artifact.get('kind'))} " - f"path={McpDevPayloadProjection.text(artifact.get('path'))} " - f"groups={ViewerValidationRenderer._sequence_text(artifact.get('group_keys'))}" - f"{suffix}" ) + suffix = f" {' '.join(source_parts)}" if source_parts else "" + return f" artifact input {artifact.name}: kind={artifact.kind} path={artifact.path} groups={_sequence_text(artifact.group_keys)}{suffix}" @classmethod def _materialization_lines( cls, - materialization: Mapping[str, JsonValue], + materialization: ArtifactMaterializationPlanSummary, ) -> list[str]: - status_parts: list[str] = [] - if optional_bool(materialization.get("disabled")) is True: - status_parts.append("disabled") - elif optional_bool(materialization.get("runtime_resolved")) is True: - status_parts.append("runtime-resolved") - else: - status_parts.append("explicit") - status_parts.append( - "persistent=" - f"{McpDevPayloadProjection.text(materialization.get('persistent_enabled'))}" + mode = ( + "disabled" + if materialization.disabled + else "runtime-resolved" + if materialization.runtime_resolved + else "explicit" ) - backend = materialization.get("persistent_backend") - if backend is not None: - status_parts.append(f"backend={McpDevPayloadProjection.text(backend)}") - analysis_output_dir = materialization.get("analysis_output_dir") - if analysis_output_dir is not None: - status_parts.append( - f"analysis_dir={McpDevPayloadProjection.text(analysis_output_dir)}" + parts = [mode, f"persistent={materialization.persistent_enabled}"] + parts.extend( + cls.optional_lines( + materialization.persistent_backend, lambda value: (f"backend={value}",) + ) + ) + parts.extend( + cls.optional_lines( + materialization.analysis_output_dir, + lambda value: (f"analysis_dir={value}",), + ) + ) + if materialization.filename_uses_source_identity: + parts.append("source-identity-filenames") + if materialization.runtime_metadata_can_refine_paths: + parts.append("runtime-metadata-filenames") + lines = [f" materialization: {' '.join(parts)}"] + for path in materialization.paths[:3]: + lines.append( + f" candidates group={McpDevPayloadProjection.text(path.group_key)}: {', '.join(path.candidate_paths)}" + ) + if len(materialization.paths) > 3: + lines.append( + f" candidates ... truncated={len(materialization.paths) - 3}" + ) + lines.extend( + cls.optional_lines( + materialization.note, lambda value: (f" note: {value}",) ) - if optional_bool(materialization.get("filename_uses_source_identity")) is True: - status_parts.append("source-identity-filenames") - if ( - optional_bool(materialization.get("runtime_metadata_can_refine_paths")) - is True - ): - status_parts.append("runtime-metadata-filenames") - - lines = [f" materialization: {' '.join(status_parts)}"] - paths = McpDevPayloadProjection.sequence_of_mappings( - materialization.get("paths") ) - for path in paths[:3]: - group_key = McpDevPayloadProjection.text(path.get("group_key")) - candidates = cls._candidate_path_text(path.get("candidate_paths")) - lines.append(" candidates " f"group={group_key}: {candidates}") - if len(paths) > 3: - lines.append(f" candidates ... truncated={len(paths) - 3}") - note = materialization.get("note") - if note is not None: - lines.append(f" note: {McpDevPayloadProjection.text(note)}") return lines - @staticmethod - def _candidate_path_text(value: JsonValue) -> str: - if isinstance(value, list | tuple): - return ", ".join(McpDevPayloadProjection.text(item) for item in value) - return McpDevPayloadProjection.text(value) - @staticmethod - def _mapping_text(mapping: Mapping[str, JsonValue]) -> str: - return ", ".join( - f"{key}={McpDevPayloadProjection.text(value)}" - for key, value in mapping.items() - ) +class OrchestratorSessionReferenceRenderer(McpDevTypedOutputRenderer): + output_contract = OrchestratorSessionRef @classmethod - def _worker_text(cls, mapping: Mapping[str, JsonValue]) -> str: - return ", ".join( - f"{worker}=[{ViewerValidationRenderer._sequence_text(axes)}]" - for worker, axes in mapping.items() - ) + def render_payload( + cls, payload: OrchestratorSessionRef, options: McpDevOutputRenderOptions + ) -> str: + return f"Session: id={payload.session_id} uri={payload.uri}" -class ExecuteSourceRenderer: - """Compact renderer for source-backed headless execution command results.""" +class ExecutionJobRenderer(McpDevTypedOutputRenderer): + """Presentation shared by the actual job-identity family, not sibling DTOs.""" + + @staticmethod + def identity_line(payload: ExecutionJobIdentity, status: str) -> str: + return f"Job: id={payload.job_id} kind={payload.kind} status={status} server_execution={McpDevPayloadProjection.text(payload.server_execution_id)}" - SESSION_TOOL = ( - agent_capabilities.create_orchestrator_session_from_pipeline_source.name - ) - SUBMIT_TOOL = agent_capabilities.submit_pipeline_execution.name + +class ExecutionJobReferenceRenderer(ExecutionJobRenderer): + output_contract = ExecutionJobRef @classmethod - def render(cls, response: JsonObject) -> str: - session_payload = McpDevPayloadProjection.tool_payload( - response, - cls.SESSION_TOOL, - ) - submit_payload = McpDevPayloadProjection.tool_payload( - response, - cls.SUBMIT_TOOL, - ) - if session_payload is None and submit_payload is None: - return json.dumps(response, indent=2, sort_keys=True) + def render_payload( + cls, payload: ExecutionJobRef, options: McpDevOutputRenderOptions + ) -> str: + return cls.identity_line(payload, payload.status) - lines = ["Headless source execution:"] - if session_payload is not None: - lines.append( - "Session: " - f"id={McpDevPayloadProjection.text(session_payload.get('session_id'))} " - f"uri={McpDevPayloadProjection.text(session_payload.get('uri'))}" - ) - ObjectStateScopeRenderer._append_messages(lines, session_payload) - else: - lines.append("Session: ") - if submit_payload is not None: +class ExecutionJobStatusRenderer(ExecutionJobRenderer): + output_contract = ExecutionJobStatus + + @classmethod + def render_payload( + cls, payload: ExecutionJobStatus, options: McpDevOutputRenderOptions + ) -> str: + lines = [cls.identity_line(payload, payload.status)] + if payload.response: lines.append( - "Job: " - f"id={McpDevPayloadProjection.text(submit_payload.get('job_id'))} " - f"kind={McpDevPayloadProjection.text(submit_payload.get('kind'))} " - f"status={McpDevPayloadProjection.text(submit_payload.get('status'))} " - "server_execution=" - f"{McpDevPayloadProjection.text(submit_payload.get('server_execution_id'))}" - ) - response_payload = McpDevPayloadProjection.nested_mapping( - submit_payload, - "response", + "Response: " + + " ".join(f"{key}={value}" for key, value in payload.response.items()) ) - if response_payload: - lines.append( - "Response: " + cls._response_summary_text(response_payload) - ) - ObjectStateScopeRenderer._append_messages(lines, submit_payload) - else: - lines.append("Job: ") + lines.extend( + cls.optional_lines(payload.progress, lambda value: (f"Progress: {value}",)) + ) + _append_messages(lines, (), payload.warnings) return "\n".join(lines) - @staticmethod - def _response_summary_text(response_payload: Mapping[str, JsonValue]) -> str: - summary_keys = ( - "status", - "state", - "message", - "execution_id", - "server_execution_id", - "completed", - "success", + +class ExecuteSourceRenderer: + @classmethod + def render(cls, response) -> str: + decoded = McpDevToolBatchResponse.for_rendering(response) + session: OrchestratorSessionRef | None = decoded.payload_for( + agent_capabilities.create_orchestrator_session_from_pipeline_source ) - parts = [ - f"{key}={McpDevPayloadProjection.text(response_payload.get(key))}" - for key in summary_keys - if key in response_payload - ] - if parts: - return " ".join(parts) - return ( - f"keys={ViewerValidationRenderer._sequence_text(sorted(response_payload))}" + job: ExecutionJobRef | ExecutionJobStatus | None = decoded.payload_for( + agent_capabilities.submit_pipeline_execution + ) + options = McpDevOutputRenderOptions() + lines = ["Headless source execution:"] + lines.append( + "Session: " + if session is None + else OrchestratorSessionReferenceRenderer.render_payload(session, options) ) + lines.append( + "Job: " + if job is None + else McpDevTypedOutputRenderer.render_payload_value(job, options) + ) + lines.extend( + McpDiagnosticRenderer.typed_error_lines(decoded.diagnostic_errors()) + ) + return "\n".join(lines) diff --git a/openhcs/mcp/dev_client_renderers/ui_bridge.py b/openhcs/mcp/dev_client_renderers/ui_bridge.py index a6ea7d8c98..6b13af6c53 100644 --- a/openhcs/mcp/dev_client_renderers/ui_bridge.py +++ b/openhcs/mcp/dev_client_renderers/ui_bridge.py @@ -192,7 +192,7 @@ def render(cls, response: JsonObject) -> str: results = McpDevPayloadProjection.sequence_of_mappings(response.get("results")) mcp_errors = sum(1 for result in results if result.get("mcp_error") is True) - lines = ["UI smoke: " f"results={len(results)} mcp_errors={mcp_errors}"] + lines = [f"UI smoke: results={len(results)} mcp_errors={mcp_errors}"] lines.append(cls._health_line(response)) lines.extend( UiBridgeStatusRenderer.render( @@ -782,7 +782,7 @@ def _action_lines( suffix = "" if disabled: suffix = ( - " disabled=" f"{McpDevPayloadProjection.text(disabled.get('code'))}" + f" disabled={McpDevPayloadProjection.text(disabled.get('code'))}" ) lines.append( "- " @@ -1250,20 +1250,13 @@ def render( source = payload.get("source") if include_source and isinstance(source, str): lines.append("Source:") - lines.append(cls._source_text(source, max_source_chars=max_source_chars)) + lines.append( + CodeDocumentRenderOptions( + max_source_chars=max_source_chars + ).source_text(source) + ) return "\n".join(lines) - @staticmethod - def _source_text(source: str, *, max_source_chars: int) -> str: - if max_source_chars < 0: - raise ValueError("max_source_chars must be nonnegative.") - if len(source) <= max_source_chars: - return source - return ( - source[:max_source_chars] - + f"\n..." - ) - class CodeDocumentValidationRenderer(McpDevOutputRenderer): """Compact renderer for UI code-document validation results.""" diff --git a/openhcs/mcp/dev_client_rendering.py b/openhcs/mcp/dev_client_rendering.py index c193e50f76..fb12c1afeb 100644 --- a/openhcs/mcp/dev_client_rendering.py +++ b/openhcs/mcp/dev_client_rendering.py @@ -4,21 +4,23 @@ import argparse import json -from collections.abc import Callable, Mapping +from collections.abc import Callable, Mapping, Sequence from dataclasses import asdict, dataclass from enum import Enum -from typing import ClassVar, TypeAlias +from typing import ClassVar, TypeAlias, TypeVar from metaclass_registry import AutoRegisterMeta +from python_introspect import dataclass_from_mapping from openhcs.agent.capabilities import ( AgentCapabilitySpec, CapabilityWorkflowGroup, get_agent_capability, ) -from openhcs.agent.dto.common import JsonObject, JsonValue +from openhcs.agent.dto.common import AgentError, JsonObject, JsonValue DEFAULT_CODE_DOCUMENT_MAX_CHARS = 2_000 +PresentationValue = TypeVar("PresentationValue") class WidgetTreeOutputFormat(str, Enum): @@ -62,6 +64,17 @@ class CodeDocumentRenderOptions(McpDevOutputRenderOptions): include_source: bool = True max_source_chars: int = DEFAULT_CODE_DOCUMENT_MAX_CHARS + def source_text(self, source: str) -> str: + """One bounded source-text policy shared by all source presentations.""" + if self.max_source_chars < 0: + raise ValueError("max_source_chars must be nonnegative.") + if len(source) <= self.max_source_chars: + return source + return ( + source[: self.max_source_chars] + + f"\n..." + ) + @dataclass(frozen=True, slots=True) class UiActionCatalogRenderOptions(McpDevOutputRenderOptions): @@ -140,14 +153,25 @@ class McpDevOutputRenderer(metaclass=AutoRegisterMeta): render_options_type: ClassVar[type[McpDevOutputRenderOptions]] = ( McpDevOutputRenderOptions ) - __renderer_types__: ClassVar[tuple[type["McpDevOutputRenderer"], ...]] = () - def __init_subclass__(cls, **kwargs: JsonValue) -> None: + def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs) - McpDevOutputRenderer.__renderer_types__ = ( - *McpDevOutputRenderer.__renderer_types__, - cls, - ) + # AutoRegisterMeta prefers an explicit (including inherited) key. A + # child must register its own declaration, not overwrite its parent's + # key merely because the metaclass assigned that key to the parent. + cls.renderer_key = mcp_dev_output_renderer_key(cls.__name__, cls) + + @classmethod + def declaration_types(cls): + """Derived hierarchy view, visiting a diamond's identity only once.""" + pending = list(cls.__subclasses__()) + seen: set[type] = set() + while pending: + member = pending.pop(0) + if member not in seen: + seen.add(member) + yield member + pending.extend(member.__subclasses__()) @classmethod def for_output_contract( @@ -161,7 +185,11 @@ def for_output_contract( ) ensure_dev_client_renderers_registered() - for renderer_type in cls.__renderer_types__: + for owner in output_contract.__mro__: + renderer_type = cls.__registry__.get(owner) + if renderer_type is not None: + return McpDevOutputRendererBinding(output_contract, renderer_type) + for renderer_type in cls.declaration_types(): binding = renderer_type.binding_for_output_contract(output_contract) if binding is not None: return binding @@ -188,6 +216,18 @@ def render_bindings(cls) -> tuple["McpDevOutputRendererBinding", ...]: """Return additional output-contract bindings owned by this renderer.""" return () + @classmethod + def decode_payload(cls, payload: object, output_contract: type) -> object: + """Unmigrated families keep their existing presentation until cutover.""" + del output_contract + return payload + + @classmethod + def render_result(cls, response, options: McpDevOutputRenderOptions) -> str: + from openhcs.serialization.json import to_jsonable + + return cls.render_with_options(to_jsonable(response), options) + @classmethod def render(cls, response: JsonObject) -> str: raise NotImplementedError @@ -210,6 +250,16 @@ class McpDevOutputRendererBinding: renderer_type: type[McpDevOutputRenderer] render_function: McpDevOutputRenderFunction | None = None + def decode_payload(self, payload: object) -> object: + return self.renderer_type.decode_payload(payload, self.output_contract) + + def render_result(self, response, options: McpDevOutputRenderOptions) -> str: + if self.render_function is not None: + from openhcs.serialization.json import to_jsonable + + return self.render_function(to_jsonable(response)) + return self.renderer_type.render_result(response, options) + def configure_cli_parser(self, parser: argparse.ArgumentParser) -> None: self.renderer_type.render_options_type.configure_cli_parser(parser) @@ -232,6 +282,65 @@ def render_with_options( return self.renderer_type.render_with_options(response, options) +class McpDevTypedOutputRenderer(McpDevOutputRenderer): + """Shared contract descent for typed presentation members, never raw readers.""" + + @staticmethod + def optional_lines( + value: PresentationValue | None, + render_lines: Callable[[PresentationValue], Sequence[str]], + ) -> Sequence[str]: + """Compose a nullable declared fact without repeating omission policy. + + None is the native absence fact; false, zero and empty strings remain + present. Members own only the presentation of a present value. + """ + return () if value is None else render_lines(value) + + @classmethod + def render_result(cls, response, options: McpDevOutputRenderOptions) -> str: + return cls.render_with_options(response, options) + + @classmethod + def render(cls, response) -> str: + return cls.render_with_options(response, cls.render_options_type()) + + @classmethod + def render_with_options(cls, response, options: McpDevOutputRenderOptions) -> str: + from openhcs.mcp.dev_client_core import McpDevToolBatchResponse + + decoded = McpDevToolBatchResponse.for_rendering(response) + payload = next( + (result.first_decoded_payload() for result in decoded.results), None + ) + lines = [ + "Result: " + if payload is None + else cls.render_payload_value(payload, options) + ] + lines.extend( + McpDiagnosticRenderer.typed_error_lines(decoded.diagnostic_errors()) + ) + return "\n".join(lines) + + @classmethod + def render_payload(cls, payload, options: McpDevOutputRenderOptions) -> str: + raise NotImplementedError + + @classmethod + def render_payload_value(cls, payload, options: McpDevOutputRenderOptions) -> str: + binding = McpDevOutputRenderer.for_output_contract(type(payload)) + if binding is None: + raise TypeError(f"No renderer declared for {type(payload).__name__}") + return binding.renderer_type.render_payload(payload, options) + + @classmethod + def decode_payload(cls, payload: object, output_contract: type) -> object: + if isinstance(payload, output_contract): + return payload + return dataclass_from_mapping(output_contract, payload) + + class McpDevPayloadProjection: """Small read helpers for dev-client JSON envelopes.""" @@ -352,15 +461,21 @@ def response_error_lines(cls, response: JsonObject) -> tuple[str, ...]: @staticmethod def error_lines(errors: tuple[Mapping[str, JsonValue], ...]) -> tuple[str, ...]: + return McpDiagnosticRenderer.typed_error_lines( + tuple(dataclass_from_mapping(AgentError, error) for error in errors) + ) + + @staticmethod + def typed_error_lines(errors) -> tuple[str, ...]: grouped_codes: dict[str, list[str]] = {} grouped_hints: dict[str, list[str]] = {} for error in errors: - message = McpDevPayloadProjection.text(error.get("message")) - hint = error.get("hint") + message = error.message + hint = error.hint hint_text = ( None if hint is None else McpDevPayloadProjection.quoted_text(hint) ) - code = McpDevPayloadProjection.text(error.get("code")) + code = error.code codes = grouped_codes.setdefault(message, []) if code not in codes: codes.append(code) @@ -373,7 +488,7 @@ def error_lines(errors: tuple[Mapping[str, JsonValue], ...]) -> tuple[str, ...]: lines: list[str] = [] for message, codes in tuple(grouped_codes.items())[:3]: code_text = codes[0] if len(codes) == 1 else ", ".join(codes) - line = f"- {code_text}: " f"{message}" + line = f"- {code_text}: {message}" hint_texts = grouped_hints.get(message, []) if len(hint_texts) == 1: line += f" hint={hint_texts[0]}" diff --git a/tests/unit/agent/test_mcp_dev_client_pipeline_results.py b/tests/unit/agent/test_mcp_dev_client_pipeline_results.py new file mode 100644 index 0000000000..bd826771a8 --- /dev/null +++ b/tests/unit/agent/test_mcp_dev_client_pipeline_results.py @@ -0,0 +1,745 @@ +"""Production command journeys with controlled wire receipts, never a runtime.""" + +from __future__ import annotations + +import ast +import asyncio +import json +import sys +from dataclasses import dataclass, replace +from pathlib import Path + +import pytest +from zmqruntime.execution import ExecutionProgressObservation + +from openhcs.agent.capabilities import agent_capabilities +from openhcs.agent.dto.common import ( + SCHEMA_VERSION, + AgentError, + AgentWarning, + RenderedSource, +) +from openhcs.agent.dto.execution import ( + ArtifactInputPlanSummary, + ArtifactMaterializationPathSummary, + ArtifactMaterializationPlanSummary, + ArtifactPlanInspection, + ArtifactPlanSummary, + CompiledStepPlanSummary, + ExecutionJobRef, + ExecutionJobStatus, + MainFlowMaterializationPlanSummary, + OrchestratorSessionRef, + SourceWorkspaceFileRecord, + SourceWorkspaceSummary, + ViewerStreamingPlanSummary, +) +from openhcs.agent.dto.pipeline import ( + FunctionSpecRef, + FunctionStepSpec, + PipelineRef, + PipelineSpec, + PipelineValidationResult, +) +from openhcs.core.streaming_config_declarations import ViewerType +from openhcs.mcp import dev_client +from openhcs.mcp.dev_client_commanding import CapabilityBackedCommandSpec +from openhcs.mcp.dev_client_core import ( + McpDevPayloadFailure, + McpDevServerSpec, + McpDevToolBatchResponse, + McpDevToolResult, +) +from openhcs.mcp.dev_client_rendering import ( + McpDevOutputRenderer, + McpDevTypedOutputRenderer, +) +from openhcs.mcp.dev_client_renderers.pipeline import PipelineArtifactPlanRenderer +from openhcs.serialization.json import to_jsonable + + +class ControlledWireSession: + """Only replace the wire peer; retain real framing, commands and rendering.""" + + def __init__(self, receipts, *, structured=True): + self.server_spec = McpDevServerSpec(sys.executable) + self.receipts = iter(receipts) + self.calls = [] + self.structured = structured + + async def call_tool(self, name, arguments, *, timeout_seconds): + self.calls.append((name, arguments, timeout_seconds)) + receipt = to_jsonable(next(self.receipts)) + if self.structured: + return {"isError": False, "structuredContent": receipt, "content": []} + return { + "isError": False, + "content": [{"type": "text", "text": json.dumps(receipt)}], + } + + +def batch(capability, value): + wire = {"isError": False, "structuredContent": to_jsonable(value), "content": []} + return McpDevToolBatchResponse.from_results( + McpDevServerSpec(sys.executable), + (McpDevToolResult.from_payload(capability.name, wire),), + ) + + +def args_for(*argv): + return dev_client._build_parser().parse_args(argv) + + +def plan_fixture(): + checkpoint = ArtifactMaterializationPlanSummary( + persistent_enabled=True, + persistent_backend="disk", + paths=tuple( + ArtifactMaterializationPathSummary( + str(index), f"image-{index}", (f"result-{index}.tif",) + ) + for index in range(4) + ), + filename_uses_source_identity=True, + note="receipt retained", + ) + workspace = SourceWorkspaceSummary( + file_count=7, + files=tuple( + SourceWorkspaceFileRecord( + f"virtual-{index}", + f"/full/virtual-{index}", + f"physical-{index}", + {"new_component": {"identity": index}}, + ) + for index in range(7) + ), + axis_file_counts={"A": 7}, + ) + step = CompiledStepPlanSummary( + 1, + "Owned step", + "A", + "result-dir", + execution_groups=("g",), + main_flow_axis_persistence_enabled=True, + main_flow_materialization=MainFlowMaterializationPlanSummary( + "result-dir", "disk", "plate-root", "sub-dir" + ), + viewer_streaming=( + ViewerStreamingPlanSummary( + "napari", ViewerType.NAPARI, "memory", {"new_option": [1, 2]} + ), + ), + artifact_inputs=( + ArtifactInputPlanSummary( + "input", + "image", + "input-path", + ("g",), + source_step_id=2, + source_step_scope_id="scope-2", + ), + ), + artifact_outputs=( + ArtifactPlanSummary( + "output", "image", "output-path", ("g",), materialization=checkpoint + ), + ), + ) + return ArtifactPlanInspection( + schema_version=SCHEMA_VERSION, + plate_path="source-only-plate", + axis_count=1, + axes=("A",), + step_count=1, + steps=(step,), + worker_assignments={"worker": ["A"]}, + source_workspace=workspace, + progress_event_count=3, + ) + + +@pytest.mark.parametrize("structured", [True, False]) +def test_draft_journey_decodes_once_then_renders_real_command(structured, monkeypatch): + ref = PipelineRef("pipeline-1", "openhcs://pipelines/pipeline-1") + spec = PipelineSpec( + SCHEMA_VERSION, + "pipeline-1", + "config-1", + ( + FunctionStepSpec( + "step-1", + "Count cells", + (FunctionSpecRef("function-1", {"extension": [1, 2]}),), + ), + ), + ) + validation = PipelineValidationResult( + schema_version=SCHEMA_VERSION, + valid=False, + pipeline_ref=ref, + errors=( + AgentError( + "missing_function_kwargs", + "Missing required agent kwargs: threshold.", + "required agent kwargs: threshold.", + ), + ), + warnings=(AgentWarning("note", "Pipeline is small."),), + ) + source = RenderedSource( + SCHEMA_VERSION, "Pipeline", "pipeline_steps = [FunctionStep(...)]\n" + ) + peer = ControlledWireSession((ref, spec, validation, source), structured=structured) + args = args_for("draft-pipeline-step", "function-1", "--max-source-chars", "10") + command = dev_client.McpDevCommandSpec.for_name(args.command) + import openhcs.mcp.dev_client_rendering as rendering + + decoded_contracts = [] + decode = rendering.dataclass_from_mapping + + def observed_decode(contract, payload): + decoded_contracts.append(contract) + return decode(contract, payload) + + monkeypatch.setattr(rendering, "dataclass_from_mapping", observed_decode) + response = asyncio.run(command.run_session(peer, args)) + assert [call[0] for call in peer.calls] == [ + agent_capabilities.create_pipeline.name, + agent_capabilities.add_function_step.name, + agent_capabilities.validate_pipeline.name, + agent_capabilities.render_pipeline_source.name, + ] + assert peer.calls[1][1]["pipeline_id"] == ref.pipeline_id + assert isinstance(response.results[1].payloads[0], PipelineSpec) + assert response.results[1].payloads[0].steps[0].functions[0].kwargs == { + "extension": [1, 2] + } + before = to_jsonable(response) + render = command.render_result(response, args) + assert "id=pipeline-1 valid=False steps=1" in render + assert "functions=function-1" in render + assert "missing_function_kwargs" in render and "threshold" in render + assert "note: Pipeline is small." in render + assert "Next: function function-1" in render and "Retry shape:" in render + assert "truncated" in render + assert to_jsonable(response) == before + assert decoded_contracts == [ + PipelineRef, + PipelineSpec, + PipelineValidationResult, + RenderedSource, + ] + assert command.render_result( + response, replace_namespace(args, json=True) + ) == json.dumps(before, indent=2, sort_keys=True) + + +def replace_namespace(args, **values): + import argparse + + return argparse.Namespace(**(vars(args) | values)) + + +def test_nested_artifact_contract_and_receipt_survive_compact_limits(): + original = plan_fixture() + response = batch(agent_capabilities.inspect_pipeline_source_artifact_plan, original) + decoded = response.results[0].payloads[0] + assert isinstance(decoded, ArtifactPlanInspection) + assert isinstance(decoded.source_workspace.files[6], SourceWorkspaceFileRecord) + assert decoded.steps[0].viewer_streaming[0].viewer_type is ViewerType.NAPARI + assert decoded.steps[0].artifact_outputs[0].materialization.paths[ + 3 + ].candidate_paths == ("result-3.tif",) + assert to_jsonable(decoded) == to_jsonable(original) + args = args_for( + "artifact-plan", "source-only-plate", "--source-text", "pipeline_steps = []" + ) + rendered = dev_client.McpDevCommandSpec.for_name(args.command).render_result( + response, args + ) + for fact in ( + "source-only-plate", + "axes=1", + "progress_events=3", + "new_component", + "source_scope=scope-2", + "backend=disk", + "sub_dir=sub-dir", + "viewer=napari", + "result-2.tif", + "truncated=1", + "receipt retained", + ): + assert fact in rendered + assert "physical-6" not in rendered + assert len(decoded.source_workspace.files) == 7 + assert len(decoded.steps[0].artifact_outputs[0].materialization.paths) == 4 + assert to_jsonable(decoded) == to_jsonable(original) + + +@pytest.mark.parametrize("waited", [True, False]) +def test_execute_journey_preserves_native_variants_progress_and_errors(waited): + session = OrchestratorSessionRef( + session_id="session-1", + schema_version=SCHEMA_VERSION, + uri="openhcs://sessions/1", + ) + identity = dict( + schema_version=SCHEMA_VERSION, + job_id="job-1", + session_id="session-1", + kind="execute", + status="failed" if waited else "queued", + uri="openhcs://jobs/1", + server_execution_id="execution-1", + ) + job = ( + ExecutionJobStatus( + **identity, + response={"native_extension": {"retain": [1, 2]}, "status": "failed"}, + progress=ExecutionProgressObservation( + 3, {"phase": "processing", "extension": {"done": 2}} + ), + errors=( + AgentError("execution_failed", "Failure retained", "Inspect receipt"), + ), + warnings=(AgentWarning("wait", "Deadline reached"),), + ) + if waited + else ExecutionJobRef(**identity) + ) + peer = ControlledWireSession((session, job)) + args = args_for( + "execute-source", + "source-only-plate", + "--source-text", + "pipeline_steps = []", + "--wait" if waited else "--no-wait", + ) + command = dev_client.McpDevCommandSpec.for_name(args.command) + response = asyncio.run(command.run_session(peer, args)) + decoded = response.results[1].payloads[0] + assert type(decoded) is type(job) + assert to_jsonable(decoded) == to_jsonable(job) + assert peer.calls[1][1]["session_id"] == "session-1" + rendered = command.render_result(response, args) + assert ( + "id=session-1" in rendered + and "id=job-1" in rendered + and "server_execution=execution-1" in rendered + ) + if waited: + assert decoded.progress.sequence == 3 + assert decoded.progress.event["extension"]["done"] == 2 + for fact in ( + "native_extension", + "processing", + "Failure retained", + "Inspect receipt", + "Deadline reached", + ): + assert fact in rendered + assert response.has_errors() + else: + assert not response.has_errors() + + +@pytest.mark.parametrize( + "broken", + [ + {"schema_version": SCHEMA_VERSION, "plate_path": "source-only", "steps": [42]}, + { + "schema_version": SCHEMA_VERSION, + "plate_path": "source-only", + "axis_count": True, + }, + { + "schema_version": SCHEMA_VERSION, + "plate_path": "source-only", + "source_workspace": {"files": [{"virtual_path": "v"}]}, + }, + { + "schema_version": SCHEMA_VERSION, + "plate_path": "source-only", + "undeclared": "not metadata", + }, + { + "schema_version": SCHEMA_VERSION, + "ok": False, + "errors": [ + { + "code": "server_stale", + "message": "Reconnect first", + "hint": "No retry", + } + ], + "restart_required": True, + }, + ], +) +def test_malformed_contract_is_explicit_and_keeps_entire_receipt(broken): + response = batch(agent_capabilities.inspect_pipeline_source_artifact_plan, broken) + failed = response.results[0].payloads[0] + assert isinstance(failed, McpDevPayloadFailure) + assert failed.receipt == broken + assert to_jsonable(failed) == broken + assert response.has_errors() + rendered = PipelineArtifactPlanRenderer.render(response) + assert "mcp_payload_invalid" in rendered and "unavailable" in rendered + if "errors" in broken: + assert "Reconnect first" in rendered and "No retry" in rendered + + +def test_not_run_missing_transport_and_mcp_errors_are_distinct(): + args = args_for("draft-pipeline-step", "function-1") + command = dev_client.McpDevCommandSpec.for_name(args.command) + empty = McpDevToolBatchResponse.from_results(McpDevServerSpec(sys.executable), ()) + assert "valid=" in command.render_result(empty, args) + missing = replace( + empty, + results=( + McpDevToolResult(agent_capabilities.validate_pipeline.name, False, ()), + ), + ) + assert "mcp_payload_missing" in command.render_result(missing, args) + tool_error = replace( + empty, + results=( + McpDevToolResult(agent_capabilities.validate_pipeline.name, True, ()), + ), + ) + assert "mcp_tool_error" in command.render_result(tool_error, args) + failure = command.transport_failure_response( + McpDevServerSpec(sys.executable), + command.execution_phase, + RuntimeError("connection lost"), + server_stderr_tail="full retained stderr", + ) + rendered = command.render_result(failure, args) + assert "connection lost" in rendered and "mcp_transport_failed" in rendered + assert failure.errors[0].server_stderr_tail == "full retained stderr" + + +def test_single_declaration_extension_uses_output_mro_without_consumer_edits(): + from openhcs.agent.capabilities import InspectPipelineSourceArtifactPlanCapability + + @dataclass(frozen=True, slots=True, kw_only=True) + class ExtendedInspection(ArtifactPlanInspection): + extension_fact: str = "retained" + + binding = McpDevOutputRenderer.for_output_contract(ExtendedInspection) + assert binding.renderer_type is PipelineArtifactPlanRenderer + value = binding.decode_payload( + to_jsonable( + ExtendedInspection(schema_version=SCHEMA_VERSION, plate_path="extended") + ) + ) + assert type(value) is ExtendedInspection and value.extension_fact == "retained" + assert "plate=extended" in binding.renderer_type.render_payload_value( + value, binding.renderer_type.render_options_type() + ) + + # One capability declaration selects the new output at real wire ingress; + # presentation inherits the existing owner, without adding a roster entry. + class ExtensionCapability(InspectPipelineSourceArtifactPlanCapability): + name = "openhcs_s1_extension_inspection" + cli_command = "s1-extension-inspection" + output_contract = ExtendedInspection + + response = batch(ExtensionCapability.to_spec(), value) + assert type(response.results[0].first_decoded_payload()) is ExtendedInspection + assert "plate=extended" in PipelineArtifactPlanRenderer.render(response) + command = CapabilityBackedCommandSpec.for_capability_name(ExtensionCapability.name) + assert "plate=extended" in command.render_result( + response, command.call_render_args({}) + ) + + +@pytest.mark.parametrize("value", [None, False, 0, ""]) +def test_shared_optional_presentation_preserves_native_absence(value): + from openhcs.mcp.dev_client_rendering import McpDevTypedOutputRenderer + + observed = [] + + def present(fact): + observed.append(fact) + return (str(fact),) + + lines = McpDevTypedOutputRenderer.optional_lines(value, present) + assert lines == (() if value is None else (str(value),)) + assert observed == ([] if value is None else [value]) + + +@pytest.mark.parametrize("reverse_order", [False, True]) +def test_cooperative_diamond_renderer_identity_is_visited_once(reverse_order): + from openhcs.agent.capabilities import InspectPipelineSourceArtifactPlanCapability + + @dataclass(frozen=True, slots=True) + class NewResult: + value: str + + events = [] + + class ValueRenderer(McpDevTypedOutputRenderer): + @classmethod + def render_payload(cls, payload: NewResult, options): + events.append("ancestor") + return payload.value + + class Left(ValueRenderer): + @classmethod + def render_payload(cls, payload, options): + events.append("left") + return super().render_payload(payload, options) + + class Right(ValueRenderer): + @classmethod + def render_payload(cls, payload, options): + events.append("right") + return super().render_payload(payload, options) + + bases = (Right, Left) if reverse_order else (Left, Right) + Diamond = type("Diamond", bases, {"output_contract": NewResult}) + + class DiamondCapability(InspectPipelineSourceArtifactPlanCapability): + name = f"openhcs_s1_diamond_{int(reverse_order)}" + cli_command = f"s1-diamond-{int(reverse_order)}" + output_contract = NewResult + + binding = McpDevOutputRenderer.for_output_contract(NewResult) + value = binding.decode_payload({"value": "one-declaration"}) + assert ( + binding.renderer_type.render_payload_value( + value, binding.renderer_type.render_options_type() + ) + == "one-declaration" + ) + expected = ["right", "left"] if reverse_order else ["left", "right"] + assert events == [*expected, "ancestor"] + events.clear() + response = batch(DiamondCapability.to_spec(), value) + command = CapabilityBackedCommandSpec.for_capability_name(DiamondCapability.name) + assert command.render_result(response, command.call_render_args({})) == value.value + assert events == [*expected, "ancestor"] + assert Diamond.__mro__.count(ValueRenderer) == 1 + assert tuple(McpDevOutputRenderer.declaration_types()).count(Diamond) == 1 + # Registry views are projections of declaration identity, not a second roster. + assert McpDevOutputRenderer.__registry__[NewResult] is Diamond + + +def test_pipeline_guard_forbids_raw_reader_reintroduction(): + import openhcs.mcp.dev_client_renderers.pipeline as pipeline + + tree = ast.parse(Path(pipeline.__file__).read_text()) + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + assert node.func.attr not in { + "get", + "nested_mapping", + "sequence_of_mappings", + "first_tool_payload", + "tool_payload", + } + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + assert node.func.id not in { + "isinstance", + "getattr", + "hasattr", + "optional_int", + "optional_bool", + } + + +def test_submission_declaration_preserves_advertised_external_contract(): + from typing import get_type_hints + from openhcs.agent.capabilities import require_agent_type_contract + from openhcs.agent.services.execution_session_service import ExecutionSessionService + from openhcs.mcp.server import _mcp_tool_meta + + for capability in ( + agent_capabilities.submit_compile, + agent_capabilities.submit_pipeline_execution, + ): + assert ( + capability.output_contract.producer is ExecutionSessionService._submit_job + ) + assert ( + capability.output_contract.result_type + == get_type_hints(ExecutionSessionService._submit_job, include_extras=True)[ + "return" + ] + ) + assert capability.output_contract_types == (ExecutionJobRef, ExecutionJobStatus) + assert ( + require_agent_type_contract(capability.output_contract) is ExecutionJobRef + ) + assert _mcp_tool_meta(capability) == { + "openhcs/outputContract": "ExecutionJobRef" + } + + +def test_typed_diagnostics_do_not_reserialize_or_rescan_error_records(monkeypatch): + import openhcs.mcp.dev_client_core as core + + error = AgentError("nested_failure", "Retained typed diagnosis", "Retained hint") + response = batch( + agent_capabilities.validate_pipeline, + PipelineValidationResult( + schema_version=SCHEMA_VERSION, + valid=False, + pipeline_ref=PipelineRef("p", "openhcs://pipelines/p"), + errors=(error,), + ), + ) + + def forbidden_serialization(value): + raise AssertionError("Typed diagnostics must not round-trip through JSON") + + monkeypatch.setattr(core, "to_jsonable", forbidden_serialization) + assert response.has_errors() + assert response.results[0].agent_error_codes() == ("nested_failure",) + assert response.diagnostic_errors()[0] == error + assert "Retained hint" in dev_client.McpDevCommandSpec.for_name( + "draft-pipeline-step" + ).render_result(response, args_for("draft-pipeline-step", "function-1")) + + +@pytest.mark.parametrize("json_output", [False, True]) +def test_actual_cli_main_renders_without_runtime(monkeypatch, capsys, json_output): + response = batch( + agent_capabilities.inspect_pipeline_source_artifact_plan, plan_fixture() + ) + + async def controlled_entry(args): + assert args.command == "artifact-plan" + return response + + monkeypatch.setattr(dev_client, "_run_async", controlled_entry) + argv = [ + "artifact-plan", + "source-only-plate", + "--source-text", + "pipeline_steps = []", + ] + if json_output: + argv.append("--json") + assert dev_client.main(argv) == 0 + printed = capsys.readouterr().out + if json_output: + assert json.loads(printed) == to_jsonable(response) + else: + assert "Owned step" in printed and "new_component" in printed + + +def test_generic_call_cli_uses_nominal_typed_contract_without_json_roundtrip( + monkeypatch, capsys +): + import openhcs.serialization.json as serialization + + response = batch( + agent_capabilities.inspect_pipeline_source_artifact_plan, plan_fixture() + ) + + async def controlled_entry(args): + return response + + def forbidden_serialization(value): + raise AssertionError("Compact migrated call must consume the typed result") + + monkeypatch.setattr(dev_client, "_run_async", controlled_entry) + monkeypatch.setattr(serialization, "to_jsonable", forbidden_serialization) + assert ( + dev_client.main( + ["call", agent_capabilities.inspect_pipeline_source_artifact_plan.name] + ) + == 0 + ) + assert "Owned step" in capsys.readouterr().out + + +def test_generic_call_preserves_pending_workflow_native_receipt(monkeypatch, capsys): + from openhcs.agent.dto.ui_bridge import ( + UiActionIdentity, + UiActionInvokeResult, + UiMutationReceipt, + UiMutationRequestToken, + UiSelectedPlateWorkflowKind, + UiSelectedPlateWorkflowResult, + ) + + native = UiSelectedPlateWorkflowResult( + SCHEMA_VERSION, + UiSelectedPlateWorkflowKind("run_plate"), + UiActionInvokeResult( + SCHEMA_VERSION, + UiActionIdentity(widget_id="plate_manager", action_id="run_plate"), + "accepted", + UiMutationReceipt(UiMutationRequestToken(), accepted=True), + target_scope_ids=("scope-1",), + ), + ) + response = batch( + agent_capabilities.ui_selected_plate_workflow, + native, + ) + + async def controlled_entry(args): + return response + + monkeypatch.setattr(dev_client, "_run_async", controlled_entry) + assert ( + dev_client.main( + [ + "call", + agent_capabilities.ui_selected_plate_workflow.name, + "--json", + "--arguments", + '{"workflow":"run_plate"}', + ] + ) + == 0 + ) + rendered = capsys.readouterr().out + assert "run_plate" in rendered and "accepted" in rendered and "scope-1" in rendered + assert json.loads(rendered) == to_jsonable(response) + + +@pytest.mark.parametrize("malformed", [False, True]) +def test_persistent_client_execute_uses_real_command_boundary_without_runtime( + monkeypatch, malformed +): + from io import StringIO + + receipt = to_jsonable(plan_fixture()) + if malformed: + receipt["axis_count"] = True + peer = ControlledWireSession((receipt,)) + + def forbidden_raw_failure_scan(value): + raise AssertionError("Production execute must query typed diagnostics") + + monkeypatch.setattr(dev_client, "_command_failed", forbidden_raw_failure_scan) + client = dev_client.McpDevClient(server_stderr=StringIO()) + # Only the peer is controlled. No process/session is launched; execute, + # run_session, framing, decoding, rendering and exit status are production. + client._session = peer + client._session_started = True + try: + execution = client.execute( + [ + "artifact-plan", + "source-only-plate", + "--source-text", + "pipeline_steps = []", + ] + ) + finally: + client.close() + assert execution.returncode == int(malformed) + assert execution.payload["results"][0]["payloads"][0] == receipt + assert ( + "mcp_payload_invalid" if malformed else "Owned step" + ) in execution.rendered_output + assert len(peer.calls) == 1 diff --git a/tests/unit/agent/test_mcp_server.py b/tests/unit/agent/test_mcp_server.py index 516adb0339..c8eac2cad5 100644 --- a/tests/unit/agent/test_mcp_server.py +++ b/tests/unit/agent/test_mcp_server.py @@ -4668,6 +4668,13 @@ def test_mcp_dev_client_authoring_context_renders_bounded_content(): assert "..." in rendered +def _dev_client_server_fixture(): + from openhcs.mcp.dev_client_core import McpDevServerIdentity, McpDevServerSpec + from openhcs.serialization.json import to_jsonable + + return to_jsonable(McpDevServerIdentity.from_spec(McpDevServerSpec(sys.executable))) + + def test_mcp_dev_client_draft_pipeline_step_command_renders_composite_summary(): if importlib.util.find_spec("mcp") is None: return @@ -4690,6 +4697,7 @@ def test_mcp_dev_client_draft_pipeline_step_command_renders_composite_summary(): ) ) response = { + "server": _dev_client_server_fixture(), "errors": [], "results": [ { @@ -4707,7 +4715,9 @@ def test_mcp_dev_client_draft_pipeline_step_command_renders_composite_summary(): "mcp_error": False, "payloads": [ { + "schema_version": "openhcs.agent.v1", "pipeline_id": "pipeline-1", + "pipeline_config_id": "config-1", "steps": [ { "step_id": "step-1", @@ -4722,7 +4732,6 @@ def test_mcp_dev_client_draft_pipeline_step_command_renders_composite_summary(): ], } ], - "errors": [], } ], }, @@ -4731,7 +4740,12 @@ def test_mcp_dev_client_draft_pipeline_step_command_renders_composite_summary(): "mcp_error": False, "payloads": [ { + "schema_version": "openhcs.agent.v1", "valid": True, + "pipeline_ref": { + "pipeline_id": "pipeline-1", + "uri": "openhcs://pipelines/pipeline-1", + }, "warnings": [{"code": "note", "message": "Pipeline is small."}], "errors": [], } @@ -4742,9 +4756,9 @@ def test_mcp_dev_client_draft_pipeline_step_command_renders_composite_summary(): "mcp_error": False, "payloads": [ { + "schema_version": "openhcs.agent.v1", "title": "Pipeline", "source": "pipeline_steps = [\\n FunctionStep(...)\\n]\\n", - "errors": [], } ], }, @@ -4783,6 +4797,7 @@ def test_mcp_dev_client_draft_pipeline_step_suggests_missing_kwargs_repair(): ) ) response = { + "server": _dev_client_server_fixture(), "errors": [], "results": [ { @@ -4800,7 +4815,9 @@ def test_mcp_dev_client_draft_pipeline_step_suggests_missing_kwargs_repair(): "mcp_error": False, "payloads": [ { + "schema_version": "openhcs.agent.v1", "pipeline_id": "pipeline-1", + "pipeline_config_id": "config-1", "steps": [ { "step_id": "step-1", @@ -4816,7 +4833,6 @@ def test_mcp_dev_client_draft_pipeline_step_suggests_missing_kwargs_repair(): ], } ], - "errors": [], } ], }, @@ -4825,7 +4841,12 @@ def test_mcp_dev_client_draft_pipeline_step_suggests_missing_kwargs_repair(): "mcp_error": False, "payloads": [ { + "schema_version": "openhcs.agent.v1", "valid": False, + "pipeline_ref": { + "pipeline_id": "pipeline-1", + "uri": "openhcs://pipelines/pipeline-1", + }, "warnings": [], "errors": [ { @@ -4918,6 +4939,7 @@ def test_mcp_dev_client_artifact_plan_command_renders_compact_summary(): ) ) response = { + "server": _dev_client_server_fixture(), "errors": [], "results": [ { @@ -4925,6 +4947,7 @@ def test_mcp_dev_client_artifact_plan_command_renders_compact_summary(): "mcp_error": False, "payloads": [ { + "schema_version": "openhcs.agent.v1", "plate_path": "/tmp/example-plate", "axes": ["A01"], "axis_count": 1, @@ -4938,6 +4961,7 @@ def test_mcp_dev_client_artifact_plan_command_renders_compact_summary(): "files": [ { "virtual_path": "A01_s001_w1_z001_t001.tif", + "full_virtual_path": "/virtual/A01_s001_w1_z001_t001.tif", "source_path": "/tmp/source/A01_w1.tif", "source_metadata": { "well": "A01", @@ -4952,6 +4976,7 @@ def test_mcp_dev_client_artifact_plan_command_renders_compact_summary(): "step_index": 0, "step_name": "Count cells", "axis_id": "A01", + "output_dir": None, "execution_groups": [None], "artifact_inputs": [ { @@ -4985,7 +5010,7 @@ def test_mcp_dev_client_artifact_plan_command_renders_compact_summary(): "paths": [ { "group_key": None, - "base_path": ( + "shared_output_stem": ( "/tmp/example-plate_openhcs/" "images_results/" "A01_cell_counts_step0.roi.zip" @@ -5073,6 +5098,7 @@ def test_mcp_dev_client_artifact_plan_explains_empty_source_workspace(): ) ) response = { + "server": _dev_client_server_fixture(), "errors": [], "results": [ { @@ -5080,6 +5106,7 @@ def test_mcp_dev_client_artifact_plan_explains_empty_source_workspace(): "mcp_error": False, "payloads": [ { + "schema_version": "openhcs.agent.v1", "plate_path": "/tmp/example-plate", "axes": ["A01"], "axis_count": 1, @@ -5167,7 +5194,7 @@ async def fake_call_tool(session, call, timeout_seconds): "uri": "openhcs://execution/sessions/session-1", }, ), - ) + ).decoded_for_rendering() return dev_client.McpDevToolResult( tool=call.name, mcp_error=False, @@ -5176,13 +5203,14 @@ async def fake_call_tool(session, call, timeout_seconds): "schema_version": "openhcs.agent.v1", "session_id": "session-1", "job_id": "job-1", + "uri": "openhcs://execution/jobs/job-1", "kind": "execute", "status": "complete", "server_execution_id": "exec-1", "response": {"status": "complete", "completed": True}, }, ), - ) + ).decoded_for_rendering() monkeypatch.setattr(knowledge_pipeline, "call_mcp_tool", fake_call_tool) @@ -12139,6 +12167,17 @@ def test_mcp_dev_client_selected_workflow_poll_renders_compact_summary(): ) assert '- plate-a_openhcs: state=created, status="", terminal=' in rendered + # The production entrypoint now retains a batch. The pending workflow's + # distinct composite view must not be replaced by its primary tool view. + from python_introspect import dataclass_from_mapping + from openhcs.mcp.dev_client_core import McpDevToolBatchResponse + + response["server"] = _dev_client_server_fixture() + framed = dataclass_from_mapping(McpDevToolBatchResponse, response) + assert dev_client.McpDevCommandSpec.for_name("selected-workflow").render_result( + framed, args + ) == rendered + def test_mcp_dev_client_selected_workflow_poll_summarizes_rejection( monkeypatch,