perf(research): avoid repeated full-file source verification

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:

tests/capabilities/test_trajectory_assessment_store.py

@@ -151,6 +151,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"

tests/presentation/test_research_analysis_api.py

@@ -97,6 +97,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"