fix(run-hardening): stop three blocked-task respawn loops (#253)

* fix(run-hardening): stop three blocked-task respawn loops

Three independent fixes for blocked-task respawn loops observed in the live
run (the bleeders behind a wedged near-complete run):

- verb runner: re-check the working task after EACH composed atomic action,
  not just at entry. A concurrent transition between a verb's precondition
  gate and execution (e.g. a racing i_am_blocked moving a root from
  needs_revision to blocked) made claim() return None mid-sequence; the next
  composed step dereferenced None.id and crashed with the opaque
  "'NoneType' object has no attribute 'id'", looping the PM. Now fails fast
  with an actionable INVALID_STATE; the savepoint rolls the partial run back.

- blocker dispatch: never dispatch a Board role (product-owner / head-
  marketing) as a blocker resolver. Board roles have no unblock verb, so the
  dispatcher respawned one forever to "resolve" a blocker it could only
  notify/triage about — one incident burned ~6400 tool calls on a single
  mis-owned root. _blocker_resolver_slug now returns None for a Board
  assignee so the dispatch skips it.

- git push: recover a missing local task-branch ref from origin/<branch>
  before push-by-name. A re-provisioned shared clone can lack the branch
  locally though its commits are on origin, so push died on
  "src refspec <branch> does not match any" and the task wedged at i_am_done.
  Now materializes the ref (no-op push when already on origin) or fails loud
  with an unclaim+reclaim instruction when the work is on neither.

Adds regression tests for all three. Full no-DB gate green (ruff, reflow,
mypy, xenon); pytest+coverage validated by CI.

* fix(verb-runner): only raise on an INTERMEDIATE composed None, not the last

The mid-composition None-guard was too aggressive: it raised for a None
returned by the LAST composed action too (e.g. start()), preempting the
caller's existing `if task is None` handler that surfaces the verb-specific
message ("start failed for task ...", the board verb's decline envelope).
Three tests asserting those messages broke in CI.

Only an INTERMEDIATE None is fatal (the next action would deref None.id). A
None from the last action is the verb's own result and must flow out as the
runner's return value. Guard now fires only for position > 0, before the
next dispatch — still prevents the crash, preserves the last-action contract.

* fix(escalation): never hand a Main-PM coordination root to the Board

The upstream cause of the board catch-22 (which the orchestrator-side
blocker-dispatch guard only backstopped): the escalation chain points
main-pm -> product-owner, and i_am_blocked/escalate REASSIGNS the task to
that chain target. apply_escalation's board-advisory guard only refused
descendant cell tasks (both predicates require parent_task_id), so a
top-level Main-PM coordination root slipped through and the whole root was
reassigned to the Product Owner + marked blocked. The board has no unblock
verb, so it spam-notified the CEO and respawn-looped (~6400 tool calls on
one root).

Add _is_coordination_task (team == main_pm — covers a delivery root AND a
MegaTask root-subtask) and a shared _board_cannot_own predicate, applied at
all four board-refusal sites (escalation, reassign, reassign_active_claim,
dependency-revival). A main_pm coordination task escalated/reassigned onto a
board role is now diverted to the pool for a role-matched (Main-PM) reclaim.

Complements the blocker-dispatch backstop in the prior commits (defense in
depth). Tests: coordination-root predicate cases + apply_escalation divert;
existing teamless-root / board-root behavior unchanged.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/unit/gateway/test_verb_runner.py

@@ -72,6 +72,42 @@ async def test_runner_rejects_none_task_or_agent() -> None:
         await runner.run_intent("i_will_plan", task, None, ctx)
 
 
+@pytest.mark.asyncio
+async def test_runner_rejects_none_returned_mid_composition() -> None:
+    """A composed action returning None mid-sequence fails loud, not a crash.
+
+    Observed in prod: i_will_plan on a task a concurrent agent had just moved to
+    `blocked` — claim() returned None (no valid transition), then
+    _do_set_plan(None, ...) crashed with "'NoneType' object has no attribute
+    'id'". The choreographer surfaced it as a cryptic "verb runner failed" and
+    the PM respawn-looped. The entry guard only covers the INITIAL task, so the
+    loop body must re-check after each composed action.
+    """
+    task_svc = AsyncMock()
+    # __aexit__ must return falsy so the savepoint context does not SUPPRESS the
+    # ValueError raised inside it (real SQLAlchemy begin_nested re-raises + rolls back).
+    task_svc.session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(), __aexit__=AsyncMock(return_value=False)
+        )
+    )
+    # claim() returns None — its source status was invalid (concurrent change).
+    task_svc.claim = AsyncMock(return_value=None)
+    task_svc.set_plan = AsyncMock()
+    task_svc.start = AsyncMock()
+    runner = VerbRunner(task_service=task_svc, git_service=AsyncMock())
+
+    task = MagicMock(id=uuid4(), status="needs_revision", plan="p", commits=[])
+    agent = MagicMock(id=uuid4(), role="main_pm")
+    ctx = spec.Context(plan="my plan")
+
+    with pytest.raises(ValueError, match="INVALID_STATE"):
+        await runner.run_intent("i_will_plan", task, agent, ctx)
+    # The downstream composed actions must NOT run on a None task.
+    task_svc.set_plan.assert_not_called()
+    task_svc.start.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_runner_runs_side_effects_after_db_commit() -> None:
     """For open_pr: composes is empty; side_effects (push_branch, create_pr) run."""

tests/unit/runtime/test_blocker_and_claimed_dispatch.py

@@ -49,14 +49,30 @@ def test_blocked_task_assigned_to_main_pm_dispatches_main_pm() -> None:
     assert orch._blocker_resolver_slug(task) == "main-pm"
 
 
-def test_blocked_task_assigned_to_board_dispatches_board() -> None:
+def test_blocked_task_assigned_to_board_is_not_dispatched() -> None:
+    # A board/advisory role (product-owner / head-marketing) has NO unblock
+    # verb — dispatching it to resolve a blocker is a futile catch-22 (it can
+    # only notify/triage, so it spam-notifies the CEO and respawns forever).
+    # The resolver must be None so the blocker dispatch SKIPS it; the task is
+    # mis-owned and must be re-routed / surfaced to the CEO out-of-band.
     orch = _orch()
     task: dict[str, Any] = {
         "id": "t1",
         "team": "backend",
         "assigned_to": AGENT_UUIDS["product-owner"],
     }
-    assert orch._blocker_resolver_slug(task) == "product-owner"
+    assert orch._blocker_resolver_slug(task) is None
+
+
+def test_blocked_task_assigned_to_head_marketing_is_not_dispatched() -> None:
+    # Same catch-22 guard for the other board role.
+    orch = _orch()
+    task: dict[str, Any] = {
+        "id": "t1",
+        "team": "backend",
+        "assigned_to": AGENT_UUIDS["head-marketing"],
+    }
+    assert orch._blocker_resolver_slug(task) is None
 
 
 def test_blocked_task_held_by_dev_falls_back_to_cell_pm() -> None:

tests/unit/services/test_escalation_board_guard.py

@@ -20,7 +20,9 @@ import pytest
 from roboco.models.base import AgentRole, TaskStatus, TaskType, Team
 from roboco.services.task import (
     TaskService,
+    _board_cannot_own,
     _is_cell_team_task,
+    _is_coordination_task,
     _is_descendant_executable_task,
 )
 
@@ -116,11 +118,83 @@ def test_documentation_task_type_as_raw_string_is_flagged() -> None:
     assert _is_descendant_executable_task(task) is True
 
 
+# ---------------------------------------------------------------------------
+# _is_coordination_task (pure) — Main-PM coordination roots / root-subtasks
+# ---------------------------------------------------------------------------
+
+
+def test_main_pm_root_is_coordination_task() -> None:
+    # A top-level delivery coordination root (no parent, main_pm team). The two
+    # descendant predicates miss it (they require parent_task_id); this catches it.
+    task = MagicMock(parent_task_id=None, team=Team.MAIN_PM)
+    assert _is_coordination_task(task) is True
+    assert _board_cannot_own(task) is True
+
+
+def test_main_pm_root_subtask_is_coordination_task() -> None:
+    # A MegaTask root-subtask is parented under the umbrella but still main_pm.
+    task = MagicMock(parent_task_id=uuid4(), team=Team.MAIN_PM)
+    assert _is_coordination_task(task) is True
+
+
+def test_main_pm_team_as_raw_string_is_coordination_task() -> None:
+    task = MagicMock(parent_task_id=None, team="main_pm")
+    assert _is_coordination_task(task) is True
+
+
+def test_board_root_is_not_coordination_task() -> None:
+    # A board/product root (e.g. a product root the PO reviews) is board-ownable.
+    task = MagicMock(parent_task_id=None, team=Team.BOARD)
+    assert _is_coordination_task(task) is False
+    assert _board_cannot_own(task) is False
+
+
+def test_cell_root_is_not_coordination_task() -> None:
+    task = MagicMock(parent_task_id=None, team=Team.FRONTEND)
+    assert _is_coordination_task(task) is False
+
+
 # ---------------------------------------------------------------------------
 # apply_escalation board-role divert
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.asyncio
+async def test_apply_escalation_diverts_main_pm_coordination_root_from_board() -> None:
+    # The confirmed catch-22: a Main PM's i_am_blocked on its own coordination
+    # ROOT escalated up the chain to product-owner (a board role). The root is
+    # neither a descendant nor a cell task, so the old guard missed it and the
+    # whole root was reassigned to the board, which respawn-looped on a blocker it
+    # could not unblock. It must now divert to the pool instead.
+    svc = _service()
+    target_id = uuid4()
+    task = MagicMock(
+        id=uuid4(),
+        parent_task_id=None,
+        team=Team.MAIN_PM,
+        task_type=TaskType.CODE,
+        assigned_to=uuid4(),
+        blocker_raised_by=None,
+        status=TaskStatus.IN_PROGRESS,
+    )
+    _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=True))
+    release_mock = AsyncMock()
+    _bind(svc, "_release_code_task_to_pool", release_mock)
+
+    await svc.apply_escalation(
+        task=task,
+        target_agent_id=target_id,
+        escalator_slug="main-pm",
+        target_slug="product-owner",
+        reason="root blocked: branch behind master",
+    )
+
+    # Diverted — NOT blocked-and-reassigned onto the board.
+    release_mock.assert_awaited_once()
+    assert task.status == TaskStatus.IN_PROGRESS
+    assert task.assigned_to != target_id
+
+
 @pytest.mark.asyncio
 async def test_apply_escalation_diverts_descendant_code_to_board() -> None:
     svc = _service()

tests/unit/services/test_git.py

@@ -186,6 +186,67 @@ async def test_push_targets_explicit_branch_not_current_checkout() -> None:
     assert "feature/frontend/OTHER" not in push_args
 
 
+@pytest.mark.asyncio
+async def test_push_recovers_missing_local_branch_from_origin() -> None:
+    """A push-by-name on a re-provisioned/shared clone missing the local ref
+    recovers it from origin instead of dying on "src refspec ... does not
+    match any".
+
+    The branch's commits are already on origin (pushed in a prior cycle/clone),
+    so after recreating the local tracking ref the push is a clean no-op.
+    """
+    svc = _service()
+    _bind(svc, "_token_for_workspace", AsyncMock(return_value=None))
+    calls: list[list[str]] = []
+
+    async def _run_git(_ws: object, args: list[str], **_kw: object) -> MagicMock:
+        calls.append(args)
+        res = MagicMock()
+        # Local ref MISSING; origin HAS it.
+        if args[:2] == ["rev-parse", "--verify"]:
+            is_local = any(a.startswith("refs/heads/") for a in args)
+            res.returncode = 1 if is_local else 0
+            res.stdout = ""
+            return res
+        res.returncode = 0
+        res.stdout = "0" if args[:2] == ["rev-list", "--count"] else ""
+        return res
+
+    _bind(svc, "_run_git", AsyncMock(side_effect=_run_git))
+
+    branch, _pushed = await svc.push(Path("/tmp/ws"), branch="feature/backend/TASK")
+
+    assert branch == "feature/backend/TASK"
+    # It fetched origin and recreated the local ref before pushing.
+    assert ["fetch", "origin", "feature/backend/TASK"] in calls
+    assert ["branch", "feature/backend/TASK", "origin/feature/backend/TASK"] in calls
+    assert any(a and a[0] == "push" for a in calls)
+
+
+@pytest.mark.asyncio
+async def test_push_fails_loud_when_branch_absent_local_and_origin() -> None:
+    """When the named branch is in neither the local clone nor origin, the work
+    is genuinely lost from this clone — fail with a recoverable instruction, not
+    the raw "src refspec does not match any"."""
+    svc = _service()
+    _bind(svc, "_token_for_workspace", AsyncMock(return_value=None))
+
+    async def _run_git(_ws: object, args: list[str], **_kw: object) -> MagicMock:
+        res = MagicMock()
+        if args[:2] == ["rev-parse", "--verify"]:
+            res.returncode = 1  # absent both locally and on origin
+            res.stdout = ""
+            return res
+        res.returncode = 0
+        res.stdout = ""
+        return res
+
+    _bind(svc, "_run_git", AsyncMock(side_effect=_run_git))
+
+    with pytest.raises(GitCommandError, match="unclaim the task and"):
+        await svc.push(Path("/tmp/ws"), branch="feature/backend/GONE")
+
+
 @pytest.mark.asyncio
 async def test_pr_head_is_task_branch_not_current() -> None:
     """The PR head is the task's recorded branch, not the workspace checkout."""