diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2d8dfb2..e6ddbee 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,4 +41,4 @@ jobs: - name: Install MCP extra run: python -m pip install --disable-pip-version-check -e '.[mcp]' - name: MCP protocol tests - run: python -m unittest tests.test_feedback_ledger tests.test_mcp_adapter -v + run: python -m unittest tests.test_feedback_ledger tests.test_mcp_adapter tests.test_real_task_shadow_v2 -v diff --git a/docs/optional-mcp-interface.md b/docs/optional-mcp-interface.md index 422c33c..497e9aa 100644 --- a/docs/optional-mcp-interface.md +++ b/docs/optional-mcp-interface.md @@ -218,6 +218,8 @@ In this deployment, trace handles expire ; feedback after expi `trace_expires_at` は自動 expiry がなければ `null`、retention があれば Unix timestamp seconds とする。 +`search` はさらに `effective_config_provenance` を必ず返す。これは実際に trace を生成した `EngineConfig` を、ranking と graph propagation に影響する `effective_config.retrieval` と、feedback mutation だけに影響する `effective_config.feedback` へ分離し、実際に選ばれた `search_surface`（`combined` / `relation`）を併記した機械可読 object である。各 config 区分と全体には、UTF-8・key sort・2-space indent・末尾 LF の canonical JSON bytes に対する `sha256:` fingerprint をそれぞれ `retrieval_config_fingerprint`、`feedback_config_fingerprint`、`full_config_fingerprint` として付ける。process default も明示値も、解決後の effective value を省略せず返す。`created_at` とこの provenance により、result-free feedback shadow v2 は同じ snapshot、query、limit、capture 時刻、retrieval config、search surface を exact replayできる。provenance の追加は serving retrieval/default、feedback policy、transport、deployment を変更しない。 + ### 5.5 Core mapping | MCP field or behavior | Current NGR API | diff --git a/docs/real-task-feedback-shadow-protocol-v2.md b/docs/real-task-feedback-shadow-protocol-v2.md new file mode 100644 index 0000000..16fb81f --- /dev/null +++ b/docs/real-task-feedback-shadow-protocol-v2.md @@ -0,0 +1,63 @@ +# Real-task feedback shadow protocol v2 + +## 1. 目的と境界 + +`real-task-feedback-shadow-v2` は、live MCP search trace を frozen SQLite snapshot 上で exact replay できる result-free capture / replay 契約である。v1 は packet に effective runtime config と検索時刻を保存せず、固定 shadow arm config と synthetic time を retrieval verification に使うため、serving retrieval と feedback policy の差を分離できない。v2 は v1 artifact を書き換えず、この欠落だけを superseding protocol として修正する。 + +この protocol commit は placeholder identity だけを含み、実 task packet、実 source identity、observed outcome、aggregate result を含めない。serving retrieval/default、local registration、remote deployment、Issue #85 の frozen/observed artifact、Issue #89 の事前登録 gate を変更しない。Issue #90 は slot を作成していないため再利用しない。 + +## 2. Capture contract + +MCP `search` success output は trace の `created_at` と `effective_config_provenance` を返す。provenance は `EngineConfig` の全 field を次の二つへ分離し、省略せず、実際に trace を生成した `search_surface`（`combined` / `relation`）を併記する。 + +- `retrieval`: `sparse_weight`、`dense_weight`、`entry_weight`、`graph_weight`、`seed_count`、`max_hops`、`hop_decay`、`activation_half_life_seconds`、`maximum_activation`、path / propagation bound、activation strategy とその係数、dense / graph switch、normalization、fusion、`rrf_k` +- `feedback`: `feedback_learning_rate`、`sibling_feedback_normalization`、`maximum_edge_weight`、`relation_feedback_evidence_quorum`、`confirmed_outcome_reinforcement`、`confirmation_decay_ratio` + +各 config 区分と全体は canonical JSON bytes（UTF-8、key sort、2-space indent、末尾 LF）の SHA-256 を持つ。packet の `capture` は `searched_at`、`search_surface`、分離済み effective config、三 fingerprint を保存する。さらに query、limit、candidate identity / content hash、credited path、検索時刻、search surface、config と fingerprint をまとめた `capture_fingerprint` を保存する。通常の field 書換え、欠落、未知 field、非 canonical config は fail closed とする。 + +packet schema は `tests/fixtures/real_task_shadow_v2.packet-schema.json`、result schema は `tests/fixtures/real_task_shadow_v2.result-schema.json` を正本とする。v1 の task、snapshot、candidate、source-use、objective evidence、efficiency、append-only correction semantics は維持する。 + +## 3. Snapshot verification + +packet verifier は次を順に満たす。 + +1. canonical packet validation と全 config field の `EngineConfig` round-trip を行う。 +2. fingerprint と capture binding を再計算する。 +3. live WAL / journal sidecar を拒否し、snapshot byte hash を照合する。 +4. snapshot の fresh clone を開き、packet の capture effective config、`search_surface`、`searched_at` で同じ query / limit を同じ検索surfaceへ渡す。 +5. candidate 順、source URL、content hash、used node、credited path を exact 比較する。 + +config、timestamp、candidate、source identity、content、path、fingerprint のいずれかが不整合なら packet は登録・replay できない。 + +## 4. Two-arm shadow replay + +ordered batch ごとに arm あたり一つの fresh snapshot clone を作り、slot 1 から累積 replay する。全 packet は同じ snapshot hash と同じ capture effective config を共有し、capture timestamp は slot 順に厳密増加する。 + +両 arm は capture config 全体を基底にし、次の feedback-only field だけを上書きする。 + +| arm | feedback override | +| --- | --- | +| `used_q3_s1` | `relation_feedback_evidence_quorum=3`、`confirmed_outcome_reinforcement=false`、`confirmation_decay_ratio=null`、`sibling_feedback_normalization=1.0` | +| `confirmed_r05_s1` | `relation_feedback_evidence_quorum=1`、`confirmed_outcome_reinforcement=true`、`confirmation_decay_ratio=0.5`、`sibling_feedback_normalization=1.0` | + +`feedback_learning_rate` と `maximum_edge_weight` は capture effective value を共通利用する。retrieval field と capture `search_surface` は一つも上書きしない。result は各 arm の full effective config、search surface、retrieval / feedback / full fingerprint、override field list を保存し、二 arm の retrieval config / search surface と capture provenance が byte-semantic に一致しなければ拒否する。 + +各 packet は capture timestamp で pre-feedback relation trace を再計算し、candidate 順と credited path を packet と照合してから source-use / outcome を適用する。同じ idempotency key の再送が同じ semantic receipt を返すこと、同じ batch の再実行が同じ semantic result を返すこと、source snapshot hash が不変であることを必須とする。stored result verifier は packet / registry と snapshot から batch 全体を再計算し、result と exact 比較する。 + +## 5. Registry と one-time observation boundary + +registry writer は exclusive lock の内側で scan、slot / correction validation、exclusive create を行う。新規 root packet は slot 1 から連番、correction は immutable task / snapshot / retrieval / source-use / capture を保持する。異なる packet ID が同一 slot を競合した場合、一つだけが作成される。 + +この result-free commit の placeholder probe は registry へ永続登録しない。実 task stream の capture と一回だけの final aggregate は別の事前登録 observation issue だけが開始できる。development / holdout、#85 result、#90 task の遡及利用、観測後の再実行・再集計は行わない。 + +## 6. Commands + +```text +python tools/run_real_task_shadow_v2.py probe --fixture tests/fixtures/real_task_shadow_v2.placeholder.json +python tools/run_real_task_shadow_v2.py capture --input PACKET.json --registry-dir LOCAL_REGISTRY +python tools/run_real_task_shadow_v2.py verify-packet --packet PACKET.json --snapshot SNAPSHOT.db +python tools/run_real_task_shadow_v2.py replay --registry-dir LOCAL_REGISTRY --snapshot SNAPSHOT.db --output RESULT.json +python tools/run_real_task_shadow_v2.py verify-result --result RESULT.json --registry-dir LOCAL_REGISTRY --snapshot SNAPSHOT.db +``` + +`probe` は modern timestamp、serving-default retrieval mix `0.55 / 0.45`、relation path、placeholder-only source を使い、実 MCP adapter search から packet capture、snapshot verification、二 arm replay、exact result verification までを一巡する。registered output は生成しない。 diff --git a/src/neuron_graph_rag/config_provenance.py b/src/neuron_graph_rag/config_provenance.py new file mode 100644 index 0000000..c9cc864 --- /dev/null +++ b/src/neuron_graph_rag/config_provenance.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict, fields +from typing import Any, Mapping + +from .evidence_feedback import EngineConfig, NeuronGraphRAG + + +FEEDBACK_CONFIG_FIELDS = ( + "feedback_learning_rate", + "sibling_feedback_normalization", + "maximum_edge_weight", + "relation_feedback_evidence_quorum", + "confirmed_outcome_reinforcement", + "confirmation_decay_ratio", +) +SEARCH_SURFACES = ("combined", "relation") + + +def config_fingerprint(value: Mapping[str, Any]) -> str: + canonical = ( + json.dumps(dict(value), ensure_ascii=False, indent=2, sort_keys=True) + "\n" + ).encode("utf-8") + return "sha256:" + hashlib.sha256(canonical).hexdigest() + + +def effective_config(config: EngineConfig) -> dict[str, dict[str, Any]]: + raw = asdict(config) + names = {field.name for field in fields(EngineConfig)} + if set(raw) != names: + raise RuntimeError("EngineConfig serialization is incomplete") + return { + "retrieval": { + name: raw[name] + for name in sorted(names - set(FEEDBACK_CONFIG_FIELDS)) + }, + "feedback": { + name: raw[name] + for name in sorted(FEEDBACK_CONFIG_FIELDS) + }, + } + + +def effective_config_provenance(config: EngineConfig) -> dict[str, Any]: + value = effective_config(config) + return { + "effective_config": value, + "retrieval_config_fingerprint": config_fingerprint(value["retrieval"]), + "feedback_config_fingerprint": config_fingerprint(value["feedback"]), + "full_config_fingerprint": config_fingerprint(value), + } + + +def effective_search_surface(config: EngineConfig) -> str: + if ( + config.confirmed_outcome_reinforcement + or config.sibling_feedback_normalization > 0.0 + ): + return "relation" + return "combined" + + +def search_with_surface( + engine: NeuronGraphRAG, + query: str, + *, + limit: int, + search_surface: str, + now: float | None = None, +) -> Any: + if search_surface == "combined": + return engine.search(query, limit=limit, now=now) + if search_surface == "relation": + return engine.search_channels(query, limit=limit, now=now).relation + raise ValueError(f"unknown search surface: {search_surface}") diff --git a/src/neuron_graph_rag/real_task_shadow_v2.py b/src/neuron_graph_rag/real_task_shadow_v2.py new file mode 100644 index 0000000..718513a --- /dev/null +++ b/src/neuron_graph_rag/real_task_shadow_v2.py @@ -0,0 +1,541 @@ +from __future__ import annotations + +import math +import shutil +import tempfile +from pathlib import Path +from typing import Any, Mapping, Sequence + +from . import real_task_shadow as v1 +from .config_provenance import ( + FEEDBACK_CONFIG_FIELDS as FEEDBACK_FIELDS, + SEARCH_SURFACES, + config_fingerprint as _fingerprint, + effective_config, + effective_config_provenance, + effective_search_surface, + search_with_surface, +) +from .evidence_feedback import EngineConfig, NeuronGraphRAG +from .feedback import FeedbackLedger +from .models import SourceUseEvent + + +PROTOCOL_ID = "real-task-feedback-shadow-v2" +SCHEMA_VERSION = 2 +ARM_IDS = v1.ARM_IDS +SOURCE_USE_STAGES = v1.SOURCE_USE_STAGES +ARM_OVERRIDES = { + "used_q3_s1": { + "relation_feedback_evidence_quorum": 3, + "confirmed_outcome_reinforcement": False, + "confirmation_decay_ratio": None, + "sibling_feedback_normalization": 1.0, + }, + "confirmed_r05_s1": { + "relation_feedback_evidence_quorum": 1, + "confirmed_outcome_reinforcement": True, + "confirmation_decay_ratio": 0.5, + "sibling_feedback_normalization": 1.0, + }, +} +def _capture_payload(packet: Mapping[str, Any]) -> dict[str, Any]: + return { + "query": packet["retrieval"]["query"], + "limit": packet["retrieval"]["limit"], + "candidates": packet["retrieval"]["candidates"], + "credited_path": packet["retrieval"]["credited_path"], + "searched_at": packet["capture"]["searched_at"], + "search_surface": packet["capture"]["search_surface"], + "effective_config": packet["capture"]["effective_config"], + "retrieval_config_fingerprint": packet["capture"]["retrieval_config_fingerprint"], + "feedback_config_fingerprint": packet["capture"]["feedback_config_fingerprint"], + "full_config_fingerprint": packet["capture"]["full_config_fingerprint"], + } + + +def bind_capture_fingerprint(packet: Mapping[str, Any]) -> str: + return _fingerprint(_capture_payload(packet)) + + +def _as_v1_packet(packet: Mapping[str, Any]) -> dict[str, Any]: + converted = {key: value for key, value in packet.items() if key != "capture"} + converted["schema_version"] = v1.SCHEMA_VERSION + converted["protocol_id"] = v1.PROTOCOL_ID + return converted + + +def validate_packet(packet: Mapping[str, Any]) -> None: + required = { + "schema_version", "protocol_id", "packet_id", "slot", + "supersedes_packet_id", "task", "database_snapshot", "retrieval", + "source_use", "outcome", "efficiency", "captured_at", "capture", + } + v1._exact_keys(packet, required, "packet") + if packet["schema_version"] != SCHEMA_VERSION or packet["protocol_id"] != PROTOCOL_ID: + raise ValueError("unknown packet protocol") + v1.validate_packet(_as_v1_packet(packet)) + capture = v1._mapping(packet["capture"], "capture") + v1._exact_keys( + capture, + { + "searched_at", "search_surface", "effective_config", + "retrieval_config_fingerprint", + "feedback_config_fingerprint", "full_config_fingerprint", + "capture_fingerprint", + }, + "capture", + ) + searched_at = capture["searched_at"] + if isinstance(searched_at, bool) or not isinstance(searched_at, (int, float)) or not math.isfinite(searched_at) or searched_at <= 0: + raise ValueError("capture.searched_at must be a positive finite Unix timestamp") + if capture["search_surface"] not in SEARCH_SURFACES: + raise ValueError("capture.search_surface is unknown") + config = v1._mapping(capture["effective_config"], "capture.effective_config") + v1._exact_keys(config, {"retrieval", "feedback"}, "capture.effective_config") + retrieval = v1._mapping(config["retrieval"], "capture.effective_config.retrieval") + feedback = v1._mapping(config["feedback"], "capture.effective_config.feedback") + engine_config = EngineConfig(**{**retrieval, **feedback}) + expected = effective_config(engine_config) + if config != expected: + raise ValueError("capture effective config is not canonical or complete") + if capture["search_surface"] != effective_search_surface(engine_config): + raise ValueError("capture search surface does not match effective config") + for key, value in ( + ("retrieval_config_fingerprint", retrieval), + ("feedback_config_fingerprint", feedback), + ("full_config_fingerprint", config), + ): + v1._sha256(capture[key], f"capture.{key}") + if capture[key] != _fingerprint(value): + raise ValueError(f"capture {key} mismatch") + v1._sha256(capture["capture_fingerprint"], "capture.capture_fingerprint") + if capture["capture_fingerprint"] != bind_capture_fingerprint(packet): + raise ValueError("capture fingerprint mismatch") + + +def _read_registry(registry: Path) -> list[dict[str, Any]]: + packets = [v1.read_canonical_json(path) for path in sorted(registry.glob("*.json"))] + for packet in packets: + validate_packet(packet) + return packets + + +def capture_packet(packet: Mapping[str, Any], registry_dir: str | Path) -> Path: + validate_packet(packet) + registry = Path(registry_dir) + with v1._registry_lock(registry): + packets = _read_registry(registry) + packet_id = str(packet["packet_id"]) + if any(item["packet_id"] == packet_id for item in packets): + raise FileExistsError(f"packet ID already registered: {packet_id}") + supersedes = packet["supersedes_packet_id"] + if supersedes is None: + expected_slot = max((int(item["slot"]) for item in packets), default=0) + 1 + if packet["slot"] != expected_slot: + raise ValueError(f"new packet must use sequential slot {expected_slot}") + else: + prior = next((item for item in packets if item["packet_id"] == supersedes), None) + if prior is None: + raise ValueError("superseded packet is not registered") + if any(item["supersedes_packet_id"] == supersedes for item in packets): + raise ValueError("superseded packet already has a successor") + if packet["slot"] != prior["slot"]: + raise ValueError("superseding packet must retain its slot") + for field in ("task", "database_snapshot", "retrieval", "source_use", "capture"): + if packet[field] != prior[field]: + raise ValueError(f"superseding packet changed immutable field: {field}") + output = registry / f"{int(packet['slot']):04d}-{packet_id}.json" + v1.write_json_exclusive(output, packet) + return output + + +def load_effective_registry(registry_dir: str | Path) -> list[dict[str, Any]]: + registry = Path(registry_dir) + with v1._registry_lock(registry): + packets = _read_registry(registry) + if not packets: + raise ValueError("packet registry is empty") + by_id = {str(packet["packet_id"]): packet for packet in packets} + if len(by_id) != len(packets): + raise ValueError("registry packet IDs must be unique") + roots: dict[int, dict[str, Any]] = {} + successors: dict[str, dict[str, Any]] = {} + for packet in packets: + prior = packet["supersedes_packet_id"] + if prior is None: + slot = int(packet["slot"]) + if slot in roots: + raise ValueError("registry contains duplicate root slot") + roots[slot] = packet + else: + if prior not in by_id or prior in successors: + raise ValueError("invalid registry correction chain") + successors[str(prior)] = packet + if sorted(roots) != list(range(1, len(roots) + 1)): + raise ValueError("registry root slots must be sequential from one") + result = [] + reachable: set[str] = set() + for slot in sorted(roots): + current = roots[slot] + while str(current["packet_id"]) in successors: + reachable.add(str(current["packet_id"])) + current = successors[str(current["packet_id"])] + if current["slot"] != slot: + raise ValueError("registry correction changed slot") + reachable.add(str(current["packet_id"])) + result.append(current) + if reachable != set(by_id): + raise ValueError("registry contains an unreachable correction") + return result + + +def _engine_config(packet: Mapping[str, Any], arm_id: str) -> EngineConfig: + if arm_id not in ARM_OVERRIDES: + raise ValueError(f"unknown arm: {arm_id}") + captured = packet["capture"]["effective_config"] + values = {**captured["retrieval"], **captured["feedback"]} + values.update(ARM_OVERRIDES[arm_id]) + return EngineConfig(**values) + + +def _capture_engine_config(packet: Mapping[str, Any]) -> EngineConfig: + captured = packet["capture"]["effective_config"] + return EngineConfig(**{**captured["retrieval"], **captured["feedback"]}) + + +def _arm_provenance(packet: Mapping[str, Any], arm_id: str) -> dict[str, Any]: + provenance = effective_config_provenance(_engine_config(packet, arm_id)) + provenance["search_surface"] = packet["capture"]["search_surface"] + provenance["overridden_feedback_fields"] = sorted(ARM_OVERRIDES[arm_id]) + return provenance + + +def verify_packet_against_snapshot(packet: Mapping[str, Any], snapshot_path: str | Path) -> None: + validate_packet(packet) + snapshot = Path(snapshot_path) + for suffix in ("-wal", "-journal"): + sidecar = Path(str(snapshot) + suffix) + if sidecar.exists() and sidecar.stat().st_size: + raise ValueError("database snapshot has an uncheckpointed live sidecar") + if v1.sha256_file(snapshot) != packet["database_snapshot"]["sha256"]: + raise ValueError("database snapshot hash mismatch") + with tempfile.TemporaryDirectory() as directory: + clone = Path(directory) / "verify.db" + shutil.copyfile(snapshot, clone) + with NeuronGraphRAG(clone, config=_capture_engine_config(packet)) as engine: + nodes = {node.node_id: node for node in engine.store.list_nodes()} + for candidate in packet["retrieval"]["candidates"]: + node = nodes.get(candidate["node_id"]) + if node is None or node.metadata.get("source_url") != candidate["source_url"]: + raise ValueError("captured source identity mismatch") + if v1.sha256_text(node.text) != candidate["content_sha256"]: + raise ValueError("captured source content hash mismatch") + trace = search_with_surface( + engine, + str(packet["retrieval"]["query"]), + limit=int(packet["retrieval"]["limit"]), + search_surface=str(packet["capture"]["search_surface"]), + now=float(packet["capture"]["searched_at"]), + ) + expected_ids = [item["node_id"] for item in packet["retrieval"]["candidates"]] + if [hit.node.node_id for hit in trace.hits] != expected_ids: + raise ValueError("captured candidates do not match exact runtime replay") + hit = next(item for item in trace.hits if item.node.node_id == packet["retrieval"]["used_node_id"]) + if v1._selected_path(hit) != packet["retrieval"]["credited_path"]: + raise ValueError("credited path mismatch") + + +def _validate_batch(packets: Sequence[Mapping[str, Any]], snapshot_path: str | Path) -> None: + if not packets: + raise ValueError("replay requires at least one packet") + for packet in packets: + verify_packet_against_snapshot(packet, snapshot_path) + if [packet["slot"] for packet in packets] != list(range(1, len(packets) + 1)): + raise ValueError("batch packets must be in sequential slot order") + if len({packet["packet_id"] for packet in packets}) != len(packets): + raise ValueError("batch packet IDs must be unique") + if len({packet["database_snapshot"]["sha256"] for packet in packets}) != 1: + raise ValueError("batch packets must share the exact replay snapshot") + capture_configs = [packet["capture"]["effective_config"] for packet in packets] + if any(config != capture_configs[0] for config in capture_configs[1:]): + raise ValueError("batch packets must share one effective capture config") + search_surfaces = { + packet["capture"]["search_surface"] + for packet in packets + } + if len(search_surfaces) != 1: + raise ValueError("batch packets must share one capture search surface") + times = [float(packet["capture"]["searched_at"]) for packet in packets] + if any(right <= left for left, right in zip(times, times[1:])): + raise ValueError("batch capture timestamps must increase in slot order") + + +def _surface_ranking(trace: Any, node_id: str) -> dict[str, Any]: + for position, hit in enumerate(trace.hits, start=1): + if hit.node.node_id == node_id: + score = ( + hit.channel_score + if hasattr(hit, "channel_score") + else hit.final_score + ) + return {"rank": getattr(hit, "rank", position), "score": score} + return {"rank": len(trace.hits) + 1, "score": 0.0} + + +def _run_packet(engine: NeuronGraphRAG, packet: Mapping[str, Any], arm_id: str) -> dict[str, Any]: + clock = float(packet["capture"]["searched_at"]) + before_edges = v1._edge_state(engine) + trace = search_with_surface( + engine, + packet["retrieval"]["query"], + limit=packet["retrieval"]["limit"], + search_surface=packet["capture"]["search_surface"], + now=clock, + ) + expected_ids = [item["node_id"] for item in packet["retrieval"]["candidates"]] + if [hit.node.node_id for hit in trace.hits] != expected_ids: + raise ValueError("cumulative replay candidates diverged from capture") + used = packet["retrieval"]["used_node_id"] + hit = next(item for item in trace.hits if item.node.node_id == used) + if v1._selected_path(hit) != packet["retrieval"]["credited_path"]: + raise ValueError("cumulative replay credited path diverged from capture") + ledger = FeedbackLedger(engine) + events = tuple(SourceUseEvent(used, stage) for stage in SOURCE_USE_STAGES) + key = f"shadow-v2:{packet['packet_id']}:{arm_id}:use" + receipt = ledger.record_source_use(trace.trace_id, events, idempotency_key=key, now=clock + 0.1) + repeat = ledger.record_source_use(trace.trace_id, events, idempotency_key=key, now=clock + 0.9) + if v1._source_use_semantics(receipt) != v1._source_use_semantics(repeat): + raise RuntimeError("source-use idempotency replay mismatch") + outcome_value = None + status = packet["outcome"]["status"] + if status != "pending": + outcome_key = f"shadow-v2:{packet['packet_id']}:{arm_id}:outcome" + outcome = ledger.record_outcome( + trace.trace_id, [used], status, packet["outcome"]["summary"], + idempotency_key=outcome_key, external_ref=packet["outcome"]["external_ref"], now=clock + 0.2, + ) + repeat_outcome = ledger.record_outcome( + trace.trace_id, [used], status, packet["outcome"]["summary"], + idempotency_key=outcome_key, external_ref=packet["outcome"]["external_ref"], now=clock + 0.9, + ) + outcome_value = v1._outcome_semantics(outcome) + if outcome_value != v1._outcome_semantics(repeat_outcome): + raise RuntimeError("outcome idempotency replay mismatch") + after_edges = v1._edge_state(engine) + post = search_with_surface( + engine, + packet["retrieval"]["query"], + limit=packet["retrieval"]["limit"], + search_surface=packet["capture"]["search_surface"], + now=clock + 0.3, + ) + before_rank = _surface_ranking(trace, used) + after_rank = _surface_ranking(post, used) + return { + "packet_id": packet["packet_id"], "slot": packet["slot"], + "before": before_rank, "after": after_rank, + "rank_delta": after_rank["rank"] - before_rank["rank"], + "score_delta": after_rank["score"] - before_rank["score"], + "edge_state_before": before_edges, "edge_state_after": after_edges, + "edge_delta": v1._edge_delta(before_edges, after_edges), + "non_target_churn": v1._non_target_churn(trace, post, used), + "source_use": v1._source_use_semantics(receipt), "outcome": outcome_value, + "idempotency_replay": True, + } + + +def _run_arm(packets: Sequence[Mapping[str, Any]], snapshot_path: str | Path, arm_id: str) -> dict[str, Any]: + provenance = _arm_provenance(packets[0], arm_id) + with tempfile.TemporaryDirectory() as directory: + clone = Path(directory) / "shadow.db" + shutil.copyfile(snapshot_path, clone) + with NeuronGraphRAG(clone, config=_engine_config(packets[0], arm_id)) as engine: + return { + "arm_id": arm_id, + "policy": "used" if arm_id == "used_q3_s1" else "confirmed", + **provenance, + "packets": [_run_packet(engine, packet, arm_id) for packet in packets], + "final_edge_state": v1._edge_state(engine), + } + + +def validate_result(result: Mapping[str, Any]) -> None: + v1._exact_keys( + result, + { + "schema_version", "protocol_id", "packet_ids", "slots", + "snapshot_sha256", "capture_config", "capture_search_surface", + "arms", "comparison", "efficiency", "replay", + }, + "result", + ) + if result["schema_version"] != SCHEMA_VERSION or result["protocol_id"] != PROTOCOL_ID: + raise ValueError("unknown result protocol") + arms = v1._mapping(result["arms"], "arms") + v1._exact_keys(arms, set(ARM_IDS), "arms") + retrievals = [] + for arm_id in ARM_IDS: + arm = v1._mapping(arms[arm_id], f"arms.{arm_id}") + if arm.get("arm_id") != arm_id or arm.get("overridden_feedback_fields") != sorted(ARM_OVERRIDES[arm_id]): + raise ValueError("arm policy provenance mismatch") + if arm.get("search_surface") != result["capture_search_surface"]: + raise ValueError("arm search surface diverged from capture") + provenance = { + "effective_config": arm.get("effective_config"), + "retrieval_config_fingerprint": arm.get("retrieval_config_fingerprint"), + "feedback_config_fingerprint": arm.get("feedback_config_fingerprint"), + "full_config_fingerprint": arm.get("full_config_fingerprint"), + } + expected = effective_config_provenance(EngineConfig(**{**provenance["effective_config"]["retrieval"], **provenance["effective_config"]["feedback"]})) + if provenance != expected: + raise ValueError("arm effective config provenance mismatch") + retrievals.append(provenance["effective_config"]["retrieval"]) + if retrievals[0] != retrievals[1] or retrievals[0] != result["capture_config"]["retrieval"]: + raise ValueError("shadow arms do not share capture retrieval config") + + +def replay_packets(packets: Sequence[Mapping[str, Any]], snapshot_path: str | Path) -> dict[str, Any]: + ordered = list(packets) + _validate_batch(ordered, snapshot_path) + source_hash = v1.sha256_file(snapshot_path) + arms = {} + for arm_id in ARM_IDS: + first = _run_arm(ordered, snapshot_path, arm_id) + if first != _run_arm(ordered, snapshot_path, arm_id): + raise RuntimeError(f"non-deterministic replay: {arm_id}") + arms[arm_id] = first + if v1.sha256_file(snapshot_path) != source_hash: + raise RuntimeError("source snapshot changed during replay") + result = { + "schema_version": SCHEMA_VERSION, "protocol_id": PROTOCOL_ID, + "packet_ids": [packet["packet_id"] for packet in ordered], + "slots": [packet["slot"] for packet in ordered], "snapshot_sha256": source_hash, + "capture_config": ordered[0]["capture"]["effective_config"], + "capture_search_surface": ordered[0]["capture"]["search_surface"], + "arms": arms, "comparison": v1._comparison(arms), + "efficiency": [{"packet_id": packet["packet_id"], **packet["efficiency"]} for packet in ordered], + "replay": { + "fresh_clone_per_arm": True, "cumulative_slot_order": True, + "capture_time_reused": True, "capture_retrieval_config_reused": True, + "repeated_semantic_replay": 2, "deterministic": True, + "source_snapshot_unchanged": True, + }, + } + validate_result(result) + return result + + +def replay_packet(packet: Mapping[str, Any], snapshot_path: str | Path) -> dict[str, Any]: + return replay_packets([packet], snapshot_path) + + +def replay_registry(registry_dir: str | Path, snapshot_path: str | Path) -> dict[str, Any]: + return replay_packets(load_effective_registry(registry_dir), snapshot_path) + + +def verify_result_against_packets(result: Mapping[str, Any], packets: Sequence[Mapping[str, Any]], snapshot_path: str | Path) -> None: + validate_result(result) + if result != replay_packets(packets, snapshot_path): + raise ValueError("stored result does not match exact semantic replay") + + +def verify_result_against_registry(result: Mapping[str, Any], registry_dir: str | Path, snapshot_path: str | Path) -> None: + verify_result_against_packets(result, load_effective_registry(registry_dir), snapshot_path) + + +def build_packet_from_search(fixture: Mapping[str, Any], search: Mapping[str, Any], snapshot_path: str | Path) -> dict[str, Any]: + if fixture.get("placeholder_only") is not True: + raise ValueError("placeholder fixture must declare placeholder_only") + if any(not str(node["node_id"]).startswith("placeholder-") or "example.invalid" not in str(node["source_url"]) for node in fixture["documents"]): + raise ValueError("fixture contains a non-placeholder source identity") + seed = fixture["packet_seed"] + hits = search["hits"] + used = seed["used_node_id"] + hit = next(item for item in hits if item["node_id"] == used) + paths = [path for path in hit["paths"] if path["steps"]] + selected = max(paths, key=lambda path: (path["contribution"], path["seed_id"])) + provenance = search["effective_config_provenance"] + packet = { + "schema_version": SCHEMA_VERSION, "protocol_id": PROTOCOL_ID, + "packet_id": seed["packet_id"], "slot": seed.get("slot", 1), + "supersedes_packet_id": None, "task": seed["task"], + "database_snapshot": {"sha256": v1.sha256_file(snapshot_path)}, + "retrieval": { + "query": search["query"], "limit": seed["limit"], + "candidates": [ + {"node_id": item["node_id"], "source_url": item["metadata"]["source_url"], "content_sha256": v1.sha256_text(item["text"])} + for item in hits + ], + "used_node_id": used, + "credited_path": { + "node_id": used, "seed_id": selected["seed_id"], + "steps": [{key: step[key] for key in ("source_id", "target_id", "edge_type")} for step in selected["steps"]], + }, + }, + "source_use": [{"stage": stage, "node_id": used} for stage in SOURCE_USE_STAGES], + "outcome": seed["outcome"], "efficiency": seed["efficiency"], + "captured_at": seed["captured_at"], + "capture": { + "searched_at": search["created_at"], + **provenance, + "capture_fingerprint": "sha256:" + "0" * 64, + }, + } + packet["capture"]["capture_fingerprint"] = bind_capture_fingerprint(packet) + validate_packet(packet) + return packet + + +def create_placeholder_snapshot(fixture: Mapping[str, Any], output: str | Path) -> None: + path = Path(output) + if path.exists(): + raise FileExistsError(path) + with NeuronGraphRAG(path) as engine: + for node in fixture["documents"]: + engine.add_document(node["node_id"], node["text"], metadata={"source_url": node["source_url"]}) + for edge in fixture["edges"]: + engine.add_edge(edge["source_id"], edge["target_id"], edge["edge_type"], weight=edge["weight"]) + + +def probe_placeholder(fixture_path: str | Path) -> dict[str, Any]: + from neuron_graph_rag_mcp.server import CONTRACT_VERSION, FeedbackMCPAdapter + + fixture = v1.read_canonical_json(fixture_path) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + snapshot = root / "placeholder.db" + create_placeholder_snapshot(fixture, snapshot) + adapter = FeedbackMCPAdapter( + snapshot, + config=EngineConfig( + relation_feedback_evidence_quorum=3, + sibling_feedback_normalization=1.0, + ), + ) + try: + search = adapter._search( + { + "contract_version": CONTRACT_VERSION, + "query": fixture["packet_seed"]["query"], + "limit": fixture["packet_seed"]["limit"], + } + ) + finally: + adapter.close() + packet = build_packet_from_search(fixture, search, snapshot) + registered = capture_packet(packet, root / "registry") + captured = v1.read_canonical_json(registered) + result = replay_packet(captured, snapshot) + output = root / "placeholder.result.json" + v1.write_json_exclusive(output, result) + verify_result_against_packets(v1.read_canonical_json(output), [captured], snapshot) + return { + "protocol_id": PROTOCOL_ID, + "placeholder_only": True, + "mcp_search_capture_round_trip": True, + "capture_time_reused": True, + "effective_config_provenance_verified": True, + "replay_round_trip": True, + "exclusive_writer_verified": True, + } diff --git a/src/neuron_graph_rag_mcp/server.py b/src/neuron_graph_rag_mcp/server.py index bf3f8d5..c207e46 100644 --- a/src/neuron_graph_rag_mcp/server.py +++ b/src/neuron_graph_rag_mcp/server.py @@ -17,6 +17,11 @@ from mcp.shared.exceptions import MCPError from neuron_graph_rag import FeedbackContractError, FeedbackLedger, SourceUseEvent +from neuron_graph_rag.config_provenance import ( + effective_config_provenance, + effective_search_surface, + search_with_surface, +) from neuron_graph_rag.evidence_feedback import EngineConfig, NeuronGraphRAG CONTRACT_VERSION = "ngr.mcp.feedback/v1" @@ -184,6 +189,28 @@ def _object(properties: dict[str, Any], required: list[str]) -> dict[str, Any]: "query": {"type": "string"}, "created_at": {"type": "number"}, "trace_expires_at": {"type": ["number", "null"]}, + "effective_config_provenance": _object( + { + "effective_config": _object( + { + "retrieval": {"type": "object"}, + "feedback": {"type": "object"}, + }, + ["retrieval", "feedback"], + ), + "search_surface": { + "type": "string", + "enum": ["combined", "relation"], + }, + "retrieval_config_fingerprint": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}, + "feedback_config_fingerprint": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}, + "full_config_fingerprint": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}, + }, + [ + "effective_config", "search_surface", "retrieval_config_fingerprint", + "feedback_config_fingerprint", "full_config_fingerprint", + ], + ), "hits": { "type": "array", "items": _object( @@ -223,7 +250,10 @@ def _object(properties: dict[str, Any], required: list[str]) -> dict[str, Any]: ), }, }, - ["contract_version", "trace_id", "query", "created_at", "trace_expires_at", "hits"], + [ + "contract_version", "trace_id", "query", "created_at", + "trace_expires_at", "effective_config_provenance", "hits", + ], ) SOURCE_USE_OUTPUT = _object( { @@ -380,13 +410,13 @@ def _search(self, data: dict[str, Any]) -> dict[str, Any]: if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 100: raise ValueError("limit must be an integer from 1 through 100") try: - if ( - self.engine.config.confirmed_outcome_reinforcement - or self.engine.config.sibling_feedback_normalization > 0.0 - ): - trace = self.engine.search_channels(query, limit=limit).relation - else: - trace = self.engine.search(query, limit=limit) + search_surface = effective_search_surface(self.engine.config) + trace = search_with_surface( + self.engine, + query, + limit=limit, + search_surface=search_surface, + ) except ValueError as error: if str(error) == "Cannot search an empty corpus": raise FeedbackContractError("empty_corpus", "local NGR corpus is empty") from error @@ -436,6 +466,10 @@ def _search(self, data: dict[str, Any]) -> dict[str, Any]: "query": trace.query, "created_at": trace.created_at, "trace_expires_at": None, + "effective_config_provenance": { + **effective_config_provenance(self.engine.config), + "search_surface": search_surface, + }, "hits": hits, } diff --git a/tests/fixtures/real_task_shadow_v2.manifest.json b/tests/fixtures/real_task_shadow_v2.manifest.json new file mode 100644 index 0000000..9736cde --- /dev/null +++ b/tests/fixtures/real_task_shadow_v2.manifest.json @@ -0,0 +1,51 @@ +{ + "arms": [ + "used_q3_s1", + "confirmed_r05_s1" + ], + "artifact_sha256": { + "docs/real-task-feedback-shadow-protocol-v2.md": "270cb257ea23a3f5d4e09ec5a522e7e1ab24171e115204fa35bfa0fdea8a25bd", + "src/neuron_graph_rag/config_provenance.py": "5cc42425c6576f330a96c629f341ac5ddf8fd3629f14181e307919d3ebd581e6", + "src/neuron_graph_rag/real_task_shadow_v2.py": "0a816da79e1f4fb97c9a7e9e80fd15657f06b5bd0787537c748c1d11cfa9cbef", + "src/neuron_graph_rag_mcp/server.py": "4f29b0b2172a6ced58b0b5a46dc0c4204dcf695d93b64e7b9e8b1124af44c571", + "tests/fixtures/real_task_shadow_v2.packet-schema.json": "13c549ae5e322d6d92105f83c5170d93bea3e1c267cbaac657fe372b9d074fe3", + "tests/fixtures/real_task_shadow_v2.placeholder.json": "7c7e6a2f8c2b811404afb5eb4fec71b8889877e80a7f1baaf5e63e3aa923ef23", + "tests/fixtures/real_task_shadow_v2.result-schema.json": "5d952ca06eed54b790aaba69d5336332f22fbcec7337d50ba1b7332c89a67add", + "tests/test_mcp_adapter.py": "68ea846d480214dbb229ec51da19cba64b2c21edd5f5e51c8fd7b971bdc4c08d", + "tests/test_real_task_shadow_v2.py": "8552b8601c38b8b63a9f692d942858c9f9b684647e5461db2b8e414ed7b567af", + "tools/run_real_task_shadow_v2.py": "c4cac4b927a50ea7100ca44e5c93e270e80e216b11fab0e3b619dc36067cb577" + }, + "capture_contract": { + "config_partition": "retrieval_and_feedback", + "fingerprints": [ + "retrieval_config_fingerprint", + "feedback_config_fingerprint", + "full_config_fingerprint", + "capture_fingerprint" + ], + "mcp_provenance": "effective_runtime_config", + "search_time": "captured_trace_created_at" + }, + "freeze_parent_commit": "062e3748d65e18a3f75b4e01ff9a183197d21c16", + "observation_status": "not_started", + "protocol_artifacts": { + "packet_schema": "tests/fixtures/real_task_shadow_v2.packet-schema.json", + "placeholder_fixture": "tests/fixtures/real_task_shadow_v2.placeholder.json", + "result_schema": "tests/fixtures/real_task_shadow_v2.result-schema.json" + }, + "protocol_id": "real-task-feedback-shadow-v2", + "registered_outputs": { + "observed_results": "artifacts/real-task-feedback-shadow-v2/observed", + "packet_registry": "artifacts/real-task-feedback-shadow-v2/packets" + }, + "registry_write": "exclusive_lock_scan_validate_reserve_write", + "replay_contract": { + "arm_clone_scope": "one_fresh_clone_per_arm_per_ordered_batch", + "capture_retrieval_config": "shared_across_arms", + "capture_time": "reused_per_packet", + "result_verification": "exact_semantic_replay_from_packets_and_snapshot", + "slot_order": "sequential_from_one" + }, + "result_free": true, + "supersedes_protocol_id": "real-task-feedback-shadow-v1" +} diff --git a/tests/fixtures/real_task_shadow_v2.packet-schema.json b/tests/fixtures/real_task_shadow_v2.packet-schema.json new file mode 100644 index 0000000..eaa8cca --- /dev/null +++ b/tests/fixtures/real_task_shadow_v2.packet-schema.json @@ -0,0 +1,135 @@ +{ + "$defs": { + "sha256": { + "pattern": "^sha256:[0-9a-f]{64}$", + "type": "string" + } + }, + "$id": "https://example.invalid/neuron-graph-rag/real-task-feedback-shadow-v2.packet.schema.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": false, + "properties": { + "capture": { + "additionalProperties": false, + "properties": { + "capture_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "effective_config": { + "additionalProperties": false, + "properties": { + "feedback": { + "type": "object" + }, + "retrieval": { + "type": "object" + } + }, + "required": [ + "feedback", + "retrieval" + ], + "type": "object" + }, + "feedback_config_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "full_config_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "retrieval_config_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "search_surface": { + "enum": [ + "combined", + "relation" + ], + "type": "string" + }, + "searched_at": { + "exclusiveMinimum": 0, + "type": "number" + } + }, + "required": [ + "capture_fingerprint", + "effective_config", + "feedback_config_fingerprint", + "full_config_fingerprint", + "retrieval_config_fingerprint", + "search_surface", + "searched_at" + ], + "type": "object" + }, + "captured_at": { + "minLength": 1, + "type": "string" + }, + "database_snapshot": { + "additionalProperties": false, + "properties": { + "sha256": { + "$ref": "#/$defs/sha256" + } + }, + "required": [ + "sha256" + ], + "type": "object" + }, + "efficiency": { + "type": "object" + }, + "outcome": { + "type": "object" + }, + "packet_id": { + "minLength": 1, + "type": "string" + }, + "protocol_id": { + "const": "real-task-feedback-shadow-v2" + }, + "retrieval": { + "type": "object" + }, + "schema_version": { + "const": 2 + }, + "slot": { + "minimum": 1, + "type": "integer" + }, + "source_use": { + "type": "array" + }, + "supersedes_packet_id": { + "type": [ + "string", + "null" + ] + }, + "task": { + "type": "object" + } + }, + "required": [ + "capture", + "captured_at", + "database_snapshot", + "efficiency", + "outcome", + "packet_id", + "protocol_id", + "retrieval", + "schema_version", + "slot", + "source_use", + "supersedes_packet_id", + "task" + ], + "title": "Real-task feedback shadow v2 packet", + "type": "object" +} diff --git a/tests/fixtures/real_task_shadow_v2.placeholder.json b/tests/fixtures/real_task_shadow_v2.placeholder.json new file mode 100644 index 0000000..2775f9d --- /dev/null +++ b/tests/fixtures/real_task_shadow_v2.placeholder.json @@ -0,0 +1,62 @@ +{ + "documents": [ + { + "node_id": "placeholder-runtime-seed", + "source_url": "https://example.invalid/placeholder/runtime-seed", + "text": "placeholder runtime provenance seed for exact MCP retrieval" + }, + { + "node_id": "placeholder-runtime-target", + "source_url": "https://example.invalid/placeholder/runtime-target", + "text": "placeholder runtime terminal reached through the captured relation" + }, + { + "node_id": "placeholder-runtime-sibling", + "source_url": "https://example.invalid/placeholder/runtime-sibling", + "text": "placeholder runtime sibling for feedback normalization" + } + ], + "edges": [ + { + "edge_type": "placeholder_runtime_link", + "source_id": "placeholder-runtime-seed", + "target_id": "placeholder-runtime-target", + "weight": 0.8 + }, + { + "edge_type": "placeholder_runtime_link", + "source_id": "placeholder-runtime-seed", + "target_id": "placeholder-runtime-sibling", + "weight": 0.2 + } + ], + "packet_seed": { + "captured_at": "2026-08-15T02:00:00Z", + "efficiency": { + "elapsed_seconds": null, + "research_count": null, + "token_count": null, + "tool_calls": null + }, + "limit": 3, + "outcome": { + "evidence": [], + "external_ref": null, + "status": "pending", + "summary": "placeholder runtime outcome pending" + }, + "packet_id": "placeholder-v2-packet-0001", + "query": "placeholder runtime provenance seed exact MCP retrieval", + "slot": 1, + "task": { + "base_commit": "0000000000000000000000000000000000000000", + "close_condition": "placeholder v2 close condition", + "eligible_at": "2026-08-15T02:00:00Z", + "repository": "example.invalid/placeholder/repository", + "task_url": "https://example.invalid/placeholder/task/v2-1" + }, + "used_node_id": "placeholder-runtime-target" + }, + "placeholder_only": true, + "protocol_id": "real-task-feedback-shadow-v2" +} diff --git a/tests/fixtures/real_task_shadow_v2.result-schema.json b/tests/fixtures/real_task_shadow_v2.result-schema.json new file mode 100644 index 0000000..243184a --- /dev/null +++ b/tests/fixtures/real_task_shadow_v2.result-schema.json @@ -0,0 +1,82 @@ +{ + "$id": "https://example.invalid/neuron-graph-rag/real-task-feedback-shadow-v2.result.schema.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": false, + "properties": { + "arms": { + "additionalProperties": false, + "properties": { + "confirmed_r05_s1": { + "type": "object" + }, + "used_q3_s1": { + "type": "object" + } + }, + "required": [ + "confirmed_r05_s1", + "used_q3_s1" + ], + "type": "object" + }, + "capture_config": { + "type": "object" + }, + "capture_search_surface": { + "enum": [ + "combined", + "relation" + ], + "type": "string" + }, + "comparison": { + "type": "object" + }, + "efficiency": { + "type": "array" + }, + "packet_ids": { + "items": { + "type": "string" + }, + "minItems": 1, + "type": "array" + }, + "protocol_id": { + "const": "real-task-feedback-shadow-v2" + }, + "replay": { + "type": "object" + }, + "schema_version": { + "const": 2 + }, + "slots": { + "items": { + "minimum": 1, + "type": "integer" + }, + "minItems": 1, + "type": "array" + }, + "snapshot_sha256": { + "pattern": "^sha256:[0-9a-f]{64}$", + "type": "string" + } + }, + "required": [ + "arms", + "capture_config", + "capture_search_surface", + "comparison", + "efficiency", + "packet_ids", + "protocol_id", + "replay", + "schema_version", + "slots", + "snapshot_sha256" + ], + "title": "Real-task feedback shadow v2 result", + "type": "object" +} diff --git a/tests/test_mcp_adapter.py b/tests/test_mcp_adapter.py index a737db1..4020ac4 100644 --- a/tests/test_mcp_adapter.py +++ b/tests/test_mcp_adapter.py @@ -76,6 +76,12 @@ async def test_tools_list_contract_and_structured_search_result(self) -> None: self.assertFalse(result.is_error) self.assertEqual(json.loads(result.content[0].text), result.structured_content) self.assertIsNone(result.structured_content["trace_expires_at"]) + provenance = result.structured_content["effective_config_provenance"] + self.assertEqual(provenance["search_surface"], "combined") + self.assertEqual(provenance["effective_config"]["retrieval"]["sparse_weight"], 0.55) + self.assertEqual(provenance["effective_config"]["feedback"]["maximum_edge_weight"], 2.0) + self.assertTrue(provenance["retrieval_config_fingerprint"].startswith("sha256:")) + self.assertTrue(provenance["full_config_fingerprint"].startswith("sha256:")) self.assertTrue(all(hit["source_use_stage"] == "retrieved" for hit in result.structured_content["hits"])) async def test_feedback_loop_and_safe_tool_errors(self) -> None: diff --git a/tests/test_real_task_shadow_v2.py b/tests/test_real_task_shadow_v2.py new file mode 100644 index 0000000..f19dd05 --- /dev/null +++ b/tests/test_real_task_shadow_v2.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +import copy +import hashlib +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from neuron_graph_rag.evidence_feedback import EngineConfig +from neuron_graph_rag.real_task_shadow import read_canonical_json, write_json_exclusive +from neuron_graph_rag.real_task_shadow_v2 import ( + PROTOCOL_ID, + bind_capture_fingerprint, + build_packet_from_search, + capture_packet, + create_placeholder_snapshot, + effective_config_provenance, + load_effective_registry, + probe_placeholder, + replay_packets, + verify_packet_against_snapshot, + verify_result_against_packets, +) +try: + from neuron_graph_rag_mcp.server import CONTRACT_VERSION, FeedbackMCPAdapter +except ImportError as error: + module = error.name or "" + if module != "mcp" and not module.startswith("mcp."): + raise + MCP_AVAILABLE = False +else: + MCP_AVAILABLE = True + + +ROOT = Path(__file__).resolve().parents[1] +FIXTURES = ROOT / "tests" / "fixtures" +PLACEHOLDER = FIXTURES / "real_task_shadow_v2.placeholder.json" +MANIFEST = FIXTURES / "real_task_shadow_v2.manifest.json" + + +@unittest.skipUnless(MCP_AVAILABLE, "optional MCP SDK is not installed") +class RealTaskShadowV2Test(unittest.TestCase): + def setUp(self) -> None: + self.fixture = read_canonical_json(PLACEHOLDER) + + def _packet(self, root: Path) -> tuple[Path, dict[str, object]]: + snapshot = root / "placeholder-v2.db" + create_placeholder_snapshot(self.fixture, snapshot) + config = EngineConfig( + relation_feedback_evidence_quorum=3, + sibling_feedback_normalization=1.0, + ) + adapter = FeedbackMCPAdapter(snapshot, config=config) + try: + search = adapter._search( + { + "contract_version": CONTRACT_VERSION, + "query": self.fixture["packet_seed"]["query"], + "limit": self.fixture["packet_seed"]["limit"], + } + ) + finally: + adapter.close() + return snapshot, build_packet_from_search(self.fixture, search, snapshot) + + @staticmethod + def _rebind(packet: dict[str, object]) -> None: + capture = packet["capture"] + config = capture["effective_config"] + capture["retrieval_config_fingerprint"] = effective_config_provenance( + EngineConfig(**{**config["retrieval"], **config["feedback"]}) + )["retrieval_config_fingerprint"] + capture["feedback_config_fingerprint"] = effective_config_provenance( + EngineConfig(**{**config["retrieval"], **config["feedback"]}) + )["feedback_config_fingerprint"] + capture["full_config_fingerprint"] = effective_config_provenance( + EngineConfig(**{**config["retrieval"], **config["feedback"]}) + )["full_config_fingerprint"] + capture["capture_fingerprint"] = bind_capture_fingerprint(packet) + + def test_runtime_mcp_search_capture_and_exact_replay_round_trip(self) -> None: + self.assertEqual( + probe_placeholder(PLACEHOLDER), + { + "protocol_id": PROTOCOL_ID, + "placeholder_only": True, + "mcp_search_capture_round_trip": True, + "capture_time_reused": True, + "effective_config_provenance_verified": True, + "replay_round_trip": True, + "exclusive_writer_verified": True, + }, + ) + + def test_capture_uses_serving_retrieval_config_and_arms_only_override_feedback(self) -> None: + with tempfile.TemporaryDirectory() as directory: + snapshot, packet = self._packet(Path(directory)) + result = replay_packets([packet], snapshot) + capture = packet["capture"] + self.assertEqual(capture["search_surface"], "relation") + self.assertGreater(capture["searched_at"], 1_700_000_000) + self.assertEqual(capture["effective_config"]["retrieval"]["sparse_weight"], 0.55) + self.assertEqual(capture["effective_config"]["retrieval"]["dense_weight"], 0.45) + retrievals = [result["arms"][arm]["effective_config"]["retrieval"] for arm in result["arms"]] + self.assertEqual(retrievals[0], retrievals[1]) + self.assertEqual(retrievals[0], capture["effective_config"]["retrieval"]) + used_feedback = result["arms"]["used_q3_s1"]["effective_config"]["feedback"] + confirmed_feedback = result["arms"]["confirmed_r05_s1"]["effective_config"]["feedback"] + self.assertEqual(used_feedback["relation_feedback_evidence_quorum"], 3) + self.assertFalse(used_feedback["confirmed_outcome_reinforcement"]) + self.assertTrue(confirmed_feedback["confirmed_outcome_reinforcement"]) + self.assertEqual(confirmed_feedback["confirmation_decay_ratio"], 0.5) + self.assertEqual(used_feedback["maximum_edge_weight"], 2.0) + self.assertEqual(confirmed_feedback["maximum_edge_weight"], 2.0) + + def test_default_combined_search_surface_is_reused_by_both_arms(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + snapshot = root / "placeholder-default.db" + create_placeholder_snapshot(self.fixture, snapshot) + adapter = FeedbackMCPAdapter(snapshot) + try: + search = adapter._search( + { + "contract_version": CONTRACT_VERSION, + "query": self.fixture["packet_seed"]["query"], + "limit": self.fixture["packet_seed"]["limit"], + } + ) + finally: + adapter.close() + packet = build_packet_from_search(self.fixture, search, snapshot) + result = replay_packets([packet], snapshot) + self.assertEqual(packet["capture"]["search_surface"], "combined") + self.assertEqual(result["capture_search_surface"], "combined") + self.assertEqual( + {arm["search_surface"] for arm in result["arms"].values()}, + {"combined"}, + ) + + def test_config_timestamp_candidate_path_and_fingerprint_tamper_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + snapshot, packet = self._packet(Path(directory)) + cases = [] + config = copy.deepcopy(packet) + config["capture"]["effective_config"]["retrieval"]["max_hops"] = 0 + cases.append(("config", config)) + timestamp = copy.deepcopy(packet) + timestamp["capture"]["searched_at"] += 60 + cases.append(("timestamp", timestamp)) + candidate = copy.deepcopy(packet) + candidate["retrieval"]["candidates"].reverse() + self._rebind(candidate) + cases.append(("candidate_order", candidate)) + path = copy.deepcopy(packet) + path["retrieval"]["credited_path"]["steps"][0]["edge_type"] = "tampered" + self._rebind(path) + cases.append(("path", path)) + surface = copy.deepcopy(packet) + surface["capture"]["search_surface"] = "combined" + self._rebind(surface) + cases.append(("search_surface", surface)) + fingerprint = copy.deepcopy(packet) + fingerprint["capture"]["full_config_fingerprint"] = "sha256:" + "0" * 64 + cases.append(("fingerprint", fingerprint)) + for name, tampered in cases: + with self.subTest(name=name): + with self.assertRaises(ValueError): + verify_packet_against_snapshot(tampered, snapshot) + + def test_cumulative_q3_and_confirmed_decay_share_one_capture_config(self) -> None: + with tempfile.TemporaryDirectory() as directory: + snapshot, first = self._packet(Path(directory)) + packets = [] + for slot in range(1, 4): + packet = copy.deepcopy(first) + packet["packet_id"] = f"placeholder-v2-packet-{slot:04d}" + packet["slot"] = slot + packet["task"]["task_url"] = f"https://example.invalid/placeholder/task/v2-{slot}" + packet["capture"]["searched_at"] = first["capture"]["searched_at"] + slot - 1 + packet["outcome"] = { + "status": "confirmed", + "summary": f"placeholder objective confirmation {slot}", + "external_ref": f"https://example.invalid/placeholder/check/{slot}", + "evidence": [ + { + "kind": "test_passed", + "node_id": "placeholder-runtime-target", + "external_ref": f"https://example.invalid/placeholder/check/{slot}", + "target_commit": "0000000000000000000000000000000000000000", + "details": {"command": f"placeholder-check-{slot}", "exit_code": 0}, + } + ], + } + self._rebind(packet) + packets.append(packet) + result = replay_packets(packets, snapshot) + used = result["arms"]["used_q3_s1"]["packets"] + self.assertEqual([row["source_use"]["feedback"]["evidence"][0]["count"] for row in used], [1, 2, 3]) + confirmed = result["arms"]["confirmed_r05_s1"]["packets"] + self.assertEqual([row["outcome"]["confirmations"][0]["multiplier"] for row in confirmed], [1.0, 0.5, 0.25]) + + def test_batch_config_divergence_registry_and_exact_result_tamper_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + snapshot, first = self._packet(root) + second = copy.deepcopy(first) + second["packet_id"] = "placeholder-v2-packet-0002" + second["slot"] = 2 + second["task"]["task_url"] = "https://example.invalid/placeholder/task/v2-2" + second["capture"]["searched_at"] += 1 + self._rebind(second) + registry = root / "registry" + capture_packet(first, registry) + capture_packet(second, registry) + self.assertEqual([item["slot"] for item in load_effective_registry(registry)], [1, 2]) + result = replay_packets([first, second], snapshot) + stored = root / "result.json" + write_json_exclusive(stored, result) + verify_result_against_packets(read_canonical_json(stored), [first, second], snapshot) + forged = copy.deepcopy(result) + forged["arms"]["used_q3_s1"]["effective_config"]["retrieval"]["seed_count"] = 1 + with self.assertRaises(ValueError): + verify_result_against_packets(forged, [first, second], snapshot) + divergent = copy.deepcopy(second) + divergent["capture"]["effective_config"]["feedback"]["maximum_edge_weight"] = 2.5 + self._rebind(divergent) + with self.assertRaisesRegex(ValueError, "share one effective capture config"): + replay_packets([first, divergent], snapshot) + + def test_cli_probe_and_result_free_manifest(self) -> None: + environment = {**os.environ, "PYTHONPATH": str(ROOT / "src")} + tool = ROOT / "tools" / "run_real_task_shadow_v2.py" + run = subprocess.run( + [sys.executable, str(tool), "probe", "--fixture", str(PLACEHOLDER)], + check=False, capture_output=True, text=True, env=environment, + ) + self.assertEqual(run.returncode, 0, run.stderr) + self.assertIn('"placeholder_only": true', run.stdout) + manifest = read_canonical_json(MANIFEST) + self.assertTrue(manifest["result_free"]) + self.assertEqual(manifest["observation_status"], "not_started") + for relative, expected in manifest["artifact_sha256"].items(): + self.assertEqual(hashlib.sha256((ROOT / relative).read_bytes()).hexdigest(), expected, relative) + for relative in manifest["registered_outputs"].values(): + path = ROOT / relative + self.assertTrue(not path.exists() or not any(path.iterdir()), relative) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/run_real_task_shadow_v2.py b/tools/run_real_task_shadow_v2.py new file mode 100644 index 0000000..899940d --- /dev/null +++ b/tools/run_real_task_shadow_v2.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from neuron_graph_rag.real_task_shadow import read_canonical_json, write_json_exclusive +from neuron_graph_rag.real_task_shadow_v2 import ( + capture_packet, + probe_placeholder, + replay_packet, + replay_registry, + verify_packet_against_snapshot, + verify_result_against_packets, + verify_result_against_registry, +) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run the result-free real-task shadow v2 protocol") + commands = parser.add_subparsers(dest="command", required=True) + probe = commands.add_parser("probe", help="run the unregistered runtime-shaped placeholder round-trip") + probe.add_argument("--fixture", type=Path, required=True) + capture = commands.add_parser("capture", help="append one v2 packet to a local registry") + capture.add_argument("--input", type=Path, required=True) + capture.add_argument("--registry-dir", type=Path, required=True) + verify_packet = commands.add_parser("verify-packet", help="verify a v2 packet against its snapshot") + verify_packet.add_argument("--packet", type=Path, required=True) + verify_packet.add_argument("--snapshot", type=Path, required=True) + replay = commands.add_parser("replay", help="cumulatively replay v2 packets into both shadow arms") + replay_input = replay.add_mutually_exclusive_group(required=True) + replay_input.add_argument("--packet", type=Path) + replay_input.add_argument("--registry-dir", type=Path) + replay.add_argument("--snapshot", type=Path, required=True) + replay.add_argument("--output", type=Path, required=True) + verify_result = commands.add_parser("verify-result", help="recompute and verify an exact v2 result") + verify_result.add_argument("--result", type=Path, required=True) + verify_input = verify_result.add_mutually_exclusive_group(required=True) + verify_input.add_argument("--packet", type=Path) + verify_input.add_argument("--registry-dir", type=Path) + verify_result.add_argument("--snapshot", type=Path, required=True) + + args = parser.parse_args() + if args.command == "probe": + print(json.dumps(probe_placeholder(args.fixture), ensure_ascii=False, sort_keys=True)) + elif args.command == "capture": + print(capture_packet(read_canonical_json(args.input), args.registry_dir)) + elif args.command == "verify-packet": + verify_packet_against_snapshot(read_canonical_json(args.packet), args.snapshot) + print("packet verified") + elif args.command == "replay": + result = replay_packet(read_canonical_json(args.packet), args.snapshot) if args.packet else replay_registry(args.registry_dir, args.snapshot) + write_json_exclusive(args.output, result) + print(args.output) + else: + result = read_canonical_json(args.result) + if args.packet: + verify_result_against_packets(result, [read_canonical_json(args.packet)], args.snapshot) + else: + verify_result_against_registry(result, args.registry_dir, args.snapshot) + print("result verified") + + +if __name__ == "__main__": + main()