From d87e2d9b4e4c7e263b169f55c80aa87e51f7b545 Mon Sep 17 00:00:00 2001 From: Renn F Date: Thu, 30 Jul 2026 23:37:27 +0200 Subject: [PATCH 1/3] =?UTF-8?q?fix(lifecycle):=20stop=20PMs=20re-claiming?= =?UTF-8?q?=20tasks=20out=20of=20the=20closure=20queue=20=E2=80=94=20kills?= =?UTF-8?q?=20the=20i=5Fwill=5Fplan=20review=20loop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cell/main PM's i_will_plan could legally re-claim its own task from awaiting_pm_review (a CLAIM_RULES edge added for post-respawn recovery), resetting the task to in_progress and re-running submit_up -> pr_pass -> awaiting_pm_review forever: one Sentinel conventions child looped eleven full laps in four hours (14 reviews on one PR, 37 agent spawns) while its root's closure check fired eighteen times and the Main PM could never close anything. The claim edge is gone from every table that carried it — CLAIM_RULES, the claim ActionSpec's source statuses, the StatusTransition row, the service-layer _ROLE_CLAIM_STATUSES twin, and the legacy enforcement shim's operational-edge/role-gate entries (left divergent, it would be the same silent two-table drift that produced this bug). The respawn case the edge existed for is now served properly: _handle_pm_reentry gained a third contract — a PM calling i_will_plan on its own awaiting_pm_review task gets a steering envelope (no claim, no state change) pointing at complete/request_changes, and give_me_work's next hint for that status says the same instead of steering back into i_will_plan. The no-transition review-claim path and the pm-review dispatch prompt were already correct and are untouched, so closure still converges through them. A new bidirectional test asserts CLAIM_RULES and _ROLE_CLAIM_STATUSES stay identical per PM role (the old comment claimed a sync test existed; it checked one direction only). Lifecycle artifacts regenerated; the parity suite's three unshaped session mocks fixed, zero AsyncMock warnings remain. --- docs/rag/lifecycle/status-transitions.md | 1 - panel/lib/lifecycle.json | 11 -- roboco/enforcement/task_lifecycle.py | 17 +- roboco/foundation/policy/lifecycle.py | 53 +++---- .../services/gateway/choreographer/_impl.py | 47 +++++- roboco/services/task.py | 24 ++- .../test_lifecycle_consumer_parity.py | 40 +++++ tests/foundation/test_lifecycle_spec.py | 30 ++-- tests/unit/enforcement/test_task_lifecycle.py | 34 ++++ .../gateway/test_claim_doc_task_checkout.py | 10 ++ .../test_pm_review_reentry_loop_fix.py | 148 ++++++++++++++++++ .../services/test_pm_claim_needs_revision.py | 29 +++- 12 files changed, 373 insertions(+), 71 deletions(-) create mode 100644 tests/unit/gateway/test_pm_review_reentry_loop_fix.py diff --git a/docs/rag/lifecycle/status-transitions.md b/docs/rag/lifecycle/status-transitions.md index 083614ab..d7f39fdf 100644 --- a/docs/rag/lifecycle/status-transitions.md +++ b/docs/rag/lifecycle/status-transitions.md @@ -11,7 +11,6 @@ | awaiting_documentation | claimed | claim | documenter | | awaiting_pm_review | awaiting_ceo_approval | escalate_to_ceo | head_marketing, main_pm, product_owner | | awaiting_pm_review | cancelled | cancel | cell_pm, ceo, main_pm | -| awaiting_pm_review | claimed | claim | cell_pm, main_pm | | awaiting_pm_review | completed | complete | cell_pm, main_pm | | awaiting_pm_review | needs_revision | request_changes | cell_pm, main_pm | | awaiting_pr_review | awaiting_pm_review | pr_pass | pr_reviewer | diff --git a/panel/lib/lifecycle.json b/panel/lib/lifecycle.json index 8d0aa4c3..c673a92e 100644 --- a/panel/lib/lifecycle.json +++ b/panel/lib/lifecycle.json @@ -2,7 +2,6 @@ "claim_rules": { "auditor": [], "cell_pm": [ - "awaiting_pm_review", "needs_revision", "pending" ], @@ -17,7 +16,6 @@ ], "head_marketing": [], "main_pm": [ - "awaiting_pm_review", "needs_revision", "pending" ], @@ -531,15 +529,6 @@ "source": "awaiting_pm_review", "target": "cancelled" }, - { - "action": "claim", - "roles": [ - "cell_pm", - "main_pm" - ], - "source": "awaiting_pm_review", - "target": "claimed" - }, { "action": "complete", "roles": [ diff --git a/roboco/enforcement/task_lifecycle.py b/roboco/enforcement/task_lifecycle.py index 257902a3..90c46ab5 100644 --- a/roboco/enforcement/task_lifecycle.py +++ b/roboco/enforcement/task_lifecycle.py @@ -92,8 +92,14 @@ _LEGACY_OPERATIONAL_EDGES: dict[Status, frozenset[Status]] = { Status.VERIFYING: frozenset({Status.NEEDS_REVISION, Status.PENDING}), # QA can park a task as blocked while waiting on dev clarification. Status.AWAITING_QA: frozenset({Status.BLOCKED}), - # PM claim + PM reject path on review queue. - Status.AWAITING_PM_REVIEW: frozenset({Status.CLAIMED, Status.NEEDS_REVISION}), + # PM reject path on review queue. No CLAIMED edge here (removed): a PM + # claiming its own awaiting_pm_review task used to legally reset it to + # in_progress via i_will_plan's composed claim, looping submit_up -> + # pr_pass -> awaiting_pm_review forever. lifecycle.CLAIM_RULES / + # _ROLE_CLAIM_STATUSES close the edge upstream of this legacy view; the + # choreographer's _handle_pm_reentry now steers a re-entering PM to + # complete/request_changes instead. + Status.AWAITING_PM_REVIEW: frozenset({Status.NEEDS_REVISION}), # Re-entry from revision back into active dev work (without re-claim), or # voluntary unclaim back to the pool (TaskService.unclaim_for_agent) — a # dev sent back for revision otherwise had no legal exit but in_progress. @@ -113,13 +119,6 @@ _LEGACY_ROLE_GATES: dict[tuple[Status, Status], tuple[str, ...]] = { "main_pm", "product_owner", ), - # PM claim of review queue. - (Status.AWAITING_PM_REVIEW, Status.CLAIMED): ( - "cell_pm", - "head_marketing", - "main_pm", - "product_owner", - ), # PM reject back to dev (needs_revision). (Status.AWAITING_PM_REVIEW, Status.NEEDS_REVISION): ( "cell_pm", diff --git a/roboco/foundation/policy/lifecycle.py b/roboco/foundation/policy/lifecycle.py index 380c3bf0..6194aa50 100644 --- a/roboco/foundation/policy/lifecycle.py +++ b/roboco/foundation/policy/lifecycle.py @@ -245,14 +245,12 @@ _STATUS_TRANSITIONS: tuple[StatusTransition, ...] = ( frozenset({Role.DOCUMENTER}), ), StatusTransition(Status.NEEDS_REVISION, Status.CLAIMED, "claim", None), - # A PM re-claims an awaiting_pm_review task it already owns (CLAIM_RULES - # grants this to CELL_PM/MAIN_PM) — see the "claim" ActionSpec comment. - StatusTransition( - Status.AWAITING_PM_REVIEW, - Status.CLAIMED, - "claim", - frozenset({Role.CELL_PM, Role.MAIN_PM}), - ), + # No AWAITING_PM_REVIEW -> CLAIMED edge: a PM re-entering its own + # review-queue task is steered by the choreographer's i_will_plan + # re-entry contract straight to complete/request_changes, never via a + # claim (a claim edge here let i_will_plan legally reset the task and + # loop the submit_up -> pr_pass -> awaiting_pm_review cycle forever — + # see the CLAIM_RULES comment below). # Start StatusTransition(Status.CLAIMED, Status.IN_PROGRESS, "start", None), # Block / pause / resume @@ -454,14 +452,6 @@ _ATOMIC_ACTIONS: dict[str, ActionSpec] = { Status.AWAITING_QA, Status.AWAITING_DOCUMENTATION, Status.AWAITING_PR_REVIEW, - # A PM re-claims a review/queue task it already owns (e.g. after - # a respawn) via i_will_plan — CLAIM_RULES[CELL_PM/MAIN_PM] - # grants AWAITING_PM_REVIEW. Without it here, can_invoke_intent - # rejected i_will_plan on an awaiting_pm_review task with - # invalid_state even though CLAIM_RULES said it was allowed. - # test_claim_rules_match_pre_gateway_table keeps this source set - # and CLAIM_RULES in sync. - Status.AWAITING_PM_REVIEW, } ), target_status=Status.CLAIMED, @@ -749,21 +739,22 @@ CLAIM_RULES: dict[Role, frozenset[Status]] = { # agent its own assigned tasks. (A per-instance ownership gate at the gateway # would diverge from this spec — the parity invariant forbids that.) # - # AWAITING_PM_REVIEW: a PM re-claims its own review-queue task (e.g. after - # a respawn) via i_will_plan. roboco/services/task.py's - # _ROLE_CLAIM_STATUSES already granted this to "cell_pm"/"main_pm" on the - # stated belief that "the spec (lifecycle.CLAIM_RULES) grants it" — but - # CLAIM_RULES never actually did, so can_invoke_intent silently rejected - # every i_will_plan attempt on an awaiting_pm_review task with - # invalid_state regardless of what the service layer allowed. Added here - # to match; test_claim_rules_match_pre_gateway_table asserts CLAIM_RULES - # and the service table stay in sync so they can't diverge again. - Role.CELL_PM: frozenset( - {Status.PENDING, Status.NEEDS_REVISION, Status.AWAITING_PM_REVIEW} - ), - Role.MAIN_PM: frozenset( - {Status.PENDING, Status.NEEDS_REVISION, Status.AWAITING_PM_REVIEW} - ), + # AWAITING_PM_REVIEW is deliberately ABSENT here — do not re-add it. A task + # in this status already passed the in-path PR gate and is waiting on the + # owning PM's merge decision (complete / request_changes), not on + # re-planning. A prior revision granted CELL_PM/MAIN_PM a claim from + # AWAITING_PM_REVIEW so a respawned PM could "re-claim its own review-queue + # task", but claim composes into i_will_plan's (claim, set_plan, start) + # sequence: every respawn legally re-claimed the task, reset it to + # in_progress, and re-ran the full submit_up -> pr_pass -> + # awaiting_pm_review cycle — looping forever with no progress (one + # production task cycled 11 times across 37 spawns in 4h before this was + # caught). A PM re-entering its own AWAITING_PM_REVIEW task is now steered + # by the choreographer's i_will_plan re-entry contract (_handle_pm_reentry) + # straight to complete/request_changes, with no claim and no status change + # — closing the edge that made the reset possible in the first place. + Role.CELL_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}), + Role.MAIN_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}), Role.PRODUCT_OWNER: frozenset(), Role.HEAD_MARKETING: frozenset(), Role.AUDITOR: frozenset(), diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 9dc14257..424e40fc 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -519,7 +519,7 @@ class Choreographer: role_str: str, briefing: dict[str, Any], ) -> Envelope | None: - """Handle two distinct re-entry contracts for i_will_plan. + """Handle three distinct re-entry contracts for i_will_plan. Idempotent heartbeat: the PM already owns the task in in_progress — touch the heartbeat and return OK without re-running the spec gate. @@ -531,10 +531,20 @@ class Choreographer: re-claim (claimed is not a valid source for the claim transition) and run set_plan+start to complete the interrupted sequence. - Returns None when neither condition applies, signalling the caller to - continue to the normal claim-plan-start path. PLR0911 budget is the - secondary reason this lives in a helper; the domain contract above is - the primary one. + Review-queue steering: the task already passed the in-path PR gate and + is sitting in awaiting_pm_review awaiting this PM's merge decision. A + respawned PM re-offered this task by give_me_work used to call + i_will_plan on it, and CLAIM_RULES used to let that legally re-claim + it — running the full composed (claim, set_plan, start) sequence and + resetting the task to in_progress, which re-ran submit_up -> pr_pass + -> awaiting_pm_review forever (one production task looped 11 cycles / + 37 spawns in 4h before this was caught). Steer to complete / + request_changes instead, with NO claim and NO status change. + + Returns None when none of the three conditions applies, signalling the + caller to continue to the normal claim-plan-start path. PLR0911 budget + is the secondary reason this lives in a helper; the domain contract + above is the primary one. """ status = str(t.status) if status == "in_progress" and t.assigned_to == pm_agent_id: @@ -550,6 +560,20 @@ class Choreographer: return await self._post_claim_journal_gate( "i_will_plan", pm_agent_id, task_id, envelope ) + if status == "awaiting_pm_review" and t.assigned_to == pm_agent_id: + return Envelope.ok( + status=status, + task_id=str(task_id), + next=( + "this task already passed the PR-review gate and is" + " awaiting your merge decision — do NOT re-plan or" + " re-submit it. Call complete(task_id) to merge the" + " assembled PR, or request_changes(task_id," + " findings=[...]) to bounce it back with concrete" + " findings." + ), + context_briefing=briefing, + ).with_introspection(task=t, role=role_str) return None async def _pm_sub_tasks_gate( @@ -830,6 +854,12 @@ class Choreographer: handed an awaiting_documentation task (or QA an awaiting_qa task) was told to call a dev verb it doesn't have — it looped. Map to the verb that actually claims the task for this role. + + awaiting_pm_review is a review-queue state, not a re-plan state: a + PM offered its own already-gated task here must be steered to + complete/request_changes, never i_will_plan (i_will_plan legally + re-claiming from this status used to reset the task and loop the + submit_up -> pr_pass -> awaiting_pm_review cycle forever). """ tid = str(getattr(task, "id", "")) status = str(getattr(task, "status", "")) @@ -837,6 +867,13 @@ class Choreographer: return f"call claim_doc_task(task_id='{tid}') to start" if status == "awaiting_qa": return f"call claim_review(task_id='{tid}') to start" + if status == "awaiting_pm_review" and role in ("cell_pm", "main_pm"): + return ( + f"this task already passed the PR-review gate — call" + f" complete(task_id='{tid}') to merge, or" + f" request_changes(task_id='{tid}', findings=[...]) to bounce" + " it back; do NOT call i_will_plan" + ) if role in ("cell_pm", "main_pm", "product_owner", "head_marketing"): return f"call i_will_plan(task_id='{tid}', plan='') to start" return f"call i_will_work_on(task_id='{tid}', plan='') to start" diff --git a/roboco/services/task.py b/roboco/services/task.py index fb3dc550..8a70e67f 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -107,15 +107,26 @@ _ROLE_CLAIM_STATUSES: dict[str, set[TaskStatus]] = { # claim() returns None -> INVALID_STATE -> the PM respawn-loops on its own # rejected root, unable to plan or idle it. Parity is locked by # tests/unit/services/test_pm_claim_needs_revision.py. + # + # AWAITING_PM_REVIEW is deliberately absent: granting it here once let a + # respawned PM's i_will_plan legally re-claim its own review-queue task, + # resetting it to in_progress and re-running submit_up -> pr_pass -> + # awaiting_pm_review forever. lifecycle.CLAIM_RULES agrees, and unlike the + # NEEDS_REVISION parity above (spec ⊆ runtime only, via + # test_runtime_pm_claim_mapping_covers_spec_claim_rules), this exact + # per-PM-role SET is pinned bidirectionally against spec.CLAIM_RULES by + # test_claim_rules_and_role_statuses_are_identical — re-adding + # AWAITING_PM_REVIEW here alone (without touching the spec) fails that + # test instead of silently reopening the loop. The choreographer's + # _handle_pm_reentry now steers a re-entering PM straight to + # complete/request_changes instead, with no claim involved. "cell_pm": { TaskStatus.PENDING, TaskStatus.NEEDS_REVISION, - TaskStatus.AWAITING_PM_REVIEW, }, "main_pm": { TaskStatus.PENDING, TaskStatus.NEEDS_REVISION, - TaskStatus.AWAITING_PM_REVIEW, }, } @@ -3741,11 +3752,18 @@ class TaskService(BaseService): if task.assigned_to and str(task.assigned_to) != str(agent.id): markers.set_original_developer(task, task.assigned_to) + # Consulted only inside _finalize_claim (both its target-status write and + # its branch-failure rollback's reversal-audit check), reached only via + # claim() -> _validate_claim_preconditions -> _get_valid_claim_statuses, + # which already screens the pre-claim status against _ROLE_CLAIM_STATUSES + # per role. AWAITING_PM_REVIEW is deliberately absent: no role's + # _ROLE_CLAIM_STATUSES grants it anymore (the i_will_plan re-claim loop + # fix), so _finalize_claim can never run with that pre-claim status — + # listing it here would be dead weight, not an independent gate. _CLAIMABLE_STATUSES: ClassVar[set[TaskStatus]] = { TaskStatus.PENDING, TaskStatus.AWAITING_QA, TaskStatus.AWAITING_DOCUMENTATION, - TaskStatus.AWAITING_PM_REVIEW, } async def _claim_blocked_by_dependencies(self, task: TaskTable) -> bool: diff --git a/tests/foundation/test_lifecycle_consumer_parity.py b/tests/foundation/test_lifecycle_consumer_parity.py index 23930ad2..86011d05 100644 --- a/tests/foundation/test_lifecycle_consumer_parity.py +++ b/tests/foundation/test_lifecycle_consumer_parity.py @@ -869,6 +869,26 @@ async def test_complete_matches_spec(role: str, status: str) -> None: id=task_id, status="awaiting_ceo_approval", assigned_to=None, team="backend" ) task_svc.all_subtasks_terminal.return_value = True + # complete's merge path runs _stamp_pm_findings_verified_or_rejection, + # which opens a session.begin_nested() savepoint and reads the findings + # ledger via ReviewFindingsRepository.list_for_task (session.execute -> + # .scalars().all()). Unconfigured, both calls resolve to auto-generated + # AsyncMock children: `async with ():` fails the async + # context-manager protocol and `.scalars()` returns an + # unawaited coroutine — caught by the verb's own except Exception, but + # the never-awaited coroutine leaks a RuntimeWarning at GC time. + task_svc.session = MagicMock() + task_svc.session.execute = AsyncMock( + return_value=MagicMock( + scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))) + ) + ) + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) git_svc = AsyncMock() git_svc.pr_merge.return_value = {"merged": True, "merge_commit_sha": "x"} git_svc.create_pr.return_value = {"pr_number": 99, "pr_url": "x"} @@ -1268,6 +1288,18 @@ async def test_claim_review_matches_spec(role: str, status: str) -> None: task_svc.qa_claim.return_value = after task_svc.list_in_progress_for_agent.return_value = [] task_svc.list_paused_for_agent.return_value = [] + # claim_review's full=True briefing reads the findings ledger + # (findings.open_findings_for_task -> session.execute -> .scalars().all()). + # Unconfigured, session.execute auto-generates as an AsyncMock child whose + # call returns another AsyncMock; .scalars() on that is itself an + # unawaited coroutine — caught by open_findings_for_task's own fail-open + # except, but the dangling coroutine leaks a RuntimeWarning at GC time. + task_svc.session = MagicMock() + task_svc.session.execute = AsyncMock( + return_value=MagicMock( + scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))) + ) + ) deps = _make_deps(task_svc=task_svc) c = Choreographer(deps) @@ -1546,6 +1578,14 @@ async def test_claim_doc_task_matches_spec(role: str, status: str) -> None: task_svc.doc_claim.return_value = after task_svc.list_in_progress_for_agent.return_value = [] task_svc.list_paused_for_agent.return_value = [] + # claim_doc_task's full=True briefing reads the findings ledger the same + # way claim_review's does (see that test's comment) — same fix. + task_svc.session = MagicMock() + task_svc.session.execute = AsyncMock( + return_value=MagicMock( + scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))) + ) + ) deps = _make_deps(task_svc=task_svc) c = Choreographer(deps) diff --git a/tests/foundation/test_lifecycle_spec.py b/tests/foundation/test_lifecycle_spec.py index 238a7ce9..cfca53e1 100644 --- a/tests/foundation/test_lifecycle_spec.py +++ b/tests/foundation/test_lifecycle_spec.py @@ -491,6 +491,11 @@ def test_claim_rules_match_pre_gateway_table() -> None: re-delegating fixes (scoped by give_me_work routing, which offers only the caller's own assigned tasks). BACKLOG → PENDING is a separate `activate` action (strict transitions; no implicit activate-on-claim). + + AWAITING_PM_REVIEW is deliberately absent from both PM roles — a claim + edge there let a respawned PM's i_will_plan legally re-claim its own + review-queue task and loop the submit_up -> pr_pass -> awaiting_pm_review + cycle forever. See test_awaiting_pm_review_not_claimable_by_any_role. """ assert spec.CLAIM_RULES[spec.Role.DEVELOPER] == frozenset( {spec.Status.PENDING, spec.Status.NEEDS_REVISION} @@ -500,21 +505,26 @@ def test_claim_rules_match_pre_gateway_table() -> None: {spec.Status.PENDING, spec.Status.AWAITING_DOCUMENTATION} ) assert spec.CLAIM_RULES[spec.Role.CELL_PM] == frozenset( - { - spec.Status.PENDING, - spec.Status.NEEDS_REVISION, - spec.Status.AWAITING_PM_REVIEW, - } + {spec.Status.PENDING, spec.Status.NEEDS_REVISION} ) assert spec.CLAIM_RULES[spec.Role.MAIN_PM] == frozenset( - { - spec.Status.PENDING, - spec.Status.NEEDS_REVISION, - spec.Status.AWAITING_PM_REVIEW, - } + {spec.Status.PENDING, spec.Status.NEEDS_REVISION} ) +def test_awaiting_pm_review_not_claimable_by_any_role() -> None: + """The claim edge that let a respawned PM's i_will_plan reset an + awaiting_pm_review task (looping submit_up -> pr_pass -> awaiting_pm_review + forever) is permanently closed: no role may claim from this status. A PM + re-entering its own review task is steered by the choreographer directly + to complete/request_changes, never through claim. + """ + for role, statuses in spec.CLAIM_RULES.items(): + assert spec.Status.AWAITING_PM_REVIEW not in statuses, role + # PR_REVIEWER's own review-gate claim (a different status) is untouched. + assert spec.Status.AWAITING_PR_REVIEW in spec.CLAIM_RULES[spec.Role.PR_REVIEWER] + + def test_team_rules_pin_team_for_seeded_agents() -> None: assert spec.ROLE_TEAM_RULES["be-dev-1"] == "backend" assert spec.ROLE_TEAM_RULES["be-pm"] == "backend" diff --git a/tests/unit/enforcement/test_task_lifecycle.py b/tests/unit/enforcement/test_task_lifecycle.py index 799794d8..c998c997 100644 --- a/tests/unit/enforcement/test_task_lifecycle.py +++ b/tests/unit/enforcement/test_task_lifecycle.py @@ -277,6 +277,40 @@ def test_status_classification_is_mutually_disjoint() -> None: assert waiting & terminal == set() +# --------------------------------------------------------------------------- +# awaiting_pm_review -> claimed is closed (the i_will_plan re-claim loop): +# a PM re-entering its own review-queue task is steered by the choreographer +# straight to complete/request_changes, never via a claim that resets the +# task and re-runs submit_up -> pr_pass -> awaiting_pm_review forever. This +# legacy shim (_LEGACY_OPERATIONAL_EDGES / _LEGACY_ROLE_GATES) used to grant +# the same edge lifecycle.CLAIM_RULES had already closed — the identical +# two-tables-drift shape that caused the incident. +# --------------------------------------------------------------------------- + + +def test_awaiting_pm_review_claim_no_longer_allowed_for_pm_roles() -> None: + assert can_agent_transition("awaiting_pm_review", "claimed", "cell_pm") is False + assert can_agent_transition("awaiting_pm_review", "claimed", "main_pm") is False + with pytest.raises(TaskLifecycleError): + validate_task_transition("awaiting_pm_review", "claimed", "cell_pm") + with pytest.raises(TaskLifecycleError): + validate_task_transition("awaiting_pm_review", "claimed", "main_pm") + + +def test_awaiting_pm_review_needs_revision_still_allowed() -> None: + """The PM reject-back-to-dev path (request_changes) is untouched.""" + assert ( + can_agent_transition("awaiting_pm_review", "needs_revision", "cell_pm") is True + ) + assert ( + can_agent_transition("awaiting_pm_review", "needs_revision", "main_pm") is True + ) + assert ( + validate_task_transition("awaiting_pm_review", "needs_revision", "cell_pm") + is True + ) + + def test_status_classification_covers_every_enum_member() -> None: """Every Status enum member must be classified by EXACTLY one of is_terminal_state / is_active_state / is_waiting_state — the coverage diff --git a/tests/unit/gateway/test_claim_doc_task_checkout.py b/tests/unit/gateway/test_claim_doc_task_checkout.py index ff2ce514..b09344f2 100644 --- a/tests/unit/gateway/test_claim_doc_task_checkout.py +++ b/tests/unit/gateway/test_claim_doc_task_checkout.py @@ -86,6 +86,16 @@ def test_claim_verb_hint_pm_for_planning() -> None: assert "i_will_plan" in hint +def test_claim_verb_hint_pm_for_awaiting_pm_review_steers_to_complete() -> None: + """A PM's own review-queue task must never hint i_will_plan — that verb + used to legally re-claim and reset it, looping submit_up -> pr_pass -> + awaiting_pm_review forever.""" + for role in ("cell_pm", "main_pm"): + hint = Choreographer._claim_verb_hint(role, _task("awaiting_pm_review")) + assert "complete" in hint + assert "call i_will_plan(" not in hint + + def test_claim_verb_hint_dev_default() -> None: hint = Choreographer._claim_verb_hint("developer", _task("pending")) assert "i_will_work_on" in hint diff --git a/tests/unit/gateway/test_pm_review_reentry_loop_fix.py b/tests/unit/gateway/test_pm_review_reentry_loop_fix.py new file mode 100644 index 00000000..c56ce90e --- /dev/null +++ b/tests/unit/gateway/test_pm_review_reentry_loop_fix.py @@ -0,0 +1,148 @@ +"""PM re-entry on an awaiting_pm_review task must steer, never re-claim. + +Live incident: an awaiting_pm_review task (already past the in-path PR gate) +kept getting re-offered to its owning PM by give_me_work. The respawned PM +called i_will_plan, and CLAIM_RULES used to grant CELL_PM/MAIN_PM a claim from +AWAITING_PM_REVIEW — so the composed (claim, set_plan, start) sequence legally +reset the task to in_progress and re-ran submit_up -> pr_pass -> +awaiting_pm_review forever (one production task looped 11 cycles across 37 +spawns in 4h). ``_handle_pm_reentry`` now recognizes this status for the +owning PM and returns a steering-only OK envelope (complete / request_changes) +with no claim and no state change; CLAIM_RULES no longer permits the claim at +all, so a non-owner (or any other caller) falls through to a normal spec +rejection. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps + +# --------------------------------------------------------------------------- +# Shared fixture helpers — same pattern as test_i_will_plan_sub_tasks_gate.py +# --------------------------------------------------------------------------- + + +def _make_deps(**overrides: Any) -> ChoreographerDeps: + base: dict[str, Any] = { + "task": AsyncMock(), + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + base.update(overrides) + task = base["task"] + task.session = MagicMock() + task.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + repo = base["evidence_repo"] + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(repo, method).return_value = [] + _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) + return ChoreographerDeps(**base) + + +def _review_task_svc(task_id: object, pm_id: object, *, role: str) -> AsyncMock: + """TaskService mock for a PM re-entering its own awaiting_pm_review task.""" + task_svc = AsyncMock() + task_svc.get.return_value = MagicMock( + id=task_id, + status="awaiting_pm_review", + plan={"text": "already planned"}, + assigned_to=pm_id, + task_type="planning", + parent_task_id=None, + sequence=0, + team="backend", + commits=["abc123"], + pr_number=42, + branch_name="feature/backend/abc", + quick_context=None, + ) + task_svc.agent_for.return_value = MagicMock( + id=pm_id, role=role, team="backend", slug=None + ) + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.get_subtasks.return_value = [] + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + return task_svc + + +async def _assert_steers_without_reclaiming(role: str) -> None: + pm_id = uuid4() + task_id = uuid4() + task_svc = _review_task_svc(task_id, pm_id, role=role) + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan(pm_id, task_id, plan="resume") + body = env.as_dict() + + assert body.get("error") is None, body + assert body["status"] == "awaiting_pm_review", body + assert "complete" in body["next"], body + + task_svc.claim.assert_not_awaited() + task_svc.set_plan.assert_not_awaited() + task_svc.start.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_cell_pm_reentry_awaiting_pm_review_steers_to_complete() -> None: + await _assert_steers_without_reclaiming("cell_pm") + + +@pytest.mark.asyncio +async def test_main_pm_reentry_awaiting_pm_review_steers_to_complete() -> None: + await _assert_steers_without_reclaiming("main_pm") + + +@pytest.mark.asyncio +async def test_non_owner_awaiting_pm_review_is_rejected_not_reclaimed() -> None: + """A PM that does NOT own the review task gets the normal spec rejection — + CLAIM_RULES no longer grants a claim from awaiting_pm_review to anyone, so + this falls straight through to invalid_state instead of resetting the task. + """ + pm_id = uuid4() + other_pm_id = uuid4() + task_id = uuid4() + task_svc = _review_task_svc(task_id, other_pm_id, role="cell_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan(pm_id, task_id, plan="resume") + body = env.as_dict() + + assert body.get("error") == "invalid_state", body + task_svc.claim.assert_not_awaited() + task_svc.set_plan.assert_not_awaited() + task_svc.start.assert_not_awaited() diff --git a/tests/unit/services/test_pm_claim_needs_revision.py b/tests/unit/services/test_pm_claim_needs_revision.py index 963062e1..d95ec57a 100644 --- a/tests/unit/services/test_pm_claim_needs_revision.py +++ b/tests/unit/services/test_pm_claim_needs_revision.py @@ -22,7 +22,11 @@ from typing import TYPE_CHECKING, cast import pytest from roboco.foundation.policy import lifecycle as spec from roboco.models.base import TaskStatus -from roboco.services.task import _default_claim_statuses, _get_valid_claim_statuses +from roboco.services.task import ( + _ROLE_CLAIM_STATUSES, + _default_claim_statuses, + _get_valid_claim_statuses, +) if TYPE_CHECKING: from roboco.db.tables import AgentTable @@ -52,3 +56,26 @@ def test_runtime_pm_claim_mapping_covers_spec_claim_rules(role: spec.Role) -> No f"runtime claim mapping for {role.value} is missing spec-allowed " f"status '{status.value}'" ) + + +@pytest.mark.parametrize("role", [spec.Role.CELL_PM, spec.Role.MAIN_PM]) +def test_claim_rules_and_role_statuses_are_identical(role: spec.Role) -> None: + """Genuine bidirectional cross-check between the two claim tables. + + ``test_runtime_pm_claim_mapping_covers_spec_claim_rules`` above only + checks spec ⊆ runtime — it would still pass if ``_ROLE_CLAIM_STATUSES`` + carried an EXTRA status the spec doesn't grant (e.g. AWAITING_PM_REVIEW + re-added to the runtime table alone, with lifecycle.CLAIM_RULES left + untouched). That silent one-sided drift — "the service table already + granted this on the belief that the spec granted it too" — is exactly the + shape that caused the awaiting_pm_review re-claim loop + (``lifecycle.py``'s ``CLAIM_RULES`` comment covers the incident). This + test imports both tables directly and asserts the per-role sets are + IDENTICAL, not just one-way-covering. + """ + spec_values = {s.value for s in spec.CLAIM_RULES[role]} + runtime_values = {s.value for s in _ROLE_CLAIM_STATUSES[role.value]} + assert spec_values == runtime_values, ( + f"spec.CLAIM_RULES[{role.value}]={spec_values} != " + f"task._ROLE_CLAIM_STATUSES[{role.value!r}]={runtime_values}" + ) From e6c9dde2a9b5e14b0de11508eee650d8d5f54d5f Mon Sep 17 00:00:00 2001 From: Renn F Date: Fri, 31 Jul 2026 01:53:07 +0200 Subject: [PATCH 2/3] 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 From 07b5eecf27b3a842ef29ab0ba51b60cbd8b5e944 Mon Sep 17 00:00:00 2001 From: Renn F Date: Fri, 31 Jul 2026 02:51:44 +0200 Subject: [PATCH 3/3] =?UTF-8?q?fix(release):=20publish=20roboco-agent-kimi?= =?UTF-8?q?=20=E2=80=94=20v0.28.0=20shipped=20without=20it?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The kimi provider (#713) added docker/agent-kimi.Dockerfile and the agent-kimi-image pull stanza in docker-compose.registry.yml, but never the entry in release.yml's build/push map — the exact regression class the map's own history records for the grok sub-images. v0.28.0 published every image except kimi and its pull-smoke job went red on precisely that pull; the red run went unactioned. The image is backfilled to both registries manually for :0.28.0/:latest; this entry covers every future release. --- .github/workflows/release.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 1004a784..6d3e7820 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -122,6 +122,12 @@ jobs: # (unlike roboco-agent-grok above) it needs no special build-order # carve-out and is just another entry here. [roboco-agent-gemini]=docker/agent-gemini.Dockerfile + # Kimi (Moonshot, official kimi-code CLI) — same shape as gemini/ + # codex: one-shot delivery roles only, plain FROM agent-base. + # Missing from this map on v0.28.0 shipped the release without the + # image; pull-smoke went red on exactly that pull. Every new + # docker/agent-*.Dockerfile MUST land here in the same PR. + [roboco-agent-kimi]=docker/agent-kimi.Dockerfile ) for name in "${!IMAGES[@]}"; do echo "::group::build ${name}"