perf(runtime): remove redundant source probes and planning encoding

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",

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:

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)

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"

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,
     )
 

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