From 99eb92496edd023c24817a1e023d5673360356c1 Mon Sep 17 00:00:00 2001 From: shangzh0 <97216392+shangzh0@users.noreply.github.com> Date: Sun, 1 Nov 2026 01:04:34 +0800 Subject: [PATCH] perf(trajectory): stream capture parsing and source verification Signed-off-by: shangzh0 <97216392+shangzh0@users.noreply.github.com> --- .../trajectory_assessment_store.py | 32 +++++-- .../test_trajectory_capture_streaming.py | 83 +++++++++++++++++++ 2 files changed, 106 insertions(+), 9 deletions(-) create mode 100644 tests/capabilities/test_trajectory_capture_streaming.py diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py index 3abf690a69..4cef534a48 100644 --- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py +++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py @@ -130,6 +130,20 @@ def _request_usage_record(value: object) -> object: ) return result + +def _usage_rows(path: Path, digest): + """Hash the exact captured bytes while parsing one JSONL record at a time.""" + with path.open("rb") as stream: + for line_number, line in enumerate(stream, 1): + digest.update(line) + yield line_number, json.loads(line.decode("utf-8")) + + +def _source_digest(path: Path) -> str: + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + def _usage_capture( source_root: Path, artifact: str, @@ -139,12 +153,12 @@ def _usage_capture( loopx_input_json: str | None = None, input_goal_id: str | None = None, ) -> dict: - data = _source_path(source_root, artifact).read_bytes() + path = _source_path(source_root, artifact) + digest = hashlib.sha256() events = [] # Transport only the provider's counter/lifecycle fields, not model text, # account metadata or tool outputs. TS owns all accounting decisions. - for line_number, line in enumerate(data.decode("utf-8").splitlines(), 1): - event = json.loads(line) + for line_number, event in _usage_rows(path, digest): if not isinstance(event, dict): raise ValueError("Usage capture contains a non-object event") payload = event.get("payload") or {} @@ -236,7 +250,7 @@ def _usage_capture( ) return { "artifact": artifact, - "artifact_sha256": hashlib.sha256(data).hexdigest(), + "artifact_sha256": digest.hexdigest(), "projection": projection, **({"input_artifacts": input_artifacts} if input_artifacts else {}), } @@ -248,14 +262,14 @@ def _captured_view(record: dict, field: str) -> dict | None: return None reason = None try: - data = _source_path( + path = _source_path( Path(record["source_root"]), captured["artifact"] - ).read_bytes() - if hashlib.sha256(data).hexdigest() != captured["artifact_sha256"]: + ) + if _source_digest(path) != captured["artifact_sha256"]: reason = "source_changed" for artifact in captured.get("input_artifacts", []): - data = _source_path(Path(record["source_root"]), artifact["artifact"]).read_bytes() - if hashlib.sha256(data).hexdigest() != artifact["artifact_sha256"]: + path = _source_path(Path(record["source_root"]), artifact["artifact"]) + if _source_digest(path) != artifact["artifact_sha256"]: reason = "source_changed" except (OSError, ValueError): reason = "source_unavailable" diff --git a/tests/capabilities/test_trajectory_capture_streaming.py b/tests/capabilities/test_trajectory_capture_streaming.py new file mode 100644 index 0000000000..2e92c31a81 --- /dev/null +++ b/tests/capabilities/test_trajectory_capture_streaming.py @@ -0,0 +1,83 @@ +"""Large private captures must not require a full text copy to observe costs.""" +import hashlib +import json +import os +import tracemalloc + +import pytest + +from loopx.capabilities.benchmark_toolkit.trajectory_assessment_store import ( + _captured_view, _usage_rows, _usage_capture, +) + + +def test_hash_and_line_provenance_preserve_utf8_crlf_and_no_final_newline(tmp_path): + path = tmp_path / 'capture.jsonl' + raw = '{"text":"合成结果"}\r\n{"value":2}'.encode() + path.write_bytes(raw) + digest = hashlib.sha256() + assert list(_usage_rows(path, digest)) == [(1, {'text': '合成结果'}), (2, {'value': 2})] + assert digest.hexdigest() == hashlib.sha256(raw).hexdigest() + path.write_bytes(b'{}\n{broken') + with pytest.raises(ValueError): + list(_usage_rows(path, hashlib.sha256())) + + +def test_current_readback_rehashes_same_size_same_timestamp_mutations(tmp_path): + path = tmp_path / 'capture.bin' + path.write_bytes(b'original') + stamp = path.stat().st_mtime_ns + captured = {'artifact': path.name, 'artifact_sha256': hashlib.sha256(b'original').hexdigest(), + 'projection': {'input_tokens': 123}} + record = {'source_root': str(tmp_path), 'solver_usage': captured} + assert _captured_view(record, 'solver_usage')['applicability'] == 'current' + path.write_bytes(b'changed!') + os.utime(path, ns=(stamp, stamp)) + historical = _captured_view(record, 'solver_usage') + assert historical['reason_code'] == 'source_changed' + assert historical['projection'] == captured['projection'] + path.unlink() + assert _captured_view(record, 'solver_usage')['reason_code'] == 'source_unavailable' + + +def test_large_ignored_text_has_bounded_observation_and_verification_memory(tmp_path): + path = tmp_path / 'large.jsonl' + digest = hashlib.sha256() + row = json.dumps({'type': 'response_item', 'payload': { + 'type': 'message', 'role': 'assistant', 'content': [{'text': 'x' * 65536}]}}).encode() + b'\n' + with path.open('wb') as stream: + for _ in range(512): + stream.write(row) + digest.update(row) + # Start after fixture construction; measure the real parsing/hash path. + tracemalloc.start() + try: + check = hashlib.sha256() + assert sum(1 for _ in _usage_rows(path, check)) == 512 + captured = {'artifact': path.name, 'artifact_sha256': digest.hexdigest(), 'projection': {}} + assert _captured_view({'source_root': str(tmp_path), 'solver_usage': captured}, 'solver_usage')['applicability'] == 'current' + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + assert check.hexdigest() == digest.hexdigest() + assert peak < 4 * 1024 * 1024, f'32 MiB capture required {peak} bytes of Python allocation' + + +def test_real_counter_projection_retains_capture_digest_and_source_lines(tmp_path): + usage = {'input_tokens': 100, 'cached_input_tokens': 60, 'output_tokens': 5} + rows = [ + {'type': 'session_meta', 'payload': {'id': 'session'}}, + {'type': 'event_msg', 'payload': {'type': 'task_started', 'turn_id': 'turn'}}, + {'type': 'response_item', 'payload': {'type': 'message', 'role': 'assistant', 'content': [{'text': '合成输出'}]}}, + {'type': 'event_msg', 'payload': {'type': 'token_count', 'info': {'total_token_usage': usage, 'last_token_usage': usage}}}, + {'type': 'event_msg', 'payload': {'type': 'task_complete', 'turn_id': 'turn'}}, + ] + for i, row in enumerate(rows): + row['timestamp'] = f'2026-01-01T00:00:0{i}Z' + raw = '\r\n'.join(json.dumps(r, ensure_ascii=False) for r in rows).encode() + (tmp_path / 'usage.jsonl').write_bytes(raw) + result = _usage_capture(tmp_path, 'usage.jsonl', 'session') + assert result['artifact_sha256'] == hashlib.sha256(raw).hexdigest() + assert result['projection']['input_tokens'] == 100 + assert result['projection']['turns'][0]['first_line'] == 2 + assert result['projection']['turns'][0]['last_line'] == 5