Collect worker changed paths from git diff

src/forge_loop/sandbox/__init__.py

@@ -1,5 +1,6 @@
 """Sandbox and capability policy contracts."""
 
+from forge_loop.sandbox.changed_paths import worker_changed_paths
 from forge_loop.sandbox.policy import (
     CapabilityPolicy,
     FilesystemScope,
@@ -21,5 +22,6 @@
     "mcp_allow_patterns",
     "policy_hash",
     "render_capability_policy",
+    "worker_changed_paths",
     "write_root_violations",
 ]

src/forge_loop/sandbox/changed_paths.py

@@ -0,0 +1,56 @@
+"""Collect a worker worktree's changed files from git output."""
+
+from __future__ import annotations
+
+import logging
+import os
+from collections.abc import Callable
+
+logger = logging.getLogger(__name__)
+
+GitOutputRunner = Callable[[tuple[str, ...], str], str]
+
+
+def _changed_path(worktree_path: str, name: str) -> str:
+    return os.path.normpath(os.path.abspath(os.path.join(worktree_path, name)))
+
+
+def worker_changed_paths(
+    run_git: GitOutputRunner,
+    worktree_path: str,
+    *,
+    base_ref: str,
+) -> tuple[str, ...]:
+    """Absolute normalized paths changed in ``worktree_path`` relative to ``base_ref``.
+
+    Uses injected git execution only: tracked modifications come from
+    ``git diff --name-only <base_ref>`` and untracked, non-ignored files from
+    ``git ls-files --others --exclude-standard``. Any git failure returns an
+    empty tuple so collection cannot abort the runner tick.
+    """
+    cwd = os.fspath(worktree_path)
+    try:
+        diff = run_git(("git", "diff", "--name-only", base_ref), cwd)
+        others = run_git(("git", "ls-files", "--others", "--exclude-standard"), cwd)
+    except Exception as exc:  # noqa: BLE001 — best-effort diff collection, never crash the tick
+        logger.warning(
+            "worker changed-path collection failed for worktree=%s base_ref=%s: %s",
+            cwd,
+            base_ref,
+            exc,
+            exc_info=True,
+        )
+        return ()
+
+    seen: set[str] = set()
+    changed: list[str] = []
+    for output in (diff, others):
+        for name in output.splitlines():
+            if not name.strip():
+                continue
+            path = _changed_path(cwd, name)
+            if path in seen:
+                continue
+            seen.add(path)
+            changed.append(path)
+    return tuple(changed)

tests/test_sandbox_changed_paths.py

@@ -0,0 +1,101 @@
+"""Unit tests for worker changed-path collection from git output (issue #442)."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+import pytest
+
+from forge_loop.sandbox.changed_paths import worker_changed_paths
+
+
+def test_worker_changed_paths_is_exported_from_sandbox_package() -> None:
+    from forge_loop.sandbox import worker_changed_paths as exported
+
+    assert exported is worker_changed_paths
+
+
+class FakeGit:
+    def __init__(
+        self,
+        *,
+        diff: str = "",
+        others: str = "",
+        raise_on: tuple[str, ...] = (),
+    ) -> None:
+        self.calls: list[tuple[tuple[str, ...], str]] = []
+        self._outputs = {"diff": diff, "ls-files": others}
+        self._raise_on = set(raise_on)
+
+    def __call__(self, argv: tuple[str, ...], cwd: str) -> str:
+        self.calls.append((argv, cwd))
+        command = argv[1]
+        if command in self._raise_on:
+            raise RuntimeError(f"{command} failed")
+        return self._outputs[command]
+
+
+def _abs(worktree: Path, *names: str) -> tuple[str, ...]:
+    return tuple(os.path.normpath(os.path.abspath(str(worktree / name))) for name in names)
+
+
+@pytest.mark.parametrize(
+    ("diff", "others", "expected"),
+    [
+        ("src/a.py\nsrc/b.py\n", "new.txt\n", ("src/a.py", "src/b.py", "new.txt")),
+        (
+            "src/a.py\nshared.txt\n",
+            "shared.txt\nnew.txt\n",
+            ("src/a.py", "shared.txt", "new.txt"),
+        ),
+        ("./src/../src/a.py\nsub/./x\n", "sub/../new.txt\n", ("src/a.py", "sub/x", "new.txt")),
+    ],
+)
+def test_worker_changed_paths_collects_deduplicates_and_normalizes(
+    tmp_path: Path, diff: str, others: str, expected: tuple[str, ...]
+) -> None:
+    git = FakeGit(diff=diff, others=others)
+
+    assert worker_changed_paths(git, str(tmp_path), base_ref="origin/trunk") == _abs(
+        tmp_path, *expected
+    )
+
+
+def test_worker_changed_paths_empty_output_is_empty_tuple(tmp_path: Path) -> None:
+    git = FakeGit(diff="\n  \n", others="")
+
+    assert worker_changed_paths(git, str(tmp_path), base_ref="origin/trunk") == ()
+
+
+def test_worker_changed_paths_preserves_significant_path_spaces(tmp_path: Path) -> None:
+    git = FakeGit(diff=" allowed/escape.py\nallowed/ok.py \n")
+
+    assert worker_changed_paths(git, str(tmp_path), base_ref="origin/trunk") == _abs(
+        tmp_path, " allowed/escape.py", "allowed/ok.py "
+    )
+
+
+@pytest.mark.parametrize(("raise_on", "diff"), [("diff", ""), ("ls-files", "src/a.py\n")])
+def test_worker_changed_paths_returns_empty_and_logs_when_git_raises(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture, raise_on: str, diff: str
+) -> None:
+    git = FakeGit(diff=diff, raise_on=(raise_on,))
+
+    caplog.set_level(logging.WARNING, logger="forge_loop.sandbox.changed_paths")
+
+    assert worker_changed_paths(git, str(tmp_path), base_ref="origin/trunk") == ()
+    assert "worktree=" in caplog.text
+    assert "base_ref=origin/trunk" in caplog.text
+
+
+def test_worker_changed_paths_invokes_expected_git_commands(tmp_path: Path) -> None:
+    git = FakeGit(diff="", others="")
+
+    worker_changed_paths(git, str(tmp_path), base_ref="origin/base")
+
+    assert git.calls == [
+        (("git", "diff", "--name-only", "origin/base"), str(tmp_path)),
+        (("git", "ls-files", "--others", "--exclude-standard"), str(tmp_path)),
+    ]