diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 2c999f4360..e0e82ffcd7 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -5,6 +5,7 @@ import os import secrets import socket +import stat import subprocess import sys import tempfile @@ -280,42 +281,38 @@ def _control_plane_root() -> Path: return Path(__file__).resolve().parent -def _scan_runtime_source_files(root: Path) -> tuple[str, ...]: - files: list[str] = [] +def _scan_runtime_source_snapshot(root: Path) -> _RuntimeSourceSnapshot: + snapshot: list[tuple[str, int, int, int]] = [] for directory, child_directories, filenames in os.walk(root): child_directories.sort() relative_directory = os.path.relpath(directory, root) if relative_directory == ".": relative_directory = "" - files.extend( - Path(relative_directory, filename).as_posix() - for filename in sorted(filenames) - if Path(filename).suffix in _RUNTIME_SOURCE_SUFFIXES - and Path(directory, filename).is_file() - ) - return tuple(sorted(files)) - - -def _runtime_file_snapshot( - root: Path, - files: tuple[str, ...], -) -> _RuntimeSourceSnapshot: - return tuple( - ( - relative, - (metadata := (root / relative).stat()).st_mtime_ns, - metadata.st_ctime_ns, - metadata.st_size, - ) - for relative in files - ) + for filename in sorted(filenames): + dot = filename.rfind(".") + if dot <= 0 or filename[dot:] not in _RUNTIME_SOURCE_SUFFIXES: + continue + source = Path(directory, filename) + try: + metadata = source.stat() + except OSError: + # Preserve pathlib's missing-file and platform-specific error policy. + if not source.is_file(): + continue + raise + except ValueError: + continue + if stat.S_ISREG(metadata.st_mode): + snapshot.append((Path(relative_directory, filename).as_posix(), + metadata.st_mtime_ns, metadata.st_ctime_ns, metadata.st_size)) + return tuple(sorted(snapshot)) def _runtime_source_snapshot(root: Path | None = None) -> _RuntimeSourceSnapshot: """Return cheap metadata that invalidates the packaged-source hash.""" source_root = (root or _control_plane_root()).resolve() - return _runtime_file_snapshot(source_root, _scan_runtime_source_files(source_root)) + return _scan_runtime_source_snapshot(source_root) def _runtime_source_files(root: Path | None = None) -> tuple[str, ...]: @@ -815,7 +812,6 @@ def _request_with_info( } with ExitStack() as stack: response_sink: Path | None = None - encoded = (json.dumps(request, separators=(",", ":")) + "\n").encode() if large_local_snapshot: if method not in LOCAL_SNAPSHOT_METHODS: raise EffectRuntimeRejected( @@ -846,6 +842,8 @@ def _request_with_info( "sha256": hashlib.sha256(params_bytes).hexdigest(), } encoded = (json.dumps(request, separators=(",", ":")) + "\n").encode() + else: + encoded = (json.dumps(request, separators=(",", ":")) + "\n").encode() if len(encoded) > MAX_REQUEST_BYTES: raise EffectRuntimeRejected( "TypeScript Effect runtime request is oversized", diff --git a/loopx/control_plane/todos/resume_planning.py b/loopx/control_plane/todos/resume_planning.py index 24a5da4f0a..418cda3dd4 100644 --- a/loopx/control_plane/todos/resume_planning.py +++ b/loopx/control_plane/todos/resume_planning.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +from collections import Counter from typing import Any from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result @@ -102,17 +103,19 @@ def build_referenced_todo_resume_planning_request( item_limit=item_limit, available_capabilities=available_capabilities, ) + distinct_rows = {id(row): row for rows in request["sources"].values() for row in rows} + id_counts = Counter(row["id"] for row in distinct_rows.values()) records: list[dict[str, Any]] = [] - references: dict[str, int] = {} + references: dict[tuple[str | None, str | None], int] = {} sources: dict[str, list[int]] = {} for source_name, source_rows in request["sources"].items(): source_refs: list[int] = [] for row in source_rows: - # Equal normalized records are semantically interchangeable. Use - # their exact JSON form instead of Todo id alone so conflicting - # historical facts remain separate and are still diagnosed. - identity = json.dumps( - row, ensure_ascii=False, sort_keys=True, separators=(",", ":") + # Distinct ids cannot share a full record; repeated ids still need exact facts. + identity = ( + row["id"], + json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + if id_counts[row["id"]] > 1 else None, ) reference = references.get(identity) if reference is None: diff --git a/tests/control_plane/test_quota_selection.py b/tests/control_plane/test_quota_selection.py index c33cee9cbf..cf80ebaaca 100644 --- a/tests/control_plane/test_quota_selection.py +++ b/tests/control_plane/test_quota_selection.py @@ -98,6 +98,21 @@ def test_quota_returned_records_do_not_alias_source_or_other_lanes(entrypoint): assert other["resume_condition"]["target_status"] == "open" +def test_fitting_request_preserves_a_pending_payload_in_an_expanded_response(): + from loopx.control_plane.todos.quota_selection import project_quota_planning + + # This currently fits the request wire but expands beyond the response wire. + row = item("todo_wait", role="agent", task_class="advancement_task", text="x" * 600_000, + required_capabilities=["shell"], resume_when="todo_done:todo_dependency", resume_ready=False, + resume_condition={"kind": "todo_done", "satisfied": False, + "target_todo_id": "todo_dependency", "target_status": "open", + "target_task_class": "advancement_task"}) + result = project_quota_planning(summary([row]), all_open_items=[row], source_open_count=1, + agent_identity={"agent_id": "agent-b"}, filter_user_gate_blocks_agent=False, + available_capabilities=[], current_time="2026-10-01T00:00:00Z") + assert result["resume_planning"]["blocked_successor_items"][0]["text"] == row["text"] + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_large_history_completes_public_quota_through_vision_wait(tmp_path, monkeypatch, provider): isolate_sqlite_runtime(tmp_path, monkeypatch) diff --git a/tests/control_plane/test_resume_planning_projection.py b/tests/control_plane/test_resume_planning_projection.py index 75cd0658ea..4267f0d0e6 100644 --- a/tests/control_plane/test_resume_planning_projection.py +++ b/tests/control_plane/test_resume_planning_projection.py @@ -58,15 +58,16 @@ def ids(items: list[dict]) -> list[str]: return [item["todo_id"] for item in items] -def test_referenced_request_deduplicates_lanes_without_pruning_payload_facts() -> None: +@pytest.mark.parametrize("copy_rows", [False, True], ids=["shared-object", "equal-copies"]) +def test_referenced_request_deduplicates_lanes_without_pruning_payload_facts(copy_rows) -> None: row = waiting( "todo_shared", content_revision="sha256:synthetic-full-projection", ) request = build_referenced_todo_resume_planning_request({ "items": [row], - "first_open_items": [row], - "deferred_items": [row], + "first_open_items": [deepcopy(row) if copy_rows else row], + "deferred_items": [deepcopy(row) if copy_rows else row], }) assert request["schema_version"] == "todo_resume_planning_request_v1" diff --git a/tests/control_plane/test_status_rollout_event_snapshot.py b/tests/control_plane/test_status_rollout_event_snapshot.py index 21b62ab319..964bcc4207 100644 --- a/tests/control_plane/test_status_rollout_event_snapshot.py +++ b/tests/control_plane/test_status_rollout_event_snapshot.py @@ -372,16 +372,16 @@ def test_collect_status_scans_effect_runtime_sources_once_per_request( monkeypatch: pytest.MonkeyPatch, ) -> None: registry_path, runtime_root, _goal = _write_fixture(tmp_path) - original_scan = effect_runtime._scan_runtime_source_files + original_scan = effect_runtime._scan_runtime_source_snapshot scanned_roots: list[Path] = [] - def counted_scan(root: Path) -> tuple[str, ...]: + def counted_scan(root: Path) -> effect_runtime._RuntimeSourceSnapshot: scanned_roots.append(root) return original_scan(root) monkeypatch.setattr( effect_runtime, - "_scan_runtime_source_files", + "_scan_runtime_source_snapshot", counted_scan, ) diff --git a/tests/control_plane/test_turn_journal_runtime_readiness.py b/tests/control_plane/test_turn_journal_runtime_readiness.py index 1d628186f9..1fc87ff040 100644 --- a/tests/control_plane/test_turn_journal_runtime_readiness.py +++ b/tests/control_plane/test_turn_journal_runtime_readiness.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os from pathlib import Path from threading import Event from typing import Any @@ -17,6 +18,64 @@ def __init__(self, *, stdout: str, returncode: int = 0) -> None: self.returncode = returncode +def test_source_snapshot_observes_each_selected_file_metadata_once(tmp_path, monkeypatch): + names = ["first.ts", "nested/data.json", "....json", "folder.ts/child.ts"] + for name in [*names, ".json", ".ts", "ignored.py", "UPPER.TS"]: + path = tmp_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("{}", encoding="utf-8") + for name in names: + os.utime(tmp_path / name, ns=(1_000_000_000, 1_000_000_000)) + selected = {tmp_path / name for name in names} + expected = [] + for name in sorted(names): + metadata = (tmp_path / name).stat() + expected.append((name, metadata.st_mtime_ns, metadata.st_ctime_ns, metadata.st_size)) + observed: list[Path] = [] + original_stat = Path.stat + + def observe_stat(path, *args, **kwargs): + if path in selected: + observed.append(path) + return original_stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", observe_stat) + assert effect_runtime._runtime_source_snapshot(tmp_path) == tuple(expected) + assert sorted(observed) == sorted(selected) + + +@pytest.mark.skipif(os.name == "nt", reason="requires filesystem symlinks") +def test_source_snapshot_follows_file_links_and_ignores_directory_or_broken_links(tmp_path): + root = tmp_path / "package" + root.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + target = outside / "actual.ts" + target.write_text("export {};", encoding="utf-8") + (root / "alias.ts").symlink_to(target) + (root / "directory").symlink_to(outside, target_is_directory=True) + (root / "missing.ts").symlink_to(outside / "missing.ts") + (root / "loop.ts").symlink_to(root / "loop.ts") + metadata = target.stat() + assert effect_runtime._runtime_source_snapshot(root) == ( + ("alias.ts", metadata.st_mtime_ns, metadata.st_ctime_ns, metadata.st_size), + ) + + +def test_source_snapshot_keeps_pathlib_policy_for_nonencodable_entries(tmp_path, monkeypatch): + source = tmp_path / "source.ts" + source.write_text("export {};", encoding="utf-8") + original_stat = Path.stat + + def nonencodable_stat(path, *args, **kwargs): + if path == source: + raise ValueError("non-encodable path") + return original_stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", nonencodable_stat) + assert effect_runtime._runtime_source_snapshot(tmp_path) == () + + def test_runtime_fingerprint_rotates_when_any_owned_source_changes( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -91,17 +150,17 @@ def test_runtime_fingerprint_rescans_when_a_discovered_file_disappears( removed = tmp_path / "removed.ts" kept.write_text("export const kept = true;\n", encoding="utf-8") removed.write_text("export const removed = true;\n", encoding="utf-8") - original_scan = effect_runtime._scan_runtime_source_files + original_scan = effect_runtime._scan_runtime_source_snapshot scans: list[tuple[str, ...]] = [] - def scan_then_remove(root: Path) -> tuple[str, ...]: + def scan_then_remove(root: Path) -> effect_runtime._RuntimeSourceSnapshot: files = original_scan(root) - scans.append(files) + scans.append(tuple(row[0] for row in files)) if len(scans) == 1: removed.unlink() return files - monkeypatch.setattr(effect_runtime, "_scan_runtime_source_files", scan_then_remove) + monkeypatch.setattr(effect_runtime, "_scan_runtime_source_snapshot", scan_then_remove) monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: tmp_path) @@ -152,15 +211,15 @@ def _install_persistent_stat_read_churn( later = tmp_path / "later.ts" first.write_text("export const first = true;\n", encoding="utf-8") later.write_text("export const later = true;\n", encoding="utf-8") - original_scan = effect_runtime._scan_runtime_source_files + original_scan = effect_runtime._scan_runtime_source_snapshot original_read_bytes = Path.read_bytes scans: list[tuple[str, ...]] = [] first_read = Event() later_reads = 0 - def record_scan(root: Path) -> tuple[str, ...]: + def record_scan(root: Path) -> effect_runtime._RuntimeSourceSnapshot: files = original_scan(root) - scans.append(files) + scans.append(tuple(row[0] for row in files)) return files def churn_after_each_later_read(path: Path) -> bytes: @@ -178,7 +237,7 @@ def churn_after_each_later_read(path: Path) -> bytes: return content monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: tmp_path) - monkeypatch.setattr(effect_runtime, "_scan_runtime_source_files", record_scan) + monkeypatch.setattr(effect_runtime, "_scan_runtime_source_snapshot", record_scan) monkeypatch.setattr(Path, "read_bytes", churn_after_each_later_read) return scans