From e6c9dde2a9b5e14b0de11508eee650d8d5f54d5f Mon Sep 17 00:00:00 2001 From: Renn F Date: Fri, 31 Jul 2026 01:53:07 +0200 Subject: [PATCH] fix(runtime): stop the chown storm from starving claims, and savepoint the PM journal auto-record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claim-shaped verbs were failing 7/7 (claim_review) and 6/6 (claim_doc_task) as silent 120s FlowVerbTimeout 504s on the NAS: the per-claim ownership repair walked the whole clone issuing two stat syscalls per entry (chown_ms 39502 vs git_ms 8 in the live log), several passes stacked per claim, and the claim transaction held the task row the whole time — so concurrent writers queued behind it into the 60s lock_timeout. The walk now does one stat per entry shared by the chown-skip and chmod-skip checks, and a .git/roboco-owned sentinel (worktree-aware via _resolve_clone_root, written only after a zero-failure pass) skips the walk entirely when the tree is already agent-owned. Every root-side git write invalidates the sentinel BEFORE its subprocess runs — GitService._run_git for scope != none, plus the three raw-subprocess paths inside WorkspaceService the adversarial pass proved bypass it deterministically on the common respawn shape (_worktree_git for mutating verbs, _fetch_branch_ref, _fetch_origin_best_effort) — so a live marker can never vouch for files a root write is about to create. One of those queued writers was the PM journal-decision auto-record: its INSERT hit the lock timeout, _ensure_pm_decision's catch-all swallowed it without rollback, and the poisoned session blew up escalate_up with PendingRollbackError (live incident). The helper's try body now runs in a savepoint — one fix covering all seven PM verbs that route through it — verified empirically against real Postgres in both directions: the failure path leaves the session healthy and the task object readable, and create_entry's internal commit inside the savepoint drains the transactional outbox exactly once. --- .../services/gateway/choreographer/_impl.py | 34 ++- roboco/services/git.py | 13 + roboco/services/workspace.py | 218 +++++++++++++-- .../unit/gateway/test_budget_unblock_guard.py | 9 + .../test_choreographer_delegate_guards.py | 9 + tests/unit/gateway/test_choreographer_pm.py | 47 ++++ .../gateway/test_delegate_ac_coverage_gate.py | 12 + .../gateway/test_delegate_incomplete_input.py | 9 + .../unit/gateway/test_delegate_parent_lock.py | 9 + .../gateway/test_delegate_project_routing.py | 9 + .../unit/gateway/test_oscillation_breaker.py | 9 + .../unit/gateway/test_unblock_flip_breaker.py | 9 + .../unit/services/test_git_ownership_scope.py | 66 +++++ ...test_workspace_ensure_agent_owned_scope.py | 251 +++++++++++++++++- .../test_workspace_owned_marker_bypass.py | 212 +++++++++++++++ 15 files changed, 874 insertions(+), 42 deletions(-) create mode 100644 tests/unit/services/test_workspace_owned_marker_bypass.py diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 424e40fc..ca8d4b64 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -3400,6 +3400,19 @@ class Choreographer: satisfies the gate, no duplicate is written. Best-effort — a journal write failure is logged and swallowed so the verb falls through to the normal gate (which rejects as before), never crashing the verb. + + Savepoint-guarded: the journal INSERT can lock-timeout on a + concurrent claim holding the task row's FK share lock (live + incident: an escalate_up request's write hit + ``LockNotAvailableError`` mid-flush). Swallowing that without a + rollback left the session's transaction poisoned — the very next + attribute touch anywhere in the request (``_escalate_up_preflight`` + reading ``t.id``) raised an unhandled ``PendingRollbackError`` + instead of the clean gate rejection this docstring promises. + ``begin_nested()`` scopes the failure to a SAVEPOINT the except + below rolls back to, leaving the outer transaction — and every + object this call didn't itself touch, e.g. the caller's ``t`` — + exactly as usable as if the write had never been attempted. """ from roboco.config import settings as _settings @@ -3407,16 +3420,17 @@ class Choreographer: if not text: return try: - latest = await self.journal.latest_decision_at(agent_id, task_id) - window = _settings.pm_decision_window_seconds - if ( - latest is not None - and (datetime.now(UTC) - latest).total_seconds() <= window - ): - return - await self.journal.write_decision( - agent_id=agent_id, task_id=task_id, content=text - ) + async with self.task.session.begin_nested(): + latest = await self.journal.latest_decision_at(agent_id, task_id) + window = _settings.pm_decision_window_seconds + if ( + latest is not None + and (datetime.now(UTC) - latest).total_seconds() <= window + ): + return + await self.journal.write_decision( + agent_id=agent_id, task_id=task_id, content=text + ) except Exception as exc: # best-effort; gate rejects normally on failure logger.warning( "auto-record pm decision failed", diff --git a/roboco/services/git.py b/roboco/services/git.py index 2f325476..25013f36 100644 --- a/roboco/services/git.py +++ b/roboco/services/git.py @@ -448,6 +448,13 @@ class GitService(BaseService): with "unable to append to .git/logs/refs/heads/...". A read-only op (status, log, diff, ...) never writes, so it skips the repair entirely. + + This is also the ONE chokepoint every root-side git write routes + through, so it's where the ownership-repair root-sentinel marker + (``_ensure_agent_owned``'s ``.git/roboco-owned``, workspace.py) gets + invalidated — BEFORE the op runs, since the op is what's about to + create new root-owned files a live marker would otherwise let a + later ``_ensure_agent_owned`` call wrongly skip. """ effective_timeout = timeout if timeout is not None else _default_git_timeout() @@ -477,6 +484,12 @@ class GitService(BaseService): loop = asyncio.get_running_loop() op = " ".join(args[:2]) + if _git_ownership_scope(args) != "none": + from roboco.services.workspace import invalidate_owned_marker + + await loop.run_in_executor( + _GIT_EXECUTOR, invalidate_owned_marker, workspace + ) t0 = time.monotonic() try: result = await loop.run_in_executor(_GIT_EXECUTOR, _run) diff --git a/roboco/services/workspace.py b/roboco/services/workspace.py index 8d9b57ee..6f006e5f 100644 --- a/roboco/services/workspace.py +++ b/roboco/services/workspace.py @@ -84,19 +84,35 @@ _PRUNE_DIRS = frozenset( ) -def _chown_entry(entry: str) -> bool: - """Chown a single entry; return True on success (or already correct).""" +def _chown_entry(entry: str, st: os.stat_result) -> bool: + """Chown a single entry to the agent uid/gid per the given (already-known) + stat; return True on success (or already correct).""" + if st.st_uid == _AGENT_UID and st.st_gid == _AGENT_GID: + return True try: - st = Path(entry).stat() - if st.st_uid != _AGENT_UID or st.st_gid != _AGENT_GID: - os.chown(entry, _AGENT_UID, _AGENT_GID) + os.chown(entry, _AGENT_UID, _AGENT_GID) except OSError: return False return True -def _make_owner_and_group_rw(entry: str) -> None: - """Best-effort chmod ensuring owner+group have rw (+x for dirs). +def _rw_mode_for(st_mode: int) -> int: + """The owner+group rw (+x for dirs) bits ``_make_owner_and_group_rw`` + ensures, given an entry's current mode. Shared with the root-sentinel + check in ``_root_already_owned`` so both agree on what "already has the + required bits" means. + """ + import stat as _stat + + mode = st_mode | _stat.S_IRUSR | _stat.S_IWUSR | _stat.S_IRGRP | _stat.S_IWGRP + if _stat.S_ISDIR(st_mode): + mode |= _stat.S_IXUSR | _stat.S_IXGRP + return mode + + +def _make_owner_and_group_rw(entry: str, st: os.stat_result) -> None: + """Best-effort chmod ensuring owner+group have rw (+x for dirs), given the + entry's already-known stat. NAS volumes with POSIX ACL inheritance can land cloned files with owner=0 (e.g. `.git/config` arriving as `----rw----`). POSIX permission @@ -107,25 +123,36 @@ def _make_owner_and_group_rw(entry: str) -> None: capabilities; if chown failed earlier (we're not root), we still can't chmod files we don't own, so this is best-effort by design. """ - import stat as _stat - - try: - st = Path(entry).stat() - new_mode = ( - st.st_mode | _stat.S_IRUSR | _stat.S_IWUSR | _stat.S_IRGRP | _stat.S_IWGRP - ) - if _stat.S_ISDIR(st.st_mode): - new_mode |= _stat.S_IXUSR | _stat.S_IXGRP - if new_mode != st.st_mode: - Path(entry).chmod(new_mode) - except OSError: - pass + new_mode = _rw_mode_for(st.st_mode) + if new_mode == st.st_mode: + return + with contextlib.suppress(OSError): + Path(entry).chmod(new_mode) def _own_and_grant_rw(entry: str) -> int: - """Chown + grant owner/group rw on one entry; return 1 if the chown failed.""" - failed = 0 if _chown_entry(entry) else 1 - _make_owner_and_group_rw(entry) + """Chown + grant owner/group rw on one entry; return 1 if the chown failed. + + One ``stat`` (follows symlinks, matching os.chown/Path.chmod's own + default follow behavior) now backs BOTH the chown-needed and + chmod-needed checks below, replacing what used to be two separate + ``Path.stat()`` calls (one inside each helper). The common case — an + agent re-claiming its own already-correctly-owned clone — costs one + read syscall and zero metadata-write syscalls per entry instead of two + stats plus, on some hosts, always re-testing each write independently. + On a NAS volume with tens of thousands of files, where every + chown/chmod is a copy-on-write metadata write, that is the difference + between a sub-second ownership pass and one that stacks tens of + seconds per claim verb. A stat failure (e.g. a broken symlink) is + treated as a chown failure, matching the prior behavior where the same + OSError surfaced from inside ``_chown_entry``'s own stat call. + """ + try: + st = Path(entry).stat() + except OSError: + return 1 + failed = 0 if _chown_entry(entry, st) else 1 + _make_owner_and_group_rw(entry, st) return failed @@ -166,10 +193,27 @@ def _ensure_agent_owned(workspace: Path) -> None: userns hosts) we log the failure instead of swallowing it, so a still-failing agent write is diagnosable rather than silent. 2. chmod owner+group rw. Belt + suspenders for ACL-inheriting NAS volumes. + + Root-sentinel short-circuit: if the workspace root itself is already + agent-owned with the right bits AND ``_root_already_owned`` finds the + marker from the last zero-failure pass, the ENTIRE walk is skipped — one + stat instead of walking tens of thousands of files, the remaining cost + of a re-claim on an already-correctly-owned NAS clone (this repo's + per-entry stat-collapse already halved the walk itself; this removes it + outright in the common case). The marker is deleted at the single + root-side git-write chokepoint (``GitService._run_git`` → + ``invalidate_owned_marker``) before any op that could create new + root-owned files, so a live marker is trustworthy: it can only ever be + stale-and-wrongly-trusted if some OTHER path creates root-owned files + without going through that chokepoint, which is the class this repo's + ownership repair exists to fix in the first place. """ if not workspace.exists(): return + if _root_already_owned(workspace): + return + failed_chowns = sum( _own_and_grant_rw(entry) for entry in _iter_ownable_entries(workspace) ) @@ -182,6 +226,8 @@ def _ensure_agent_owned(workspace: Path) -> None: workspace=str(workspace), failures=failed_chowns, ) + else: + _write_owned_marker(workspace) def _resolve_clone_root(workspace: Path) -> Path: @@ -198,6 +244,111 @@ def _resolve_clone_root(workspace: Path) -> Path: return workspace +# Sentinel filename recording "the last full _ensure_agent_owned pass over +# this clone found zero wrong-owned entries". Lives under `.git/` — git +# never tracks its own metadata dir, so this needs no .gitignore entry and +# sits outside every `_PRUNE_DIRS` exemption — instead of the working tree, +# so writing/deleting it never touches a tracked file. +_OWNED_MARKER_NAME = "roboco-owned" + + +def _owned_marker_path(workspace: Path) -> Path: + """The sentinel's path for a workspace or one of its worktrees. + + Worktree-aware via ``_resolve_clone_root``: a worktree checkout and its + clone root share the ONE ``.git`` they both ultimately read/write, so + they share one marker too. + """ + return _resolve_clone_root(workspace) / ".git" / _OWNED_MARKER_NAME + + +def _root_already_owned(workspace: Path) -> bool: + """True iff the workspace root is already agent-owned with the required + rw(+x) bits AND the marker attests the last full walk found zero + wrong-owned entries anywhere under the tree. + + Only the root gets stat'd here — the marker stands in for "every other + entry was already correct as of the last zero-failure pass", so the + common re-claim case costs one stat instead of walking the whole clone. + A missing/unreadable root or marker just falls through to the real walk + (safe default — this is a pure perf short-circuit, never a correctness + one). + """ + try: + st = workspace.stat() + except OSError: + return False + if st.st_uid != _AGENT_UID or st.st_gid != _AGENT_GID: + return False + if _rw_mode_for(st.st_mode) != st.st_mode: + return False + return _owned_marker_path(workspace).is_file() + + +def _write_owned_marker(workspace: Path) -> None: + """Record a zero-failure ``_ensure_agent_owned`` pass so the next call + can trust ``_root_already_owned`` and skip the walk entirely. + + Best-effort: a write failure just means the next call re-walks — it + only costs the perf win, never correctness. + """ + with contextlib.suppress(OSError): + marker = _owned_marker_path(workspace) + marker.parent.mkdir(parents=True, exist_ok=True) + marker.touch() + + +def invalidate_owned_marker(workspace: Path) -> None: + """Delete the ownership sentinel so the next ``_ensure_agent_owned`` + call re-walks instead of trusting stale state. + + Called from every root-side git-write chokepoint before the git + invocation that could create new root-owned files runs: + ``GitService._run_git`` (orchestrator-driven git ops) and, in this + module, ``WorkspaceService._worktree_git`` (mutating verbs only), + ``_fetch_branch_ref``, and ``_fetch_origin_best_effort`` — the + raw-subprocess worktree/fetch paths the spawn-time self-heal flow + (``ensure_worktree_self_heal`` -> ``_refresh_present_worktree``) hits on + (nearly) every spawn. A marker written BEFORE those files land would let + the very next ``_ensure_agent_owned`` call (concurrent or later) wrongly + skip them, stranding root-owned files the agent can't write. Best-effort: + a missing marker/workspace is a silent no-op, never an error. + """ + with contextlib.suppress(OSError): + _owned_marker_path(workspace).unlink() + + +# Verbs `_worktree_git` receives that never write anything, hand-enumerated +# against every real call site in this module (rev-parse --verify, rev-list +# --count, status --porcelain, symbolic-ref --short — the SET form of +# symbolic-ref is never used here). `branch` is the one ambiguous verb this +# helper is called with: `--show-current` only reads, while a create +# (`branch `) or delete (`branch -d/-D `) writes — handled +# separately in `_worktree_git_is_mutating` below rather than folded into +# this set. `git.py`'s `_git_ownership_scope` can't be imported here (git.py +# imports FROM workspace.py; the reverse would cycle). +_WORKTREE_GIT_ALWAYS_READ_ONLY = frozenset( + {"rev-parse", "rev-list", "status", "symbolic-ref"} +) + + +def _worktree_git_is_mutating(args: list[str]) -> bool: + """True iff a ``_worktree_git`` invocation could write anything. + + Everything this helper is ever called with besides the always-read-only + set and `branch` is mutating (checkout, reset, worktree add/remove/ + prune) — the safe default for an unrecognized/empty verb too. + """ + if not args: + return True + verb = args[0] + if verb in _WORKTREE_GIT_ALWAYS_READ_ONLY: + return False + if verb == "branch": + return any(not a.startswith("-") for a in args[1:]) + return True + + def _iter_git_dir_entries(clone_root: Path) -> Iterator[str]: """Yield ``clone_root/.git`` and every entry beneath it. @@ -517,6 +668,19 @@ class WorkspaceService: def _worktree_git( clone_root: Path, args: list[str], check: bool = True ) -> subprocess.CompletedProcess[str]: + """Run a raw ``git -C `` invocation (sync, no + token injection — internal worktree plumbing only). + + A mutating verb (anything but rev-parse/rev-list/status/ + symbolic-ref, or a `branch` create/delete) invalidates the + ownership-sentinel marker BEFORE the subprocess runs — this is one + of the root-side git-write chokepoints ``invalidate_owned_marker`` + documents; without it a stale marker lets ``_ensure_agent_owned`` + skip the walk that would repair the root-owned files this call is + about to create. + """ + if _worktree_git_is_mutating(args): + invalidate_owned_marker(clone_root) return subprocess.run( ["git", "-C", str(clone_root), *args], capture_output=True, @@ -684,6 +848,10 @@ class WorkspaceService: prefix = ["-c", f"http.extraheader=Authorization: Basic {basic}"] def _do_fetch() -> subprocess.CompletedProcess[str]: + # fetch always writes .git/objects + refs — a root-side git-write + # chokepoint (see invalidate_owned_marker's docstring). Invalidate + # BEFORE the subprocess runs so a marker can never straddle it. + invalidate_owned_marker(clone_root) return subprocess.run( [ "git", @@ -1133,6 +1301,10 @@ class WorkspaceService: return refs or ["master"] def _do_fetch() -> subprocess.CompletedProcess[str]: + # fetch always writes .git/objects + refs — a root-side git-write + # chokepoint (see invalidate_owned_marker's docstring). Invalidate + # BEFORE the subprocess runs so a marker can never straddle it. + invalidate_owned_marker(workspace) return subprocess.run( ["git", "fetch", "--no-tags", "--prune", "origin", *_scoped_refs()], cwd=str(workspace), diff --git a/tests/unit/gateway/test_budget_unblock_guard.py b/tests/unit/gateway/test_budget_unblock_guard.py index 84d6e234..54cc1340 100644 --- a/tests/unit/gateway/test_budget_unblock_guard.py +++ b/tests/unit/gateway/test_budget_unblock_guard.py @@ -34,6 +34,15 @@ def _make_deps(task_svc: AsyncMock) -> ChoreographerDeps: } base["journal"].has_decision_for_task.return_value = True base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/gateway/test_choreographer_delegate_guards.py b/tests/unit/gateway/test_choreographer_delegate_guards.py index f3df065a..12468d85 100644 --- a/tests/unit/gateway/test_choreographer_delegate_guards.py +++ b/tests/unit/gateway/test_choreographer_delegate_guards.py @@ -53,6 +53,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps: _ldef = base["journal"].latest_decision_at.return_value if type(_ldef).__name__ in ("MagicMock", "AsyncMock"): base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/gateway/test_choreographer_pm.py b/tests/unit/gateway/test_choreographer_pm.py index 03f70b6a..ef7740ba 100644 --- a/tests/unit/gateway/test_choreographer_pm.py +++ b/tests/unit/gateway/test_choreographer_pm.py @@ -12,6 +12,7 @@ from uuid import uuid4 import pytest from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps +from sqlalchemy.exc import OperationalError def _make_deps(**overrides: Any) -> ChoreographerDeps: @@ -969,6 +970,52 @@ async def test_escalate_up_blocks_without_journal_decision() -> None: assert "journal:decision" in body["missing"] +@pytest.mark.asyncio +async def test_escalate_up_survives_journal_write_lock_timeout() -> None: + """Regression: a journal:decision INSERT that lock-times out (a + concurrent claim transaction holding the task row's FK share lock — + live production 500) used to be swallowed by ``_ensure_pm_decision`` + with no rollback/savepoint, poisoning the session so the very next + attribute touch (``_escalate_up_preflight`` reading ``t.id``) raised an + unhandled ``PendingRollbackError``. The write is now savepoint-guarded + (``begin_nested()``): the failure is contained, the verb falls through + cleanly to the normal tracing_gap rejection (no decision was actually + persisted), and the task stays fully readable — no unhandled exception + escapes ``escalate_up``.""" + pm_id = uuid4() + task_id = uuid4() + t = MagicMock(id=task_id, status="blocked", assigned_to=pm_id, team="backend") + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock( + role="cell_pm", escalation_target="main-pm" + ) + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = False + journal_svc.latest_decision_at.return_value = None + journal_svc.write_decision.side_effect = OperationalError( + "INSERT INTO journal_entries (id, ...) VALUES (...)", + {}, + Exception("canceling statement due to lock timeout"), + ) + deps = _make_deps(task=task_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.escalate_up(pm_id, task_id, reason="needs cross-cell coordination") + + # The savepoint was actually engaged — proves the fix is wired in, not + # merely that AsyncMock happened to swallow the raise on its own. + task_svc.session.begin_nested.assert_called() + # No unhandled exception escaped escalate_up: the gate falls through to + # its normal clean rejection since the decision write never landed. + body = env.as_dict() + assert body["error"] == "tracing_gap" + assert "journal:decision" in body["missing"] + # The task is still fully readable afterward — this is exactly where + # the production trace crashed with PendingRollbackError on t.id. + assert t.id == task_id + + @pytest.mark.asyncio async def test_escalate_up_no_target_returns_invalid_state() -> None: """Verb-specific preflight: PM whose escalation_target is unconfigured. diff --git a/tests/unit/gateway/test_delegate_ac_coverage_gate.py b/tests/unit/gateway/test_delegate_ac_coverage_gate.py index 645bbca1..272ac532 100644 --- a/tests/unit/gateway/test_delegate_ac_coverage_gate.py +++ b/tests/unit/gateway/test_delegate_ac_coverage_gate.py @@ -64,6 +64,18 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps: _ldef = base["journal"].latest_decision_at.return_value if type(_ldef).__name__ in ("MagicMock", "AsyncMock"): base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. Only stub it for a mocked + # task service: one test below passes a REAL TaskService (get_task_service) + # over a live db_session, whose genuine begin_nested must stay intact. + if isinstance(base["task"], AsyncMock | MagicMock): + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/gateway/test_delegate_incomplete_input.py b/tests/unit/gateway/test_delegate_incomplete_input.py index 79823eb9..7f97dd64 100644 --- a/tests/unit/gateway/test_delegate_incomplete_input.py +++ b/tests/unit/gateway/test_delegate_incomplete_input.py @@ -52,6 +52,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps: _ldef = base["journal"].latest_decision_at.return_value if type(_ldef).__name__ in ("MagicMock", "AsyncMock"): base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/gateway/test_delegate_parent_lock.py b/tests/unit/gateway/test_delegate_parent_lock.py index 23a09ac9..733b2f32 100644 --- a/tests/unit/gateway/test_delegate_parent_lock.py +++ b/tests/unit/gateway/test_delegate_parent_lock.py @@ -50,6 +50,15 @@ def _make_deps(task: AsyncMock) -> ChoreographerDeps: # A fresh decision within the recency window so the delegate tracing gate # (journal:decision required) passes without a separate write. base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/gateway/test_delegate_project_routing.py b/tests/unit/gateway/test_delegate_project_routing.py index 819304d4..92b8b944 100644 --- a/tests/unit/gateway/test_delegate_project_routing.py +++ b/tests/unit/gateway/test_delegate_project_routing.py @@ -38,6 +38,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps: _ldef = base["journal"].latest_decision_at.return_value if type(_ldef).__name__ in ("MagicMock", "AsyncMock"): base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/gateway/test_oscillation_breaker.py b/tests/unit/gateway/test_oscillation_breaker.py index 38441b01..16ff7332 100644 --- a/tests/unit/gateway/test_oscillation_breaker.py +++ b/tests/unit/gateway/test_oscillation_breaker.py @@ -41,6 +41,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps: base.update(overrides) base["journal"].has_decision_for_task.return_value = True base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/gateway/test_unblock_flip_breaker.py b/tests/unit/gateway/test_unblock_flip_breaker.py index 53de9a58..12391fcc 100644 --- a/tests/unit/gateway/test_unblock_flip_breaker.py +++ b/tests/unit/gateway/test_unblock_flip_breaker.py @@ -38,6 +38,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps: base.update(overrides) base["journal"].has_decision_for_task.return_value = True base["journal"].latest_decision_at.return_value = datetime.now(UTC) + # _ensure_pm_decision's journal write is savepoint-guarded — an + # unconfigured AsyncMock's begin_nested() call returns a raw unawaited + # coroutine, which `async with` cannot use. + base["task"].session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) return ChoreographerDeps(**base) diff --git a/tests/unit/services/test_git_ownership_scope.py b/tests/unit/services/test_git_ownership_scope.py index db3ee082..5e56fa51 100644 --- a/tests/unit/services/test_git_ownership_scope.py +++ b/tests/unit/services/test_git_ownership_scope.py @@ -194,6 +194,72 @@ async def test_full_scope_op_calls_full_repair_not_git_repair( git_repair.assert_not_called() +@pytest.mark.asyncio +async def test_read_only_op_does_not_invalidate_owned_marker( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A read-only op writes nothing, so the ownership-sentinel marker + (`_ensure_agent_owned`'s root short-circuit) stays valid — invalidating + it here would force a needless full walk on the very next call.""" + (tmp_path / ".git").mkdir() + monkeypatch.setattr( + "roboco.services.git.subprocess.run", lambda *_a, **_k: _ok(["status"]) + ) + invalidate = MagicMock() + monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate) + + await _svc()._run_git(tmp_path, ["status", "--porcelain"]) + + invalidate.assert_not_called() + + +@pytest.mark.asyncio +async def test_git_scoped_op_invalidates_owned_marker_before_running( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A `.git`-only-writing op (commit) can still create new root-owned + files, so it must invalidate the marker too, not just checkout/reset/ + etc. — and it must do so BEFORE the subprocess runs, so a marker still + trusted by a concurrent `_ensure_agent_owned` call can never straddle + the write.""" + (tmp_path / ".git").mkdir() + order: list[str] = [] + + def _run_subprocess(*_a: object, **_k: object) -> subprocess.CompletedProcess[str]: + order.append("subprocess.run") + return _ok(["commit"]) + + monkeypatch.setattr("roboco.services.git.subprocess.run", _run_subprocess) + monkeypatch.setattr( + "roboco.services.workspace.invalidate_owned_marker", + lambda _ws: order.append("invalidate_owned_marker"), + ) + monkeypatch.setattr("roboco.services.workspace._ensure_git_dir_owned", MagicMock()) + + await _svc()._run_git(tmp_path, ["commit", "-m", "msg"]) + + assert order == ["invalidate_owned_marker", "subprocess.run"] + + +@pytest.mark.asyncio +async def test_full_scope_op_invalidates_owned_marker( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """checkout/reset/rebase/pull can create root-owned working-tree files + too — the marker invalidation isn't scoped to `.git`-only writes.""" + (tmp_path / ".git").mkdir() + monkeypatch.setattr( + "roboco.services.git.subprocess.run", lambda *_a, **_k: _ok(["checkout"]) + ) + invalidate = MagicMock() + monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate) + monkeypatch.setattr("roboco.services.workspace._ensure_agent_owned", MagicMock()) + + await _svc()._run_git(tmp_path, ["checkout", "some-branch"]) + + invalidate.assert_called_once_with(tmp_path) + + @pytest.mark.asyncio async def test_reown_after_git_op_returns_zero_ms_when_skipped() -> None: """The instrumentation must see a true near-zero cost for a skipped repair, diff --git a/tests/unit/services/test_workspace_ensure_agent_owned_scope.py b/tests/unit/services/test_workspace_ensure_agent_owned_scope.py index ae625286..692a46a3 100644 --- a/tests/unit/services/test_workspace_ensure_agent_owned_scope.py +++ b/tests/unit/services/test_workspace_ensure_agent_owned_scope.py @@ -10,14 +10,19 @@ approach) left the working tree root-owned and broke every agent file write. from __future__ import annotations -from typing import TYPE_CHECKING +import os +import stat as stat_module +from pathlib import Path +from unittest.mock import MagicMock import pytest from roboco.services import workspace as workspace_module -from roboco.services.workspace import _ensure_agent_owned - -if TYPE_CHECKING: - from pathlib import Path +from roboco.services.workspace import ( + _AGENT_GID, + _AGENT_UID, + _ensure_agent_owned, + _own_and_grant_rw, +) def _build_workspace(root: Path) -> None: @@ -48,11 +53,11 @@ def _record_touched(monkeypatch: pytest.MonkeyPatch) -> list[str]: """Record every path _ensure_agent_owned tries to chown/chmod.""" touched: list[str] = [] - def fake_chown_entry(entry: str) -> bool: + def fake_chown_entry(entry: str, _st: os.stat_result) -> bool: touched.append(entry) return True - def fake_make_rw(entry: str) -> None: + def fake_make_rw(entry: str, _st: os.stat_result) -> None: touched.append(entry) monkeypatch.setattr(workspace_module, "_chown_entry", fake_chown_entry) @@ -105,9 +110,11 @@ def test_chown_failure_falls_back_to_chmod_and_warns( (tmp_path / "file.py").write_text("x = 1\n") chmod_calls: list[str] = [] - monkeypatch.setattr(workspace_module, "_chown_entry", lambda _entry: False) + monkeypatch.setattr(workspace_module, "_chown_entry", lambda _entry, _st: False) monkeypatch.setattr( - workspace_module, "_make_owner_and_group_rw", chmod_calls.append + workspace_module, + "_make_owner_and_group_rw", + lambda entry, _st: chmod_calls.append(entry), ) warning_calls: list[tuple[str, dict[str, object]]] = [] monkeypatch.setattr( @@ -123,3 +130,229 @@ def test_chown_failure_falls_back_to_chmod_and_warns( # The failure is surfaced, not swallowed. assert warning_calls assert warning_calls[0][1]["failures"] + + +# --------------------------------------------------------------------------- +# `_own_and_grant_rw`: one shared stat now backs both the chown-needed and +# chmod-needed checks (previously two separate `Path.stat()` calls, one per +# helper) — the fix for chown_ms: 39502 on the production NAS, where every +# chown/chmod is a copy-on-write metadata write and the tree is almost always +# ALREADY correctly owned on a re-claim. These exercise the real (unmocked) +# `_own_and_grant_rw` / `_chown_entry` / `_make_owner_and_group_rw` at the +# os-syscall boundary. +# --------------------------------------------------------------------------- + + +def _stat_result(mode: int, uid: int = 0, gid: int = 0) -> os.stat_result: + """A real ``os.stat_result`` exposing only the fields the ownership + helpers read (st_mode/st_uid/st_gid) — no filesystem entry needed.""" + return os.stat_result((mode, 0, 0, 0, uid, gid, 0, 0, 0, 0)) + + +def _fake_stat(result: os.stat_result) -> object: + """A stand-in for ``os.stat`` accepting the ``(path, *, follow_symlinks)`` + signature ``Path(entry).stat()`` actually calls it with underneath.""" + + def _stat(_path: object, **_kwargs: object) -> os.stat_result: + return result + + return _stat + + +_ALREADY_RW_MODE = ( + stat_module.S_IFREG + | stat_module.S_IRUSR + | stat_module.S_IWUSR + | stat_module.S_IRGRP + | stat_module.S_IWGRP +) +_ROOT_NARROW_MODE = stat_module.S_IFREG | stat_module.S_IRUSR | stat_module.S_IWUSR + + +def test_own_and_grant_rw_skips_syscalls_when_already_correct( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An already agent-owned, already rw entry costs one stat and ZERO + metadata-write syscalls — the no-op-write cost that stacked to + chown_ms: 39502 across tens of thousands of files when the tree was + already correctly owned, as it almost always is on a re-claim.""" + chown_mock = MagicMock() + chmod_mock = MagicMock() + monkeypatch.setattr( + workspace_module.os, + "stat", + _fake_stat(_stat_result(_ALREADY_RW_MODE, _AGENT_UID, _AGENT_GID)), + ) + monkeypatch.setattr(workspace_module.os, "chown", chown_mock) + monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock) + + failed = _own_and_grant_rw("/fake/already-owned") + + assert failed == 0 + chown_mock.assert_not_called() + chmod_mock.assert_not_called() + + +def test_own_and_grant_rw_still_repairs_a_wrong_owned_entry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A genuinely wrong-owned entry (fresh root-side clone/fetch/checkout + output — root uid/gid, no group-write bit) still gets chowned AND + chmodded exactly as before the single-stat merge.""" + chown_mock = MagicMock() + chmod_mock = MagicMock() + monkeypatch.setattr( + workspace_module.os, "stat", _fake_stat(_stat_result(_ROOT_NARROW_MODE)) + ) + monkeypatch.setattr(workspace_module.os, "chown", chown_mock) + monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock) + + failed = _own_and_grant_rw("/fake/root-owned") + + assert failed == 0 + chown_mock.assert_called_once_with("/fake/root-owned", _AGENT_UID, _AGENT_GID) + expected_mode = _ROOT_NARROW_MODE | stat_module.S_IRGRP | stat_module.S_IWGRP + # chmod runs via Path(entry).chmod(...), which calls os.chmod with a + # Path-wrapped first arg + follow_symlinks=True — not the bare string. + chmod_mock.assert_called_once_with( + Path("/fake/root-owned"), expected_mode, follow_symlinks=True + ) + + +def test_own_and_grant_rw_chown_failure_still_counted_and_chmod_still_attempted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Unchanged failure-counting contract: a rejected chown (rootless / + userns host) still counts as one failure AND the chmod best-effort + fallback still runs (belt + suspenders for ACL-inheriting NAS volumes).""" + monkeypatch.setattr( + workspace_module.os, "stat", _fake_stat(_stat_result(_ROOT_NARROW_MODE)) + ) + + def _raise_chown(*_args: object, **_kwargs: object) -> None: + raise OSError("Operation not permitted") + + chmod_mock = MagicMock() + monkeypatch.setattr(workspace_module.os, "chown", _raise_chown) + monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock) + + failed = _own_and_grant_rw("/fake/rootless-host") + + assert failed == 1 + chmod_mock.assert_called_once() + + +def test_own_and_grant_rw_broken_symlink_counts_as_failure_without_crashing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Symlink decision: the shared stat call follows symlinks — matching + os.chown/os.chmod's own default follow behavior, unchanged from before + the merge (the old code's two separate `Path(entry).stat()` calls also + followed). A dangling symlink's stat raises OSError exactly as it did + from inside the old `_chown_entry`'s own stat call: counted as one + chown failure, and chmod is never attempted — the old chmod path hit + the identical OSError from its own separate stat call and silently + swallowed it, so the net effect (one counted failure, no chmod) is + unchanged.""" + broken_link = tmp_path / "dangling" + broken_link.symlink_to(tmp_path / "does-not-exist") + chown_mock = MagicMock() + chmod_mock = MagicMock() + monkeypatch.setattr(workspace_module.os, "chown", chown_mock) + monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock) + + failed = _own_and_grant_rw(str(broken_link)) + + assert failed == 1 + chown_mock.assert_not_called() + chmod_mock.assert_not_called() + + +# --------------------------------------------------------------------------- +# Root-sentinel short-circuit: `_ensure_agent_owned` skips the ENTIRE walk +# when the workspace root is already agent-owned with the right bits AND a +# `.git/roboco-owned` marker from the last zero-failure pass exists. The +# marker is invalidated by `GitService._run_git` +# (tests/unit/services/test_git_ownership_scope.py) before any root-side git +# write, so a live marker is trustworthy — this only removes the remaining +# per-entry-stat cost the earlier collapse (above) couldn't, the walk itself. +# --------------------------------------------------------------------------- + +_OWNED_DIR_MODE = ( + stat_module.S_IFDIR + | stat_module.S_IRUSR + | stat_module.S_IWUSR + | stat_module.S_IXUSR + | stat_module.S_IRGRP + | stat_module.S_IWGRP + | stat_module.S_IXGRP +) + + +def _fake_root_owned_stat(root: Path) -> object: + """Real ``os.stat`` for every path except ``root``, which reports as + agent-owned with the required rw+x bits. Lets the marker file's own + existence check (``Path.is_file()``, which also routes through + ``os.stat``) reflect the real filesystem instead of a blanket fake.""" + real_stat = os.stat + + def _stat(path: Path, *, follow_symlinks: bool = True) -> os.stat_result: + if path == root: + return _stat_result(_OWNED_DIR_MODE, _AGENT_UID, _AGENT_GID) + return real_stat(path, follow_symlinks=follow_symlinks) + + return _stat + + +def test_skips_walk_when_root_owned_and_marker_present( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _record_touched: list[str] +) -> None: + _build_workspace(tmp_path) + (tmp_path / ".git" / "roboco-owned").touch() + monkeypatch.setattr(workspace_module.os, "stat", _fake_root_owned_stat(tmp_path)) + walk_mock = MagicMock(return_value=iter(())) + monkeypatch.setattr(workspace_module.os, "walk", walk_mock) + + _ensure_agent_owned(tmp_path) + + walk_mock.assert_not_called() + assert _record_touched == [] + + +def test_full_walk_when_marker_absent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _record_touched: list[str] +) -> None: + """Root already agent-owned, but no marker: the last pass over this + clone is unattested, so the real walk still runs.""" + _build_workspace(tmp_path) + monkeypatch.setattr(workspace_module.os, "stat", _fake_root_owned_stat(tmp_path)) + + _ensure_agent_owned(tmp_path) + + assert str(tmp_path) in _record_touched + assert str(tmp_path / "README.md") in _record_touched + + +def test_marker_written_only_on_zero_failure_pass(tmp_path: Path) -> None: + _build_workspace(tmp_path) + marker = tmp_path / ".git" / "roboco-owned" + assert not marker.exists() + + # A real pass: chown to uid 1000 fails under the test's real (non-root) + # uid, exactly like a rootless/userns host — so no marker should land. + _ensure_agent_owned(tmp_path) + assert not marker.exists() + + +def test_marker_written_after_successful_pass( + tmp_path: Path, _record_touched: list[str] +) -> None: + """`_record_touched`'s fakes report every chown/chmod as succeeding, so + this exercises the zero-failure branch without needing real root.""" + _build_workspace(tmp_path) + marker = tmp_path / ".git" / "roboco-owned" + + _ensure_agent_owned(tmp_path) + + assert marker.is_file() + assert _record_touched # the walk actually ran (no marker existed yet) diff --git a/tests/unit/services/test_workspace_owned_marker_bypass.py b/tests/unit/services/test_workspace_owned_marker_bypass.py new file mode 100644 index 00000000..c4598dbc --- /dev/null +++ b/tests/unit/services/test_workspace_owned_marker_bypass.py @@ -0,0 +1,212 @@ +"""The ownership-sentinel marker (`_ensure_agent_owned`'s root short-circuit, +see test_workspace_ensure_agent_owned_scope.py) must be invalidated by EVERY +root-side git-write path, not just `GitService._run_git` +(test_git_ownership_scope.py covers that one). + +Adversarial review found a deterministic hole: `WorkspaceService`'s raw- +subprocess git helpers — `_worktree_git`, `_fetch_branch_ref`, +`_fetch_origin_best_effort` — never invalidated the marker, and the most +common spawn path hits them on (nearly) every respawn +(`ensure_worktree_self_heal` -> `_refresh_present_worktree` -> +`_fetch_branch_ref` + `_worktree_git(["reset", "--hard", ...])`). A stale +marker then let `_ensure_agent_owned` skip the walk that would have repaired +the root-owned files those calls had just created — a live Permission +denied for the agent. These tests cover the bypass paths directly. +""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from roboco.services.workspace import WorkspaceService, _ensure_agent_owned +from tests.unit.services.test_workspace_ensure_agent_owned_scope import ( + _build_workspace, +) + +if TYPE_CHECKING: + from pathlib import Path + + +def _svc() -> WorkspaceService: + return WorkspaceService(MagicMock()) + + +def _ok(args: list[str]) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + args=["git", *args], returncode=0, stdout="", stderr="" + ) + + +# --------------------------------------------------------------------------- +# `_worktree_git`: mutating verbs invalidate, read-only verbs don't. +# --------------------------------------------------------------------------- + + +def test_worktree_git_reset_hard_invalidates_marker( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr( + "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["reset"]) + ) + invalidate = MagicMock() + monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate) + + WorkspaceService._worktree_git(tmp_path, ["reset", "--hard", "origin/x"]) + + invalidate.assert_called_once_with(tmp_path) + + +def test_worktree_git_rev_parse_does_not_invalidate_marker( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr( + "roboco.services.workspace.subprocess.run", + lambda *_a, **_k: _ok(["rev-parse"]), + ) + invalidate = MagicMock() + monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate) + + WorkspaceService._worktree_git( + tmp_path, ["rev-parse", "--verify", "--quiet", "refs/heads/x"], check=False + ) + + invalidate.assert_not_called() + + +def test_worktree_git_branch_show_current_does_not_invalidate_marker( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The ambiguous verb's query form: `branch --show-current` only reads.""" + monkeypatch.setattr( + "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["branch"]) + ) + invalidate = MagicMock() + monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate) + + WorkspaceService._worktree_git(tmp_path, ["branch", "--show-current"], check=False) + + invalidate.assert_not_called() + + +def test_worktree_git_branch_delete_invalidates_marker( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The ambiguous verb's write forms: `branch -d/-D ` writes.""" + monkeypatch.setattr( + "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["branch"]) + ) + invalidate = MagicMock() + monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate) + + WorkspaceService._worktree_git( + tmp_path, ["branch", "-D", "task-branch"], check=False + ) + + invalidate.assert_called_once_with(tmp_path) + + +# --------------------------------------------------------------------------- +# `_fetch_branch_ref` — always mutating (fetch writes .git/objects + refs). +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fetch_branch_ref_invalidates_marker_before_subprocess( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + order: list[str] = [] + + def _run_subprocess(*_a: object, **_k: object) -> subprocess.CompletedProcess[str]: + order.append("subprocess.run") + return _ok(["fetch"]) + + monkeypatch.setattr("roboco.services.workspace.subprocess.run", _run_subprocess) + monkeypatch.setattr( + "roboco.services.workspace.invalidate_owned_marker", + lambda _ws: order.append("invalidate_owned_marker"), + ) + mock_project_service = MagicMock() + mock_project_service.get_by_slug = AsyncMock(return_value=None) + + with patch( + "roboco.services.project.get_project_service", + return_value=mock_project_service, + ): + await _svc()._fetch_branch_ref(tmp_path, "task-branch", "roboco-api") + + assert order == ["invalidate_owned_marker", "subprocess.run"] + + +# --------------------------------------------------------------------------- +# `_fetch_origin_best_effort` — same shape, scoped multi-ref fetch. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fetch_origin_best_effort_invalidates_marker( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr( + "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["fetch"]) + ) + invalidate = MagicMock() + monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate) + + await WorkspaceService._fetch_origin_best_effort(tmp_path, "roboco-api") + + invalidate.assert_called_once_with(tmp_path) + + +# --------------------------------------------------------------------------- +# End-to-end-shaped repro: a full zero-failure pass writes the marker; a +# bypass-path root write (through the now-fixed helper) invalidates it; the +# NEXT _ensure_agent_owned call walks again instead of trusting stale state. +# --------------------------------------------------------------------------- + + +def test_bypass_path_write_forces_next_ensure_agent_owned_to_walk( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _build_workspace(tmp_path) + touched_pass_1: list[str] = [] + touched_pass_2: list[str] = [] + + def make_fakes(sink: list[str]) -> tuple[object, object]: + def fake_chown_entry(entry: str, _st: object) -> bool: + sink.append(entry) + return True + + def fake_make_rw(entry: str, _st: object) -> None: + sink.append(entry) + + return fake_chown_entry, fake_make_rw + + # Pass 1: a normal zero-failure ensure_agent_owned — writes the marker. + fake_chown, fake_rw = make_fakes(touched_pass_1) + monkeypatch.setattr("roboco.services.workspace._chown_entry", fake_chown) + monkeypatch.setattr("roboco.services.workspace._make_owner_and_group_rw", fake_rw) + _ensure_agent_owned(tmp_path) + marker = tmp_path / ".git" / "roboco-owned" + assert marker.is_file() + assert touched_pass_1 # the walk actually ran + + # Bypass-path root write: a mutating _worktree_git call (the exact class + # of call ensure_worktree_self_heal's self-heal makes on nearly every + # respawn) — with subprocess mocked so no real git repo is needed, but + # invalidate_owned_marker running for real. + monkeypatch.setattr( + "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["reset"]) + ) + WorkspaceService._worktree_git(tmp_path, ["reset", "--hard", "origin/x"]) + assert not marker.exists() # the fix: the bypass path invalidated it + + # Pass 2: _ensure_agent_owned must walk again (marker gone), not trust + # the stale "fully owned" state from before the bypass-path write. + fake_chown_2, fake_rw_2 = make_fakes(touched_pass_2) + monkeypatch.setattr("roboco.services.workspace._chown_entry", fake_chown_2) + monkeypatch.setattr("roboco.services.workspace._make_owner_and_group_rw", fake_rw_2) + _ensure_agent_owned(tmp_path) + assert touched_pass_2 # the walk ran again — nothing was silently skipped