diff --git a/packages/flow/openadapt_flow/compile_agent/emit.py b/packages/flow/openadapt_flow/compile_agent/emit.py index ba887b3de..94b852550 100644 --- a/packages/flow/openadapt_flow/compile_agent/emit.py +++ b/packages/flow/openadapt_flow/compile_agent/emit.py @@ -40,6 +40,7 @@ from pydantic import Field as {field_type} {sdk_import} +from openadapt_flow.request_binding import mcp_request_scope as {scoped_server_type} {task_symbol} = {task_dir!r} @@ -59,9 +60,22 @@ def {func}({signature}) -> dict: """ from openadapt_flow.compile_agent import run_compiled_params as {dispatch_symbol} + if {generation_symbol} == 1: + with {scoped_server_type}({server_symbol}.get_context()): + return {dispatch_symbol}({task_symbol}, {values}).to_dict() return {dispatch_symbol}({task_symbol}, {values}).to_dict() +async def {bound_call_symbol}(self, name, arguments, context=None): + with {scoped_server_type}(context): + return await {server_type}.call_tool(self, name, arguments, context) + + +if {generation_symbol} == 2: + from types import MethodType as {method_type_symbol} + + {server_symbol}.call_tool = {method_type_symbol}({bound_call_symbol}, {server_symbol}) + if __name__ == "__main__": {server_symbol}.run() ''' @@ -86,6 +100,13 @@ def emit_mcp(compiler: "FirstRunCompiler", out_path: Path) -> Path: field_type = symbols.allocate("Field") server_type = symbols.allocate("FastMCP") import_error = symbols.allocate("mcp_import_error") + scoped_server_type = symbols.allocate("request_scope_for_context") + bound_call_symbol = symbols.allocate("call_tool_with_request_context") + method_type_symbol = symbols.allocate("MethodType") + generation_symbol = symbols.allocate("mcp_generation") + parse_context_symbol = symbols.allocate("parse_request_context") + request_scope_symbol = symbols.allocate("request_scope") + meta_symbol = symbols.allocate("request_meta") task_symbol = symbols.allocate("TASK_DIR") server_symbol = symbols.allocate("mcp") dispatch_symbol = symbols.allocate("run_compiled_params") @@ -102,7 +123,11 @@ def emit_mcp(compiler: "FirstRunCompiler", out_path: Path) -> Path: annotated_type=annotated_type, field_type=field_type, server_type=server_type, - sdk_import=server_import(server_type, import_error), + sdk_import=server_import(server_type, import_error, generation_symbol), + scoped_server_type=scoped_server_type, + bound_call_symbol=bound_call_symbol, + method_type_symbol=method_type_symbol, + generation_symbol=generation_symbol, task_symbol=task_symbol, server_symbol=server_symbol, dispatch_symbol=dispatch_symbol, @@ -110,6 +135,9 @@ def emit_mcp(compiler: "FirstRunCompiler", out_path: Path) -> Path: f"{binding}: {annotated_type}[str, {field_type}(validation_alias={name!r})]" for name, binding in bindings.items() ), + parse_context_symbol=parse_context_symbol, + request_scope_symbol=request_scope_symbol, + meta_symbol=meta_symbol, doc=" ".join(instruction.split()), values="{" + ", ".join(f"{name!r}: {binding}" for name, binding in bindings.items()) diff --git a/packages/flow/openadapt_flow/compile_agent/pipeline.py b/packages/flow/openadapt_flow/compile_agent/pipeline.py index f2b61b3dd..1e501fa00 100644 --- a/packages/flow/openadapt_flow/compile_agent/pipeline.py +++ b/packages/flow/openadapt_flow/compile_agent/pipeline.py @@ -1060,6 +1060,7 @@ def _governed( judge the confirmed checks against the system of record.""" from openadapt_flow import __version__ + from openadapt_flow.request_binding import current_request_context from openadapt_flow.verify_run import verify_run, write_result cfg = self._confirmed_config() @@ -1083,6 +1084,9 @@ def _governed( }, authorization_ref=f"compile-agent:{self.task.key}:{kind}", baseline_path=self._receipt_baseline, + # A canary is a distinct operation, even during first admission. + # Never relabel its receipt as the outer request's effect. + request_context=current_request_context() if kind == "run" else None, ) return result, receipt, write_result(receipt, work) diff --git a/packages/flow/openadapt_flow/emit/mcp_sdk.py b/packages/flow/openadapt_flow/emit/mcp_sdk.py index e5707835f..9ca75bcbd 100644 --- a/packages/flow/openadapt_flow/emit/mcp_sdk.py +++ b/packages/flow/openadapt_flow/emit/mcp_sdk.py @@ -7,12 +7,16 @@ from __future__ import annotations -def server_import(server_type: str, import_error: str) -> str: +def server_import( + server_type: str, import_error: str, generation_symbol: str | None = None +) -> str: """Render imports using symbols reserved by the emitter's name allocator.""" + modern_generation = f" {generation_symbol} = 2\n" if generation_symbol else "" + legacy_generation = f" {generation_symbol} = 1\n" if generation_symbol else "" return f"""try: from mcp.server.mcpserver import MCPServer as {server_type} -except ModuleNotFoundError as {import_error}: +{modern_generation}except ModuleNotFoundError as {import_error}: if {import_error}.name != "mcp.server.mcpserver": raise from mcp.server.fastmcp import FastMCP as {server_type} -""" +{legacy_generation}""" diff --git a/packages/flow/openadapt_flow/integrations/agents/skills.py b/packages/flow/openadapt_flow/integrations/agents/skills.py index 21f0ac890..f185bf10b 100644 --- a/packages/flow/openadapt_flow/integrations/agents/skills.py +++ b/packages/flow/openadapt_flow/integrations/agents/skills.py @@ -422,6 +422,11 @@ def _compiler(self, prompt: str, params: Mapping[str, Any]) -> Any: ) return compiler, key + def receipt_root(self, prompt: str, params: Mapping[str, Any]) -> Path: + """The producer's receipt subtree, without opening a halted server.""" + compiler, _key = self._compiler(prompt, params) + return compiler.dir + def server(self, prompt: str, params: Mapping[str, Any]) -> Optional[Path]: """The task's compiled MCP server, when it has an admitted one.""" @@ -1051,7 +1056,14 @@ async def _call_skill(python: str, server: Path, params: dict[str, Any]) -> dict tools = (await session.list_tools()).tools if len(tools) != 1: raise RuntimeError("a compiled skill's server has exactly one tool") - result = await session.call_tool(tools[0].name, params) + from openadapt_flow.request_binding import request_meta + + metadata = request_meta() + if metadata is None: + result = await session.call_tool(tools[0].name, params) + else: + # Unsupported metadata APIs refuse; never retry without context. + result = await session.call_tool(tools[0].name, params, meta=metadata) # mcp 2 names these fields in snake case, mcp 1 in camel case. if getattr(result, "is_error", None) or getattr(result, "isError", None): text = " ".join(getattr(c, "text", "") for c in result.content) diff --git a/packages/flow/openadapt_flow/request_binding.py b/packages/flow/openadapt_flow/request_binding.py new file mode 100644 index 000000000..a0ae8a916 --- /dev/null +++ b/packages/flow/openadapt_flow/request_binding.py @@ -0,0 +1,498 @@ +"""Associate a normal transport reservation with verifier receipt bytes. + +These local bindings do not authenticate a caller, issuer or signing key. +Execution verification and request/receipt association are separate results. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import stat +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +CONTEXT_SCHEMA = "openadapt.request-context/1" +BINDING_SCHEMA = "openadapt.receipt-binding/1" +MCP_META_KEY = "ai.openadapt/request-context" +MAX_RECEIPT_BYTES = 8_000_000 +MAX_JSON_DEPTH = 64 +MAX_JSON_NODES = 100_000 +MAX_JSON_INTEGER_BITS = 4096 +MAX_PATH_COMPONENTS = 128 +_HEX = re.compile(r"[0-9a-f]{64}\Z") +_OPAQUE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z") + + +class BindingError(ValueError): + """A fixed refusal category; never includes receipt values or paths.""" + + +@dataclass(frozen=True) +class RequestContext: + run_id: str + request_ref: str + workflow: str + fingerprint: str + + def to_dict(self) -> dict[str, str]: + return { + "schema": CONTEXT_SCHEMA, + "run_id": self.run_id, + "request_ref": self.request_ref, + "workflow": self.workflow, + "fingerprint": self.fingerprint, + } + + @classmethod + def parse(cls, value: object) -> RequestContext: + if ( + type(value) is not dict + or any(type(key) is not str for key in value) + or set(value) + != {"schema", "run_id", "request_ref", "workflow", "fingerprint"} + ): + raise BindingError("request_context_shape") + if any(type(v) is not str for v in value.values()): + raise BindingError("request_context_type") + if ( + value["schema"] != CONTEXT_SCHEMA + or not _OPAQUE.fullmatch(value["run_id"]) + or ".." in value["run_id"] + or not value["request_ref"].startswith("sha256:") + or not _HEX.fullmatch(value["request_ref"][7:]) + or not value["workflow"].startswith("wrap:") + or not _HEX.fullmatch(value["workflow"][5:]) + or not _HEX.fullmatch(value["fingerprint"]) + ): + raise BindingError("request_context_value") + return cls( + value["run_id"], + value["request_ref"], + value["workflow"], + value["fingerprint"], + ) + + @classmethod + def from_record(cls, record: dict[str, Any]) -> RequestContext: + if record.get("mode") != "skill-cache" or record.get("state") not in { + "running", + "finished", + }: + raise BindingError("request_reservation") + request_id = record.get("request_id") + if type(request_id) is not str or not 1 <= len(request_id) <= 128: + raise BindingError("request_reference") + return cls.parse( + { + "schema": CONTEXT_SCHEMA, + "run_id": record.get("run_id"), + "request_ref": "sha256:" + + hashlib.sha256(request_id.encode("utf-8")).hexdigest(), + "workflow": record.get("workflow"), + "fingerprint": record.get("fingerprint"), + } + ) + + +_CURRENT: ContextVar[RequestContext | None] = ContextVar( + "openadapt_request_context", default=None +) + + +def current_request_context() -> RequestContext | None: + return _CURRENT.get() + + +def request_meta() -> dict[str, Any] | None: + """The exact namespaced transport projection of the active owned scope.""" + context = current_request_context() + return None if context is None else {MCP_META_KEY: context.to_dict()} + + +@contextmanager +def request_scope(context: RequestContext | None) -> Iterator[None]: + if context is not None: + if type(context) is not RequestContext: + raise BindingError("request_context_type") + context = RequestContext.parse(context.to_dict()) + previous = _CURRENT.get() + if previous is not None and previous != context: + raise BindingError("request_context_conflict") + token = _CURRENT.set(context) + try: + yield + finally: + _CURRENT.reset(token) + + +def context_from_mcp(meta: object) -> RequestContext | None: + if meta is None: + return None + # MCP 1 has a Pydantic Meta object; MCP 2 uses a mapping. This is not a + # generic callback protocol and does not interpret business arguments. + if type(meta) is not dict: + from pydantic import BaseModel + + if not isinstance(meta, BaseModel): + raise BindingError("mcp_context_metadata") + meta = meta.model_dump(by_alias=True) + if type(meta) is not dict: + raise BindingError("mcp_context_metadata") + document = _json_value(meta) + if type(document) is not dict: + raise BindingError("mcp_context_metadata") + if MCP_META_KEY not in document: + return None + return RequestContext.parse(document[MCP_META_KEY]) + + +@contextmanager +def mcp_request_scope(context: Any) -> Iterator[None]: + """Use native transport metadata with the existing strict context model.""" + try: + meta = context.request_context.meta if context is not None else None + except ValueError: + # Native direct calls have no incoming request context. + meta = None + with request_scope(context_from_mcp(meta)): + yield + + +def _json_value(value: Any) -> Any: + """Detach bounded exact JSON values without serialization coercions. + + This is a representation check, not a receipt or caller authority check. + Keys and values share one aggregate budget. Cycles exceed the depth bound; + repeated finite subtrees count once per occurrence, as they do in JSON. + No caller-owned mutable container is retained in the returned snapshot. + """ + nodes = 0 + string_bytes = 0 + + def visit(item: Any, depth: int) -> Any: + nonlocal nodes, string_bytes + nodes += 1 + if nodes > MAX_JSON_NODES or depth > MAX_JSON_DEPTH: + raise BindingError("json_bound") + kind = type(item) + if item is None or kind is bool: + return item + if kind is int: + if item.bit_length() > MAX_JSON_INTEGER_BITS: + raise BindingError("json_bound") + elif kind is float: + if not math.isfinite(item): + raise BindingError("json_number") + elif kind is str: + try: + string_bytes += len(item.encode("utf-8")) + except UnicodeError: + raise BindingError("json_string") from None + if string_bytes > MAX_RECEIPT_BYTES: + raise BindingError("json_bound") + elif kind is list: + if nodes + len(item) > MAX_JSON_NODES: + raise BindingError("json_bound") + return [visit(child, depth + 1) for child in item] + elif kind is dict: + if nodes + 2 * len(item) > MAX_JSON_NODES: + raise BindingError("json_bound") + detached: dict[str, Any] = {} + for key, child in item.items(): + if type(key) is not str: + raise BindingError("json_key") + visit(key, depth + 1) + detached[key] = visit(child, depth + 1) + return detached + else: + raise BindingError("json_type") + return item + + try: + return visit(value, 0) + except (RecursionError, RuntimeError): + raise BindingError("json_changed") from None + + +def _canonical(value: Any) -> str: + detached = _json_value(value) + try: + result = json.dumps( + detached, sort_keys=True, separators=(",", ":"), allow_nan=False + ) + except (TypeError, ValueError, RecursionError): + raise BindingError("terminal_result") from None + if len(result.encode("utf-8")) > MAX_RECEIPT_BYTES: + raise BindingError("json_bound") + return result + + +def _no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise BindingError("receipt_duplicate_key") + value[key] = item + return value + + +def _read_receipt_impl(path: object, under: Path) -> tuple[bytes, dict[str, Any]]: + """Read through a retained, no-follow directory-FD chain. + + Every component, including the supplied root's ancestors, must be a real + directory. Relative names are resolved from retained parent descriptors, + never from a later pathname resolution. The chain and leaf must still name + the opened inodes after reading. This is not an atomic filesystem snapshot + or a lock against another writer with access to those same inodes. + """ + if type(path) is not str: + raise BindingError("receipt_path") + if ( + os.open not in os.supports_dir_fd + or os.stat not in os.supports_dir_fd + or os.stat not in os.supports_follow_symlinks + or any( + not hasattr(os, flag) + for flag in ("O_NOFOLLOW", "O_DIRECTORY", "O_NONBLOCK") + ) + ): + raise BindingError("receipt_primitives") + file, supplied_root = Path(path), Path(under) + if ".." in file.parts or ".." in supplied_root.parts: + raise BindingError("receipt_path") + root = Path(os.path.abspath(supplied_root)) + file = Path(os.path.abspath(file)) + if not file.is_relative_to(root): + raise BindingError("receipt_owner") + relative = file.relative_to(root) + if file.name != "effect-receipt.json" or "receipts" not in relative.parts[:-1]: + raise BindingError("receipt_path") + components = file.parts[1:] + if not 1 < len(components) <= MAX_PATH_COMPONENTS: + raise BindingError("receipt_path") + descriptors: list[int] = [] + chain: list[tuple[int, str, os.stat_result]] = [] + primary: BaseException | None = None + try: + directory = os.open(file.anchor, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + descriptors.append(directory) + for name in components[:-1]: + before = os.stat(name, dir_fd=directory, follow_symlinks=False) + if not stat.S_ISDIR(before.st_mode): + raise BindingError("receipt_owner") + child = os.open( + name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory + ) + descriptors.append(child) + opened = os.fstat(child) + if not stat.S_ISDIR(opened.st_mode) or (opened.st_dev, opened.st_ino) != ( + before.st_dev, + before.st_ino, + ): + raise BindingError("receipt_changed") + chain.append((directory, name, opened)) + directory = child + name = components[-1] + before = os.stat(name, dir_fd=directory, follow_symlinks=False) + if ( + not stat.S_ISREG(before.st_mode) + or not 0 < before.st_size <= MAX_RECEIPT_BYTES + ): + raise BindingError("receipt_header") + fd = os.open( + name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory + ) + descriptors.append(fd) + opened = os.fstat(fd) + if not stat.S_ISREG(opened.st_mode) or ( + opened.st_dev, + opened.st_ino, + opened.st_size, + ) != (before.st_dev, before.st_ino, before.st_size): + raise BindingError("receipt_changed") + chunks: list[bytes] = [] + remaining = before.st_size + 1 + while remaining: + chunk = os.read(fd, min(65536, remaining)) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + raw = b"".join(chunks) + after = os.fstat(fd) + named = os.stat(name, dir_fd=directory, follow_symlinks=False) + if ( + len(raw) != before.st_size + or (after.st_size, after.st_mtime_ns, after.st_ctime_ns) + != (opened.st_size, opened.st_mtime_ns, opened.st_ctime_ns) + or not stat.S_ISREG(named.st_mode) + or (named.st_dev, named.st_ino) != (opened.st_dev, opened.st_ino) + ): + raise BindingError("receipt_changed") + for parent, component, expected in chain: + named = os.stat(component, dir_fd=parent, follow_symlinks=False) + if not stat.S_ISDIR(named.st_mode) or (named.st_dev, named.st_ino) != ( + expected.st_dev, + expected.st_ino, + ): + raise BindingError("receipt_changed") + except BaseException as error: + primary = error + raise + finally: + closing: BaseException | None = None + for descriptor in reversed(descriptors): + try: + os.close(descriptor) + except BaseException as error: + if closing is None: + closing = error + if primary is None and closing is not None: + raise closing + try: + document = json.loads(raw, object_pairs_hook=_no_duplicate_keys) + except (ValueError, UnicodeError, RecursionError): + raise BindingError("receipt_json") from None + document = _json_value(document) + if type(document) is not dict: + raise BindingError("receipt_shape") + return raw, document + + +def _read_receipt(path: object, under: Path) -> tuple[bytes, dict[str, Any]]: + try: + return _read_receipt_impl(path, under) + except BindingError: + raise + except (OSError, RuntimeError, TypeError, ValueError): + raise BindingError("receipt_unavailable") from None + + +def _check_document( + document: dict[str, Any], expected: RequestContext, outcome: object +) -> None: + from openadapt_flow.verify_run import evidence_problem, receipt_problem + + receipt, evidence = document.get("receipt"), document.get("evidence") + if type(receipt) is not dict or type(evidence) is not dict: + raise BindingError("receipt_shape") + if ( + RequestContext.parse(receipt.get("request_context")) != expected + or RequestContext.parse(evidence.get("request_context")) != expected + ): + raise BindingError("receipt_context_mismatch") + if ( + receipt_problem(receipt) is not None + or evidence_problem(evidence, receipt) is not None + ): + raise BindingError("receipt_evidence") + if ( + type(outcome) is not str + or type(receipt.get("outcome")) is not str + or receipt["outcome"].upper() != outcome + or receipt.get("executor_kind") != "flow_replay" + ): + raise BindingError("receipt_outcome") + + +def unbound(reason: str) -> dict[str, str]: + if reason not in { + "NO_REQUEST_CONTEXT", + "NO_RECEIPT", + "NO_OWNED_RECEIPT_ROOT", + "LEGACY_TERMINAL", + }: + raise BindingError("binding_reason") + return {"schema": BINDING_SCHEMA, "status": "UNBOUND", "reason": reason} + + +def capture_receipt_binding( + result: dict[str, Any], expected: RequestContext, under: Path | None +) -> dict[str, Any]: + """Capture actual producer bytes before the durable terminal is written. + + An absent receipt is explicitly unbound, including first-run admission; + its canary receipt must never be substituted for this operation. + """ + result = _json_value(result) + if result.get("receipt") is None: + return unbound("NO_RECEIPT") + if under is None: + return unbound("NO_OWNED_RECEIPT_ROOT") + raw, document = _read_receipt(result.get("receipt"), under) + _check_document(document, expected, result.get("receipt_outcome")) + return { + "schema": BINDING_SCHEMA, + "status": "BOUND", + "request_context": expected.to_dict(), + "receipt_sha256": hashlib.sha256(raw).hexdigest(), + "receipt_bytes": len(raw), + } + + +def verify_receipt_binding( + record: dict[str, Any], reply: dict[str, Any], under: Path +) -> dict[str, Any]: + """Low-level check against a trusted normal RunStore record. + + Supplied arbitrary records are not authority. WrappedTask.verify_reply + reads the actual record and also binds the caller's request and arguments. + Legacy, unbound and refused metadata never succeeds here. + """ + record, reply = _json_value(record), _json_value(reply) + if type(record) is not dict or type(reply) is not dict: + raise BindingError("terminal_result") + if record.get("state") != "finished": + raise BindingError("terminal_unfinished") + expected = RequestContext.from_record(record) + stored = record.get("result") + if type(stored) is not dict or type(reply) is not dict: + raise BindingError("terminal_result") + compared = dict(reply) + # The normal replay branch changes this one transport field only. + if compared.get("replayed") is True and stored.get("replayed") is False: + compared["replayed"] = False + if _canonical(stored) != _canonical(compared): + raise BindingError("terminal_result_mismatch") + if reply.get("run_id") != expected.run_id or reply.get("request_id") != record.get( + "request_id" + ): + raise BindingError("terminal_identity") + binding = stored.get("receipt_binding") + if type(binding) is not dict or set(binding) != { + "schema", + "status", + "request_context", + "receipt_sha256", + "receipt_bytes", + }: + raise BindingError("receipt_unbound") + if binding["schema"] != BINDING_SCHEMA or binding["status"] != "BOUND": + raise BindingError("receipt_unbound") + if RequestContext.parse(binding["request_context"]) != expected: + raise BindingError("terminal_context_mismatch") + if type(binding["receipt_sha256"]) is not str or not _HEX.fullmatch( + binding["receipt_sha256"] + ): + raise BindingError("receipt_digest") + if ( + type(binding["receipt_bytes"]) is not int + or not 0 < binding["receipt_bytes"] <= MAX_RECEIPT_BYTES + ): + raise BindingError("receipt_size") + raw, document = _read_receipt(reply.get("receipt"), under) + if ( + len(raw) != binding["receipt_bytes"] + or hashlib.sha256(raw).hexdigest() != binding["receipt_sha256"] + ): + raise BindingError("receipt_digest_mismatch") + _check_document(document, expected, reply.get("receipt_outcome")) + return dict(binding) diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 348ada344..ce5505244 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -75,6 +75,7 @@ load_coverage, ) from openadapt_flow.qualify.result import verify_config_sha256 +from openadapt_flow.request_binding import RequestContext from openadapt_flow.runtime.effects import late_write from openadapt_flow.runtime.effects.adapter import verifier_identity from openadapt_flow.runtime.effects.effect import ( @@ -201,6 +202,7 @@ _RECEIPT_VALIDATION_REASONS = frozenset( { + "request_context", "schema_version", "unknown_field", "missing_field", @@ -2717,6 +2719,7 @@ def verify_run( second_extractor: Optional[Any] = None, post_state_out: Optional[dict[str, Any]] = None, read_provider: Optional[CompleteReadProvider] = None, + request_context: Optional[RequestContext] = None, ) -> dict[str, Any]: """Read the system of record and return ``{"receipt", "evidence"}``. @@ -2726,6 +2729,10 @@ def verify_run( CONFIRMED by a system-of-record read at or above ``minimum_effect_strength``; anything weaker is never upgraded. + ``request_context`` is an explicit producer correlation context, separate + from ``execution_id``. It does not authenticate a caller. An ambient + wrapper scope alone does not bind direct verifier or canary receipts. + With ``baseline_path`` the run continues the one the baseline started (its ``execution_id`` and salt) and the baseline is consumed. Without one, a new salt is drawn and written to ``private_dir/effect-run.private.json`` @@ -2752,6 +2759,17 @@ def verify_run( if not isinstance(run_record, Mapping): raise VerifyRunError("run record must be a JSON object") + if request_context is not None: + from openadapt_flow.request_binding import BindingError + + try: + if type(request_context) is not RequestContext: + raise BindingError("request_context_type") + request_context = RequestContext.parse(request_context.to_dict()) + except BindingError: + raise VerifyRunError( + "invalid request context", validation_reason="request_context" + ) from None if not isinstance(authorization_ref, str) or not authorization_ref: raise VerifyRunError("an authorization reference is required") remediates = _opaque_or_none("remediates", remediates) @@ -3167,6 +3185,8 @@ def verify_run( model_used=executor_kind != "flow_replay", external_network_used=executor_kind != "flow_replay", ) + if request_context is not None: + evidence["request_context"] = request_context.to_dict() contract_hex = contract_digest.split(":", 1)[1] issued_at = _now() v1 = ExecuteEvidenceReceiptV1( @@ -3199,6 +3219,8 @@ def verify_run( receipt["correlation_id"] = correlation_id if recheck_due_at is not None: receipt[RECHECK_RECEIPT_FIELD] = recheck_due_at + if request_context is not None: + receipt["request_context"] = request_context.to_dict() problem = receipt_problem(receipt) or evidence_problem(evidence, receipt) if problem is not None: # pragma: no cover - a producer bug, never input raise VerifyRunError( @@ -3583,12 +3605,19 @@ def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: v1_names | set(_NEW_RECEIPT_FIELDS) | set(_OPTIONAL_RECEIPT_FIELDS) - | {"checked_how", "checked", RECHECK_RECEIPT_FIELD} + | {"checked_how", "checked", RECHECK_RECEIPT_FIELD, "request_context"} ) if set(receipt) - allowed: return "unknown_field" if any(name not in receipt for name in _NEW_RECEIPT_FIELDS): return "missing_field" + if "request_context" in receipt: + from openadapt_flow.request_binding import BindingError, RequestContext + + try: + RequestContext.parse(receipt["request_context"]) + except BindingError: + return "request_context" v1 = {k: v for k, v in receipt.items() if k in v1_names and k != "seal_signature"} v1["schema_version"] = EXECUTE_EVIDENCE_RECEIPT_SCHEMA try: @@ -3679,6 +3708,16 @@ def evidence_problem( if sha256_tagged(dict(evidence)) != receipt.get("evidence_digest"): return "evidence_mismatch" + if "request_context" in receipt or "request_context" in evidence: + from openadapt_flow.request_binding import BindingError, RequestContext + + try: + if RequestContext.parse( + receipt.get("request_context") + ) != RequestContext.parse(evidence.get("request_context")): + return "request_context" + except BindingError: + return "request_context" executor = evidence.get("executor") or {} readback = evidence.get("readback") or {} authorization = evidence.get("authorization") or {} diff --git a/packages/flow/openadapt_flow/wrap/__init__.py b/packages/flow/openadapt_flow/wrap/__init__.py index 13aa6d84c..d915a54aa 100644 --- a/packages/flow/openadapt_flow/wrap/__init__.py +++ b/packages/flow/openadapt_flow/wrap/__init__.py @@ -89,13 +89,72 @@ def _values(self, arguments: Mapping[str, Any]) -> dict[str, str]: raise TaskArgumentsError("task parameter values must be strings") return values + def _workflow(self, values: Mapping[str, str]) -> str: + identity = self.cache.key(self.instruction, values) + return ( + "wrap:" + + hashlib.sha256( + json.dumps(identity, sort_keys=True).encode("utf-8") + ).hexdigest() + ) + + def verify_reply( + self, request_id: str, arguments: Mapping[str, Any], reply: dict[str, Any] + ) -> dict[str, Any]: + """Strict local request/receipt readback against the normal RunStore. + + The caller supplies its actual expected request and arguments, not an + arbitrary purported run record. UNBOUND and REFUSED never pass. This + checks local association, not a signing issuer or store completeness. + """ + from openadapt_flow.agent.runs import RunStore + from openadapt_flow.request_binding import BindingError, verify_receipt_binding + + if ( + type(request_id) is not str + or not request_id.strip() + or len(request_id) > 128 + ): + raise BindingError("request_reference") + values = self._values(arguments) + if type(reply) is not dict or reply.get("request_id") != request_id: + raise BindingError("terminal_request") + store = RunStore(self.runs_dir) + record = store.read(reply.get("run_id", "")) + if record is None or record.get("request_id") != request_id: + raise BindingError("terminal_request") + workflow = self._workflow(values) + if record.get("workflow") != workflow or record.get( + "fingerprint" + ) != store.fingerprint(workflow, values): + raise BindingError("terminal_workflow") + root_reader = getattr(self.cache, "receipt_root", None) + root = root_reader(self.instruction, values) if callable(root_reader) else None + if not isinstance(root, Path): + raise BindingError("receipt_unbound") + return verify_receipt_binding(record, reply, root) + def call(self, request_id: str, arguments: Mapping[str, Any]) -> dict[str, Any]: """Reserve a transport request through the existing durable RunStore. Retrying the same request reads its original outcome. An interrupted owner leaves a reservation for reconciliation, never another write. + + ``verified`` remains the cache's effect result. ``receipt_binding`` + separately states whether its receipt is associated with this actual + request and durable byte digest. First-run, direct and legacy results + may be UNBOUND; that is never a successful request-binding check. Use + :meth:`verify_reply` with the actual request and arguments for a + strict local readback. These bindings do not verify a signing issuer. """ from openadapt_flow.agent.runs import RunStore + from openadapt_flow.request_binding import ( + BINDING_SCHEMA, + RequestContext, + capture_receipt_binding, + request_scope, + unbound, + ) if ( not isinstance(request_id, str) @@ -104,13 +163,7 @@ def call(self, request_id: str, arguments: Mapping[str, Any]) -> dict[str, Any]: ): raise TaskArgumentsError("request_id must contain 1 to 128 characters") values = self._values(arguments) - identity = self.cache.key(self.instruction, values) - workflow = ( - "wrap:" - + hashlib.sha256( - json.dumps(identity, sort_keys=True).encode("utf-8") - ).hexdigest() - ) + workflow = self._workflow(values) store = RunStore(self.runs_dir) begin = store.begin( request_id=request_id, workflow=workflow, inputs=values, mode="skill-cache" @@ -126,15 +179,42 @@ def call(self, request_id: str, arguments: Mapping[str, Any]) -> dict[str, Any]: record = begin.record or {} result = record.get("result") if record.get("state") == "finished" and isinstance(result, dict): - return dict(result, replayed=True) + replay = dict(result, replayed=True) + if "receipt_binding" not in replay: + replay["receipt_binding"] = unbound("LEGACY_TERMINAL") + return replay return dict(envelope, status="RECONCILIATION_REQUIRED", verified=False) assert begin.record is not None try: - result = dict(envelope, **_result(self.run(values), self.agent)) + context = RequestContext.from_record(begin.record) + with request_scope(context): + result = dict(envelope, **_result(self.run(values), self.agent)) + try: + root_reader = getattr(self.cache, "receipt_root", None) + root = ( + root_reader(self.instruction, values) + if callable(root_reader) + else None + ) + if root is not None and not isinstance(root, Path): + raise TypeError("receipt root must be a Path") + result["receipt_binding"] = capture_receipt_binding( + result, context, root + ) + except Exception: + # Binding diagnostics never replace the actual effect result. + # A strict binding consumer refuses this fixed state. + result["receipt_binding"] = { + "schema": BINDING_SCHEMA, + "status": "REFUSED", + "reason": "RECEIPT_BINDING_UNAVAILABLE", + } except Exception: # noqa: BLE001 - preserve uncertain writes without reflecting values # The call may have written before failing. A terminal, nonretryable # result keeps this request from being sent again. result = dict(envelope, status="RECONCILIATION_REQUIRED", verified=False) + if "receipt_binding" not in result: + result["receipt_binding"] = unbound("NO_RECEIPT") result["reason"] = "cached_tool_result" store.finish(begin.record, result) return result diff --git a/packages/flow/openadapt_flow/wrap/mcp.py b/packages/flow/openadapt_flow/wrap/mcp.py index 035f8463b..a90b4ca81 100644 --- a/packages/flow/openadapt_flow/wrap/mcp.py +++ b/packages/flow/openadapt_flow/wrap/mcp.py @@ -4,6 +4,7 @@ import functools import json +from importlib import import_module from typing import Any from openadapt_flow.wrap import TaskArgumentsError, WrappedTask @@ -12,10 +13,24 @@ def build_server(task: WrappedTask) -> Any: """Adapt the existing task/cache/ledger to MCP 1 and MCP 2.""" import anyio - import mcp.types as types from jsonschema import ValidationError, validate from mcp.server.lowlevel import Server + types: Any + try: + types = import_module("mcp.types") + except ModuleNotFoundError as error: + if error.name != "mcp.types": + raise + types = import_module("mcp_types") + + if types.__name__ not in {"mcp.types", "mcp_types"}: + raise RuntimeError("Unsupported MCP protocol types") + modern = types.__name__ == "mcp_types" + input_schema_field = "input_schema" if modern else "inputSchema" + error_field = "is_error" if modern else "isError" + structured_field = "structured_content" if modern else "structuredContent" + schema = { "type": "object", "properties": { @@ -35,7 +50,7 @@ async def list_tools() -> list[Any]: "read its original result. Only verified=true confirms a read-back. " "A halt or reconciliation result requires review, not another ID." ), - inputSchema=schema, + **{input_schema_field: schema}, ) ] @@ -56,7 +71,7 @@ async def call_tool(name: str, arguments: dict[str, Any] | None) -> Any: content=[ types.TextContent(type="text", text='{"error":"invalid task call"}') ], - isError=True, + **{error_field: True}, ) except Exception: # noqa: BLE001 - no raw exception values on the protocol return types.CallToolResult( @@ -66,11 +81,11 @@ async def call_tool(name: str, arguments: dict[str, Any] | None) -> Any: text='{"error":"local execution unavailable; retain request_id and check the run"}', ) ], - isError=True, + **{error_field: True}, ) return types.CallToolResult( content=[types.TextContent(type="text", text=json.dumps(result))], - structuredContent=result, + **{structured_field: result}, ) instructions = ( diff --git a/packages/flow/tests/test_request_binding.py b/packages/flow/tests/test_request_binding.py new file mode 100644 index 000000000..7c6c6eac1 --- /dev/null +++ b/packages/flow/tests/test_request_binding.py @@ -0,0 +1,759 @@ +"""One finite context model; no store effect or caller authority is inferred.""" + +from __future__ import annotations + +import asyncio +import hashlib +import threading +from dataclasses import FrozenInstanceError + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st +from hypothesis.stateful import RuleBasedStateMachine, invariant, rule + +from openadapt_flow.request_binding import ( + MCP_META_KEY, + BindingError, + RequestContext, + context_from_mcp, + current_request_context, + request_scope, +) + + +def reservation(index: int = 0) -> dict: + return { + "mode": "skill-cache", + "state": "running", + "run_id": f"run-{index}", + "request_id": f"request-{index}", + "workflow": "wrap:" + hashlib.sha256(b"synthetic-task").hexdigest(), + "fingerprint": hashlib.sha256(str(index).encode()).hexdigest(), + } + + +@given(index=st.integers(min_value=0, max_value=100000)) +def test_context_is_immutable_closed_and_roundtrips(index): + context = RequestContext.from_record(reservation(index)) + assert RequestContext.parse(context.to_dict()) == context + assert context_from_mcp({MCP_META_KEY: context.to_dict()}) == context + assert set(context.to_dict()) == { + "schema", + "run_id", + "request_ref", + "workflow", + "fingerprint", + } + with pytest.raises(FrozenInstanceError): + context.run_id = "changed" + for key in context.to_dict(): + changed = context.to_dict() + changed[key] = None + with pytest.raises(BindingError): + RequestContext.parse(changed) + with pytest.raises(BindingError): + RequestContext.parse({**context.to_dict(), "unexpected": "field"}) + + +@pytest.mark.parametrize("value", [None, {}, {"unrelated/protocol": "metadata"}]) +def test_absent_mcp_context_is_unbound(value): + assert context_from_mcp(value) is None + + +@pytest.mark.parametrize( + "value", + [None, {}, [], "context", True, 1], + ids=("null", "empty-object", "array", "text", "boolean", "integer"), +) +def test_present_invalid_mcp_context_refuses(value): + with pytest.raises(BindingError): + context_from_mcp({MCP_META_KEY: value}) + + +def test_scope_preserves_primary_and_resets_nested_context(): + context = RequestContext.from_record(reservation()) + primary = KeyboardInterrupt() + with pytest.raises(KeyboardInterrupt) as caught: + with request_scope(context): + assert current_request_context() == context + with request_scope(context): + assert current_request_context() == context + with pytest.raises(BindingError): + with request_scope(RequestContext.from_record(reservation(1))): + raise AssertionError("conflicting scope entered") + with pytest.raises(BindingError): + with request_scope(None): + raise AssertionError("scope downgraded") + raise primary + assert caught.value is primary + assert current_request_context() is None + + +def test_threads_do_not_inherit_or_share_another_request(): + barrier = threading.Barrier(2) + observed = {} + errors = [] + + def worker(index): + try: + assert current_request_context() is None + context = RequestContext.from_record(reservation(index)) + with request_scope(context): + barrier.wait(timeout=5) + observed[index] = current_request_context() + assert current_request_context() is None + except BaseException as error: + errors.append(error) + + threads = [threading.Thread(target=worker, args=(i,)) for i in (1, 2)] + with request_scope(RequestContext.from_record(reservation(99))): + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=6) + assert not errors and not any(thread.is_alive() for thread in threads) + assert observed == {i: RequestContext.from_record(reservation(i)) for i in (1, 2)} + assert current_request_context() is None + + +def test_async_requests_have_independent_scopes_and_reset_after_failure(): + async def worker(index): + context = RequestContext.from_record(reservation(index)) + with request_scope(context): + await asyncio.sleep(0) + assert current_request_context() == context + assert current_request_context() is None + return context + + async def run(): + return await asyncio.gather(worker(1), worker(2)) + + assert asyncio.run(run()) == [ + RequestContext.from_record(reservation(i)) for i in (1, 2) + ] + assert current_request_context() is None + + primary = RuntimeError("synthetic interruption") + + async def interrupted(): + try: + with request_scope(RequestContext.from_record(reservation(3))): + await asyncio.sleep(0) + raise primary + finally: + assert current_request_context() is None + + with pytest.raises(RuntimeError) as caught: + asyncio.run(interrupted()) + assert caught.value is primary and current_request_context() is None + + +class ContextLifetime(RuleBasedStateMachine): + """Independent finite owner oracle: a completed scope leaves no owner.""" + + @rule(index=st.integers(min_value=0, max_value=20), interrupted=st.booleans()) + def request(self, index, interrupted): + context = RequestContext.from_record(reservation(index)) + primary = SystemExit(3) + try: + with request_scope(context): + assert current_request_context() == context + if interrupted: + raise primary + except SystemExit as caught: + assert interrupted and caught is primary + assert current_request_context() is None + + @invariant() + def no_ambient_request_after_transition(self): + assert current_request_context() is None + + +TestContextLifetime = ContextLifetime.TestCase + + +def test_pydantic_mcp_metadata_and_request_snapshot_do_not_alias_mutable_inputs(): + from pydantic import BaseModel, ConfigDict + + class Meta(BaseModel): + model_config = ConfigDict(extra="allow") + + record = reservation(5) + context = RequestContext.from_record(record) + values = context.to_dict() + meta = Meta(**{MCP_META_KEY: values}) + parsed = context_from_mcp(meta) + values["run_id"] = "changed" + record["run_id"] = "changed-again" + assert parsed == context + assert context.run_id == "run-5" + + +def test_emitted_context_symbols_never_add_or_replace_business_parameters(): + from pathlib import Path + from tempfile import TemporaryDirectory + + from tests.test_mcp_sdk_compatibility import _emit, _load + + @given(data=st.data()) + @settings(deadline=None) + def check(data): + names = data.draw( + st.lists( + st.sampled_from( + ( + "MCPContext", + "request_context", + "parse_request_context", + "request_scope", + "request_meta", + ) + ), + min_size=1, + max_size=5, + unique=True, + ) + ) + with TemporaryDirectory() as directory: + path = _emit(Path(directory), "compiled", names) + _namespace, tool = _load(path) + properties = tool.parameters["properties"] + assert set(properties) == set(names) + assert set(tool.parameters["required"]) == set(names) + assert all(schema["type"] == "string" for schema in properties.values()) + + check() + + +_JSON_LEAF = st.one_of( + st.none(), + st.booleans(), + st.integers(min_value=-(2**128), max_value=2**128), + st.floats(allow_nan=False, allow_infinity=False), + st.text(alphabet=st.characters(blacklist_categories=("Cs",)), max_size=32), +) +_JSON_TREE = st.recursive( + _JSON_LEAF, + lambda child: st.one_of( + st.lists(child, max_size=5), + st.dictionaries( + st.text(alphabet=st.characters(blacklist_categories=("Cs",)), max_size=16), + child, + max_size=5, + ), + ), + max_leaves=30, +) + + +@given(value=_JSON_TREE) +def test_exact_json_preserves_roundtrip_types_and_mapping_order(value): + import json + + from openadapt_flow.request_binding import _canonical + + encoded = _canonical(value) + decoded = json.loads(encoded) + assert _canonical(decoded) == encoded + if type(value) is dict: + assert _canonical(dict(reversed(list(value.items())))) == encoded + assert len({_canonical(item) for item in (None, False, 0, 0.0, "0", [], {})}) == 7 + + +@given( + value=st.lists(_JSON_LEAF, max_size=8), + key=st.integers(min_value=-100, max_value=100), +) +def test_json_python_coercions_never_certify_a_terminal(value, key): + from openadapt_flow.request_binding import _canonical + + assert _canonical(value) + assert _canonical({str(key): value}) + for malformed in (tuple(value), {key: value}, {"nested": tuple(value)}): + with pytest.raises(BindingError): + _canonical(malformed) + + +@pytest.mark.parametrize( + "value", + [float("nan"), float("inf"), -float("inf"), b"bytes", {"set"}, object(), "\ud800"], + ids=( + "nan", + "infinity", + "negative-infinity", + "bytes", + "set", + "opaque-object", + "invalid-unicode", + ), +) +def test_json_unsupported_or_nonfinite_values_refuse(value): + from openadapt_flow.request_binding import _canonical + + with pytest.raises(BindingError): + _canonical(value) + + +def test_json_exact_types_and_each_bound_refuse_without_a_coercion_callback(): + import openadapt_flow.request_binding as model + + class Text(str): + pass + + class Number(int): + pass + + class Array(list): + pass + + class Mapping(dict): + pass + + recursive = [] + recursive.append(recursive) + deep = None + for _ in range(model.MAX_JSON_DEPTH + 1): + deep = [deep] + values = [ + Text("text"), + Number(1), + Array([]), + Mapping({}), + recursive, + deep, + 1 << model.MAX_JSON_INTEGER_BITS, + [None] * model.MAX_JSON_NODES, + "x" * (model.MAX_RECEIPT_BYTES + 1), + ] + for value in values: + with pytest.raises(BindingError): + model._canonical(value) + assert model._canonical((1 << model.MAX_JSON_INTEGER_BITS) - 1) + assert model._canonical([None] * (model.MAX_JSON_NODES - 1)) + + +class ExactJSONLifetime(RuleBasedStateMachine): + """Independent JSON operation model: no mutable identity memoization.""" + + def __init__(self): + super().__init__() + self.value = {"items": []} + self.items = [] + + @rule(value=_JSON_LEAF) + def append(self, value): + self.value["items"].append(value) + self.items.append(value) + + @rule() + def temporarily_replace_with_non_json(self): + from openadapt_flow.request_binding import _canonical + + original = self.value["items"] + self.value["items"] = tuple(original) + with pytest.raises(BindingError): + _canonical(self.value) + self.value["items"] = original + + @invariant() + def exact_mutable_contents_are_rechecked(self): + import json + + from openadapt_flow.request_binding import _canonical + + assert json.loads(_canonical(self.value)) == {"items": self.items} + + +TestExactJSONLifetime = ExactJSONLifetime.TestCase + + +def receipt_file(root, raw=b'{"receipt":{},"evidence":{}}'): + path = root / "receipts" / "operation" / "effect-receipt.json" + path.parent.mkdir(parents=True) + path.write_bytes(raw) + return path + + +@pytest.mark.parametrize( + "raw", + [ + b'{"receipt":{},"evidence":{}}', + b'{"receipt":{"x":[null,true,1,1.0]},"evidence":{}}', + ], + ids=("empty-document", "typed-values"), +) +def test_anchored_reader_retains_exact_bytes_without_claiming_receipt_validity( + tmp_path, raw +): + from openadapt_flow.request_binding import _read_receipt + + path = receipt_file(tmp_path / "owner", raw) + actual, document = _read_receipt(str(path), tmp_path / "owner") + assert actual == raw and set(document) == {"receipt", "evidence"} + + +@pytest.mark.parametrize( + "raw", + [ + b'{"x":NaN}', + b'{"x":Infinity}', + b'{"x":1,"x":2}', + b'{"x":"\\ud800"}', + b"[]", + b"not-json", + ], + ids=( + "nan", + "infinity", + "duplicate-key", + "invalid-unicode", + "array", + "invalid-json", + ), +) +def test_receipt_json_boundary_refuses_non_json_and_ambiguous_documents(tmp_path, raw): + from openadapt_flow.request_binding import _read_receipt + + path = receipt_file(tmp_path / "owner", raw) + with pytest.raises(BindingError): + _read_receipt(str(path), tmp_path / "owner") + + +@pytest.mark.parametrize("component", ["owner", "receipts", "operation", "leaf"]) +def test_no_follow_refuses_symlinks_at_every_owned_path_component(tmp_path, component): + from openadapt_flow.request_binding import _read_receipt + + root = tmp_path / "owner" + path = receipt_file(root) + source = { + "owner": root, + "receipts": root / "receipts", + "operation": path.parent, + "leaf": path, + }[component] + moved = source.with_name(source.name + "-actual") + source.rename(moved) + source.symlink_to(moved, target_is_directory=component != "leaf") + with pytest.raises(BindingError): + _read_receipt(str(path), root) + + +@pytest.mark.parametrize( + "change", ["directory-symlink", "directory-rename", "leaf-replace", "leaf-grow"] +) +def test_anchored_read_detects_changed_namespace_or_leaf_before_publication( + tmp_path, monkeypatch, change +): + import openadapt_flow.request_binding as model + + root = tmp_path / "owner" + path = receipt_file(root) + real_read = model.os.read + changed = False + + def read_then_change(fd, count): + nonlocal changed + raw = real_read(fd, count) + if not changed: + changed = True + if change in {"directory-symlink", "directory-rename"}: + moved = tmp_path / "moved-operation" + path.parent.rename(moved) + if change == "directory-symlink": + path.parent.symlink_to(moved, target_is_directory=True) + elif change == "leaf-replace": + replacement = path.with_name("replacement") + replacement.write_bytes(raw) + replacement.replace(path) + else: + with path.open("ab") as stream: + stream.write(b" ") + return raw + + with monkeypatch.context() as patch: + patch.setattr(model.os, "read", read_then_change) + with pytest.raises(BindingError): + model._read_receipt(str(path), root) + assert changed + + +def test_anchored_open_refuses_parent_swap_between_header_and_open( + tmp_path, monkeypatch +): + import openadapt_flow.request_binding as model + + root = tmp_path / "owner" + path = receipt_file(root) + real_open = model.os.open + swapped = False + + def open_after_swap(name, flags, *, dir_fd=None): + nonlocal swapped + if name == "operation" and dir_fd is not None and not swapped: + swapped = True + moved = tmp_path / "outside-operation" + path.parent.rename(moved) + path.parent.symlink_to(moved, target_is_directory=True) + if dir_fd is None: + return real_open(name, flags) + return real_open(name, flags, dir_fd=dir_fd) + + with monkeypatch.context() as patch: + patch.setattr( + model.os, "supports_dir_fd", model.os.supports_dir_fd | {open_after_swap} + ) + patch.setattr(model.os, "open", open_after_swap) + with pytest.raises(BindingError): + model._read_receipt(str(path), root) + assert swapped + + +@pytest.mark.parametrize( + "mode", ["unsupported", "empty", "oversize", "outside", "dotdot"] +) +def test_reader_refuses_unsupported_headers_and_ownership_without_reading( + tmp_path, monkeypatch, mode +): + import openadapt_flow.request_binding as model + + root = tmp_path / "owner" + path = receipt_file(root) + if mode == "empty": + path.write_bytes(b"") + elif mode == "oversize": + with path.open("wb") as stream: + stream.truncate(model.MAX_RECEIPT_BYTES + 1) + elif mode == "outside": + path = receipt_file(tmp_path / "outside") + elif mode == "dotdot": + path = path.parent / ".." / "operation" / path.name + + def unexpected_read(_fd, _count): + raise AssertionError("refused header reached body read") + + with monkeypatch.context() as patch: + patch.setattr(model.os, "read", unexpected_read) + if mode == "unsupported": + patch.setattr(model.os, "supports_dir_fd", set()) + with pytest.raises(BindingError): + model._read_receipt(str(path), root) + + +def test_all_opened_descriptors_close_on_primary_failure_or_close_failure( + tmp_path, monkeypatch +): + import openadapt_flow.request_binding as model + + path = receipt_file(tmp_path / "owner") + real_open, real_close = model.os.open, model.os.close + opened, closed = [], [] + primary = KeyboardInterrupt() + + def tracked_open(name, flags, *, dir_fd=None): + fd = ( + real_open(name, flags) + if dir_fd is None + else real_open(name, flags, dir_fd=dir_fd) + ) + opened.append(fd) + return fd + + def failed_read(_fd, _count): + raise primary + + def close_then_fail(fd): + real_close(fd) + closed.append(fd) + if len(closed) == 1: + raise OSError("secondary descriptor cleanup") + + with monkeypatch.context() as patch: + patch.setattr( + model.os, "supports_dir_fd", model.os.supports_dir_fd | {tracked_open} + ) + patch.setattr(model.os, "open", tracked_open) + patch.setattr(model.os, "read", failed_read) + patch.setattr(model.os, "close", close_then_fail) + with pytest.raises(KeyboardInterrupt) as caught: + model._read_receipt(str(path), tmp_path / "owner") + assert caught.value is primary + assert len(opened) > 2 and closed == list(reversed(opened)) + for fd in opened: + with pytest.raises(OSError): + model.os.fstat(fd) + + +def test_json_snapshot_does_not_retain_mutable_caller_containers(monkeypatch): + import openadapt_flow.request_binding as model + + original = {"items": [{"value": [1]}]} + detached = model._json_value(original) + assert detached == original and detached is not original + original["items"][0]["value"].append(2) + assert detached == {"items": [{"value": [1]}]} + real_dumps = model.json.dumps + + def mutate_after_snapshot(value, **kwargs): + original["items"] = ("not-json",) + return real_dumps(value, **kwargs) + + with monkeypatch.context() as patch: + patch.setattr(model.json, "dumps", mutate_after_snapshot) + encoded = model._canonical(original) + assert encoded == '{"items":[{"value":[1,2]}]}' + with pytest.raises(BindingError): + model._canonical(original) + + +def test_context_and_mcp_metadata_use_exact_json_keys_without_key_equality_callbacks(): + calls = [] + + class Key: + def __hash__(self): + return hash("schema") + + def __eq__(self, other): + calls.append(other) + return other == "schema" + + context = RequestContext.from_record(reservation(9)) + malformed = context.to_dict() + del malformed["schema"] + malformed[Key()] = context.to_dict()["schema"] + calls.clear() + with pytest.raises(BindingError): + RequestContext.parse(malformed) + assert calls == [] + with pytest.raises(BindingError): + context_from_mcp(malformed) + assert calls == [] + with pytest.raises(BindingError): + context_from_mcp({MCP_META_KEY: context.to_dict(), "other/protocol": (1, 2)}) + assert ( + context_from_mcp({MCP_META_KEY: context.to_dict(), "other/protocol": [1, 2]}) + == context + ) + + +def _native_mcp_metadata(value): + from tests._mcp_tools import sdk_modules + + _server, _error, types = sdk_modules() + + request = types.CallToolRequestParams.model_validate( + {"name": "synthetic_context_binding", "arguments": {}, "_meta": value} + ) + assert request.meta is not None + return request.meta + + +@pytest.mark.parametrize( + "value", + [None, {}, [], "context", True, 1], + ids=("null", "empty-object", "array", "text", "boolean", "integer"), +) +def test_native_mcp_present_invalid_context_matches_exact_dict_refusal(value): + metadata = {"other/protocol": None, MCP_META_KEY: value} + for representation in (metadata, _native_mcp_metadata(metadata)): + with pytest.raises(BindingError): + context_from_mcp(representation) + assert current_request_context() is None + + +@given( + index=st.integers(min_value=0, max_value=100), + unrelated=st.one_of( + st.none(), + st.booleans(), + st.integers(min_value=-5, max_value=5), + st.text(alphabet="abc", max_size=5), + ), +) +def test_native_mcp_absent_and_valid_context_parity_preserves_unrelated_nulls( + index, unrelated +): + context = RequestContext.from_record(reservation(index)) + for present in (False, True): + metadata = {"other/protocol": unrelated} + if present: + metadata[MCP_META_KEY] = context.to_dict() + native = _native_mcp_metadata(metadata) + expected = context if present else None + assert context_from_mcp(metadata) == expected + assert context_from_mcp(native) == expected + assert current_request_context() is None + + +def test_native_mcp_reserved_progress_null_is_distinct_from_binding_metadata(): + from pydantic import ValidationError + + from tests._mcp_tools import sdk_modules + + _server, _error, types = sdk_modules() + context = RequestContext.from_record(reservation(10)) + metadata = {"progressToken": None, MCP_META_KEY: context.to_dict()} + assert context_from_mcp(metadata) == context + if types.__name__ == "mcp_types": + with pytest.raises(ValidationError): + _native_mcp_metadata(metadata) + else: + assert types.__name__ == "mcp.types" + assert context_from_mcp(_native_mcp_metadata(metadata)) == context + valid = {"progressToken": "synthetic-progress", MCP_META_KEY: context.to_dict()} + assert context_from_mcp(_native_mcp_metadata(valid)) == context + assert current_request_context() is None + + +@pytest.mark.parametrize( + "value", + [None, {}, [], "context", True, 1], + ids=("null", "empty-object", "array", "text", "boolean", "integer"), +) +def test_mcp_scope_uses_the_same_strict_metadata_refusal(value): + from types import SimpleNamespace + + from openadapt_flow.request_binding import mcp_request_scope + + holder = SimpleNamespace( + request_context=SimpleNamespace(meta={MCP_META_KEY: value}) + ) + effects = [] + with pytest.raises(BindingError): + with mcp_request_scope(holder): + effects.append("dispatched") + assert effects == [] + assert current_request_context() is None + + +def test_mcp_scope_preserves_primary_reset_and_owned_unbound_refusal(): + from types import SimpleNamespace + + from openadapt_flow.request_binding import mcp_request_scope + + context = RequestContext.from_record(reservation(11)) + holder = SimpleNamespace( + request_context=SimpleNamespace(meta={MCP_META_KEY: context.to_dict()}) + ) + primary = KeyboardInterrupt() + with pytest.raises(KeyboardInterrupt) as caught: + with mcp_request_scope(holder): + assert current_request_context() == context + raise primary + assert caught.value is primary + assert current_request_context() is None + + class DirectCall: + @property + def request_context(self): + raise ValueError("no native request context") + + for absent in (None, DirectCall()): + with mcp_request_scope(absent): + assert current_request_context() is None + with request_scope(context): + with pytest.raises(BindingError): + with mcp_request_scope(absent): + raise AssertionError("owned context downgraded") + assert current_request_context() == context + assert current_request_context() is None diff --git a/packages/flow/tests/test_request_binding_integration.py b/packages/flow/tests/test_request_binding_integration.py new file mode 100644 index 000000000..70cd1b153 --- /dev/null +++ b/packages/flow/tests/test_request_binding_integration.py @@ -0,0 +1,766 @@ +"""Normal wrapper/RunStore/verifier receipts over a public synthetic store. + +The deterministic action is an explicit component fixture, not an admitted +program, a person, a model run or a claim of receipt-signing authority. +""" + +from __future__ import annotations + +import hashlib +import json +import threading +from copy import deepcopy +from pathlib import Path +from types import SimpleNamespace + +import anyio +import pytest + +from openadapt_flow.agent.runs import RunStore +from openadapt_flow.compile_agent import pipeline +from openadapt_flow.integrations.agents.skills import SkillRun +from openadapt_flow.request_binding import ( + BindingError, + capture_receipt_binding, + current_request_context, + verify_receipt_binding, +) +from openadapt_flow.verify_run import capture_baseline +from openadapt_flow.wrap import WrappedTask +from tests.test_verify_run import _actor_writes +from tests.test_wrap_integration import mcp_session, payload + +pytest_plugins = ("tests.test_verify_run", "tests.test_wrap_integration") + + +class ReceiptCache: + """Real producer/verifier boundary with an explicit synthetic action.""" + + def __init__(self, root, cfg, base): + self.root, self.cfg, self.base = root, cfg, base + self.root.mkdir() + self.calls = 0 + self.contexts = [] + self.operation_kind = "run" + + def key(self, instruction, params): + return { + "instruction": instruction, + "parameters": sorted(params), + "verify": "fixed-test-contract", + } + + def receipt_root(self, instruction, params): + return self.root + + def run(self, instruction, params, agent): + self.calls += 1 + self.contexts.append(current_request_context()) + producer = SimpleNamespace( + dir=self.root, + task=SimpleNamespace(key="synthetic-receipt-component"), + _confirmed_config=lambda: self.cfg, + _receipt_baseline=None, + ) + + def action(): + baseline = producer._pending[2] + capture_baseline(self.cfg, params, baseline) + producer._receipt_baseline = baseline + _actor_writes(self.base, patient=params["patient_id"], note=params["note"]) + return SimpleNamespace(status="COMPONENT_COMPLETED") + + # The actual pipeline creates its receipt with the current request + # context. Neither this fixture nor a caller constructs a pass receipt. + result, receipt, path = pipeline.FirstRunCompiler._governed( + producer, self.operation_kind, params, action + ) + assert receipt is not None and path is not None + outcome = receipt["receipt"]["outcome"].upper() + return SkillRun( + path="skill", + status=result.status, + verified=outcome == "VERIFIED", + model_calls=0, + cost_usd=0.0, + seconds=0, + usage=[], + usage_known=True, + replay={"receipt": str(path), "receipt_outcome": outcome}, + ) + + +@pytest.fixture +def task_case(tmp_path, cfg, sor): + base, _store = sor + cache = ReceiptCache(tmp_path / "producer", cfg, base) + task = WrappedTask( + cache=cache, + agent=None, + instruction="Create {note}", + parameters=("patient_id", "note"), + runs_dir=tmp_path / "runs", + ) + return task, cache, tmp_path / "runs" + + +def invoke(task, request_id, values, transport, monkeypatch, capsys): + if transport == "sdk": + return task.call(request_id, values) + if transport == "cli": + from openadapt_flow.wrap import cli + + with monkeypatch.context() as patch: + patch.setattr(cli, "_task", lambda args, parser: task) + args = ["--task", "test-owned-task", "--request-id", request_id] + for key, value in values.items(): + args += ["--param", key + "=" + value] + assert cli.main(args) == 0 + return json.loads(capsys.readouterr().out) + + async def steps(session): + return payload( + await session.call_tool( + "run_task", {"request_id": request_id, "parameters": values} + ) + ) + + return anyio.run(mcp_session, task, steps) + + +@pytest.mark.parametrize("transport", ["sdk", "cli", "mcp"]) +def test_two_genuine_requests_bind_and_same_outcome_receipt_swap_refuses( + task_case, monkeypatch, capsys, transport +): + task, cache, runs = task_case + replies = [ + invoke( + task, + "request-" + letter, + {"patient_id": "p1", "note": "synthetic " + letter}, + transport, + monkeypatch, + capsys, + ) + for letter in ("a", "b") + ] + store = RunStore(runs) + records = [store.read(reply["run_id"]) for reply in replies] + assert all(record is not None for record in records) + assert all( + reply["verified"] is True and reply["receipt_outcome"] == "VERIFIED" + for reply in replies + ) + assert all( + verify_receipt_binding(record, reply, cache.root)["status"] == "BOUND" + for record, reply in zip(records, replies) + ) + for letter, reply in zip(("a", "b"), replies): + assert ( + task.verify_reply( + "request-" + letter, + {"patient_id": "p1", "note": "synthetic " + letter}, + reply, + )["status"] + == "BOUND" + ) + with pytest.raises(BindingError, match="^terminal_request$"): + task.verify_reply( + "request-a", {"patient_id": "p1", "note": "synthetic a"}, replies[1] + ) + with pytest.raises(BindingError, match="^terminal_workflow$"): + task.verify_reply( + "request-a", {"patient_id": "p1", "note": "changed arguments"}, replies[0] + ) + assert cache.calls == 2 and cache.contexts[0] != cache.contexts[1] + assert current_request_context() is None + paths = [Path(reply["receipt"]) for reply in replies] + original = paths[0].read_bytes() + replacement = paths[1].read_bytes() + try: + paths[0].write_bytes(replacement) + with pytest.raises(BindingError, match="^receipt_digest_mismatch$"): + verify_receipt_binding(records[0], replies[0], cache.root) + # Independently test context: account for replacement bytes in copies + # of the consumer inputs, without altering genuine RunStore records. + changed_record, changed_reply = deepcopy(records[0]), deepcopy(replies[0]) + changed_reply["receipt_binding"]["receipt_sha256"] = hashlib.sha256( + replacement + ).hexdigest() + changed_reply["receipt_binding"]["receipt_bytes"] = len(replacement) + changed_record["result"] = deepcopy(changed_reply) + with pytest.raises(BindingError, match="^receipt_context_mismatch$"): + verify_receipt_binding(changed_record, changed_reply, cache.root) + finally: + paths[0].write_bytes(original) + assert ( + verify_receipt_binding(records[0], replies[0], cache.root)["status"] == "BOUND" + ) + replay = invoke( + task, + "request-a", + {"patient_id": "p1", "note": "synthetic a"}, + transport, + monkeypatch, + capsys, + ) + assert replay["replayed"] is True and replay["run_id"] == replies[0]["run_id"] + assert verify_receipt_binding(records[0], replay, cache.root)["status"] == "BOUND" + assert cache.calls == 2 + + +def test_changed_context_and_digest_independently_refuse(task_case): + from openadapt_flow.request_binding import RequestContext + + task, cache, runs = task_case + reply = task.call( + "negative-fields", {"patient_id": "p1", "note": "synthetic field controls"} + ) + record = RunStore(runs).read(reply["run_id"]) + assert record is not None + assert verify_receipt_binding(record, reply, cache.root)["status"] == "BOUND" + expected = RequestContext.from_record(record) + for field, value in ( + ("run_id", "different-run"), + ("request_ref", "sha256:" + "0" * 64), + ("workflow", "wrap:" + "0" * 64), + ("fingerprint", "0" * 64), + ): + changed = expected.to_dict() + assert changed[field] != value + changed[field] = value + with pytest.raises(BindingError, match="^receipt_context_mismatch$"): + capture_receipt_binding(reply, RequestContext.parse(changed), cache.root) + for field, value in (("receipt_sha256", "0" * 64), ("receipt_bytes", 1)): + changed_reply, changed_record = deepcopy(reply), deepcopy(record) + assert changed_reply["receipt_binding"][field] != value + changed_reply["receipt_binding"][field] = value + changed_record["result"] = deepcopy(changed_reply) + with pytest.raises(BindingError, match="^receipt_digest_mismatch$"): + verify_receipt_binding(changed_record, changed_reply, cache.root) + + +def test_interrupted_real_write_retains_request_and_resets_context(task_case): + task, cache, runs = task_case + original_run = cache.run + primary = KeyboardInterrupt() + + def interrupted(instruction, params, agent): + original_run(instruction, params, agent) + raise primary + + cache.run = interrupted + with pytest.raises(KeyboardInterrupt) as caught: + task.call("interrupted", {"patient_id": "p1", "note": "synthetic interrupted"}) + assert caught.value is primary and current_request_context() is None + calls = cache.calls + retry = task.call( + "interrupted", {"patient_id": "p1", "note": "synthetic interrupted"} + ) + assert retry["status"] == "RECONCILIATION_REQUIRED" and retry["verified"] is False + assert cache.calls == calls + record = RunStore(runs).read(retry["run_id"]) + assert record is not None + with pytest.raises(BindingError, match="^terminal_unfinished$"): + verify_receipt_binding(record, retry, cache.root) + + +def test_direct_sdk_run_has_no_fabricated_transport_binding(task_case): + task, cache, _runs = task_case + run = task.run({"patient_id": "p1", "note": "synthetic direct"}) + document = json.loads(Path(run.replay["receipt"]).read_text()) + assert "request_context" not in document["receipt"] + assert "request_context" not in document["evidence"] + assert cache.contexts == [None] and current_request_context() is None + + +def test_distinct_concurrent_sdk_calls_keep_their_reserved_context(task_case): + task, cache, runs = task_case + replies, errors = [], [] + ready = threading.Barrier(2) + + def worker(index): + try: + ready.wait(timeout=5) + replies.append( + task.call( + "concurrent-" + str(index), + {"patient_id": "p1", "note": "synthetic concurrent " + str(index)}, + ) + ) + assert current_request_context() is None + except BaseException as error: + errors.append(error) + + threads = [threading.Thread(target=worker, args=(index,)) for index in (1, 2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=15) + assert not errors and not any(thread.is_alive() for thread in threads) + assert len(replies) == cache.calls == 2 + store = RunStore(runs) + for reply in replies: + assert ( + verify_receipt_binding(store.read(reply["run_id"]), reply, cache.root)[ + "status" + ] + == "BOUND" + ) + assert len(set(cache.contexts)) == 2 and current_request_context() is None + + +def test_first_miss_record_check_does_not_gain_request_receipt_authority(component): + task = component.task() + reply = task.call("first-miss", {"note": "synthetic first record"}) + assert ( + reply["path"] == "agent" + and reply["status"] == "admitted" + and reply["verified"] is True + ) + assert reply["receipt_binding"] == { + "schema": "openadapt.receipt-binding/1", + "status": "UNBOUND", + "reason": "NO_RECEIPT", + } + record = RunStore(component.root / "requests").read(reply["run_id"]) + assert record is not None + with pytest.raises(BindingError, match="^receipt_unbound$"): + verify_receipt_binding(record, reply, component.root) + + +def test_canary_inside_request_scope_is_not_relabelled_as_request_effect(task_case): + task, cache, runs = task_case + cache.operation_kind = "canary" + reply = task.call( + "separate-canary", {"patient_id": "p1", "note": "synthetic canary"} + ) + assert reply["verified"] is True # Existing effect check, not request binding. + document = json.loads(Path(reply["receipt"]).read_text()) + assert "request_context" not in document["receipt"] + assert "request_context" not in document["evidence"] + assert reply["receipt_binding"]["status"] == "REFUSED" + record = RunStore(runs).read(reply["run_id"]) + with pytest.raises(BindingError, match="^receipt_unbound$"): + verify_receipt_binding(record, reply, cache.root) + assert current_request_context() is None + + +def test_receipt_read_failure_keeps_primary_when_descriptor_close_also_fails( + task_case, monkeypatch +): + import openadapt_flow.request_binding as model + + task, cache, runs = task_case + reply = task.call("file-cleanup", {"patient_id": "p1", "note": "synthetic cleanup"}) + record = RunStore(runs).read(reply["run_id"]) + primary = KeyboardInterrupt() + descriptor = [] + real_close = model.os.close + + def failed_read(fd, _count): + descriptor.append(fd) + raise primary + + def close_then_fail(fd): + real_close(fd) + if descriptor and fd == descriptor[0]: + raise OSError("synthetic secondary close") + + with monkeypatch.context() as patch: + patch.setattr(model.os, "read", failed_read) + patch.setattr(model.os, "close", close_then_fail) + with pytest.raises(KeyboardInterrupt) as caught: + verify_receipt_binding(record, reply, cache.root) + assert caught.value is primary and len(descriptor) == 1 + + +class InnerMCPReceiptCache(ReceiptCache): + """Actual emitted inner MCP/SDK boundary, with synthetic action only. + + The SDK's memory streams exercise wire messages without claiming native + subprocess ownership or child-import provenance. The unchanged producer + and independent store verifier still create every receipt. + """ + + def run(self, instruction, params, agent): + import contextvars + import importlib.util + import uuid + + import openadapt_flow.compile_agent as api + from openadapt_flow.compile_agent.emit import emit_mcp + from openadapt_flow.request_binding import request_meta + + try: + from mcp.server.mcpserver import MCPServer + except ModuleNotFoundError as error: + if error.name != "mcp.server.mcpserver": + raise + from mcp.server.fastmcp import FastMCP as MCPServer + + sdk_generation = 1 + else: + sdk_generation = 2 + + output = self.root / ("server-" + uuid.uuid4().hex + ".py") + compiler = SimpleNamespace( + task=SimpleNamespace( + params=("patient_id", "note"), + task_id="synthetic_binding", + instruction=instruction, + ), + dir=self.root, + ) + emit_mcp(compiler, output) + spec = importlib.util.spec_from_file_location( + "synthetic_binding_" + uuid.uuid4().hex, output + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + servers = [ + value for value in vars(module).values() if isinstance(value, MCPServer) + ] + assert len(servers) == 1 + lowlevel = ( + servers[0]._lowlevel_server + if sdk_generation == 2 + else servers[0]._mcp_server + ) + real_run = api.run_compiled_params + + def dispatch(task_dir, values): + assert Path(task_dir) == self.root + run = super(InnerMCPReceiptCache, self).run(instruction, values, agent) + return pipeline.ReplayResult( + status=run.status, + verified=run.verified, + model_calls=run.model_calls, + seconds=run.seconds, + run_dir=None, + receipt_outcome=run.replay["receipt_outcome"], + receipt=Path(run.replay["receipt"]), + ) + + async def call(): + from mcp import ClientSession + from mcp.shared.memory import create_client_server_memory_streams + + metadata = request_meta() + assert metadata is not None + replies = [] + async with create_client_server_memory_streams() as (client, transport): + async with anyio.create_task_group() as group: + # Neither endpoint starts in the caller's ambient scope. + # SDK2 memory streams also copy the sender's ContextVars. + contextvars.Context().run( + group.start_soon, + lowlevel.run, + *transport, + lowlevel.create_initialization_options(), + ) + + async def client_request(): + assert current_request_context() is None + async with ClientSession(*client) as session: + await session.initialize() + tools = (await session.list_tools()).tools + assert len(tools) == 1 + result = payload( + await session.call_tool( + tools[0].name, dict(params), meta=metadata + ) + ) + assert current_request_context() is None + replies.append(result) + group.cancel_scope.cancel() + + contextvars.Context().run(group.start_soon, client_request) + assert len(replies) == 1 + return replies[0] + + api.run_compiled_params = dispatch + try: + reply = anyio.run(call) + finally: + api.run_compiled_params = real_run + return SkillRun( + path="skill", + status=reply["status"], + verified=reply["verified"], + model_calls=reply["model_calls"], + cost_usd=0.0, + seconds=reply["seconds"], + replay=reply, + usage=reply["usage"], + usage_known=reply["usage_known"], + ) + + +def test_emitted_inner_mcp_carries_real_reserved_context_to_normal_producer( + tmp_path, cfg, sor +): + base, _store = sor + cache = InnerMCPReceiptCache(tmp_path / "inner-producer", cfg, base) + task = WrappedTask( + cache=cache, + agent=None, + instruction="Create {note}", + parameters=("patient_id", "note"), + runs_dir=tmp_path / "inner-runs", + ) + store = RunStore(tmp_path / "inner-runs") + replies = [ + task.call( + "inner-" + name, {"patient_id": "p1", "note": "synthetic inner " + name} + ) + for name in ("a", "b") + ] + assert len(cache.contexts) == 2 and len(set(cache.contexts)) == 2 + for reply in replies: + assert reply["verified"] is True + record = store.read(reply["run_id"]) + assert record is not None + assert verify_receipt_binding(record, reply, cache.root)["status"] == "BOUND" + assert current_request_context() is None + + +def test_legacy_finished_reply_is_explicitly_unbound_without_rewriting_record( + task_case, +): + task, cache, runs = task_case + values = {"patient_id": "p1", "note": "synthetic legacy"} + identity = cache.key(task.instruction, values) + workflow = ( + "wrap:" + + hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest() + ) + store = RunStore(runs) + begin = store.begin( + request_id="legacy", workflow=workflow, inputs=values, mode="skill-cache" + ) + assert begin.kind == "new" and begin.record is not None + old = { + "run_id": begin.run_id, + "request_id": "legacy", + "replayed": False, + "status": "VERIFIED", + "verified": True, + } + store.finish(begin.record, old) + reply = task.call("legacy", values) + assert reply["receipt_binding"] == { + "schema": "openadapt.receipt-binding/1", + "status": "UNBOUND", + "reason": "LEGACY_TERMINAL", + } + record = store.read(begin.run_id) + assert record is not None and record["result"] == old and cache.calls == 0 + with pytest.raises(BindingError): + verify_receipt_binding(record, reply, cache.root) + + +@pytest.mark.parametrize("transport", ["sdk", "cli", "mcp"]) +def test_real_terminal_json_type_changes_never_equal_the_stored_reply( + task_case, monkeypatch, capsys, transport +): + task, cache, runs = task_case + values = {"patient_id": "p1", "note": "synthetic typed terminal"} + reply = invoke(task, "typed-terminal", values, transport, monkeypatch, capsys) + record = RunStore(runs).read(reply["run_id"]) + assert type(record["result"]["usage"]) is list + assert task.verify_reply("typed-terminal", values, reply)["status"] == "BOUND" + variants = [ + ("usage", tuple(reply["usage"])), + ("verified", 1), + ("seconds", False), + ("seconds", float("nan")), + ] + for field, replacement in variants: + changed = deepcopy(reply) + changed[field] = replacement + with pytest.raises(BindingError): + task.verify_reply("typed-terminal", values, changed) + assert cache.calls == 1 and current_request_context() is None + assert verify_receipt_binding(record, reply, cache.root)["status"] == "BOUND" + + +@pytest.mark.parametrize( + "invalid", + [None, {}, [], "context", True, 1], + ids=("null", "empty-object", "array", "text", "boolean", "integer"), +) +def test_generated_mcp_rejects_invalid_present_namespace_before_dispatch( + tmp_path, cfg, sor, monkeypatch, invalid +): + import sys + + import mcp.shared.memory as memory + + import openadapt_flow.request_binding as model + + base, _store = sor + cache = InnerMCPReceiptCache(tmp_path / "metadata-producer", cfg, base) + task = WrappedTask( + cache=cache, + agent=None, + instruction="Create {note}", + parameters=("patient_id", "note"), + runs_dir=tmp_path / "metadata-runs", + ) + original_payload = payload + actual_errors = [] + + def observe_payload(result): + from tests._mcp_tools import sdk_modules + + _server, _error, types = sdk_modules() + assert types.__name__ in {"mcp.types", "mcp_types"} + # Read the native SDK field; an absent or non-boolean field must fail. + error = result.is_error if types.__name__ == "mcp_types" else result.isError + assert type(error) is bool + actual_errors.append(error) + return original_payload(result) + + with monkeypatch.context() as patch: + patch.setattr(sys.modules[__name__], "payload", observe_payload) + warmup = {"patient_id": "p1", "note": "synthetic metadata warmup"} + first = task.call("metadata-valid-before", warmup) + assert first["verified"] is True + assert ( + task.verify_reply("metadata-valid-before", warmup, first)["status"] + == "BOUND" + ) + assert cache.calls == 1 and actual_errors == [False] + with monkeypatch.context() as malformed: + wire_faults = [] + malformed.setattr( + memory, + "create_client_server_memory_streams", + _memory_wire_namespace_fault( + memory.create_client_server_memory_streams, + model.MCP_META_KEY, + invalid, + wire_faults, + ), + ) + rejected = task.call( + "metadata-invalid", + {"patient_id": "p1", "note": "synthetic must not dispatch"}, + ) + assert wire_faults == [True] + assert actual_errors == [False, True] + assert cache.calls == 1 and len(cache.contexts) == 1 + assert rejected["verified"] is False + with pytest.raises(BindingError): + task.verify_reply( + "metadata-invalid", + {"patient_id": "p1", "note": "synthetic must not dispatch"}, + rejected, + ) + assert current_request_context() is None + values = {"patient_id": "p1", "note": "synthetic metadata valid after"} + after = task.call("metadata-valid-after", values) + assert after["verified"] is True + assert ( + task.verify_reply("metadata-valid-after", values, after)["status"] + == "BOUND" + ) + assert cache.calls == 2 and len(cache.contexts) == 2 + assert actual_errors == [False, True, False] + assert current_request_context() is None + + +def _memory_wire_namespace_fault(real_factory, namespace, invalid, observations): + """Fault only an encoded tools/call request on genuine SDK memory streams.""" + from contextlib import asynccontextmanager + + from mcp.shared.message import SessionMessage + + from tests._mcp_tools import sdk_modules + + _server, _error, types = sdk_modules() + assert types.__name__ in {"mcp.types", "mcp_types"} + modern = types.__name__ == "mcp_types" + + @asynccontextmanager + async def streams(): + async with real_factory() as (client, transport): + send, read = anyio.create_memory_object_stream(1) + async with send, read, anyio.create_task_group() as forwarding: + + async def relay(): + async for message in transport[0]: + request = None + if isinstance(message, SessionMessage): + request = ( + message.message if modern else message.message.root + ) + if ( + isinstance(message, SessionMessage) + and isinstance(request, types.JSONRPCRequest) + and request.method == "tools/call" + ): + params = deepcopy(request.params) + assert ( + type(params) is dict + and type(params.get("_meta")) is dict + ) + assert namespace in params["_meta"] + params["_meta"][namespace] = invalid + changed = request.model_copy(update={"params": params}) + # Match the receiver's actual JSONRPCRequest dump. + # Unlike a Meta model field, a raw dict entry must + # retain present null before model validation. + encoded = changed.model_dump( + by_alias=True, mode="json", exclude_none=True + ) + observed = encoded["params"]["_meta"] + assert namespace in observed + assert ( + type(observed[namespace]) is type(invalid) + and observed[namespace] == invalid + ) + assert ( + changed.id == request.id + and changed.method == request.method + ) + assert ( + changed.params["arguments"] + == request.params["arguments"] + ) + assert { + key: value + for key, value in changed.params.items() + if key != "_meta" + } == { + key: value + for key, value in request.params.items() + if key != "_meta" + } + assert { + key: value + for key, value in changed.params["_meta"].items() + if key != namespace + } == { + key: value + for key, value in request.params["_meta"].items() + if key != namespace + } + message = SessionMessage( + message=( + changed if modern else types.JSONRPCMessage(changed) + ), + metadata=message.metadata, + ) + observations.append(True) + await send.send(message) + + forwarding.start_soon(relay) + try: + yield client, (read, transport[1]) + finally: + forwarding.cancel_scope.cancel() + + return streams diff --git a/packages/flow/tests/test_wrap_cli.py b/packages/flow/tests/test_wrap_cli.py index 352bb49af..c54898924 100644 --- a/packages/flow/tests/test_wrap_cli.py +++ b/packages/flow/tests/test_wrap_cli.py @@ -229,7 +229,11 @@ def build(selected, **kwargs): assert observed["max_input_bytes"] == 1048576 -@pytest.mark.parametrize("value", ["0", str(providers.MAX_INPUT_BYTES_CEILING + 1)]) +@pytest.mark.parametrize( + "value", + ["0", str(providers.MAX_INPUT_BYTES_CEILING + 1)], + ids=("zero-bound", "over-ceiling"), +) def test_max_input_bytes_outside_the_hard_cap_is_refused(monkeypatch, tmp_path, value): setup = SimpleNamespace( cache=SimpleNamespace(server=lambda *_: None, held=lambda *_: False), diff --git a/packages/flow/tests/test_wrap_recording.py b/packages/flow/tests/test_wrap_recording.py index d8e4a4edf..5229c4828 100644 --- a/packages/flow/tests/test_wrap_recording.py +++ b/packages/flow/tests/test_wrap_recording.py @@ -128,7 +128,9 @@ def test_malformed_task_refuses_before_materializing(task, key, value): @pytest.mark.parametrize( - "data", [b'{"instruction":1,"instruction":2}', b'{"x":NaN}', b"\xff", b" " * 65537] + "data", + [b'{"instruction":1,"instruction":2}', b'{"x":NaN}', b"\xff", b" " * 65537], + ids=("duplicate-key", "nonfinite-number", "invalid-utf8", "oversize-json"), ) def test_bounded_closed_json(task, data): path, _, _ = task