From f8cf6b60fcc7c53848b1ab51465cc263498d1bd5 Mon Sep 17 00:00:00 2001 From: ProfRandom92 Date: Fri, 22 May 2026 09:48:14 +0200 Subject: [PATCH 1/3] feat: add lightweight MCP replay context layer --- docs/mcp_context_layer.md | 86 ++++++++++ src/comptext_v7/mcp/__init__.py | 14 ++ src/comptext_v7/mcp/compressor.py | 132 +++++++++++++++ src/comptext_v7/mcp/context_store.py | 67 ++++++++ src/comptext_v7/mcp/models.py | 47 ++++++ src/comptext_v7/mcp/replay_payload.py | 83 ++++++++++ src/comptext_v7/mcp/validator.py | 115 +++++++++++++ tests/test_mcp_context_layer.py | 226 ++++++++++++++++++++++++++ 8 files changed, 770 insertions(+) create mode 100644 docs/mcp_context_layer.md create mode 100644 src/comptext_v7/mcp/__init__.py create mode 100644 src/comptext_v7/mcp/compressor.py create mode 100644 src/comptext_v7/mcp/context_store.py create mode 100644 src/comptext_v7/mcp/models.py create mode 100644 src/comptext_v7/mcp/replay_payload.py create mode 100644 src/comptext_v7/mcp/validator.py create mode 100644 tests/test_mcp_context_layer.py diff --git a/docs/mcp_context_layer.md b/docs/mcp_context_layer.md new file mode 100644 index 0000000..c4f4c8a --- /dev/null +++ b/docs/mcp_context_layer.md @@ -0,0 +1,86 @@ +# MCP Context Layer + +CompText V7 includes a lightweight replay-aware context layer for MCP-style +runtime systems. The layer preserves compact operational context for +deterministic replay validation; it does not execute tools, orchestrate agents, +call external APIs, or judge semantic quality. + +## Scope + +The context layer is additive to the existing replay fixtures and contracts. It +builds a compact payload from explicit trace, state, and dependency-graph fields: + +```json +{ + "task": "mcp_trace_replay_v1", + "constraints": [ + "execute_external_action:requires_human_approval", + "execute_external_action:requires_validation_passed" + ], + "required_order": [ + "capability_scope_checked", + "tool_schema_validated", + "read_context", + "validate_external_action", + "execute_external_action" + ], + "blockers": [ + ["capability_scope_checked", "execute_external_action"] + ], + "dependency_chains": [ + ["validate_external_action", "execute_external_action"] + ], + "recovery": [ + ["execute_external_action", "recovery_path_registered"] + ] +} +``` + +## Public API + +- `build_replay_payload(trace)` extracts compact deterministic commitments. +- `render_prompt_context(payload)` renders a compact prompt-safe text view. +- `validate_replay_payload(payload)` validates replay admissibility from explicit + payload fields. +- `ContextStore(root).save_context(trace)` writes a stable JSON payload. +- `ContextStore(root).load_context(task_id)` restores the compact payload. +- `save_context(trace, store_dir=...)` and `load_context(task_id, store_dir=...)` + provide module-level convenience wrappers. + +## Deterministic checks + +`validate_replay_payload` detects: + +- missing preserved constraints as `CONSTRAINT_DRIFT` +- dependency-order drift as `TOOL_ORDER_VIOLATION` +- dependency collapse as `DEPENDENCY_CHAIN_BREAK` +- missing recovery paths as `RECOVERY_PATH_LOSS` + +The validator uses exact strings, ordered lists, and explicit dependency edges. +It performs no embedding lookup, fuzzy scoring, probabilistic reasoning, LLM +judging, runtime blocking, or policy enforcement. + +## Prompt context rendering + +`render_prompt_context(payload)` produces a token-light text form without dumping +raw trace, state, or dependency-graph documents: + +```text +task: mcp_trace_replay_v1 +admissible: true +constraints: +- execute_external_action:requires_validation_passed +required_order: +- validate_external_action +- execute_external_action +dependencies: +- validate_external_action -> execute_external_action +recovery: +- execute_external_action -> recovery_path_registered +``` + +## Relationship to MCP + +This layer augments context integrity for MCP-compatible systems. It is not an +MCP implementation and does not replace MCP transport, runtime execution, or +tool semantics. diff --git a/src/comptext_v7/mcp/__init__.py b/src/comptext_v7/mcp/__init__.py new file mode 100644 index 0000000..73d4bc0 --- /dev/null +++ b/src/comptext_v7/mcp/__init__.py @@ -0,0 +1,14 @@ +"""Replay-aware context layer for MCP-style runtime systems.""" + +from .context_store import ContextStore, load_context, save_context +from .replay_payload import build_replay_payload, render_prompt_context +from .validator import validate_replay_payload + +__all__ = [ + "ContextStore", + "build_replay_payload", + "load_context", + "render_prompt_context", + "save_context", + "validate_replay_payload", +] diff --git a/src/comptext_v7/mcp/compressor.py b/src/comptext_v7/mcp/compressor.py new file mode 100644 index 0000000..d4109c7 --- /dev/null +++ b/src/comptext_v7/mcp/compressor.py @@ -0,0 +1,132 @@ +"""Deterministic operational commitment extraction for MCP-style traces.""" + +from __future__ import annotations + +from typing import Any + +Edge = tuple[str, str] + + +def _object(value: Any, field: str) -> dict[str, Any]: + if isinstance(value, dict): + return value + raise ValueError(f"{field} must be an object") + + +def _list(value: Any, field: str) -> list[Any]: + if value is None: + return [] + if isinstance(value, list): + return value + raise ValueError(f"{field} must be a list") + + +def _task_id(context: dict[str, Any]) -> str: + for key in ("task", "task_id", "fixture_id"): + value = context.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + raise ValueError("context requires task, task_id, or fixture_id") + + +def _trace(context: dict[str, Any]) -> dict[str, Any]: + trace = context.get("trace", context) + return _object(trace, "trace") + + +def _state(context: dict[str, Any]) -> dict[str, Any]: + state = context.get("state", {}) + return _object(state, "state") + + +def _dependency_graph(context: dict[str, Any]) -> dict[str, Any]: + graph = context.get("dependency_graph", {}) + return _object(graph, "dependency_graph") + + +def extract_required_order(context: dict[str, Any]) -> list[str]: + events = _list(_trace(context).get("events"), "trace.events") + + def event_key(event: dict[str, Any], index: int) -> tuple[int, int]: + step = event.get("step") + return (step if isinstance(step, int) else index, index) + + keyed_events: list[tuple[tuple[int, int], str]] = [] + for index, raw_event in enumerate(events): + event = _object(raw_event, "trace.events[]") + action = event.get("action", event.get("tool")) + if isinstance(action, str) and action: + keyed_events.append((event_key(event, index), action)) + return [action for _, action in sorted(keyed_events)] + + +def extract_constraints(context: dict[str, Any]) -> list[str]: + permission_scopes = _state(context).get("permission_scopes", {}) + if permission_scopes is None: + return [] + scopes_by_action = _object(permission_scopes, "state.permission_scopes") + + constraints: set[str] = set() + for action, scopes in scopes_by_action.items(): + if not isinstance(action, str) or not action: + continue + for scope in _list(scopes, f"state.permission_scopes.{action}"): + if isinstance(scope, str) and scope: + constraints.add(f"{action}:{scope}") + return sorted(constraints) + + +def _graph_edges(context: dict[str, Any]) -> list[dict[str, Any]]: + return [ + _object(edge, "dependency_graph.edges[]") + for edge in _list(_dependency_graph(context).get("edges"), "dependency_graph.edges") + ] + + +def _edge_tuple(edge: dict[str, Any]) -> Edge | None: + source = edge.get("source") + target = edge.get("target") + if isinstance(source, str) and source and isinstance(target, str) and target: + return source, target + return None + + +def extract_dependency_chains(context: dict[str, Any]) -> list[list[str]]: + edges: set[Edge] = set() + for edge in _graph_edges(context): + relation = edge.get("relation") + edge_tuple = _edge_tuple(edge) + if relation != "TEMPORAL" and edge_tuple is not None: + edges.add(edge_tuple) + return [list(edge) for edge in sorted(edges)] + + +def extract_blockers(context: dict[str, Any]) -> list[list[str]]: + blockers: set[Edge] = set() + for edge in _graph_edges(context): + if edge.get("relation") == "BLOCKER": + edge_tuple = _edge_tuple(edge) + if edge_tuple is not None: + blockers.add(edge_tuple) + return [list(edge) for edge in sorted(blockers)] + + +def extract_recovery_paths(context: dict[str, Any]) -> list[list[str]]: + recovery_edges: set[Edge] = set() + for edge in _graph_edges(context): + if edge.get("relation") == "RECOVERY": + edge_tuple = _edge_tuple(edge) + if edge_tuple is not None: + recovery_edges.add(edge_tuple) + return [list(edge) for edge in sorted(recovery_edges)] + + +def compress_context(context: dict[str, Any]) -> dict[str, Any]: + return { + "task": _task_id(context), + "constraints": extract_constraints(context), + "required_order": extract_required_order(context), + "blockers": extract_blockers(context), + "dependency_chains": extract_dependency_chains(context), + "recovery": extract_recovery_paths(context), + } diff --git a/src/comptext_v7/mcp/context_store.py b/src/comptext_v7/mcp/context_store.py new file mode 100644 index 0000000..be92d44 --- /dev/null +++ b/src/comptext_v7/mcp/context_store.py @@ -0,0 +1,67 @@ +"""File-backed context store for compact replay-aware MCP payloads.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .models import StoredContext +from .replay_payload import build_replay_payload + + +class ContextStore: + def __init__(self, root: Path | str | None = None) -> None: + self.root = Path(root) if root is not None else None + self._memory: dict[str, dict[str, Any]] = {} + + def save_context(self, trace: dict[str, Any]) -> StoredContext: + payload = build_replay_payload(trace) + task = str(payload["task"]) + if self.root is None: + self._memory[task] = payload + return StoredContext(task=task, payload=payload) + + self.root.mkdir(parents=True, exist_ok=True) + path = self.root / f"{task}.json" + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return StoredContext(task=task, payload=payload) + + def load_context(self, task_id: str) -> dict[str, Any]: + if self.root is None: + try: + return self._memory[task_id] + except KeyError as exc: + raise KeyError(f"context not found: {task_id}") from exc + + path = self.root / f"{task_id}.json" + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise KeyError(f"context not found: {task_id}") from exc + if not isinstance(payload, dict): + raise ValueError(f"context payload must be an object: {path}") + return payload + + +_DEFAULT_STORE: ContextStore | None = None + + +def _store(store_dir: Path | str | None) -> ContextStore: + global _DEFAULT_STORE + if store_dir is not None: + return ContextStore(store_dir) + if _DEFAULT_STORE is None: + raise RuntimeError("module-level context store is not configured") + return _DEFAULT_STORE + + +def save_context(trace: dict[str, Any], store_dir: Path | str | None = None) -> StoredContext: + global _DEFAULT_STORE + if store_dir is None and _DEFAULT_STORE is None: + _DEFAULT_STORE = ContextStore() + return _store(store_dir).save_context(trace) + + +def load_context(task_id: str, store_dir: Path | str | None = None) -> dict[str, Any]: + return _store(store_dir).load_context(task_id) diff --git a/src/comptext_v7/mcp/models.py b/src/comptext_v7/mcp/models.py new file mode 100644 index 0000000..8c26781 --- /dev/null +++ b/src/comptext_v7/mcp/models.py @@ -0,0 +1,47 @@ +"""Small data models for replay-aware MCP context payloads.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +ReplayPayloadDict = dict[str, Any] + + +@dataclass(frozen=True, slots=True) +class ValidationIssue: + field: str + failure_label: str + message: str + evidence: tuple[tuple[str, str], ...] = () + + def to_dict(self) -> dict[str, object]: + payload: dict[str, object] = { + "field": self.field, + "failure_label": self.failure_label, + "message": self.message, + } + if self.evidence: + payload["evidence"] = [list(edge) for edge in self.evidence] + return payload + + +@dataclass(frozen=True, slots=True) +class ValidationReport: + admissible: bool + failure_labels: tuple[str, ...] + issues: tuple[ValidationIssue, ...] + + def to_dict(self) -> dict[str, object]: + return { + "admissible": self.admissible, + "failure_labels": list(self.failure_labels), + "issues": [issue.to_dict() for issue in self.issues], + } + + +@dataclass(frozen=True, slots=True) +class StoredContext: + task: str + payload: ReplayPayloadDict diff --git a/src/comptext_v7/mcp/replay_payload.py b/src/comptext_v7/mcp/replay_payload.py new file mode 100644 index 0000000..639cf0f --- /dev/null +++ b/src/comptext_v7/mcp/replay_payload.py @@ -0,0 +1,83 @@ +"""Replay payload construction for MCP-compatible context handoff.""" + +from __future__ import annotations + +from typing import Any + +from .compressor import compress_context + +Edge = tuple[str, str] + + +def build_replay_payload(trace: dict[str, Any]) -> dict[str, Any]: + """Build a compact deterministic replay payload from MCP-style context.""" + return compress_context(trace) + + +def _string_items(value: Any) -> list[str]: + if not isinstance(value, list): + return [] + return [item for item in value if isinstance(item, str) and item] + + +def _edge_items(value: Any) -> list[Edge]: + if not isinstance(value, list): + return [] + edges: set[Edge] = set() + for item in value: + if not isinstance(item, list) or len(item) != 2: + continue + source, target = item + if isinstance(source, str) and source and isinstance(target, str) and target: + edges.add((source, target)) + return sorted(edges) + + +def _append_string_section(lines: list[str], title: str, values: list[str]) -> None: + if not values: + return + lines.append(f"{title}:") + lines.extend(f"- {value}" for value in values) + + +def _append_edge_section(lines: list[str], title: str, edges: list[Edge]) -> None: + if not edges: + return + lines.append(f"{title}:") + lines.extend(f"- {source} -> {target}" for source, target in edges) + + +def _append_validation_status(lines: list[str], payload: dict[str, Any]) -> None: + validation = payload.get("validation") + if not isinstance(validation, dict): + return + + admissible = validation.get("admissible") + if isinstance(admissible, bool): + lines.append(f"admissible: {str(admissible).lower()}") + + failure_labels = _string_items(validation.get("failure_labels")) + if failure_labels: + lines.append(f"failures: {', '.join(sorted(failure_labels))}") + + +def render_prompt_context(payload: dict[str, Any]) -> str: + """Render a replay payload as deterministic token-light prompt context.""" + lines: list[str] = [] + + task = payload.get("task") + if isinstance(task, str) and task: + lines.append(f"task: {task}") + + objective = payload.get("objective", payload.get("current_objective")) + if isinstance(objective, str) and objective: + lines.append(f"objective: {objective}") + + _append_validation_status(lines, payload) + _append_string_section(lines, "constraints", sorted(_string_items(payload.get("constraints")))) + _append_string_section(lines, "required_order", _string_items(payload.get("required_order"))) + _append_edge_section(lines, "dependencies", _edge_items(payload.get("dependency_chains"))) + _append_edge_section(lines, "blockers", _edge_items(payload.get("blockers"))) + _append_edge_section(lines, "recovery", _edge_items(payload.get("recovery"))) + + return "\n".join(lines) diff --git a/src/comptext_v7/mcp/validator.py b/src/comptext_v7/mcp/validator.py new file mode 100644 index 0000000..9a6001e --- /dev/null +++ b/src/comptext_v7/mcp/validator.py @@ -0,0 +1,115 @@ +"""Deterministic replay admissibility checks for compact MCP context payloads.""" + +from __future__ import annotations + +from typing import Any + +from .models import ValidationIssue, ValidationReport + +Edge = tuple[str, str] + + +def _edge_list(value: Any, field: str) -> tuple[Edge, ...]: + if value is None: + return () + if not isinstance(value, list): + raise ValueError(f"{field} must be a list") + + edges: list[Edge] = [] + for item in value: + if not isinstance(item, list) or len(item) != 2: + raise ValueError(f"{field} entries must be [source, target]") + source, target = item + if not isinstance(source, str) or not source or not isinstance(target, str) or not target: + raise ValueError(f"{field} entries must contain non-empty strings") + edges.append((source, target)) + return tuple(sorted(edges)) + + +def _string_list(value: Any, field: str) -> tuple[str, ...]: + if value is None: + return () + if not isinstance(value, list): + raise ValueError(f"{field} must be a list") + strings: list[str] = [] + for item in value: + if not isinstance(item, str) or not item: + raise ValueError(f"{field} entries must be non-empty strings") + strings.append(item) + return tuple(strings) + + +def _order_violations(required_order: tuple[str, ...], dependency_edges: tuple[Edge, ...]) -> tuple[Edge, ...]: + positions = {action: index for index, action in enumerate(required_order)} + violations = [ + (source, target) + for source, target in dependency_edges + if source in positions and target in positions and positions[source] > positions[target] + ] + return tuple(sorted(violations)) + + +def _collapsed_dependencies(required_order: tuple[str, ...], dependency_edges: tuple[Edge, ...]) -> tuple[Edge, ...]: + present = set(required_order) + collapsed = [ + (source, target) + for source, target in dependency_edges + if source not in present or target not in present + ] + return tuple(sorted(collapsed)) + + +def validate_replay_payload(payload: dict[str, Any]) -> dict[str, object]: + """Return deterministic admissibility evidence for a compact replay payload.""" + if not isinstance(payload.get("task"), str) or not payload["task"]: + raise ValueError("payload.task must be a non-empty string") + + constraints = _string_list(payload.get("constraints"), "constraints") + required_order = _string_list(payload.get("required_order"), "required_order") + dependency_chains = _edge_list(payload.get("dependency_chains"), "dependency_chains") + recovery = _edge_list(payload.get("recovery"), "recovery") + + issues: list[ValidationIssue] = [] + if not constraints: + issues.append( + ValidationIssue( + field="constraints", + failure_label="CONSTRAINT_DRIFT", + message="payload has no preserved constraints", + ) + ) + + order_violations = _order_violations(required_order, dependency_chains) + if order_violations: + issues.append( + ValidationIssue( + field="required_order", + failure_label="TOOL_ORDER_VIOLATION", + message="required order violates dependency edges", + evidence=order_violations, + ) + ) + + collapsed_dependencies = _collapsed_dependencies(required_order, dependency_chains) + if collapsed_dependencies: + issues.append( + ValidationIssue( + field="dependency_chains", + failure_label="DEPENDENCY_CHAIN_BREAK", + message="dependency edges reference actions absent from required_order", + evidence=collapsed_dependencies, + ) + ) + + if not recovery: + issues.append( + ValidationIssue( + field="recovery", + failure_label="RECOVERY_PATH_LOSS", + message="payload has no preserved recovery paths", + ) + ) + + labels = tuple(sorted({issue.failure_label for issue in issues})) + report = ValidationReport(admissible=not issues, failure_labels=labels, issues=tuple(issues)) + return report.to_dict() diff --git a/tests/test_mcp_context_layer.py b/tests/test_mcp_context_layer.py new file mode 100644 index 0000000..eaa3607 --- /dev/null +++ b/tests/test_mcp_context_layer.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from src.comptext_v7.mcp import ( + ContextStore, + build_replay_payload, + load_context, + render_prompt_context, + save_context, + validate_replay_payload, +) + + +FIXTURE_ROOT = Path("fixtures/mcp_trace_replay_v1/original") + + +def _load_fixture_context() -> dict[str, object]: + return { + "task": "mcp_trace_replay_v1", + "trace": json.loads((FIXTURE_ROOT / "trace.json").read_text(encoding="utf-8")), + "state": json.loads((FIXTURE_ROOT / "state.json").read_text(encoding="utf-8")), + "dependency_graph": json.loads((FIXTURE_ROOT / "dependency_graph.json").read_text(encoding="utf-8")), + } + + +def test_build_replay_payload_extracts_ordered_operational_commitments() -> None: + payload = build_replay_payload(_load_fixture_context()) + + assert payload == { + "task": "mcp_trace_replay_v1", + "constraints": [ + "execute_external_action:requires_human_approval", + "execute_external_action:requires_validation_passed", + ], + "required_order": [ + "user_request_received", + "capability_scope_checked", + "tool_schema_validated", + "read_context", + "validate_external_action", + "execute_external_action", + "verify_result", + "recovery_path_registered", + ], + "blockers": [["capability_scope_checked", "execute_external_action"]], + "dependency_chains": [ + ["capability_scope_checked", "execute_external_action"], + ["capability_scope_checked", "tool_schema_validated"], + ["capability_scope_checked", "validate_external_action"], + ["execute_external_action", "recovery_path_registered"], + ["execute_external_action", "verify_result"], + ["read_context", "validate_external_action"], + ["tool_schema_validated", "read_context"], + ["user_request_received", "capability_scope_checked"], + ["validate_external_action", "execute_external_action"], + ], + "recovery": [["execute_external_action", "recovery_path_registered"]], + } + + +def test_validate_replay_payload_accepts_fixture_payload() -> None: + payload = build_replay_payload(_load_fixture_context()) + + result = validate_replay_payload(payload) + + assert result["admissible"] is True + assert result["failure_labels"] == [] + assert result["issues"] == [] + + +def test_validate_replay_payload_detects_deterministic_corruptions() -> None: + payload = build_replay_payload(_load_fixture_context()) + payload["constraints"] = [] + payload["required_order"] = [ + "execute_external_action", + "validate_external_action", + "verify_result", + ] + payload["dependency_chains"] = [ + ["validate_external_action", "execute_external_action"], + ["read_context", "validate_external_action"], + ] + payload["recovery"] = [] + + result = validate_replay_payload(payload) + + assert result["admissible"] is False + assert result["failure_labels"] == [ + "CONSTRAINT_DRIFT", + "DEPENDENCY_CHAIN_BREAK", + "RECOVERY_PATH_LOSS", + "TOOL_ORDER_VIOLATION", + ] + assert result["issues"] == [ + { + "field": "constraints", + "failure_label": "CONSTRAINT_DRIFT", + "message": "payload has no preserved constraints", + }, + { + "field": "required_order", + "failure_label": "TOOL_ORDER_VIOLATION", + "message": "required order violates dependency edges", + "evidence": [["validate_external_action", "execute_external_action"]], + }, + { + "field": "dependency_chains", + "failure_label": "DEPENDENCY_CHAIN_BREAK", + "message": "dependency edges reference actions absent from required_order", + "evidence": [["read_context", "validate_external_action"]], + }, + { + "field": "recovery", + "failure_label": "RECOVERY_PATH_LOSS", + "message": "payload has no preserved recovery paths", + }, + ] + + +def test_context_store_persists_compact_payload_by_task(tmp_path: Path) -> None: + store = ContextStore(tmp_path) + saved = store.save_context(_load_fixture_context()) + + assert saved.task == "mcp_trace_replay_v1" + assert store.load_context("mcp_trace_replay_v1") == saved.payload + + +def test_module_level_save_load_requires_configured_store(tmp_path: Path) -> None: + with pytest.raises(RuntimeError, match="not configured"): + load_context("missing") + + saved = save_context(_load_fixture_context(), store_dir=tmp_path) + + assert load_context("mcp_trace_replay_v1", store_dir=tmp_path) == saved.payload + + +def test_render_prompt_context_is_deterministic_and_token_light() -> None: + payload = build_replay_payload(_load_fixture_context()) + payload["objective"] = "continue replay validation" + payload["validation"] = validate_replay_payload(payload) + + rendered = render_prompt_context(payload) + + assert rendered == "\n".join( + [ + "task: mcp_trace_replay_v1", + "objective: continue replay validation", + "admissible: true", + "constraints:", + "- execute_external_action:requires_human_approval", + "- execute_external_action:requires_validation_passed", + "required_order:", + "- user_request_received", + "- capability_scope_checked", + "- tool_schema_validated", + "- read_context", + "- validate_external_action", + "- execute_external_action", + "- verify_result", + "- recovery_path_registered", + "dependencies:", + "- capability_scope_checked -> execute_external_action", + "- capability_scope_checked -> tool_schema_validated", + "- capability_scope_checked -> validate_external_action", + "- execute_external_action -> recovery_path_registered", + "- execute_external_action -> verify_result", + "- read_context -> validate_external_action", + "- tool_schema_validated -> read_context", + "- user_request_received -> capability_scope_checked", + "- validate_external_action -> execute_external_action", + "blockers:", + "- capability_scope_checked -> execute_external_action", + "recovery:", + "- execute_external_action -> recovery_path_registered", + ] + ) + assert "events" not in rendered + assert "dependency_graph" not in rendered + assert "permission_scopes" not in rendered + + +def test_render_prompt_context_omits_empty_fields_consistently() -> None: + rendered = render_prompt_context( + { + "task": "minimal_task", + "constraints": [], + "required_order": ["validate", "deploy"], + "dependency_chains": [], + "blockers": [], + "recovery": [], + } + ) + + assert rendered == "\n".join( + [ + "task: minimal_task", + "required_order:", + "- validate", + "- deploy", + ] + ) + + +def test_render_prompt_context_includes_validation_status_only_when_present() -> None: + payload = { + "task": "failed_task", + "required_order": ["deploy", "validate"], + "dependency_chains": [["validate", "deploy"]], + "validation": { + "admissible": False, + "failure_labels": ["TOOL_ORDER_VIOLATION"], + }, + } + + rendered = render_prompt_context(payload) + + assert rendered.splitlines()[:3] == [ + "task: failed_task", + "admissible: false", + "failures: TOOL_ORDER_VIOLATION", + ] + assert "issues" not in rendered From c6de8e2ba3134ef2a056fa721c7010aac8745d2f Mon Sep 17 00:00:00 2001 From: ProfRandom92 Date: Fri, 22 May 2026 09:56:32 +0200 Subject: [PATCH 2/3] artifact: add MCP context layer example artifact --- artifacts/mcp_context_layer_example.json | 86 +++++++++++++++++++ docs/mcp_context_layer.md | 7 ++ ...rate_mcp_context_layer_example_artifact.py | 72 ++++++++++++++++ tests/test_mcp_context_layer.py | 60 +++++++++++++ 4 files changed, 225 insertions(+) create mode 100644 artifacts/mcp_context_layer_example.json create mode 100644 scripts/generate_mcp_context_layer_example_artifact.py diff --git a/artifacts/mcp_context_layer_example.json b/artifacts/mcp_context_layer_example.json new file mode 100644 index 0000000..be865fa --- /dev/null +++ b/artifacts/mcp_context_layer_example.json @@ -0,0 +1,86 @@ +{ + "artifact_id": "mcp_context_layer_example_v1", + "evaluation_mode": "deterministic", + "example": { + "fixture_id": "mcp_trace_replay_v1", + "prompt_context": "task: mcp_trace_replay_v1\nadmissible: true\nconstraints:\n- execute_external_action:requires_human_approval\n- execute_external_action:requires_validation_passed\nrequired_order:\n- user_request_received\n- capability_scope_checked\n- tool_schema_validated\n- read_context\n- validate_external_action\n- execute_external_action\n- verify_result\n- recovery_path_registered\ndependencies:\n- capability_scope_checked -> execute_external_action\n- capability_scope_checked -> tool_schema_validated\n- capability_scope_checked -> validate_external_action\n- execute_external_action -> recovery_path_registered\n- execute_external_action -> verify_result\n- read_context -> validate_external_action\n- tool_schema_validated -> read_context\n- user_request_received -> capability_scope_checked\n- validate_external_action -> execute_external_action\nblockers:\n- capability_scope_checked -> execute_external_action\nrecovery:\n- execute_external_action -> recovery_path_registered", + "replay_payload": { + "blockers": [ + [ + "capability_scope_checked", + "execute_external_action" + ] + ], + "constraints": [ + "execute_external_action:requires_human_approval", + "execute_external_action:requires_validation_passed" + ], + "dependency_chains": [ + [ + "capability_scope_checked", + "execute_external_action" + ], + [ + "capability_scope_checked", + "tool_schema_validated" + ], + [ + "capability_scope_checked", + "validate_external_action" + ], + [ + "execute_external_action", + "recovery_path_registered" + ], + [ + "execute_external_action", + "verify_result" + ], + [ + "read_context", + "validate_external_action" + ], + [ + "tool_schema_validated", + "read_context" + ], + [ + "user_request_received", + "capability_scope_checked" + ], + [ + "validate_external_action", + "execute_external_action" + ] + ], + "recovery": [ + [ + "execute_external_action", + "recovery_path_registered" + ] + ], + "required_order": [ + "user_request_received", + "capability_scope_checked", + "tool_schema_validated", + "read_context", + "validate_external_action", + "execute_external_action", + "verify_result", + "recovery_path_registered" + ], + "task": "mcp_trace_replay_v1" + }, + "source_fixture_path": "fixtures/mcp_trace_replay_v1/original", + "validation": { + "admissible": true, + "failure_labels": [], + "issues": [] + } + }, + "external_apis": "none", + "generated_by": "McpContextLayerExampleArtifactGenerator", + "llm_judges": "none", + "schema_version": "mcp_context_layer_example.v1", + "version": "1.0" +} diff --git a/docs/mcp_context_layer.md b/docs/mcp_context_layer.md index c4f4c8a..d1361d5 100644 --- a/docs/mcp_context_layer.md +++ b/docs/mcp_context_layer.md @@ -79,6 +79,13 @@ recovery: - execute_external_action -> recovery_path_registered ``` +## Example artifact + +`artifacts/mcp_context_layer_example.json` is a fixture-bound deterministic +example generated from `fixtures/mcp_trace_replay_v1/original`. It contains the +compact replay payload, prompt-context rendering, and validation result for the +baseline MCP trace replay fixture. + ## Relationship to MCP This layer augments context integrity for MCP-compatible systems. It is not an diff --git a/scripts/generate_mcp_context_layer_example_artifact.py b/scripts/generate_mcp_context_layer_example_artifact.py new file mode 100644 index 0000000..ba0cf9f --- /dev/null +++ b/scripts/generate_mcp_context_layer_example_artifact.py @@ -0,0 +1,72 @@ +"""Generate a deterministic MCP context-layer example artifact.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from src.comptext_v7.mcp import build_replay_payload, render_prompt_context, validate_replay_payload + +ARTIFACT_ID = "mcp_context_layer_example_v1" +EXAMPLE_FIXTURE_ID = "mcp_trace_replay_v1" +EXAMPLE_FIXTURE_PATH = REPO_ROOT / "fixtures" / EXAMPLE_FIXTURE_ID / "original" +OUTPUT_PATH = REPO_ROOT / "artifacts" / "mcp_context_layer_example.json" + + +def _load_json(path: Path) -> dict[str, Any]: + return json.loads(path.read_text(encoding="utf-8")) + + +def _load_fixture_context() -> dict[str, Any]: + return { + "task": EXAMPLE_FIXTURE_ID, + "trace": _load_json(EXAMPLE_FIXTURE_PATH / "trace.json"), + "state": _load_json(EXAMPLE_FIXTURE_PATH / "state.json"), + "dependency_graph": _load_json(EXAMPLE_FIXTURE_PATH / "dependency_graph.json"), + } + + +def build_mcp_context_layer_example_artifact() -> dict[str, Any]: + replay_payload = build_replay_payload(_load_fixture_context()) + validation = validate_replay_payload(replay_payload) + prompt_context = render_prompt_context({**replay_payload, "validation": validation}) + + return { + "artifact_id": ARTIFACT_ID, + "evaluation_mode": "deterministic", + "example": { + "fixture_id": EXAMPLE_FIXTURE_ID, + "prompt_context": prompt_context, + "replay_payload": replay_payload, + "source_fixture_path": f"fixtures/{EXAMPLE_FIXTURE_ID}/original", + "validation": validation, + }, + "external_apis": "none", + "generated_by": "McpContextLayerExampleArtifactGenerator", + "llm_judges": "none", + "schema_version": "mcp_context_layer_example.v1", + "version": "1.0", + } + + +def generate_mcp_context_layer_example_artifact(output_path: Path = OUTPUT_PATH) -> Path: + artifact = build_mcp_context_layer_example_artifact() + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return output_path + + +def main() -> int: + output_path = generate_mcp_context_layer_example_artifact() + print(output_path.relative_to(REPO_ROOT).as_posix()) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_mcp_context_layer.py b/tests/test_mcp_context_layer.py index eaa3607..d18803a 100644 --- a/tests/test_mcp_context_layer.py +++ b/tests/test_mcp_context_layer.py @@ -13,9 +13,15 @@ save_context, validate_replay_payload, ) +from scripts.generate_mcp_context_layer_example_artifact import ( + ARTIFACT_ID, + EXAMPLE_FIXTURE_ID, + generate_mcp_context_layer_example_artifact, +) FIXTURE_ROOT = Path("fixtures/mcp_trace_replay_v1/original") +ARTIFACT_PATH = Path("artifacts/mcp_context_layer_example.json") def _load_fixture_context() -> dict[str, object]: @@ -224,3 +230,57 @@ def test_render_prompt_context_includes_validation_status_only_when_present() -> "failures: TOOL_ORDER_VIOLATION", ] assert "issues" not in rendered + + +def test_mcp_context_layer_artifact_matches_generator_output(tmp_path: Path) -> None: + output_path = tmp_path / "mcp_context_layer_example.json" + generate_mcp_context_layer_example_artifact(output_path) + + assert json.loads(output_path.read_text(encoding="utf-8")) == json.loads(ARTIFACT_PATH.read_text(encoding="utf-8")) + + +def test_mcp_context_layer_artifact_has_stable_schema_and_content() -> None: + artifact = json.loads(ARTIFACT_PATH.read_text(encoding="utf-8")) + + assert set(artifact.keys()) == { + "artifact_id", + "evaluation_mode", + "example", + "external_apis", + "generated_by", + "llm_judges", + "schema_version", + "version", + } + assert artifact["artifact_id"] == ARTIFACT_ID + assert artifact["schema_version"] == "mcp_context_layer_example.v1" + assert artifact["version"] == "1.0" + assert artifact["evaluation_mode"] == "deterministic" + assert artifact["llm_judges"] == "none" + assert artifact["external_apis"] == "none" + + example = artifact["example"] + assert example["fixture_id"] == EXAMPLE_FIXTURE_ID + assert example["source_fixture_path"] == "fixtures/mcp_trace_replay_v1/original" + assert example["validation"] == { + "admissible": True, + "failure_labels": [], + "issues": [], + } + assert example["replay_payload"] == build_replay_payload(_load_fixture_context()) + assert example["prompt_context"] == render_prompt_context( + { + **example["replay_payload"], + "validation": example["validation"], + } + ) + + +def test_mcp_context_layer_artifact_excludes_raw_trace_state_and_graph() -> None: + artifact_text = ARTIFACT_PATH.read_text(encoding="utf-8") + + assert '"events"' not in artifact_text + assert '"state_version"' not in artifact_text + assert '"graph_version"' not in artifact_text + assert '"dependency_graph"' not in artifact_text + assert '"permission_scopes"' not in artifact_text From c98307af2635be094b9462fb7eae5cb39b2bc462 Mon Sep 17 00:00:00 2001 From: ProfRandom92 Date: Fri, 22 May 2026 10:01:12 +0200 Subject: [PATCH 3/3] test: validate MCP context artifact regeneration --- tests/test_mcp_context_layer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_mcp_context_layer.py b/tests/test_mcp_context_layer.py index d18803a..5b2423b 100644 --- a/tests/test_mcp_context_layer.py +++ b/tests/test_mcp_context_layer.py @@ -236,7 +236,7 @@ def test_mcp_context_layer_artifact_matches_generator_output(tmp_path: Path) -> output_path = tmp_path / "mcp_context_layer_example.json" generate_mcp_context_layer_example_artifact(output_path) - assert json.loads(output_path.read_text(encoding="utf-8")) == json.loads(ARTIFACT_PATH.read_text(encoding="utf-8")) + assert output_path.read_text(encoding="utf-8") == ARTIFACT_PATH.read_text(encoding="utf-8") def test_mcp_context_layer_artifact_has_stable_schema_and_content() -> None: