diff --git a/loopx/control_plane/runtime/run_index_rebuild.py b/loopx/control_plane/runtime/run_index_rebuild.py index 04cff9e49..65c765bb9 100644 --- a/loopx/control_plane/runtime/run_index_rebuild.py +++ b/loopx/control_plane/runtime/run_index_rebuild.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import Any +from ...file_lock import exclusive_run_index_lock from .run_artifacts import run_file_stem from .run_index_duplicates import classify_index_duplicate_records, index_identity @@ -189,75 +190,84 @@ def apply_reviewed_collision_rebuild( digest_prefix = plan_sha256[:16] for raw_index_path, groups in groups_by_index.items(): index_path = Path(raw_index_path) - raw_lines, parsed_rows = read_index_rows(index_path) - rows_by_line = dict(parsed_rows) - replacements: dict[int, dict[str, Any]] = {} - recovery_paths: list[str] = [] + with exclusive_run_index_lock( + index_path, + operation="history_index_collision_rebuild", + ): + raw_lines, parsed_rows = read_index_rows(index_path) + rows_by_line = dict(parsed_rows) + replacements: dict[int, dict[str, Any]] = {} + recovery_paths: list[str] = [] - for group in groups: - source_identity = group.get("source_identity") or {} - generated_at = str(source_identity.get("generated_at") or "") - stem = run_file_stem(generated_at) - group_token = _sha256(source_identity)[:8] - for ordinal, row_spec in enumerate(group.get("rows") or [], start=1): - line_number = int(row_spec["line_number"]) - source_record = rows_by_line.get(line_number) - if source_record is None or _sha256(source_record) != row_spec.get("row_sha256"): - raise ValueError( - f"reviewed collision row changed before rebuild: {index_path}:{line_number}" + for group in groups: + source_identity = group.get("source_identity") or {} + generated_at = str(source_identity.get("generated_at") or "") + stem = run_file_stem(generated_at) + group_token = _sha256(source_identity)[:8] + for ordinal, row_spec in enumerate(group.get("rows") or [], start=1): + line_number = int(row_spec["line_number"]) + source_record = rows_by_line.get(line_number) + if source_record is None or _sha256(source_record) != row_spec.get( + "row_sha256" + ): + raise ValueError( + f"reviewed collision row changed before rebuild: {index_path}:{line_number}" + ) + recovery_stem = f"{stem}-collision-rebuild-{digest_prefix}-{group_token}-{ordinal}" + json_path = index_path.parent / f"{recovery_stem}.json" + markdown_path = index_path.parent / f"{recovery_stem}.md" + rebuilt_record = dict(source_record) + rebuilt_record["json_path"] = str(json_path) + rebuilt_record["markdown_path"] = str(markdown_path) + rebuilt_record["artifact_rebuild"] = { + "schema_version": COLLISION_RECOVERY_ARTIFACT_SCHEMA, + "review_plan_sha256": plan_sha256, + "source_identity": source_identity, + "source_line_number": line_number, + "event_identity": row_spec.get("event_identity") or {}, + "ambiguous_legacy_artifact_claimed": False, + } + recovery_payload = { + "schema_version": COLLISION_RECOVERY_ARTIFACT_SCHEMA, + "artifact_rebuild": rebuilt_record["artifact_rebuild"], + "index_record": source_record, + } + _write_new_or_verify( + json_path, + json.dumps(recovery_payload, ensure_ascii=False, indent=2) + + "\n", ) - recovery_stem = ( - f"{stem}-collision-rebuild-{digest_prefix}-{group_token}-{ordinal}" - ) - json_path = index_path.parent / f"{recovery_stem}.json" - markdown_path = index_path.parent / f"{recovery_stem}.md" - rebuilt_record = dict(source_record) - rebuilt_record["json_path"] = str(json_path) - rebuilt_record["markdown_path"] = str(markdown_path) - rebuilt_record["artifact_rebuild"] = { - "schema_version": COLLISION_RECOVERY_ARTIFACT_SCHEMA, - "review_plan_sha256": plan_sha256, - "source_identity": source_identity, - "source_line_number": line_number, - "event_identity": row_spec.get("event_identity") or {}, - "ambiguous_legacy_artifact_claimed": False, - } - recovery_payload = { - "schema_version": COLLISION_RECOVERY_ARTIFACT_SCHEMA, - "artifact_rebuild": rebuilt_record["artifact_rebuild"], - "index_record": source_record, - } - _write_new_or_verify( - json_path, - json.dumps(recovery_payload, ensure_ascii=False, indent=2) + "\n", - ) - _write_new_or_verify(markdown_path, _recovery_markdown(rebuilt_record)) - replacements[line_number] = rebuilt_record - recovery_paths.extend((str(json_path), str(markdown_path))) + _write_new_or_verify( + markdown_path, _recovery_markdown(rebuilt_record) + ) + replacements[line_number] = rebuilt_record + recovery_paths.extend((str(json_path), str(markdown_path))) - backup_path = index_path.with_name( - f"index.pre-collision-rebuild-{digest_prefix}.jsonl" - ) - original_content = "".join(line + "\n" for line in raw_lines) - _write_new_or_verify(backup_path, original_content) - rebuilt_lines = [ - json.dumps(replacements[line_number], ensure_ascii=False) - if line_number in replacements - else line - for line_number, line in enumerate(raw_lines, start=1) - ] - tmp_path = index_path.with_name(f"index.collision-rebuild-{digest_prefix}.tmp") - tmp_path.write_text( - "".join(line + "\n" for line in rebuilt_lines), - encoding="utf-8", - ) - tmp_path.replace(index_path) - rebuilt_indexes.append( - { - "index_path": str(index_path), - "backup_path": str(backup_path), - "preserved_row_count": len(replacements), - "recovery_paths": recovery_paths, - } - ) + backup_path = index_path.with_name( + f"index.pre-collision-rebuild-{digest_prefix}.jsonl" + ) + original_content = "".join(line + "\n" for line in raw_lines) + _write_new_or_verify(backup_path, original_content) + rebuilt_lines = [ + json.dumps(replacements[line_number], ensure_ascii=False) + if line_number in replacements + else line + for line_number, line in enumerate(raw_lines, start=1) + ] + tmp_path = index_path.with_name( + f"index.collision-rebuild-{digest_prefix}.tmp" + ) + tmp_path.write_text( + "".join(line + "\n" for line in rebuilt_lines), + encoding="utf-8", + ) + tmp_path.replace(index_path) + rebuilt_indexes.append( + { + "index_path": str(index_path), + "backup_path": str(backup_path), + "preserved_row_count": len(replacements), + "recovery_paths": recovery_paths, + } + ) return rebuilt_indexes diff --git a/tests/test_history_index_write_serialization.py b/tests/test_history_index_write_serialization.py index 70cb42112..3931f87f6 100644 --- a/tests/test_history_index_write_serialization.py +++ b/tests/test_history_index_write_serialization.py @@ -11,6 +11,7 @@ from __future__ import annotations import json +import threading from contextlib import contextmanager from pathlib import Path from typing import Any, Iterator @@ -18,6 +19,7 @@ import pytest from loopx import history +from loopx.control_plane.runtime import run_index_rebuild GOAL_ID = "goal-history-lock" @@ -127,3 +129,121 @@ def test_repair_dry_run_does_not_take_the_write_lock( assert result["dry_run"] is True assert record_lock_calls == [], "a read-only repair must not block writers" + + +def test_collision_rebuild_preserves_append_started_after_snapshot( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + index_path = runs_dir / "index.jsonl" + generated_at = "2026-09-16T00:00:00+00:00" + shared_paths = { + "json_path": str(runs_dir / "legacy.json"), + "markdown_path": str(runs_dir / "legacy.md"), + } + collision_rows = [ + { + "goal_id": GOAL_ID, + "generated_at": generated_at, + "classification": "quota_monitor_poll", + "todo_id": todo_id, + **shared_paths, + } + for todo_id in ("todo-a", "todo-b") + ] + index_path.write_text( + "".join(json.dumps(row) + "\n" for row in collision_rows), + encoding="utf-8", + ) + groups = run_index_rebuild.collision_review_groups(index_path, GOAL_ID) + plan = run_index_rebuild.build_collision_rebuild_plan( + groups, + goal_filter=GOAL_ID, + total_collision_group_count=len(groups), + truncated=False, + ) + + rebuild_paused = threading.Event() + allow_rebuild = threading.Event() + append_started = threading.Event() + append_finished = threading.Event() + thread_errors: list[BaseException] = [] + real_write_new_or_verify = run_index_rebuild._write_new_or_verify + + def pause_before_backup(path: Path, content: str) -> None: + if path.name.startswith("index.pre-collision-rebuild-"): + rebuild_paused.set() + if not allow_rebuild.wait(timeout=5): + raise TimeoutError("test did not release the collision rebuild") + real_write_new_or_verify(path, content) + + monkeypatch.setattr( + run_index_rebuild, + "_write_new_or_verify", + pause_before_backup, + ) + + def capture_errors(operation: Any) -> None: + try: + operation() + except BaseException as exc: + thread_errors.append(exc) + + rebuild_thread = threading.Thread( + target=lambda: capture_errors( + lambda: run_index_rebuild.apply_reviewed_collision_rebuild( + plan, + plan_sha256=plan["plan_sha256"], + ) + ) + ) + + def append_run() -> None: + append_started.set() + history.write_reserved_run_artifacts( + runs_dir=runs_dir, + generated_at="2026-09-16T00:00:01+00:00", + record={ + "goal_id": GOAL_ID, + "generated_at": "2026-09-16T00:00:01+00:00", + }, + index_record={ + "goal_id": GOAL_ID, + "generated_at": "2026-09-16T00:00:01+00:00", + "classification": "state_refreshed", + }, + payload={"goal_id": GOAL_ID}, + render_markdown=lambda _payload: "# appended run", + ) + append_finished.set() + + append_thread = threading.Thread( + target=lambda: capture_errors(append_run), + ) + rebuild_thread.start() + assert rebuild_paused.wait(timeout=5), ( + "rebuild did not reach the pre-replace window" + ) + append_thread.start() + assert append_started.wait(timeout=5), "append thread did not start" + try: + assert not append_finished.wait(timeout=0.2), ( + "append must wait while collision rebuild owns the index" + ) + finally: + allow_rebuild.set() + rebuild_thread.join(timeout=5) + append_thread.join(timeout=5) + + assert not rebuild_thread.is_alive() + assert not append_thread.is_alive() + assert thread_errors == [] + rows = [ + json.loads(line) + for line in index_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert len(rows) == 3 + assert rows[-1]["classification"] == "state_refreshed"