From 7a76b57d3f1b53ac38a12d67ab54e4072eab95d2 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 29 Sep 2026 14:31:38 -0400 Subject: [PATCH 1/4] Repair retained decorator history and add bounded MCP memory receipts --- benchmark/mcp_memory_diagnostic.py | 322 ++++++++++++++++++ docs/validation/memory-session-20260929.md | 92 +++++ external/arraybridge | 2 +- .../test_durable_decorated_history.py | 96 ++++++ tests/unit/test_mcp_memory_diagnostic.py | 29 ++ 5 files changed, 540 insertions(+), 1 deletion(-) create mode 100644 benchmark/mcp_memory_diagnostic.py create mode 100644 docs/validation/memory-session-20260929.md create mode 100644 tests/unit/pyqt_gui/test_durable_decorated_history.py create mode 100644 tests/unit/test_mcp_memory_diagnostic.py diff --git a/benchmark/mcp_memory_diagnostic.py b/benchmark/mcp_memory_diagnostic.py new file mode 100644 index 0000000000..6a7c0e02b0 --- /dev/null +++ b/benchmark/mcp_memory_diagnostic.py @@ -0,0 +1,322 @@ +"""Bounded, provider-free retention measurements through the real MCP server. + +Run with ``python -m benchmark.mcp_memory_diagnostic --output receipt.json``. +Only this disposable diagnostic entrypoint adds the process-local sampling tool; +normal OpenHCS MCP capabilities, transport, context, and services are unchanged. +No execution servers, JVMs, viewers, or GPU workloads are started by the sequence. +""" + +from __future__ import annotations + +import argparse +import asyncio +import gc +import json +import os +import sys +import tempfile +import time +from collections.abc import Sequence +from dataclasses import asdict, dataclass +from pathlib import Path + +PROBE_TOOL = "openhcs_diagnostic_process_memory" +SCRATCH_ROOT = Path( + "/home/ts/.cache/agent-scratch/openhcs-issue-memory-session-20260929" +) + + +def require_ram_headroom() -> None: + """Refuse further requests under the batch's 8 GiB available-RAM floor.""" + for line in Path("/proc/meminfo").read_text().splitlines(): + if line.startswith("MemAvailable:"): + available_kib = int(line.split()[1]) + if available_kib < 8 * 1024**2: + raise RuntimeError(f"Available RAM below 8 GiB: {available_kib} KiB") + return + raise RuntimeError("Linux MemAvailable accounting is unavailable") + + +@dataclass(frozen=True) +class ProcessMemoryReceipt: + """Linux process accounting, not host swap or execution-tree accounting.""" + + pid: int + monotonic_seconds: float + rss_kib: int + pss_kib: int + private_clean_kib: int + private_dirty_kib: int + swap_kib: int + threads: int + collected_objects: int + imported_modules: int + native_library_paths: tuple[str, ...] + imported_frameworks: tuple[str, ...] + catalog_entries: int + custom_declarations: int + history_snapshots: int + imported_source_paths: tuple[str, ...] + + @classmethod + def capture(cls, *, collect: bool) -> ProcessMemoryReceipt: + collected = gc.collect() if collect else 0 + pid = os.getpid() + proc = Path("/proc") / str(pid) + accounting = {} + for line in (proc / "smaps_rollup").read_text().splitlines(): + name, separator, value = line.partition(":") + if separator: + accounting[name] = int(value.split()[0]) + libraries = tuple( + sorted( + { + line.split()[-1] + for line in (proc / "maps").read_text().splitlines() + if len(line.split()) >= 6 and ".so" in line.split()[-1] + } + ) + ) + modules = sys.modules + registry = modules.get( + "openhcs.processing.backends.lib_registry.registry_service" + ) + custom = modules.get("openhcs.processing.custom_functions.runtime_registry") + state = modules.get("objectstate.object_state_registry") + catalog = registry.RegistryService._metadata_cache if registry else None + return cls( + pid=pid, + monotonic_seconds=time.monotonic(), + rss_kib=accounting["Rss"], + pss_kib=accounting["Pss"], + private_clean_kib=accounting["Private_Clean"], + private_dirty_kib=accounting["Private_Dirty"], + swap_kib=accounting["Swap"], + threads=len(tuple((proc / "task").iterdir())), + collected_objects=collected, + imported_modules=len(modules), + native_library_paths=libraries, + imported_frameworks=tuple( + name + for name in ( + "numpy", + "scipy", + "torch", + "cupy", + "tensorflow", + "jax", + "jpype", + "scyjava", + ) + if name in modules + ), + catalog_entries=len(catalog) if catalog is not None else 0, + custom_declarations=len( + custom.CustomFunctionRuntimeRegistry._declarations_by_name + ) + if custom + else 0, + history_snapshots=len(state.ObjectStateRegistry._snapshots) if state else 0, + imported_source_paths=tuple( + modules[name].__file__ + for name in ("openhcs", "objectstate", "arraybridge", "pyqt_reactive") + if name in modules + ), + ) + + +def serve() -> None: + """Instrument the real server without substituting any application owner.""" + from openhcs.agent.capabilities import FullLocalCapabilitySurfaceProfile + from openhcs.mcp.server import build_server + from openhcs.mcp.stdio import McpStdioTransport + + with McpStdioTransport.reserve_process_stdio() as transport: + server = build_server( + capability_surface_profile=FullLocalCapabilitySurfaceProfile() + ) + + @server.tool(name=PROBE_TOOL) + def process_memory(collect: bool = False) -> dict: + """Sample this diagnostic MCP process, optionally after full Python GC.""" + return asdict(ProcessMemoryReceipt.capture(collect=collect)) + + transport.run(server) + + +def retained_slope_kib(samples: Sequence[dict], field: str) -> float: + """Least-squares per-round slope; it is a measurement, not a leak verdict.""" + if len(samples) < 2: + raise ValueError("A retention slope requires at least two repeated rounds.") + center = (len(samples) - 1) / 2 + mean = sum(sample[field] for sample in samples) / len(samples) + return sum( + (index - center) * (sample[field] - mean) + for index, sample in enumerate(samples) + ) / sum((index - center) ** 2 for index in range(len(samples))) + + +async def diagnose(rounds: int, timeout: float, output: Path) -> dict: + from openhcs.agent.capabilities import agent_capabilities + from openhcs.mcp.dev_client_core import ( + McpDevServerSpec, + McpDevStdioSession, + McpDevToolResult, + ) + + @dataclass(frozen=True, slots=True) + class DiagnosticServerSpec(McpDevServerSpec): + data_directory: str = "" + + def process_args(self) -> tuple[str, ...]: + return ("-m", "benchmark.mcp_memory_diagnostic", "--server") + + def environment(self) -> dict[str, str]: + return { + **McpDevServerSpec.environment(self), + "PYTHONPATH": os.environ["PYTHONPATH"], + "XDG_DATA_HOME": self.data_directory, + "XDG_CACHE_HOME": self.data_directory, + "XDG_CONFIG_HOME": self.data_directory, + } + + report = { + "scope": "one disposable real MCP process; no execution tree or host tmpfs attribution", + "rounds": rounds, + "events": [], + } + output.parent.mkdir(parents=True, exist_ok=True) + + def persist() -> None: + output.write_text(json.dumps(report, indent=2) + "\n") + + # stderr is bounded to this owned temporary stream; it is never copied into + # a shared environment or another process's log. + SCRATCH_ROOT.mkdir(parents=True, exist_ok=True) + with ( + tempfile.TemporaryDirectory(prefix="mcp-", dir=SCRATCH_ROOT) as scratch, + tempfile.TemporaryFile(mode="w+", encoding="utf-8", dir=scratch) as stderr, + ): + spec = DiagnosticServerSpec( + python_executable=sys.executable, data_directory=scratch + ) + try: + async with McpDevStdioSession(spec, stderr) as session: + await session.initialize(timeout_seconds=timeout) + + async def call(name: str, arguments: dict) -> dict: + require_ram_headroom() + result = await session.call_tool( + name, arguments, timeout_seconds=timeout + ) + if result.get("isError"): + raise RuntimeError( + f"Diagnostic MCP call failed: {name}: {result}" + ) + response = McpDevToolResult.from_payload(name, result) + if response.has_errors(): + raise RuntimeError( + f"Diagnostic MCP operation failed: {name}: {response.agent_error_codes()}" + ) + payload = response.payloads[0] + if not isinstance(payload, dict): + raise TypeError(f"MCP result is not an object: {name}") + if payload.get("status") in {"error", "failed"}: + raise RuntimeError( + f"Diagnostic MCP operation failed: {name}: {payload}" + ) + return payload + + health = await call(agent_capabilities.health_check.name, {}) + report["health"] = health + if ( + health["server_process_id"] != session.require_process().pid + or health["restart_required"] + ): + raise RuntimeError( + "Diagnostic process identity/staleness check failed." + ) + await call( + agent_capabilities.get_authoring_context.name, {"kind": "first_use"} + ) + + async def sample(label: str, collect: bool) -> dict: + receipt = await call(PROBE_TOOL, {"collect": collect}) + if receipt["rss_kib"] > 2 * 1024**2: + raise RuntimeError( + "Diagnostic MCP process exceeded its 2 GiB RSS budget" + ) + report["events"].append( + {"label": label, "after_gc": collect, **receipt} + ) + persist() + return receipt + + await sample("baseline", True) + # These are read-only discovery calls. The first round includes + # warm-up; subsequent rounds isolate repeated-request retention. + sequence = ( + (agent_capabilities.health_check.name, {}), + ( + agent_capabilities.get_authoring_context.name, + {"kind": "first_use"}, + ), + (agent_capabilities.search_capabilities.name, {"text": "pipeline"}), + ) + repeated = [] + for round_index in range(rounds + 1): + for name, arguments in sequence: + await sample(f"round-{round_index}/before/{name}", False) + await call(name, arguments) + await sample(f"round-{round_index}/after/{name}", False) + await sample(f"round-{round_index}/gc/{name}", True) + end = await sample(f"round-{round_index}/end", True) + if round_index: + repeated.append(end) + report["post_warmup_slopes_kib_per_round"] = { + field: retained_slope_kib(repeated, field) + for field in ("rss_kib", "pss_kib", "private_dirty_kib", "swap_kib") + } + report["interpretation"] = ( + "Slopes describe this bounded discovery sequence only. Growth is not proof of a leak; " + "flatness does not exonerate untested catalog, compile, custom-source, execution, JVM or GPU paths. " + "Compare import/library/cache counts and repeat longer before attributing retention." + ) + except BaseException as error: + report["failure"] = {"type": type(error).__name__, "message": str(error)} + stderr.seek(0, 2) + stderr.seek(max(0, stderr.tell() - 12000)) + report["server_stderr_tail"] = stderr.read() + persist() + raise + persist() + return report + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--server", action="store_true", help=argparse.SUPPRESS) + parser.add_argument("--rounds", type=int, default=4) + parser.add_argument("--timeout", type=float, default=30) + parser.add_argument("--output", type=Path, default=Path("mcp-memory-receipt.json")) + args = parser.parse_args() + if args.server: + serve() + return + if not 2 <= args.rounds <= 10 or not 0 < args.timeout <= 60: + parser.error("rounds must be 2..10 and timeout must be >0..60 seconds") + report = asyncio.run( + asyncio.wait_for(diagnose(args.rounds, args.timeout, args.output), timeout=180) + ) + print( + json.dumps( + { + "receipt": str(args.output), + "slopes": report["post_warmup_slopes_kib_per_round"], + } + ) + ) + + +if __name__ == "__main__": + main() diff --git a/docs/validation/memory-session-20260929.md b/docs/validation/memory-session-20260929.md new file mode 100644 index 0000000000..a35ba77021 --- /dev/null +++ b/docs/validation/memory-session-20260929.md @@ -0,0 +1,92 @@ +# Memory/session issue batch: #131 and #169 + +Implementation owner: Linnaeus. Integration and installed/live validation owner: +the OpenHCS coordinator. Worktree: `/home/ts/wt/openhcs-memory-session-20260929`. +No saved H001 files, Euler inputs/results, installed source, or shared main were +opened or modified. No GUI, JVM, GPU workload, provider or execution server was +started by this worker. + +## #169: declaration and runtime ownership + +An unpublished NumPy-decorated function reproduces `TypeError: cannot pickle +'_thread._local' object` in both `dill.dumps(function)` and a real ObjectState +history export. Its `__globals__` belongs to `arraybridge.decorators`, while its +declaration module belongs to the source execution namespace. Dill therefore +serializes decorator globals by value, including `_thread_gpu_contexts`. +Published importable functions and plain current global/UI/pipeline configuration +history do not reproduce the failure. This is a reproduction of the reported +failure mode, not reconstruction of the disposed H001 in-memory graph. + +The existing ArrayBridge `ThreadGPUContext` now owns its thread-local storage. +Decorator globals carry the importable runtime owner, not its runtime handle. +There is no history filtering, alternate registry, permissive pickle handler, +legacy reader, or change to ObjectState's durable document format. The obsolete +global storage and forwarding function are deleted. GPU invocation uses the +same context/stream behavior through `ThreadGPUContext.current()`. + +Regression coverage includes unpublished CPU/GPU-decorated callable persistence, +stable isolated per-thread contexts, and a continuous native document/history +save, fresh-registry load, historical undo and return to current declarations. +The retained runtime context is populated with an unpickleable thread-local +handle and must remain alive and identical. The native history test uses real +pipeline document, scope and ObjectState owners; it is not desktop live proof. + +The minimal repaired callable round trip executed successfully (14,424-byte +payload); its restored NumPy result was `[0, 1]` for thresholding `[1, 4]` at 3. +The paired upstream's three focused decorator/context tests passed (0.15s), +and changed-source Ruff/diff checks passed. The OpenHCS native regression and +real desktop journey are pending at this initial checkpoint. Paired upstream +draft: https://github.com/OpenHCSDev/ArrayBridge/pull/1. +H001 was already closed by explicit owner disposition; +there is no request or authorization to recover its lost history. + +## #131: bounded retention diagnostic + +`benchmark/mcp_memory_diagnostic.py` constructs the real full-surface OpenHCS MCP +server and reserves the actual stdio transport. One harness-only tool samples +that disposable process, optionally after `gc.collect()`. It does not replace +the UI, application, capability services or protocol path with mocks, and does +not add a diagnostic capability to production MCP. + +The receipt records PID, monotonic time, RSS, PSS, private clean/dirty, swap, +threads, imports/native libraries/frameworks, catalog/custom declarations and +ObjectState snapshots before/after every operation and after full GC. Repeated +read-only health/authoring/capability queries are measured after a separate +warm-up round. Per-round slopes describe this sequence only. Flat measurements +cannot prove catalog/compile/custom-source/execution/GPU/JVM paths leak-free; +positive slopes alone cannot establish causation or a leak. + +Run only with the validation slot and at least 8 GiB available RAM: + +```sh +flock -n /home/ts/wt/openhcs-issue-batch-20260929/validation.lock \ + /home/ts/code/projects/openhcs/.venv/bin/python -m benchmark.mcp_memory_diagnostic \ + --rounds 4 --output docs/validation/mcp-memory-receipt-20260929.json +``` + +Set `PYTHONPATH` to this worktree and its recorded submodule `src` directories, +and verify imported paths before running. The diagnostic owns temporary XDG +data/cache/config and stderr under +`/home/ts/.cache/agent-scratch/openhcs-issue-memory-session-20260929/mcp-*`. +Each run removes that disposable directory on success or failure, retaining the +JSON receipt and bounded stderr tail at the requested output path. The harness +enforces an 8 GiB available-RAM floor, 2 GiB server RSS budget and 180-second +overall timeout. +The diagnostic has no access to another server's process lifecycle; transport +teardown affects only its own fresh subprocess. No services or tmpfs data are +removed to influence results. + +## Boundaries and remaining scope + +- Source review/AST coverage is focused on history serialization and the new + diagnostic. No full NRA scan or global architecture-clean claim is made. +- ArrayBridge source-path verification and the minimal reproducer/repair ran in + the existing environment. Initial combined test collection exposed duplicate + `tests.conftest` package names across repositories; run the two repositories + separately. No test assertion was weakened. +- Resource guard reported about 12-13 GiB available RAM but exits 2 for existing + host swap pressure. Large validation is deferred, and the nonblocking batch + lock has been busy during attempted focused runs. No unrelated cleanup occurs. +- Coordinator must serialize a new desktop capture/restore and verify the paired + installed entrypoint after review/integration. No worker merge/install and no + extra display use are authorized. Neither issue is marked Closes at checkpoint. diff --git a/external/arraybridge b/external/arraybridge index fddd9857bc..06837ec267 160000 --- a/external/arraybridge +++ b/external/arraybridge @@ -1 +1 @@ -Subproject commit fddd9857bcd21468d20bc8547a769e0c6c9f1906 +Subproject commit 06837ec267e3ca0b1734461d09919ff65cb1b1fe diff --git a/tests/unit/pyqt_gui/test_durable_decorated_history.py b/tests/unit/pyqt_gui/test_durable_decorated_history.py new file mode 100644 index 0000000000..3370b2bb4c --- /dev/null +++ b/tests/unit/pyqt_gui/test_durable_decorated_history.py @@ -0,0 +1,96 @@ +"""Continuous native declaration/history restart with a retained runtime wrapper. + +This exercises the real document and ObjectState owners, without a GUI or data +execution. Live desktop capture/restore remains a separate acceptance boundary. +""" + +import threading +from pathlib import Path + +import numpy as np +from arraybridge.decorators import ThreadGPUContext +from arraybridge.types import MemoryType +from objectstate import ObjectStateRegistry +from pyqt_reactive.services.scope_token_service import ScopeTokenService + +from openhcs.core.config import PipelineConfig +from openhcs.core.memory import numpy +from openhcs.core.pipeline_document import PipelineDocumentAuthority +from openhcs.core.steps.function_step import FunctionStep +from openhcs.processing.backends.processors.numpy_processor import tophat +from openhcs.pyqt_gui.services.pipeline_object_state_binding import ( + PipelineObjectStateBinding, +) + +SCOPE = "/isolated/durable-decorated-history" + + +def test_declarations_and_retained_callable_history_survive_native_restart( + tmp_path: Path, +): + ObjectStateRegistry.clear() + ScopeTokenService.clear_scope(SCOPE) + runtime = ThreadGPUContext.current() + runtime_key = (MemoryType.CUPY, -2) + runtime_handle = threading.local() + runtime._streams[runtime_key] = runtime_handle + try: + # An old callable no longer exported by its original package is retained + # by undo. Its wrapper's globals own thread-local GPU context storage. + @numpy + def old(image, scale=3): + return image * scale + + with ObjectStateRegistry.atomic("retained custom declaration"): + PipelineObjectStateBinding.update_plate_steps( + SCOPE, [FunctionStep(name="old custom", func=(old, {"scale": 7}))] + ) + PipelineObjectStateBinding.commit_plate_state(SCOPE) + baseline_id = ObjectStateRegistry.get_branch_history()[-1].id + with ObjectStateRegistry.atomic("current native declaration"): + PipelineObjectStateBinding.update_plate_steps( + SCOPE, + [FunctionStep(name="current", func=(tophat, {"selem_radius": 17}))], + ) + PipelineObjectStateBinding.commit_plate_state(SCOPE) + head_id = ObjectStateRegistry.get_branch_history()[-1].id + snapshot_ids = tuple(s.id for s in ObjectStateRegistry.get_branch_history()) + source = PipelineDocumentAuthority.render( + PipelineDocumentAuthority.from_values( + pipeline_config=PipelineConfig(), + pipeline_steps=PipelineObjectStateBinding.steps_for_plate(SCOPE), + ) + ) + declarations = tmp_path / "current.py" + declarations.write_text(source) + history = tmp_path / "history.objectstate" + ObjectStateRegistry.save_history_to_file(str(history)) + + ObjectStateRegistry.clear() + ScopeTokenService.clear_scope(SCOPE) + parsed = PipelineDocumentAuthority.from_source(declarations.read_text()) + PipelineObjectStateBinding.update_plate_steps(SCOPE, parsed.pipeline_steps) + ObjectStateRegistry.load_history_from_file(str(history)) + assert ( + tuple(s.id for s in ObjectStateRegistry.get_branch_history()) + == snapshot_ids + ) + current = PipelineObjectStateBinding.steps_for_plate(SCOPE)[0] + assert current.name == "current" + assert current.func == (tophat, {"selem_radius": 17}) + assert ObjectStateRegistry.time_travel_to_snapshot(baseline_id) + restored = PipelineObjectStateBinding.steps_for_plate(SCOPE)[0] + historical, kwargs = restored.func + assert restored.name == "old custom" + assert kwargs == {"scale": 7} + np.testing.assert_array_equal( + historical(np.array([1, 2], dtype=np.uint16), **kwargs), [7, 14] + ) + assert ObjectStateRegistry.time_travel_to_snapshot(head_id) + assert PipelineObjectStateBinding.steps_for_plate(SCOPE)[0].func == current.func + assert ThreadGPUContext.current() is runtime + assert runtime._streams[runtime_key] is runtime_handle + finally: + runtime._streams.pop(runtime_key, None) + ObjectStateRegistry.clear() + ScopeTokenService.clear_scope(SCOPE) diff --git a/tests/unit/test_mcp_memory_diagnostic.py b/tests/unit/test_mcp_memory_diagnostic.py new file mode 100644 index 0000000000..f82541ce0c --- /dev/null +++ b/tests/unit/test_mcp_memory_diagnostic.py @@ -0,0 +1,29 @@ +"""Process-only memory receipts and measured (not causal) retention slopes.""" + +import os + +import pytest + +from benchmark.mcp_memory_diagnostic import ProcessMemoryReceipt, retained_slope_kib + + +def test_process_receipt_uses_current_process_and_reports_gc_boundary(): + before = ProcessMemoryReceipt.capture(collect=False) + after = ProcessMemoryReceipt.capture(collect=True) + assert before.pid == after.pid == os.getpid() + assert before.collected_objects == 0 + assert after.collected_objects >= 0 + assert after.pss_kib > 0 + assert after.rss_kib >= after.pss_kib + assert after.private_dirty_kib >= 0 + assert after.swap_kib >= 0 + assert after.threads >= 1 + assert after.monotonic_seconds >= before.monotonic_seconds + + +def test_slope_distinguishes_warmup_from_repeated_request_growth(): + repeated = [{"rss": 20}, {"rss": 23}, {"rss": 26}] + assert retained_slope_kib(repeated, "rss") == 3 + assert retained_slope_kib([{"rss": 30}] * 3, "rss") == 0 + with pytest.raises(ValueError, match="at least two"): + retained_slope_kib([{"rss": 30}], "rss") From dd5f177f0c381ebac1d470f9a05f8423c2c40dfb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 29 Sep 2026 14:52:09 -0400 Subject: [PATCH 2/4] Keep MCP memory samples and metric selection with typed owners --- benchmark/mcp_memory_diagnostic.py | 343 ++++++++++++------ docs/validation/memory-session-20260929.md | 96 ++++- scripts/check_mcp_memory_ownership.py | 77 ++++ .../test_durable_decorated_history.py | 6 +- tests/unit/test_mcp_memory_diagnostic.py | 69 +++- 5 files changed, 465 insertions(+), 126 deletions(-) create mode 100644 scripts/check_mcp_memory_ownership.py diff --git a/benchmark/mcp_memory_diagnostic.py b/benchmark/mcp_memory_diagnostic.py index 6a7c0e02b0..2416cf7b2d 100644 --- a/benchmark/mcp_memory_diagnostic.py +++ b/benchmark/mcp_memory_diagnostic.py @@ -16,13 +16,26 @@ import sys import tempfile import time -from collections.abc import Sequence -from dataclasses import asdict, dataclass +from collections.abc import Callable, Mapping, Sequence +from dataclasses import asdict, dataclass, field from pathlib import Path +from typing import TYPE_CHECKING, Annotated, get_args, get_type_hints + +from python_introspect import dataclass_from_mapping + +if TYPE_CHECKING: + from openhcs.agent.dto.mcp import McpServerHealthResult + from openhcs.mcp.dev_client_core import ( + McpDevStdioSession, + McpDevToolCall, + McpDevToolResult, + ) PROBE_TOOL = "openhcs_diagnostic_process_memory" -SCRATCH_ROOT = Path( - "/home/ts/.cache/agent-scratch/openhcs-issue-memory-session-20260929" +SCRATCH_ROOT = ( + Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) + / "agent-scratch" + / "openhcs-mcp-memory-diagnostic" ) @@ -37,17 +50,36 @@ def require_ram_headroom() -> None: raise RuntimeError("Linux MemAvailable accounting is unavailable") +@dataclass(frozen=True) +class RetentionMetric: + """A sample-field declaration owning its projection and slope calculation.""" + + project: Callable[[ProcessMemoryReceipt], int] + + def slope(self, samples: Sequence[ProcessMemoryReceipt]) -> float: + if len(samples) < 2: + raise ValueError("A retention slope requires at least two repeated rounds.") + values = tuple(self.project(sample) for sample in samples) + center = (len(values) - 1) / 2 + mean = sum(values) / len(values) + return sum( + (index - center) * (value - mean) for index, value in enumerate(values) + ) / sum((index - center) ** 2 for index in range(len(values))) + + @dataclass(frozen=True) class ProcessMemoryReceipt: """Linux process accounting, not host swap or execution-tree accounting.""" pid: int monotonic_seconds: float - rss_kib: int - pss_kib: int + rss_kib: Annotated[int, RetentionMetric(lambda sample: sample.rss_kib)] + pss_kib: Annotated[int, RetentionMetric(lambda sample: sample.pss_kib)] private_clean_kib: int - private_dirty_kib: int - swap_kib: int + private_dirty_kib: Annotated[ + int, RetentionMetric(lambda sample: sample.private_dirty_kib) + ] + swap_kib: Annotated[int, RetentionMetric(lambda sample: sample.swap_kib)] threads: int collected_objects: int imported_modules: int @@ -58,8 +90,27 @@ class ProcessMemoryReceipt: history_snapshots: int imported_source_paths: tuple[str, ...] + @classmethod + def from_payload(cls, payload: Mapping[str, object]) -> ProcessMemoryReceipt: + """Decode the MCP sample once using the declared field codec.""" + return dataclass_from_mapping(cls, payload) + + @classmethod + def retention_slopes( + cls, samples: Sequence[ProcessMemoryReceipt] + ) -> dict[str, float]: + """Derive eligibility and projections from the sample field declarations.""" + return { + name: metric.slope(samples) + for name, annotation in get_type_hints(cls, include_extras=True).items() + for metric in get_args(annotation)[1:] + if isinstance(metric, RetentionMetric) + } + @classmethod def capture(cls, *, collect: bool) -> ProcessMemoryReceipt: + from arraybridge import MemoryType + collected = gc.collect() if collect else 0 pid = os.getpid() proc = Path("/proc") / str(pid) @@ -83,7 +134,9 @@ def capture(cls, *, collect: bool) -> ProcessMemoryReceipt: ) custom = modules.get("openhcs.processing.custom_functions.runtime_registry") state = modules.get("objectstate.object_state_registry") - catalog = registry.RegistryService._metadata_cache if registry else None + catalog_entries = ( + len(registry.RegistryService.cached_metadata_snapshot()) if registry else 0 + ) return cls( pid=pid, monotonic_seconds=time.monotonic(), @@ -97,22 +150,13 @@ def capture(cls, *, collect: bool) -> ProcessMemoryReceipt: imported_modules=len(modules), native_library_paths=libraries, imported_frameworks=tuple( - name - for name in ( - "numpy", - "scipy", - "torch", - "cupy", - "tensorflow", - "jax", - "jpype", - "scyjava", - ) - if name in modules + memory_type.import_name + for memory_type in MemoryType + if memory_type.loaded_module() is not None ), - catalog_entries=len(catalog) if catalog is not None else 0, + catalog_entries=catalog_entries, custom_declarations=len( - custom.CustomFunctionRuntimeRegistry._declarations_by_name + custom.CustomFunctionRuntimeRegistry.metadata_by_name() ) if custom else 0, @@ -125,6 +169,83 @@ def capture(cls, *, collect: bool) -> ProcessMemoryReceipt: ) +@dataclass(frozen=True) +class MemorySampleEvent: + label: str + after_gc: bool + receipt: ProcessMemoryReceipt + + +@dataclass(frozen=True) +class DiagnosticFailure: + exception_type: str + message: str + + +@dataclass +class MemoryDiagnosticReport: + rounds: int + request_sequence: tuple[McpDevToolCall, ...] + events: list[MemorySampleEvent] = field(default_factory=list) + health: McpServerHealthResult | None = None + post_warmup_slopes_kib_per_round: dict[str, float] = field(default_factory=dict) + failure: DiagnosticFailure | None = None + server_stderr_tail: str | None = None + scope: str = ( + "one disposable real MCP process; no execution tree or host tmpfs attribution" + ) + interpretation: str = ( + "Slopes describe the recorded bounded request sequence only. Growth is not proof of a leak; " + "flatness does not exonerate untested catalog, compile, custom-source, execution, JVM or GPU paths. " + "Compare import/library/cache counts and repeat longer before attributing retention." + ) + + def write(self, output: Path) -> None: + output.write_text(json.dumps(asdict(self), indent=2) + "\n") + + def record( + self, label: str, after_gc: bool, receipt: ProcessMemoryReceipt, output: Path + ) -> None: + self.events.append(MemorySampleEvent(label, after_gc, receipt)) + self.write(output) + + +@dataclass +class MemoryDiagnosticMcpClient: + """One MCP boundary: wire/error owners in, typed memory/health owners out.""" + + session: McpDevStdioSession + timeout_seconds: float + + async def call(self, name: str, arguments: dict) -> McpDevToolResult: + from openhcs.mcp.dev_client_core import McpDevToolResult + + require_ram_headroom() + wire = await self.session.call_tool( + name, arguments, timeout_seconds=self.timeout_seconds + ) + result = McpDevToolResult.from_payload(name, wire) + if result.has_errors(): + raise RuntimeError( + f"Diagnostic MCP operation failed: {name}: {result.agent_error_codes()}" + ) + return result + + async def health(self) -> McpServerHealthResult: + from openhcs.agent.capabilities import agent_capabilities + from openhcs.agent.dto.mcp import McpServerHealthResult + + result = await self.call(agent_capabilities.health_check.name, {}) + return dataclass_from_mapping(McpServerHealthResult, result.payloads[0]) + + async def sample(self, *, collect: bool) -> ProcessMemoryReceipt: + result = await self.call(PROBE_TOOL, {"collect": collect}) + receipt = ProcessMemoryReceipt.from_payload(result.payloads[0]) + if receipt.rss_kib > 2 * 1024**2: + raise RuntimeError("Diagnostic MCP process exceeded its 2 GiB RSS budget") + return receipt + + def serve() -> None: """Instrument the real server without substituting any application owner.""" from openhcs.agent.capabilities import FullLocalCapabilitySurfaceProfile @@ -144,24 +265,53 @@ def process_memory(collect: bool = False) -> dict: transport.run(server) -def retained_slope_kib(samples: Sequence[dict], field: str) -> float: - """Least-squares per-round slope; it is a measurement, not a leak verdict.""" - if len(samples) < 2: - raise ValueError("A retention slope requires at least two repeated rounds.") - center = (len(samples) - 1) / 2 - mean = sum(sample[field] for sample in samples) / len(samples) - return sum( - (index - center) * (sample[field] - mean) - for index, sample in enumerate(samples) - ) / sum((index - center) ** 2 for index in range(len(samples))) +def read_request_sequence(path: Path) -> tuple[McpDevToolCall, ...]: + """Decode bounded read-only calls through the existing MCP request owner.""" + from openhcs.agent.capabilities import get_agent_capability + from openhcs.mcp.dev_client_core import McpDevToolCall + + if path.stat().st_size > 64 * 1024: + raise ValueError("Diagnostic request sequence exceeds 64 KiB") + payload = json.loads(path.read_text()) + if not isinstance(payload, list) or not 1 <= len(payload) <= 16: + raise ValueError("Diagnostic sequence must contain 1..16 MCP request objects") + calls = tuple(dataclass_from_mapping(McpDevToolCall, item) for item in payload) + for call in calls: + capability = get_agent_capability(call.name) + if capability.side_effects: + raise ValueError( + f"Diagnostic sequence requires read-only calls: {call.name}" + ) + return calls -async def diagnose(rounds: int, timeout: float, output: Path) -> dict: +async def diagnose( + rounds: int, + timeout: float, + output: Path, + *, + scratch_root: Path = SCRATCH_ROOT, + sequence_path: Path | None = None, +) -> MemoryDiagnosticReport: from openhcs.agent.capabilities import agent_capabilities from openhcs.mcp.dev_client_core import ( McpDevServerSpec, McpDevStdioSession, - McpDevToolResult, + McpDevToolCall, + ) + + sequence = ( + read_request_sequence(sequence_path) + if sequence_path is not None + else ( + McpDevToolCall(agent_capabilities.health_check.name, {}), + McpDevToolCall( + agent_capabilities.get_authoring_context.name, {"kind": "first_use"} + ), + McpDevToolCall( + agent_capabilities.search_capabilities.name, {"text": "pipeline"} + ), + ) ) @dataclass(frozen=True, slots=True) @@ -180,21 +330,15 @@ def environment(self) -> dict[str, str]: "XDG_CONFIG_HOME": self.data_directory, } - report = { - "scope": "one disposable real MCP process; no execution tree or host tmpfs attribution", - "rounds": rounds, - "events": [], - } + report = MemoryDiagnosticReport(rounds=rounds, request_sequence=sequence) output.parent.mkdir(parents=True, exist_ok=True) - def persist() -> None: - output.write_text(json.dumps(report, indent=2) + "\n") - # stderr is bounded to this owned temporary stream; it is never copied into # a shared environment or another process's log. - SCRATCH_ROOT.mkdir(parents=True, exist_ok=True) + require_ram_headroom() + scratch_root.mkdir(parents=True, exist_ok=True) with ( - tempfile.TemporaryDirectory(prefix="mcp-", dir=SCRATCH_ROOT) as scratch, + tempfile.TemporaryDirectory(prefix="mcp-", dir=scratch_root) as scratch, tempfile.TemporaryFile(mode="w+", encoding="utf-8", dir=scratch) as stderr, ): spec = DiagnosticServerSpec( @@ -203,93 +347,53 @@ def persist() -> None: try: async with McpDevStdioSession(spec, stderr) as session: await session.initialize(timeout_seconds=timeout) - - async def call(name: str, arguments: dict) -> dict: - require_ram_headroom() - result = await session.call_tool( - name, arguments, timeout_seconds=timeout - ) - if result.get("isError"): - raise RuntimeError( - f"Diagnostic MCP call failed: {name}: {result}" - ) - response = McpDevToolResult.from_payload(name, result) - if response.has_errors(): - raise RuntimeError( - f"Diagnostic MCP operation failed: {name}: {response.agent_error_codes()}" - ) - payload = response.payloads[0] - if not isinstance(payload, dict): - raise TypeError(f"MCP result is not an object: {name}") - if payload.get("status") in {"error", "failed"}: - raise RuntimeError( - f"Diagnostic MCP operation failed: {name}: {payload}" - ) - return payload - - health = await call(agent_capabilities.health_check.name, {}) - report["health"] = health + client = MemoryDiagnosticMcpClient(session, timeout) + health = await client.health() + report.health = health if ( - health["server_process_id"] != session.require_process().pid - or health["restart_required"] + health.server_process_id != session.require_process().pid + or health.restart_required ): raise RuntimeError( "Diagnostic process identity/staleness check failed." ) - await call( + await client.call( agent_capabilities.get_authoring_context.name, {"kind": "first_use"} ) - async def sample(label: str, collect: bool) -> dict: - receipt = await call(PROBE_TOOL, {"collect": collect}) - if receipt["rss_kib"] > 2 * 1024**2: - raise RuntimeError( - "Diagnostic MCP process exceeded its 2 GiB RSS budget" - ) - report["events"].append( - {"label": label, "after_gc": collect, **receipt} - ) - persist() + async def sample(label: str, collect: bool) -> ProcessMemoryReceipt: + receipt = await client.sample(collect=collect) + report.record(label, collect, receipt, output) return receipt await sample("baseline", True) # These are read-only discovery calls. The first round includes # warm-up; subsequent rounds isolate repeated-request retention. - sequence = ( - (agent_capabilities.health_check.name, {}), - ( - agent_capabilities.get_authoring_context.name, - {"kind": "first_use"}, - ), - (agent_capabilities.search_capabilities.name, {"text": "pipeline"}), - ) - repeated = [] + for request in sequence: + request.require_surface_profile(spec.surface_profile) + repeated: list[ProcessMemoryReceipt] = [] for round_index in range(rounds + 1): - for name, arguments in sequence: - await sample(f"round-{round_index}/before/{name}", False) - await call(name, arguments) - await sample(f"round-{round_index}/after/{name}", False) - await sample(f"round-{round_index}/gc/{name}", True) + for request in sequence: + await sample( + f"round-{round_index}/before/{request.name}", False + ) + await client.call(request.name, request.arguments) + await sample(f"round-{round_index}/after/{request.name}", False) + await sample(f"round-{round_index}/gc/{request.name}", True) end = await sample(f"round-{round_index}/end", True) if round_index: repeated.append(end) - report["post_warmup_slopes_kib_per_round"] = { - field: retained_slope_kib(repeated, field) - for field in ("rss_kib", "pss_kib", "private_dirty_kib", "swap_kib") - } - report["interpretation"] = ( - "Slopes describe this bounded discovery sequence only. Growth is not proof of a leak; " - "flatness does not exonerate untested catalog, compile, custom-source, execution, JVM or GPU paths. " - "Compare import/library/cache counts and repeat longer before attributing retention." + report.post_warmup_slopes_kib_per_round = ( + ProcessMemoryReceipt.retention_slopes(repeated) ) except BaseException as error: - report["failure"] = {"type": type(error).__name__, "message": str(error)} + report.failure = DiagnosticFailure(type(error).__name__, str(error)) stderr.seek(0, 2) stderr.seek(max(0, stderr.tell() - 12000)) - report["server_stderr_tail"] = stderr.read() - persist() + report.server_stderr_tail = stderr.read() + report.write(output) raise - persist() + report.write(output) return report @@ -299,6 +403,12 @@ def main() -> None: parser.add_argument("--rounds", type=int, default=4) parser.add_argument("--timeout", type=float, default=30) parser.add_argument("--output", type=Path, default=Path("mcp-memory-receipt.json")) + parser.add_argument("--scratch-root", type=Path, default=SCRATCH_ROOT) + parser.add_argument( + "--sequence-json", + type=Path, + help="JSON list of 1..16 existing read-only MCP calls (name/arguments)", + ) args = parser.parse_args() if args.server: serve() @@ -306,13 +416,22 @@ def main() -> None: if not 2 <= args.rounds <= 10 or not 0 < args.timeout <= 60: parser.error("rounds must be 2..10 and timeout must be >0..60 seconds") report = asyncio.run( - asyncio.wait_for(diagnose(args.rounds, args.timeout, args.output), timeout=180) + asyncio.wait_for( + diagnose( + args.rounds, + args.timeout, + args.output, + scratch_root=args.scratch_root, + sequence_path=args.sequence_json, + ), + timeout=180, + ) ) print( json.dumps( { "receipt": str(args.output), - "slopes": report["post_warmup_slopes_kib_per_round"], + "slopes": report.post_warmup_slopes_kib_per_round, } ) ) diff --git a/docs/validation/memory-session-20260929.md b/docs/validation/memory-session-20260929.md index a35ba77021..fa6ac0258e 100644 --- a/docs/validation/memory-session-20260929.md +++ b/docs/validation/memory-session-20260929.md @@ -34,8 +34,17 @@ pipeline document, scope and ObjectState owners; it is not desktop live proof. The minimal repaired callable round trip executed successfully (14,424-byte payload); its restored NumPy result was `[0, 1]` for thresholding `[1, 4]` at 3. The paired upstream's three focused decorator/context tests passed (0.15s), -and changed-source Ruff/diff checks passed. The OpenHCS native regression and -real desktop journey are pending at this initial checkpoint. Paired upstream +and changed-source Ruff/diff checks passed. OpenHCS source validation passed: + +- 12 tests (15.61s): the native durability regression, memory diagnostic tests + and existing `test_pipeline_history_occurrence_restart_regression.py`. +- 6 tests (3.16s), after the ownership revision: five diagnostic regressions + and `test_durable_decorated_history.py`. + +The first native test assertion omitted normalized processor default kwargs; +it now compares the complete declaration captured before persistence against +the restored declaration. Snapshot identities and historical execution remain +asserted. The real desktop journey is still pending. Paired upstream draft: https://github.com/OpenHCSDev/ArrayBridge/pull/1. H001 was already closed by explicit owner disposition; there is no request or authorization to recover its lost history. @@ -52,7 +61,11 @@ The receipt records PID, monotonic time, RSS, PSS, private clean/dirty, swap, threads, imports/native libraries/frameworks, catalog/custom declarations and ObjectState snapshots before/after every operation and after full GC. Repeated read-only health/authoring/capability queries are measured after a separate -warm-up round. Per-round slopes describe this sequence only. Flat measurements +warm-up round. Optional `--sequence-json` decodes 1..16 requests through the +existing `McpDevToolCall` owner and rejects capabilities declaring side effects; +it does not introduce a second capability registry. The JSON report retains the +exact request sequence and typed sample events (`events[].receipt`). +Per-round slopes describe this sequence only. Flat measurements cannot prove catalog/compile/custom-source/execution/GPU/JVM paths leak-free; positive slopes alone cannot establish causation or a leak. @@ -61,7 +74,9 @@ Run only with the validation slot and at least 8 GiB available RAM: ```sh flock -n /home/ts/wt/openhcs-issue-batch-20260929/validation.lock \ /home/ts/code/projects/openhcs/.venv/bin/python -m benchmark.mcp_memory_diagnostic \ - --rounds 4 --output docs/validation/mcp-memory-receipt-20260929.json + --rounds 4 \ + --scratch-root /home/ts/.cache/agent-scratch/openhcs-issue-memory-session-20260929 \ + --output docs/validation/mcp-memory-receipt-20260929.json ``` Set `PYTHONPATH` to this worktree and its recorded submodule `src` directories, @@ -76,6 +91,66 @@ The diagnostic has no access to another server's process lifecycle; transport teardown affects only its own fresh subprocess. No services or tmpfs data are removed to influence results. +No real MCP measurement has run yet. There is no leak or no-leak finding at this +checkpoint, and the source tests do not substitute for that run. + +## Catalog-pattern review and new-case ownership check + +The comparison baseline is OpenHCS checkpoint `7a76b57d3`; review covers the +changed diagnostic, native history regression and paired ArrayBridge change. +This is focused source/AST/catalog review, not a full NRA proof or repository +audit. The current archived refactor-audit catalog supplies the pattern IDs. + +- **BOUND-2 / BOUND-1:** baseline `diagnose.sample` returns a raw mapping even + though `ProcessMemoryReceipt` already declares its exact shape. Health reads + also bypass `McpServerHealthResult`. `MemoryDiagnosticMcpClient.sample()` now + decodes once with the existing `python_introspect.dataclass_from_mapping` + mechanism into `ProcessMemoryReceipt`; health uses its existing DTO. Samples, + events and reports stay typed. Linux `/proc` field parsing and the inherited + child environment remain legitimate external boundaries, not violations. +- **BOUND-2:** baseline `call()` separately classifies wire `isError` and payload + `.get("status")`, alongside `McpDevToolResult.has_errors()`. Source review of + that owner confirms it already owns transport and structured agent errors. + The diagnostic now calls only that owner; it does not restate a status roster + or add another failure contract. +- **MEMB-2:** baseline slope selection spells a tuple of four field names and + `retained_slope_kib(Sequence[dict], field)` reads each by string. Eligibility + and projection now live in `RetentionMetric` metadata on receipt fields; + `ProcessMemoryReceipt.retention_slopes()` derives selection from those + declarations. The standalone string-dispatched slope helper is deleted. +- **MEMB-1:** the hand-written framework roster mixed array frameworks with + SciPy/JVM packages and omitted an existing ArrayBridge member. Array-framework + observations now derive from `MemoryType` and its `loaded_module()` behavior, + without importing an optional framework. All native library paths and total + module counts remain recorded; this is not comprehensive JVM/GPU attribution. +- **Runtime versus durable owner:** paired ArrayBridge code moves the handle + onto the existing importable `ThreadGPUContext`, deleting the module-global + thread-local and forwarding function. No second store, pickle fallback or + history exclusion is added. The native regression exercises retained callable + history while an unpickleable runtime handle remains alive. + +`scripts/check_mcp_memory_ownership.py` is a bounded AST regression guard, not a +general detector. Against `7a76b57d3` it reports five literal mapping reads plus +one duplicate status classification (six findings); against the revision it +passes. The baseline's variable-key slope helper was additionally reviewed +directly, not counted by this guard. + +The new-case test adds slope eligibility to the previously unmarked +`private_clean_kib` field on a receipt subclass. It decodes the same JSON shape +and obtains the new slope without changing the MCP client, metric selection, +report or a registry. Only the field declaration changes. Separate tests reject +unknown fields, boolean-as-integer input and mutating request sequences. + +Lightweight repeatable audit commands: + +```sh +/home/ts/code/projects/openhcs/.venv/bin/python scripts/check_mcp_memory_ownership.py +git show 7a76b57d3:benchmark/mcp_memory_diagnostic.py | \ + /home/ts/code/projects/openhcs/.venv/bin/python scripts/check_mcp_memory_ownership.py --stdin +``` + +The second command intentionally exits 1 and reports the original violations. + ## Boundaries and remaining scope - Source review/AST coverage is focused on history serialization and the new @@ -84,9 +159,16 @@ removed to influence results. the existing environment. Initial combined test collection exposed duplicate `tests.conftest` package names across repositories; run the two repositories separately. No test assertion was weakened. -- Resource guard reported about 12-13 GiB available RAM but exits 2 for existing - host swap pressure. Large validation is deferred, and the nonblocking batch - lock has been busy during attempted focused runs. No unrelated cleanup occurs. +- The focused runs above used the shared nonblocking validation lock and have + finished; no heavy worker run or runtime handle is being restarted. The latest + coordinator resource checkpoint reports 14.8 GiB available RAM, zero PSI and + 11.5 GiB historical swap. New MCP/JVM/GUI/heavy runs are deferred until Euler + releases validation for H003. No unrelated cleanup occurs. +- Owned disposable pytest artifacts were under this worktree's + `.validation/typed-native-check-2` (60 KiB of generated declarations/history + and sequence fixtures, not user sessions). They were removed after recording + the result; tests regenerate them. The diagnostic has not yet created a + scratch process directory. No large output is retained. - Coordinator must serialize a new desktop capture/restore and verify the paired installed entrypoint after review/integration. No worker merge/install and no extra display use are authorized. Neither issue is marked Closes at checkpoint. diff --git a/scripts/check_mcp_memory_ownership.py b/scripts/check_mcp_memory_ownership.py new file mode 100644 index 0000000000..0dbf076c8b --- /dev/null +++ b/scripts/check_mcp_memory_ownership.py @@ -0,0 +1,77 @@ +"""Focused BOUND-1/BOUND-2 guard for the MCP memory diagnostic. + +Literal mapping reads remain legitimate in Linux /proc parsing and the child +environment boundary. Other diagnostic consumers must use decoded owners. +``git show REV:benchmark/mcp_memory_diagnostic.py | python ... --stdin`` proves +the guard against the original violation without modifying a checkout. +""" + +from __future__ import annotations + +import argparse +import ast +import sys +from pathlib import Path + + +class MemoryOwnershipGuard(ast.NodeVisitor): + def __init__(self) -> None: + self.scope: list[str] = [] + self.violations: list[str] = [] + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self.scope.append(node.name) + self.generic_visit(node) + self.scope.pop() + + def visit_FunctionDef(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None: + self.scope.append(node.name) + self.generic_visit(node) + self.scope.pop() + + visit_AsyncFunctionDef = visit_FunctionDef + + def visit_Subscript(self, node: ast.Subscript) -> None: + if ( + isinstance(node.ctx, ast.Load) + and isinstance(node.slice, ast.Constant) + and isinstance(node.slice.value, str) + ): + owner = ".".join(self.scope) + if owner not in { + "ProcessMemoryReceipt.capture", + "diagnose.DiagnosticServerSpec.environment", + }: + self.violations.append( + f"BOUND-1/BOUND-2:{node.lineno}: raw field {node.slice.value!r} in {owner}" + ) + self.generic_visit(node) + + def visit_Call(self, node: ast.Call) -> None: + match node.func, node.args: + case ast.Attribute(attr="get"), [ast.Constant(value="status"), *_]: + self.violations.append( + f"BOUND-2:{node.lineno}: status classification bypasses McpDevToolResult" + ) + self.generic_visit(node) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--stdin", action="store_true") + parser.add_argument( + "--path", type=Path, default=Path("benchmark/mcp_memory_diagnostic.py") + ) + args = parser.parse_args() + source = sys.stdin.read() if args.stdin else args.path.read_text() + guard = MemoryOwnershipGuard() + guard.visit(ast.parse(source)) + for violation in guard.violations: + print(violation) + if guard.violations: + raise SystemExit(1) + print("Focused MCP memory ownership guard passed") + + +if __name__ == "__main__": + main() diff --git a/tests/unit/pyqt_gui/test_durable_decorated_history.py b/tests/unit/pyqt_gui/test_durable_decorated_history.py index 3370b2bb4c..3fe6de1fc3 100644 --- a/tests/unit/pyqt_gui/test_durable_decorated_history.py +++ b/tests/unit/pyqt_gui/test_durable_decorated_history.py @@ -47,6 +47,7 @@ def old(image, scale=3): ) PipelineObjectStateBinding.commit_plate_state(SCOPE) baseline_id = ObjectStateRegistry.get_branch_history()[-1].id + expected_baseline = PipelineObjectStateBinding.steps_for_plate(SCOPE)[0] with ObjectStateRegistry.atomic("current native declaration"): PipelineObjectStateBinding.update_plate_steps( SCOPE, @@ -54,6 +55,7 @@ def old(image, scale=3): ) PipelineObjectStateBinding.commit_plate_state(SCOPE) head_id = ObjectStateRegistry.get_branch_history()[-1].id + expected_current = PipelineObjectStateBinding.steps_for_plate(SCOPE)[0] snapshot_ids = tuple(s.id for s in ObjectStateRegistry.get_branch_history()) source = PipelineDocumentAuthority.render( PipelineDocumentAuthority.from_values( @@ -77,12 +79,12 @@ def old(image, scale=3): ) current = PipelineObjectStateBinding.steps_for_plate(SCOPE)[0] assert current.name == "current" - assert current.func == (tophat, {"selem_radius": 17}) + assert current.func == expected_current.func assert ObjectStateRegistry.time_travel_to_snapshot(baseline_id) restored = PipelineObjectStateBinding.steps_for_plate(SCOPE)[0] historical, kwargs = restored.func assert restored.name == "old custom" - assert kwargs == {"scale": 7} + assert kwargs == expected_baseline.func[1] np.testing.assert_array_equal( historical(np.array([1, 2], dtype=np.uint16), **kwargs), [7, 14] ) diff --git a/tests/unit/test_mcp_memory_diagnostic.py b/tests/unit/test_mcp_memory_diagnostic.py index f82541ce0c..fc18f2c50e 100644 --- a/tests/unit/test_mcp_memory_diagnostic.py +++ b/tests/unit/test_mcp_memory_diagnostic.py @@ -1,10 +1,17 @@ """Process-only memory receipts and measured (not causal) retention slopes.""" +import json import os +from dataclasses import asdict, dataclass, replace +from typing import Annotated import pytest -from benchmark.mcp_memory_diagnostic import ProcessMemoryReceipt, retained_slope_kib +from benchmark.mcp_memory_diagnostic import ( + ProcessMemoryReceipt, + RetentionMetric, + read_request_sequence, +) def test_process_receipt_uses_current_process_and_reports_gc_boundary(): @@ -22,8 +29,60 @@ def test_process_receipt_uses_current_process_and_reports_gc_boundary(): def test_slope_distinguishes_warmup_from_repeated_request_growth(): - repeated = [{"rss": 20}, {"rss": 23}, {"rss": 26}] - assert retained_slope_kib(repeated, "rss") == 3 - assert retained_slope_kib([{"rss": 30}] * 3, "rss") == 0 + baseline = ProcessMemoryReceipt.capture(collect=False) + repeated = [replace(baseline, rss_kib=value) for value in (20, 23, 26)] + assert ProcessMemoryReceipt.retention_slopes(repeated)["rss_kib"] == 3 + assert ProcessMemoryReceipt.retention_slopes([baseline] * 3)["rss_kib"] == 0 with pytest.raises(ValueError, match="at least two"): - retained_slope_kib([{"rss": 30}], "rss") + ProcessMemoryReceipt.retention_slopes([baseline]) + + +def test_memory_payload_decodes_once_and_rejects_shape_and_type_errors(): + receipt = ProcessMemoryReceipt.capture(collect=False) + wire = json.loads(json.dumps(asdict(receipt))) + restored = ProcessMemoryReceipt.from_payload(wire) + assert restored == receipt + assert isinstance(restored.native_library_paths, tuple) + with pytest.raises(ValueError, match="undeclared"): + ProcessMemoryReceipt.from_payload({**wire, "extra_metric": 0}) + wire["rss_kib"] = True + with pytest.raises(TypeError, match="rss_kib"): + ProcessMemoryReceipt.from_payload(wire) + + +def test_new_metric_requires_only_its_receipt_field_declaration(): + @dataclass(frozen=True) + class ExtendedReceipt(ProcessMemoryReceipt): + private_clean_kib: Annotated[ + int, RetentionMetric(lambda sample: sample.private_clean_kib) + ] + + baseline = ProcessMemoryReceipt.capture(collect=False) + first = ExtendedReceipt.from_payload(asdict(baseline)) + repeated = [replace(first, private_clean_kib=value) for value in (10, 15, 20)] + slopes = ExtendedReceipt.retention_slopes(repeated) + assert slopes["private_clean_kib"] == 5 + assert "private_clean_kib" not in ProcessMemoryReceipt.retention_slopes( + [baseline] * 3 + ) + + +def test_request_sequence_decodes_existing_owner_and_rejects_mutations(tmp_path): + from openhcs.agent.capabilities import agent_capabilities + from openhcs.mcp.dev_client_core import McpDevToolCall + + path = tmp_path / "sequence.json" + path.write_text( + json.dumps([{"name": agent_capabilities.health_check.name, "arguments": {}}]) + ) + assert read_request_sequence(path) == ( + McpDevToolCall(agent_capabilities.health_check.name, {}), + ) + path.write_text( + json.dumps([{"name": agent_capabilities.create_config.name, "arguments": {}}]) + ) + with pytest.raises(ValueError, match="read-only"): + read_request_sequence(path) + path.write_text("[]") + with pytest.raises(ValueError, match="1..16"): + read_request_sequence(path) From d69d65adbd56633e1d1fd2a3b65678728cd6ada8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 29 Sep 2026 16:17:11 -0400 Subject: [PATCH 3/4] Honor diagnostic capability and runtime import authorities --- docs/validation/memory-session-20260929.md | 90 ++++++++++++++-- .../mcp/memory_diagnostic.py | 99 ++++++++++++----- openhcs/mcp/memory_diagnostic_launch.py | 25 +++++ scripts/check_mcp_memory_ownership.py | 59 +++++++--- tests/unit/test_mcp_memory_diagnostic.py | 102 +++++++++++++++++- 5 files changed, 329 insertions(+), 46 deletions(-) rename benchmark/mcp_memory_diagnostic.py => openhcs/mcp/memory_diagnostic.py (84%) create mode 100644 openhcs/mcp/memory_diagnostic_launch.py diff --git a/docs/validation/memory-session-20260929.md b/docs/validation/memory-session-20260929.md index fa6ac0258e..cc804ea4bc 100644 --- a/docs/validation/memory-session-20260929.md +++ b/docs/validation/memory-session-20260929.md @@ -51,7 +51,7 @@ there is no request or authorization to recover its lost history. ## #131: bounded retention diagnostic -`benchmark/mcp_memory_diagnostic.py` constructs the real full-surface OpenHCS MCP +`openhcs/mcp/memory_diagnostic.py` constructs the real full-surface OpenHCS MCP server and reserves the actual stdio transport. One harness-only tool samples that disposable process, optionally after `gc.collect()`. It does not replace the UI, application, capability services or protocol path with mocks, and does @@ -62,7 +62,7 @@ threads, imports/native libraries/frameworks, catalog/custom declarations and ObjectState snapshots before/after every operation and after full GC. Repeated read-only health/authoring/capability queries are measured after a separate warm-up round. Optional `--sequence-json` decodes 1..16 requests through the -existing `McpDevToolCall` owner and rejects capabilities declaring side effects; +existing `McpDevToolCall` owner and admits only `AgentCapabilitySpec.read_only`; it does not introduce a second capability registry. The JSON report retains the exact request sequence and typed sample events (`events[].receipt`). Per-round slopes describe this sequence only. Flat measurements @@ -73,7 +73,7 @@ Run only with the validation slot and at least 8 GiB available RAM: ```sh flock -n /home/ts/wt/openhcs-issue-batch-20260929/validation.lock \ - /home/ts/code/projects/openhcs/.venv/bin/python -m benchmark.mcp_memory_diagnostic \ + /home/ts/code/projects/openhcs/.venv/bin/python -m openhcs.mcp.memory_diagnostic \ --rounds 4 \ --scratch-root /home/ts/.cache/agent-scratch/openhcs-issue-memory-session-20260929 \ --output docs/validation/mcp-memory-receipt-20260929.json @@ -151,6 +151,82 @@ git show 7a76b57d3:benchmark/mcp_memory_diagnostic.py | \ The second command intentionally exits 1 and reports the original violations. +## Independent review correction: admission and source authority + +Review: https://github.com/OpenHCSDev/openhcs/pull/208#issuecomment-5897134270, +against `dd5f177f0c381ebac1d470f9a05f8423c2c40dfb`. Latest main +`283b21275553c54261cf9aeedb7374c113d07f2b` (including #212) was fetched and +merged normally as `32decce9af2b7b4f51aa3dc8de39b4e0e58c5694` before correction. + +- **BOUND-2:** diagnostic admission now queries `capability.read_only`, not + either of its component fields. The new-case test changes only an existing + typed capability declaration to `mutating=True, side_effects=()` and proves + rejection without a consumer edit or registry. Integrated artifact-plan + inspection is also rejected using its actual current declaration from #212. +- **IMPL-13 / BOUND-2:** `DiagnosticServerSpec` inherits the exact-root launch + projection from `McpDevServerSpec.process_args()`, which already delegates to + `OpenHCSRuntimeImportAuthority`. It adds only the instrumentation flag and + owned XDG isolation. Bare `-m` child launch and ambient PYTHONPATH copying are + deleted. The entrypoint moved under the supported OpenHCS module namespace; + the old benchmark module is deleted, with no alias or second launcher. +- Observed diagnostic/OpenHCS paths are one `DiagnosticSourceIdentity` in every + typed memory receipt. The MCP client checks both against its parent authority, + independently of the existing health PID/staleness checks. This receipt records + observations, not a second import-root authority or source registry. +- `--source-identity` reports those same source observations and exits before + importing the application client or constructing a server. Client launch + declarations live in `memory_diagnostic_launch.py`, so lightweight provenance + inspection does not import that client/scientific dependency graph. A new + diagnostic subclass extends the existing module/argument/environment hooks, + not a copied import bootstrap or a changed shared launch owner. + +Focused regression command, using the existing Python 3.12.3 venv with explicit +worktree/recorded-submodule source paths verified first: + +```sh +PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 OPENHCS_CPU_ONLY=true OPENHCS_HEADLESS=true \ + /home/ts/code/projects/openhcs/.venv/bin/python -m pytest \ + --noconftest -c /dev/null -p no:cacheprovider \ + --basetemp=/home/ts/wt/openhcs-memory-session-20260929/.validation/review-5897134270 \ + tests/unit/test_mcp_memory_diagnostic.py tests/unit/test_runtime_import_authority.py -q +``` + +Result: **15 passed in 4.34s**. This excludes application cleanup/Qt fixtures and +plugin startup; it does not replace full application acceptance. Two tests launch +the actual diagnostic entrypoint with normal server-spec arguments plus +provenance-only mode, from a competing checkout cwd: first with PYTHONPATH absent, +then with a hostile competing PYTHONPATH. Both diagnostic and OpenHCS paths must +equal this selected source tree. Separate tests reject either foreign receipt +path. Existing import-authority tests also pass. The focused AST guard now catches +all three returned old mechanisms at `dd5f177f` and passes both current modules; +changed-source Ruff correctness and diff checks pass. + +Real source CLI check, from the selected worktree with no PYTHONPATH: + +```sh +env -u PYTHONPATH OPENHCS_CPU_ONLY=true OPENHCS_HEADLESS=true \ + PYTHONDONTWRITEBYTECODE=1 /home/ts/code/projects/openhcs/.venv/bin/python \ + -m openhcs.mcp.memory_diagnostic --source-identity +``` + +It reports this worktree's `openhcs/mcp/memory_diagnostic.py` and +`openhcs/__init__.py`; `/usr/bin/time -v` records 0.16s, 53,500 KiB peak RSS, +zero swaps and exit 0. No MCP, JVM, GUI or fitting/execution workload started. +The preliminary client-import path check loaded NumPy/CuPy modules (232,720 KiB +RSS) but did not execute GPU work; the provenance entrypoint's application-client +import was subsequently removed via the explicit launch-module boundary above. + +Before validation the guard reported 11.8 GiB available RAM and a warning for +11.9 GiB historical swap (exit 2); no heavy validation was started. Confucius owns +the shared live slot and frozen installed harness. Owned generated sequence and +competing-package fixtures under `.validation/review-5897134270` were 164 KiB; +they are disposable and removed after retaining this receipt. No installed +package, skill, authoring process or scientific input/output was changed. + +These checks close the returned **source/launch defects**, not #131's real MCP +retention measurement or #169's installed continuous desktop/history journey. +Both implementation PRs remain draft pending those serialized live gates. + ## Boundaries and remaining scope - Source review/AST coverage is focused on history serialization and the new @@ -160,10 +236,10 @@ The second command intentionally exits 1 and reports the original violations. `tests.conftest` package names across repositories; run the two repositories separately. No test assertion was weakened. - The focused runs above used the shared nonblocking validation lock and have - finished; no heavy worker run or runtime handle is being restarted. The latest - coordinator resource checkpoint reports 14.8 GiB available RAM, zero PSI and - 11.5 GiB historical swap. New MCP/JVM/GUI/heavy runs are deferred until Euler - releases validation for H003. No unrelated cleanup occurs. + finished; no heavy worker run or runtime handle is being restarted. The older + Euler resource receipt is historical; current restrictions and the latest + resource snapshot are recorded above. Confucius owns the live slot and frozen + installation; no new MCP/JVM/GUI/heavy run starts here. No unrelated cleanup occurs. - Owned disposable pytest artifacts were under this worktree's `.validation/typed-native-check-2` (60 KiB of generated declarations/history and sequence fixtures, not user sessions). They were removed after recording diff --git a/benchmark/mcp_memory_diagnostic.py b/openhcs/mcp/memory_diagnostic.py similarity index 84% rename from benchmark/mcp_memory_diagnostic.py rename to openhcs/mcp/memory_diagnostic.py index 2416cf7b2d..d51f3d8ccb 100644 --- a/benchmark/mcp_memory_diagnostic.py +++ b/openhcs/mcp/memory_diagnostic.py @@ -1,6 +1,6 @@ """Bounded, provider-free retention measurements through the real MCP server. -Run with ``python -m benchmark.mcp_memory_diagnostic --output receipt.json``. +Run with ``python -m openhcs.mcp.memory_diagnostic --output receipt.json``. Only this disposable diagnostic entrypoint adds the process-local sampling tool; normal OpenHCS MCP capabilities, transport, context, and services are unchanged. No execution servers, JVMs, viewers, or GPU workloads are started by the sequence. @@ -23,7 +23,11 @@ from python_introspect import dataclass_from_mapping +import openhcs +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority + if TYPE_CHECKING: + from openhcs.agent.capabilities import LocalCapabilitySurfaceProfile from openhcs.agent.dto.mcp import McpServerHealthResult from openhcs.mcp.dev_client_core import ( McpDevStdioSession, @@ -32,6 +36,7 @@ ) PROBE_TOOL = "openhcs_diagnostic_process_memory" +DIAGNOSTIC_MODULE = "openhcs.mcp.memory_diagnostic" SCRATCH_ROOT = ( Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "agent-scratch" @@ -39,6 +44,35 @@ ) +@dataclass(frozen=True) +class DiagnosticSourceIdentity: + """Observed child paths, checked against the parent's import authority.""" + + diagnostic_source_path: str + openhcs_source_path: str + + @classmethod + def capture(cls) -> DiagnosticSourceIdentity: + return cls( + diagnostic_source_path=str(Path(__file__).resolve()), + openhcs_source_path=str(Path(openhcs.__file__).resolve()), + ) + + def require_authority(self, authority: OpenHCSRuntimeImportAuthority) -> None: + expected_diagnostic = authority.import_root.joinpath( + *DIAGNOSTIC_MODULE.split(".") + ).with_suffix(".py") + expected_package = ( + authority.import_root / authority.package_name / "__init__.py" + ) + if Path(self.diagnostic_source_path).resolve() != expected_diagnostic: + raise RuntimeError( + "Diagnostic child imported a different diagnostic source" + ) + if Path(self.openhcs_source_path).resolve() != expected_package: + raise RuntimeError("Diagnostic child imported a different OpenHCS source") + + def require_ram_headroom() -> None: """Refuse further requests under the batch's 8 GiB available-RAM floor.""" for line in Path("/proc/meminfo").read_text().splitlines(): @@ -89,6 +123,7 @@ class ProcessMemoryReceipt: custom_declarations: int history_snapshots: int imported_source_paths: tuple[str, ...] + source_identity: DiagnosticSourceIdentity @classmethod def from_payload(cls, payload: Mapping[str, object]) -> ProcessMemoryReceipt: @@ -166,6 +201,7 @@ def capture(cls, *, collect: bool) -> ProcessMemoryReceipt: for name in ("openhcs", "objectstate", "arraybridge", "pyqt_reactive") if name in modules ), + source_identity=DiagnosticSourceIdentity.capture(), ) @@ -241,21 +277,21 @@ async def health(self) -> McpServerHealthResult: async def sample(self, *, collect: bool) -> ProcessMemoryReceipt: result = await self.call(PROBE_TOOL, {"collect": collect}) receipt = ProcessMemoryReceipt.from_payload(result.payloads[0]) + receipt.source_identity.require_authority( + OpenHCSRuntimeImportAuthority.current() + ) if receipt.rss_kib > 2 * 1024**2: raise RuntimeError("Diagnostic MCP process exceeded its 2 GiB RSS budget") return receipt -def serve() -> None: +def serve(surface_profile: LocalCapabilitySurfaceProfile) -> None: """Instrument the real server without substituting any application owner.""" - from openhcs.agent.capabilities import FullLocalCapabilitySurfaceProfile from openhcs.mcp.server import build_server from openhcs.mcp.stdio import McpStdioTransport with McpStdioTransport.reserve_process_stdio() as transport: - server = build_server( - capability_surface_profile=FullLocalCapabilitySurfaceProfile() - ) + server = build_server(capability_surface_profile=surface_profile) @server.tool(name=PROBE_TOOL) def process_memory(collect: bool = False) -> dict: @@ -278,7 +314,7 @@ def read_request_sequence(path: Path) -> tuple[McpDevToolCall, ...]: calls = tuple(dataclass_from_mapping(McpDevToolCall, item) for item in payload) for call in calls: capability = get_agent_capability(call.name) - if capability.side_effects: + if not capability.read_only: raise ValueError( f"Diagnostic sequence requires read-only calls: {call.name}" ) @@ -290,15 +326,16 @@ async def diagnose( timeout: float, output: Path, *, + surface_profile: LocalCapabilitySurfaceProfile, scratch_root: Path = SCRATCH_ROOT, sequence_path: Path | None = None, ) -> MemoryDiagnosticReport: from openhcs.agent.capabilities import agent_capabilities from openhcs.mcp.dev_client_core import ( - McpDevServerSpec, McpDevStdioSession, McpDevToolCall, ) + from openhcs.mcp.memory_diagnostic_launch import DiagnosticServerSpec sequence = ( read_request_sequence(sequence_path) @@ -314,22 +351,6 @@ async def diagnose( ) ) - @dataclass(frozen=True, slots=True) - class DiagnosticServerSpec(McpDevServerSpec): - data_directory: str = "" - - def process_args(self) -> tuple[str, ...]: - return ("-m", "benchmark.mcp_memory_diagnostic", "--server") - - def environment(self) -> dict[str, str]: - return { - **McpDevServerSpec.environment(self), - "PYTHONPATH": os.environ["PYTHONPATH"], - "XDG_DATA_HOME": self.data_directory, - "XDG_CACHE_HOME": self.data_directory, - "XDG_CONFIG_HOME": self.data_directory, - } - report = MemoryDiagnosticReport(rounds=rounds, request_sequence=sequence) output.parent.mkdir(parents=True, exist_ok=True) @@ -342,7 +363,9 @@ def environment(self) -> dict[str, str]: tempfile.TemporaryFile(mode="w+", encoding="utf-8", dir=scratch) as stderr, ): spec = DiagnosticServerSpec( - python_executable=sys.executable, data_directory=scratch + python_executable=sys.executable, + data_directory=scratch, + surface_profile=surface_profile, ) try: async with McpDevStdioSession(spec, stderr) as session: @@ -398,8 +421,18 @@ async def sample(label: str, collect: bool) -> ProcessMemoryReceipt: def main() -> None: + # Provenance inspection uses the same source owner without importing the + # application client or its scientific dependency graph. parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--server", action="store_true", help=argparse.SUPPRESS) + parser.add_argument( + "--source-identity", + action="store_true", + help="Print loaded diagnostic/OpenHCS paths and exit without starting MCP", + ) + parser.add_argument( + "--surface", help="Existing capability surface used by the diagnostic server" + ) parser.add_argument("--rounds", type=int, default=4) parser.add_argument("--timeout", type=float, default=30) parser.add_argument("--output", type=Path, default=Path("mcp-memory-receipt.json")) @@ -410,8 +443,21 @@ def main() -> None: help="JSON list of 1..16 existing read-only MCP calls (name/arguments)", ) args = parser.parse_args() + if args.source_identity: + print(json.dumps(asdict(DiagnosticSourceIdentity.capture()))) + return + from openhcs.agent.capabilities import ( + FullLocalCapabilitySurfaceProfile, + LocalCapabilitySurfaceProfile, + ) + + surface_profile = ( + FullLocalCapabilitySurfaceProfile() + if args.surface is None + else LocalCapabilitySurfaceProfile.for_name(args.surface) + ) if args.server: - serve() + serve(surface_profile) return if not 2 <= args.rounds <= 10 or not 0 < args.timeout <= 60: parser.error("rounds must be 2..10 and timeout must be >0..60 seconds") @@ -421,6 +467,7 @@ def main() -> None: args.rounds, args.timeout, args.output, + surface_profile=surface_profile, scratch_root=args.scratch_root, sequence_path=args.sequence_json, ), diff --git a/openhcs/mcp/memory_diagnostic_launch.py b/openhcs/mcp/memory_diagnostic_launch.py new file mode 100644 index 0000000000..e10fde93ee --- /dev/null +++ b/openhcs/mcp/memory_diagnostic_launch.py @@ -0,0 +1,25 @@ +"""Diagnostic client launch projection; no alternative import authority.""" + +from dataclasses import dataclass + +from openhcs.mcp.dev_client_core import McpDevServerSpec +from openhcs.mcp.memory_diagnostic import DIAGNOSTIC_MODULE + + +@dataclass(frozen=True, slots=True) +class DiagnosticServerSpec(McpDevServerSpec): + """Instrumented server using the normal exact-root launch projection.""" + + module_name: str = DIAGNOSTIC_MODULE + data_directory: str = "" + + def process_args(self) -> tuple[str, ...]: + return (*McpDevServerSpec.process_args(self), "--server") + + def environment(self) -> dict[str, str]: + return { + **McpDevServerSpec.environment(self), + "XDG_DATA_HOME": self.data_directory, + "XDG_CACHE_HOME": self.data_directory, + "XDG_CONFIG_HOME": self.data_directory, + } diff --git a/scripts/check_mcp_memory_ownership.py b/scripts/check_mcp_memory_ownership.py index 0dbf076c8b..3bba3192c9 100644 --- a/scripts/check_mcp_memory_ownership.py +++ b/scripts/check_mcp_memory_ownership.py @@ -1,7 +1,7 @@ """Focused BOUND-1/BOUND-2 guard for the MCP memory diagnostic. -Literal mapping reads remain legitimate in Linux /proc parsing and the child -environment boundary. Other diagnostic consumers must use decoded owners. +Literal mapping reads remain legitimate in Linux /proc parsing. Other +diagnostic consumers must use decoded owners and existing launch/admission authorities. ``git show REV:benchmark/mcp_memory_diagnostic.py | python ... --stdin`` proves the guard against the original violation without modifying a checkout. """ @@ -38,15 +38,32 @@ def visit_Subscript(self, node: ast.Subscript) -> None: and isinstance(node.slice.value, str) ): owner = ".".join(self.scope) - if owner not in { - "ProcessMemoryReceipt.capture", - "diagnose.DiagnosticServerSpec.environment", - }: + if owner != "ProcessMemoryReceipt.capture": self.violations.append( f"BOUND-1/BOUND-2:{node.lineno}: raw field {node.slice.value!r} in {owner}" ) self.generic_visit(node) + def visit_Attribute(self, node: ast.Attribute) -> None: + if ".".join(self.scope) == "read_request_sequence" and node.attr in { + "mutating", + "side_effects", + }: + self.violations.append( + f"BOUND-2:{node.lineno}: read-only admission bypasses AgentCapabilitySpec.read_only" + ) + self.generic_visit(node) + + def visit_Return(self, node: ast.Return) -> None: + if self.scope[-2:] == ["DiagnosticServerSpec", "process_args"] and any( + isinstance(item, ast.Constant) and item.value == "-m" + for item in ast.walk(node) + ): + self.violations.append( + f"IMPL-13:{node.lineno}: bare module launch bypasses source authority" + ) + self.generic_visit(node) + def visit_Call(self, node: ast.Call) -> None: match node.func, node.args: case ast.Attribute(attr="get"), [ast.Constant(value="status"), *_]: @@ -60,15 +77,33 @@ def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--stdin", action="store_true") parser.add_argument( - "--path", type=Path, default=Path("benchmark/mcp_memory_diagnostic.py") + "--path", + type=Path, + action="append", ) args = parser.parse_args() - source = sys.stdin.read() if args.stdin else args.path.read_text() - guard = MemoryOwnershipGuard() - guard.visit(ast.parse(source)) - for violation in guard.violations: + sources = ( + [sys.stdin.read()] + if args.stdin + else [ + path.read_text() + for path in ( + args.path + or [ + Path("openhcs/mcp/memory_diagnostic.py"), + Path("openhcs/mcp/memory_diagnostic_launch.py"), + ] + ) + ] + ) + violations = [] + for source in sources: + guard = MemoryOwnershipGuard() + guard.visit(ast.parse(source)) + violations.extend(guard.violations) + for violation in violations: print(violation) - if guard.violations: + if violations: raise SystemExit(1) print("Focused MCP memory ownership guard passed") diff --git a/tests/unit/test_mcp_memory_diagnostic.py b/tests/unit/test_mcp_memory_diagnostic.py index fc18f2c50e..851a746845 100644 --- a/tests/unit/test_mcp_memory_diagnostic.py +++ b/tests/unit/test_mcp_memory_diagnostic.py @@ -2,16 +2,25 @@ import json import os +import subprocess +import sys from dataclasses import asdict, dataclass, replace +from pathlib import Path from typing import Annotated import pytest -from benchmark.mcp_memory_diagnostic import ( +from python_introspect import dataclass_from_mapping + +import openhcs +from openhcs.mcp.memory_diagnostic import ( + DiagnosticSourceIdentity, ProcessMemoryReceipt, RetentionMetric, read_request_sequence, ) +from openhcs.mcp.memory_diagnostic_launch import DiagnosticServerSpec +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority def test_process_receipt_uses_current_process_and_reports_gc_boundary(): @@ -86,3 +95,94 @@ def test_request_sequence_decodes_existing_owner_and_rejects_mutations(tmp_path) path.write_text("[]") with pytest.raises(ValueError, match="1..16"): read_request_sequence(path) + + +def test_mutating_declaration_without_side_effects_is_rejected(tmp_path, monkeypatch): + from openhcs.agent import capabilities + + capability = replace( + capabilities.get_agent_capability( + capabilities.agent_capabilities.health_check.name + ), + mutating=True, + side_effects=(), + ) + assert not capability.read_only + monkeypatch.setattr(capabilities, "get_agent_capability", lambda name: capability) + path = tmp_path / "sequence.json" + path.write_text(json.dumps([{"name": capability.name, "arguments": {}}])) + with pytest.raises(ValueError, match="read-only"): + read_request_sequence(path) + + +def test_integrated_artifact_plan_is_not_admitted_as_read_only(tmp_path): + from openhcs.agent.capabilities import agent_capabilities, get_agent_capability + + capability = get_agent_capability( + agent_capabilities.inspect_pipeline_source_artifact_plan.name + ) + assert capability.mutating + assert not capability.read_only + path = tmp_path / "sequence.json" + path.write_text(json.dumps([{"name": capability.name, "arguments": {}}])) + with pytest.raises(ValueError, match="read-only"): + read_request_sequence(path) + + +@pytest.mark.parametrize("ambient_path", [None, "competing"]) +def test_diagnostic_child_retains_selected_source_from_competing_cwd( + tmp_path, monkeypatch, ambient_path +): + # Actual source entrypoint and launch owner, but provenance-only mode: no + # MCP server, GUI, scientific framework, or execution runtime is started. + competing = tmp_path / "competing" + package = competing / "openhcs" / "mcp" + package.mkdir(parents=True) + (package.parent / "__init__.py").write_text( + "raise RuntimeError('foreign OpenHCS')\n" + ) + (package / "__init__.py").write_text("") + (package / "memory_diagnostic.py").write_text( + "raise RuntimeError('foreign diagnostic')\n" + ) + monkeypatch.delenv("PYTHONPATH", raising=False) + if ambient_path is not None: + monkeypatch.setenv("PYTHONPATH", str(competing)) + spec = DiagnosticServerSpec( + python_executable=sys.executable, data_directory=str(tmp_path) + ) + child_environment = spec.environment() + assert "PYTHONPATH" not in child_environment + # Even a hostile ambient path must not override the selected source root. + if ambient_path is not None: + child_environment["PYTHONPATH"] = str(competing) + completed = subprocess.run( + [spec.python_executable, *spec.process_args(), "--source-identity"], + cwd=competing, + env=child_environment, + capture_output=True, + text=True, + check=True, + timeout=10, + ) + identity = dataclass_from_mapping( + DiagnosticSourceIdentity, json.loads(completed.stdout) + ) + authority = OpenHCSRuntimeImportAuthority.current() + identity.require_authority(authority) + assert ( + Path(identity.diagnostic_source_path) + == authority.import_root / "openhcs/mcp/memory_diagnostic.py" + ) + assert Path(identity.openhcs_source_path) == Path(openhcs.__file__).resolve() + + +@pytest.mark.parametrize( + "path_field", ["diagnostic_source_path", "openhcs_source_path"] +) +def test_source_receipt_rejects_either_foreign_source(path_field, tmp_path): + identity = replace( + DiagnosticSourceIdentity.capture(), **{path_field: str(tmp_path / "foreign.py")} + ) + with pytest.raises(RuntimeError, match="different.*source"): + identity.require_authority(OpenHCSRuntimeImportAuthority.current()) From a5f639e80b1f305c4e7f87294c77d01ee070acd6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 30 Sep 2026 08:06:06 -0400 Subject: [PATCH 4/4] Integrate current main and merged retained-callable dependency --- ...mory_current_main_integration_20260930.rst | 75 +++++++++++++++++++ external/arraybridge | 2 +- 2 files changed, 76 insertions(+), 1 deletion(-) create mode 100644 docs/validation/memory_current_main_integration_20260930.rst diff --git a/docs/validation/memory_current_main_integration_20260930.rst b/docs/validation/memory_current_main_integration_20260930.rst new file mode 100644 index 0000000000..36f91a6c98 --- /dev/null +++ b/docs/validation/memory_current_main_integration_20260930.rst @@ -0,0 +1,75 @@ +Memory/history source integration checkpoint +=========================================== + +Parent integrated PR208 ``d69d65adbd56633e1d1fd2a3b65678728cd6ada8`` +normally with current remote OpenHCS main +``4a9c9b3e1d7698554b2bfa3a487bfdef2ba7a14a`` in a separate persistent +worktree. The original implementation tree remains unchanged. ArrayBridge PR1 +is merged at ``409b1e0831f815fa9ef91d4ecc04989f9fbb89c5``; its tree is +identical to the reviewed ``06837ec267e3ca0b1734461d09919ff65cb1b1fe``. +This integration records the merged dependency, not an unpublished source pin. + +Sixteen source tests PASS in 5.34 seconds +---------------------------------------- + +Existing diagnostic, runtime import authority and full ObjectState/document +history tests passed together with a 15-second process bound. No skips or +deselections. Two warnings concern disabled asyncio plugin configuration. +The diagnostic provenance cases actually launch child Python processes from +a competing cwd with absent/hostile PYTHONPATH and verify the selected source. +The history test clears/loads the registry within the same process; it is NOT +fresh application startup or actual GUI capture/restore evidence. + +The unbuilt worktree initially failed collection because _tabular_native was +absent. That XML is retained separately; no test assertion was weakened. +For the passing source-only run, the parent launcher verified both C++ sources +byte-for-byte against the existing installed source and loaded those real ABI3 +modules under their original qualified names. It copied no binary, installed +no package and added no product fallback. Seven other dependency sources were +explicitly selected from already-existing checkouts at the current recorded +pins; ArrayBridge came from this integration's exact merged checkout. +This source harness does NOT establish fresh installed-process readiness. + +Actual CLI: ``python -m openhcs.mcp.memory_diagnostic --source-identity`` +returns both selected source paths, exits0 in 0.13 seconds, peak RSS25,312 KiB. +No MCP, GUI, JVM or execution runtime is launched in provenance-only mode. + +Structural and ownership evidence +---------------------------------- + +Focused existing MCP ownership guard PASS. The original packaged structural +ratchet from authenticated agent-comms3b03785f was run against current main +and the integration, across all production OpenHCS context: 5101 projected +metrics. It exits1, NOT a global structural PASS: + +* StringSubscript +5: capture boundary reads Rss, Pss, Private_Clean, + Private_Dirty and Swap from Linux's external smaps_rollup format. +* ForeignAbsenceProbe +2: ``not capability.read_only`` in request admission, + and ``args.surface is None`` at the optional argparse boundary. + +All seven sites were inspected. The first five are external-format decoding +into the existing typed ProcessMemoryReceipt, not raw downstream record reads +(BOUND-1/BOUND-2). Admission queries the existing capability owner instead of +restating mutation/side-effect logic (MEMB-2/BOUND-2). Optional CLI input is +decoded once into the existing LocalCapabilitySurfaceProfile owner. Original +source/import launch authority is retained (IMPL-13). No new roster, string +dispatch, copied controller or compatibility reader is introduced. + +These are source-backed classifications of screening leads, NOT a change to +the ratchet, an exception entry, a complete NRA proof or a passing ratchet +claim. The PR retains the exact findings for the refactoring-owner review. + +Remaining user path +------------------- + +Parent owns integration, installed capture/restore and memory measurements; +the original PR208 implementation owner remains Linnaeus. No installed source +or retained H002 runtime/viewer is changed. Issues169 and131 stay open: live +desktop history and real bounded MCP retention/slope measurements remain. +No biological acceptance or leak/no-leak finding is inferred from this work. + +Resource assertion still exits2 for swap14.6 GiB, with RAM18.5 GiB available. +No heavy/native/new-parallel job was started. The protected H002 declaration +exports and full in-memory history remain intact; pending close consent was +not inferred. Full archive scope and the parent blind-analysis goal remain +incomplete. RST is the authoritative prose receipt; historical MD is retained. diff --git a/external/arraybridge b/external/arraybridge index 06837ec267..409b1e0831 160000 --- a/external/arraybridge +++ b/external/arraybridge @@ -1 +1 @@ -Subproject commit 06837ec267e3ca0b1734461d09919ff65cb1b1fe +Subproject commit 409b1e0831f815fa9ef91d4ecc04989f9fbb89c5