From 4efc94dd41182044ab55d6c73a39313d9f03eef7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 24 Oct 2026 17:52:49 +0800 Subject: [PATCH] perf(research): verify each artifact once per pack read Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../trajectory_assessment_store.py | 25 ++++++---- .../test_trajectory_assessment_store.py | 46 +++++++++++++++++++ .../test_research_analysis_api.py | 32 +++++++++++++ 3 files changed, 95 insertions(+), 8 deletions(-) diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py index a50e86fa42..6814862e34 100644 --- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py +++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py @@ -68,16 +68,25 @@ def _prepared(record: dict, output_language: str | None = None) -> dict: def _verify_sources(record: dict) -> None: + # One byte observation per declared artifact in this verification only. + # Check every digest/excerpt; later calls still read and verify fresh bytes. + artifacts: dict[str, list[dict]] = {} for source in record["input"]["sources"]: - path = _source_path(Path(record["source_root"]), source["artifact"]) + artifacts.setdefault(source["artifact"], []).append(source) + for artifact, sources in artifacts.items(): + path = _source_path(Path(record["source_root"]), artifact) data = path.read_bytes() - if hashlib.sha256(data).hexdigest() != source["artifact_sha256"]: - raise PackUnavailable("source_changed") - lines = data.decode("utf-8").split("\n") - start = source["first_line"] - 1 - count = len(source["text"].split("\n")) - if "\n".join(lines[start : start + count]) != source["text"]: - raise PackUnavailable("excerpt_mismatch") + digest = hashlib.sha256(data).hexdigest() + lines = None + for source in sources: + if digest != source["artifact_sha256"]: + raise PackUnavailable("source_changed") + if lines is None: + lines = data.decode("utf-8").split("\n") + start = source["first_line"] - 1 + count = len(source["text"].split("\n")) + if "\n".join(lines[start : start + count]) != source["text"]: + raise PackUnavailable("excerpt_mismatch") def _source_path(source_root: Path, artifact: str) -> Path: diff --git a/tests/capabilities/test_trajectory_assessment_store.py b/tests/capabilities/test_trajectory_assessment_store.py index f19aee5790..220660bdc9 100644 --- a/tests/capabilities/test_trajectory_assessment_store.py +++ b/tests/capabilities/test_trajectory_assessment_store.py @@ -146,6 +146,52 @@ def test_scope_discovery_execute_read_citation_and_changed_source_history(pack): ) +def test_repeated_artifact_excerpts_share_one_fresh_read(pack, monkeypatch): + scope, _, _, input_path, sources, _ = pack + data = json.loads(input_path.read_text()) + data["sources"].append({**data["sources"][0], "id": "prior", "first_line": 1, "text": "Prior question"}) + data["episodes"][0]["source_ids"].append("prior") + input_path.write_text(json.dumps(data)) + calls = [] + original = type(input_path).read_bytes + + def observed(path): + if path == sources / "result.txt": + calls.append(path) + return original(path) + + monkeypatch.setattr(type(input_path), "read_bytes", observed) + registered = register_pack(**scope, input_path=input_path, source_root=sources, title="Multiple excerpts") + assert len(calls) == 1 + calls.clear() + before = inspect_pack(**scope, pack_id=registered["pack_id"]) + assert before["availability"] == "available" and len(calls) == 1 + # No cross-request cache: same-length replacement with restored timestamps + # must still invalidate the pack, rather than trusting path/size/mtime. + path = sources / "result.txt" + stat = path.stat() + original_bytes = original(path) + path.write_bytes(original_bytes.replace(b"Prior", b"Other")) + os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns)) + calls.clear() + after = inspect_pack(**scope, pack_id=registered["pack_id"]) + assert after["reason_code"] == "source_changed" and len(calls) == 1 + + +@pytest.mark.parametrize("correction,reason", [ + ({"artifact_sha256": "0" * 64}, "source_changed"), + ({"text": "An unsupported second excerpt"}, "excerpt_mismatch"), +]) +def test_shared_artifact_does_not_skip_later_digest_or_excerpt(pack, correction, reason): + scope, _, _, input_path, sources, _ = pack + data = json.loads(input_path.read_text()) + data["sources"].append({**data["sources"][0], "id": "later", **correction}) + data["episodes"][0]["source_ids"].append("later") + input_path.write_text(json.dumps(data)) + with pytest.raises(PackUnavailable, match=reason): + register_pack(**scope, input_path=input_path, source_root=sources, title="Invalid later excerpt") + + def test_continuation_reuses_identical_excerpts_despite_other_capture_bytes(pack): scope, operation, registered, path, sources, provider = pack assert execute_pack(**operation, codex=provider)["status"] == "succeeded" diff --git a/tests/presentation/test_research_analysis_api.py b/tests/presentation/test_research_analysis_api.py index 48f80d4ab2..611d012b4c 100644 --- a/tests/presentation/test_research_analysis_api.py +++ b/tests/presentation/test_research_analysis_api.py @@ -95,6 +95,38 @@ def request(*, goal="research-owner", body=None, suffix="", headers=None, **quer server.server_close() +def test_shared_excerpts_reverify_through_http_without_dispatch(workspace, monkeypatch): + w = workspace + path = w["tmp_path"] / "input.json" + data = json.loads(path.read_text()) + data["sources"].append({**data["sources"][0], "id": "prior", "first_line": 1, "text": "Prior question"}) + data["episodes"][0]["source_ids"].append("prior") + path.write_text(json.dumps(data)) + registered = register_pack(runtime_root=w["tmp_path"] / "runtime", owner_goal_id="research-owner", + input_path=path, source_root=w["sources"], title="Two excerpts of one artifact") + original = type(path).read_bytes + reads = [] + + def observed(source): + if source == w["sources"] / "facts.txt": + reads.append(source) + return original(source) + + monkeypatch.setattr(type(path), "read_bytes", observed) + view = w["request"](pack_id=registered["pack_id"]) + assert view["pack"]["availability"] == "available" + # Two catalog entries plus the selected pack; each verification reads once. + assert len(reads) == 3 and not w["count"].exists() + source = w["sources"] / "facts.txt" + stat = source.stat() + source.write_bytes(original(source).replace(b"Prior", b"Other")) + os.utime(source, ns=(stat.st_atime_ns, stat.st_mtime_ns)) + reads.clear() + changed = w["request"](pack_id=registered["pack_id"]) + assert changed["pack"]["reason_code"] == "source_changed" + assert len(reads) == 3 and not w["count"].exists() + + def test_display_context_uses_registered_sources_and_the_existing_http_reader(workspace): w, read = workspace, workspace["request"] path = w["tmp_path"] / "input.json"