Hotfixes (#293)

* fix(mcp): delegate tool carries the collision surface the B1a gate demands

TASK_AT_DELEGATE (5fc85419) requires intends_to_touch on code delegations,
but the MCP delegate tool never gained the parameter — PMs were rejected
with incomplete_input and could never comply (live fleet-wide delegation
wall, 2026-07-02). Adds intends_to_touch / adds_migration / touches_shared /
depends_on to the tool and forwards them; parity test locks the invariant.

* fix(git): assembly-integrity guard accepts squash-merged children

git cherry patch-matches each child commit individually, so a squash merge
(N patches -> one commit, new patch-id) read as 'work missing' and the #11
guard refused every legitimate submit_up (live 2026-07-02: S6 cell, three
squash-merged children at the branch tip). A parent commit carrying the
child's [taskid8] prefix now proves the child landed; children with no
marker stay flagged — the original incident the guard exists for.

* fix(git): diff head prefers origin when the local ref is behind it

Assembled branches advance on ORIGIN as child PRs squash-merge on GitHub,
but _resolve_head_ref preferred the inspecting clone's parked local ref —
the PR-gate reviewer's evidence diff was built from a pre-merge snapshot
and re-flagged work that had already landed (two false pr_fail verdicts
on the S6 cell PR, live 2026-07-02). When both refs exist and the local
ref is strictly behind origin, resolve to origin/<branch>; local-ahead
(unpushed) and diverged refs keep priority, single-ref cases unchanged.

* test(mcp): plan-gate fields must be tool parameters (parity class lock)

Extends the delegate parity test to every choreographer plan-depth gate:
a gate that can reject with missing=[field] must name only fields the
corresponding MCP tool can send, else the agent can never comply.

* perf(api): wire TaskSummaryResponse into a bounded /tasks/summary route

The panel fetched /api/tasks unbounded and full-fat — 2MB per refresh
measured live (2026-07-02), ~21KB/task, and the trimmed
TaskSummaryResponse was dead code. /tasks/summary returns exactly the
fields list views render (~50x lighter); the status-only branch of
/tasks now honors its limit, and the eleven unbounded task list routes
are capped.

* perf(panel): kill the per-page request flood and fat payloads

Every page load funneled ~85 default-prefetch RSC requests + 665KB of
images + the 2MB task list through the browser's six HTTP/1.1
connections — real data calls queued ~2s before being sent (measured
via Playwright resource timing, 2026-07-02).

- prefetch={false} on all 59 Links (sidebar, task rows, kanban cards,
  list rows) — ~85 requests/refresh down to a handful
- icon/apple-icon/logo resized to render size: 665KB -> 54KB; unused
  219KB PNG removed
- task list fetches the trimmed /tasks/summary (2MB -> ~100KB),
  normalized into the Task shape so list consumers keep their types
- ReactQueryDevtools rendered only in development

* fix(api): Annotated limit defaults so direct-call tests get real ints

Query(...) positional defaults arrive as Query objects when a route
function is invoked outside the HTTP layer (integration tests call
handlers directly) and broke the new [:limit] slices.

* fix(api,panel): summary carries completed_at + board_review_complete

The metrics page computes velocity client-side from completed_at and the
CEO approval queue gates on board_review_complete — both were nulled by
the summary normalizer, so Completed Today/Week read 0 against 63 real
completions and approved-board tasks could vanish from the queue. The
queue also renders quick_context, so it fetches the full list (small,
status-scoped) via tasksApi.listFull instead of the summary.

* fix(runtime): spawn manifest workspace_path follows the task's project

_build_manifest_for_agent hardcoded the roboco project workspace for
every agent; a guard-core task's manifest claimed /data/workspaces/roboco
while the container cwd sat in the task worktree. The manifest now takes
the same _resolve_workspace_cwd the container -w uses — one resolver,
both surfaces agree by construction.

* fix(runtime): respawn breaker catches status ping-pong loops

Any status CHANGE fully reset the strike counter, so a blocked <->
in_progress oscillation — which changes status on every spawn while
advancing nothing — never tripped the gate (live 2026-07-02: 8 spawns
over two hours). A status never seen on the (agent, task) still fully
resets; a REVISITED status gets a bounded reset budget mirroring
tracing_resets, after which strikes accrue and the gate fires.

* fix(runtime): unassigned-QA dispatch spawns without pre-claiming

The transitioning pre-claim moved awaiting_qa -> claimed before the QA
agent existed; the spawned agent's claim_review/pass_review both demand
awaiting_qa, so it bounced twice and unclaimed (live 2026-07-02,
ba7b751c). Matches _spawn_assigned_qa and the external-PR reviewer
dispatch: no pre-claim, the agent claims itself via claim_review.

* fix(tests): narrow await_args before kwargs access (mypy union-attr)

* Minor upgrades

* fix(policy): team-match gate gains org-wide exemption; resume/unblock/activate now team-matched

needs_team_match sat in its permissive fallback since shipping (no
caller supplied Context.agent_team) and three PM verbs opted out
entirely — a misrouted frontend cell PM blocked, escalated, and held a
backend task through exactly that gap (live 2026-07-02). Org-wide roles
(main_pm, board, CEO, PR reviewer) are exempt so escalation handling
and root-PR gating keep working; cell-scoped roles are now enforced
wherever the caller supplies the team.

---------

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

tests/foundation/test_team_match_context.py

@@ -0,0 +1,104 @@
+"""Team-match must actually fire: the spec gate rejects cross-team actors.
+
+Live 2026-07-02: a frontend cell PM was dispatched onto a BACKEND task's
+review, then blocked it, escalated it, and briefly held it — a 40-minute
+ownership tug-of-war. Seventeen ActionSpecs carry needs_team_match=True and
+_check_team_match enforces it — but only when the caller supplies
+Context.agent_team, which no choreographer site did, so the gate sat in its
+permissive fallback forever. These tests pin the policy behavior the
+choreographer sweep wires up.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any, cast
+from uuid import uuid4
+
+from roboco.foundation.policy.lifecycle import (
+    Context,
+    Role,
+    can_invoke_intent,
+)
+from roboco.models.base import TaskStatus
+
+
+def _task(**overrides: Any) -> Any:
+    base: dict[str, Any] = {
+        "id": uuid4(),
+        "status": TaskStatus.IN_PROGRESS,
+        "team": "backend",
+        "assigned_to": None,
+        "task_type": "code",
+    }
+    base.update(overrides)
+    return cast("Any", SimpleNamespace(**base))
+
+
+def test_cross_team_developer_is_rejected_when_team_supplied() -> None:
+    decision = can_invoke_intent(
+        Role.DEVELOPER,
+        "i_am_blocked",
+        _task(team="backend", status=TaskStatus.IN_PROGRESS, task_type="code"),
+        Context(actor_id=uuid4(), agent_team="frontend"),
+    )
+    assert not decision.allowed
+    assert "team" in (decision.message or "").lower()
+
+
+def test_cross_team_cell_pm_resume_is_rejected() -> None:
+    decision = can_invoke_intent(
+        Role.CELL_PM,
+        "resume",
+        _task(team="backend", status=TaskStatus.PAUSED),
+        Context(actor_id=uuid4(), agent_team="frontend"),
+    )
+    assert not decision.allowed
+    assert "team" in (decision.message or "").lower()
+
+
+def test_same_team_developer_is_allowed() -> None:
+    decision = can_invoke_intent(
+        Role.DEVELOPER,
+        "i_am_blocked",
+        _task(team="backend", status=TaskStatus.IN_PROGRESS, task_type="code"),
+        Context(actor_id=uuid4(), agent_team="backend"),
+    )
+    assert decision.allowed
+
+
+def test_missing_team_keeps_permissive_fallback() -> None:
+    """Absent agent_team defers to the service layer (backward compatible)."""
+    decision = can_invoke_intent(
+        Role.DEVELOPER,
+        "i_am_blocked",
+        _task(team="backend", status=TaskStatus.IN_PROGRESS, task_type="code"),
+        Context(actor_id=uuid4()),
+    )
+    assert decision.allowed
+
+
+def test_org_wide_roles_are_exempt_cross_team() -> None:
+    """Main PM handles every cell's escalations; the exemption keeps that."""
+    for role, verb, status in (
+        (Role.MAIN_PM, "resume", TaskStatus.PAUSED),
+        (Role.MAIN_PM, "unblock", TaskStatus.BLOCKED),
+    ):
+        decision = can_invoke_intent(
+            role,
+            verb,
+            _task(team="backend", status=status),
+            Context(actor_id=uuid4(), agent_team="main_pm"),
+        )
+        assert decision.allowed, f"{role} {verb} must stay org-wide"
+
+
+def test_cross_team_cell_pm_unblock_is_rejected() -> None:
+    decision = can_invoke_intent(
+        Role.CELL_PM,
+        "unblock",
+        _task(team="backend", status=TaskStatus.BLOCKED),
+        Context(actor_id=uuid4(), agent_team="frontend"),
+    )
+    assert not decision.allowed
+    assert "team" in (decision.message or "").lower()

tests/unit/api/test_task_summary_response.py

@@ -0,0 +1,144 @@
+"""Task list summary mode — trimmed payloads for panel list views.
+
+The panel fetched /api/tasks unbounded and full-fat (2MB measured live,
+2026-07-02): every list row shipped description, plan, progress_updates,
+commits, notes. TaskSummaryResponse existed but was dead code. These tests
+pin the wired-up summary path: the converter carries exactly the fields
+list views render (tree, kanban card, git badge), excludes the fat columns,
+and the /summary route is registered before /{task_id} so it can't be
+swallowed by the UUID path match.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any, cast
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.api.routes import tasks as routes_mod
+from roboco.api.routes.tasks import router
+from roboco.api.schemas.tasks import (
+    _SUMMARY_SNIPPET_LEN,
+    task_list_to_summary_response,
+    task_to_summary_response,
+)
+from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
+_LIMIT = 2
+
+
+def _stub_task(**overrides: Any) -> TaskTable:
+    base: dict[str, Any] = {
+        "id": uuid4(),
+        "title": "t",
+        "description": "d" * (_SUMMARY_SNIPPET_LEN * 2 + 100),
+        "status": TaskStatus.PENDING,
+        "priority": 3,
+        "sequence": 1,
+        "nature": TaskNature.TECHNICAL,
+        "task_type": TaskType.CODE,
+        "team": Team.BACKEND,
+        "assigned_to": uuid4(),
+        "parent_task_id": uuid4(),
+        "batch_id": None,
+        "project_id": uuid4(),
+        "product_id": None,
+        "branch_name": "feature/backend/x",
+        "pr_number": 42,
+        "pr_url": "https://github.com/x/y/pull/42",
+        "pr_created": True,
+        "docs_complete": False,
+        "created_at": datetime.now(UTC),
+        "updated_at": datetime.now(UTC),
+        "completed_at": datetime.now(UTC),
+        "board_review_complete": True,
+        "estimated_complexity": Complexity.MEDIUM,
+    }
+    base.update(overrides)
+    return cast("TaskTable", SimpleNamespace(**base))
+
+
+def test_summary_carries_every_list_view_field() -> None:
+    t = _stub_task()
+    s = task_to_summary_response(t)
+    assert (s.id, s.title, s.status) == (t.id, "t", TaskStatus.PENDING)
+    assert s.parent_task_id == t.parent_task_id  # tree build
+    assert s.sequence == 1 and s.task_type is TaskType.CODE  # kanban card
+    assert (s.pr_number, s.pr_created, s.docs_complete) == (
+        42,
+        True,
+        False,
+    )  # git badge
+    assert s.branch_name == "feature/backend/x"
+    assert s.project_id == t.project_id and s.product_id is None
+    # velocity metrics filter on completion time; the CEO approval queue
+    # gates on board_review_complete — both burned as gaps on 2026-07-02
+    assert s.completed_at == t.completed_at
+    assert s.board_review_complete is True
+
+
+def test_summary_excludes_fat_fields_and_truncates_snippet() -> None:
+    s = task_to_summary_response(_stub_task())
+    dump = s.model_dump()
+    for fat in (
+        "description",
+        "plan",
+        "progress_updates",
+        "commits",
+        "quick_context",
+        "checkpoints",
+        "notes_structured",
+        "dev_notes",
+        "acceptance_criteria",
+    ):
+        assert fat not in dump, f"summary must not carry {fat}"
+    assert len(s.description_snippet or "") == _SUMMARY_SNIPPET_LEN
+
+
+def test_summary_snippet_none_safe() -> None:
+    assert (
+        task_to_summary_response(_stub_task(description=None)).description_snippet
+        is None
+    )
+    assert (
+        task_to_summary_response(_stub_task(description="")).description_snippet is None
+    )
+
+
+def test_summary_list_converter() -> None:
+    stubs = [_stub_task() for _ in range(_LIMIT)]
+    assert len(task_list_to_summary_response(stubs)) == len(stubs)
+
+
+def test_summary_route_registered_before_task_id_route() -> None:
+    """/tasks/summary must not be swallowed by /tasks/{task_id} UUID parsing."""
+    paths = [getattr(r, "path", "") for r in router.routes]
+    assert "/summary" in paths
+    assert paths.index("/summary") < paths.index("/{task_id}")
+
+
+@pytest.mark.asyncio
+async def test_summary_route_status_branch_respects_limit() -> None:
+    service = AsyncMock()
+    service.list_by_status.return_value = [_stub_task() for _ in range(_LIMIT * 3)]
+    permissions = MagicMock()
+    permissions.can_perform_task_action.return_value = True
+    agent = MagicMock(team=Team.BACKEND)
+    with (
+        patch.object(routes_mod, "get_task_service", return_value=service),
+        patch.object(routes_mod, "get_permission_service", return_value=permissions),
+    ):
+        out = await routes_mod.list_tasks_summary(
+            db=MagicMock(),
+            agent=agent,
+            team=None,
+            status=TaskStatus.PENDING,
+            limit=_LIMIT,
+        )
+    assert len(out) == _LIMIT

tests/unit/mcp_servers/test_flow_server_delegate_surface_parity.py

@@ -0,0 +1,147 @@
+"""The delegate MCP tool must be able to send every field the gate demands.
+
+Original bug (2026-07-02 live): TASK_AT_DELEGATE required ``intends_to_touch``
+on code delegations, but the MCP ``delegate`` tool had no such parameter —
+PMs were rejected 4x with ``incomplete_input``, could never comply, and
+blocked/escalated. Fleet-wide code-delegation wall.
+
+Invariant: every FieldRequirement in TASK_AT_DELEGATE (and TASK_AT_CREATE,
+which it extends) is either a parameter of the MCP delegate tool or
+server-resolved (never demanded from the caller).
+"""
+
+from __future__ import annotations
+
+import importlib
+import inspect
+import json
+from typing import TYPE_CHECKING, Any
+from unittest.mock import MagicMock
+
+import pytest
+from roboco.foundation.policy.task_completeness import TASK_AT_DELEGATE
+
+if TYPE_CHECKING:
+    import types
+    from pathlib import Path
+
+# Fields the choreographer resolves server-side; the tool never sends them.
+_SERVER_RESOLVED = {"project_id"}
+
+
+def _pm_manifest() -> dict[str, object]:
+    return {
+        "agent_id": "00000000-0000-0000-0000-000000000098",
+        "role": "cell_pm",
+        "team": "frontend",
+        "workspace_path": "/tmp/test",
+        "flow_tools": ["delegate", "i_am_idle"],
+        "do_tools": [],
+        "read_tools": [],
+        "write_tools": [],
+        "bash_allowed": True,
+        "subagent_allowed": False,
+        "subagent_model": None,
+        "env": {},
+    }
+
+
+@pytest.fixture()
+def flow_module_pm(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> types.ModuleType:
+    manifest_path = tmp_path / "tool-manifest.json"
+    manifest_path.write_text(json.dumps(_pm_manifest()))
+    monkeypatch.setenv("ROBOCO_AGENT_ID", "00000000-0000-0000-0000-000000000098")
+    monkeypatch.setenv("ROBOCO_AGENT_ROLE", "cell_pm")
+    monkeypatch.setenv("ROBOCO_ORCHESTRATOR_URL", "http://test-orchestrator:8000")
+    monkeypatch.setenv("ROBOCO_SDK_URL", "http://test-sdk:9000")
+    monkeypatch.setenv("ROBOCO_TOOL_MANIFEST_PATH", str(manifest_path))
+
+    import roboco.mcp.flow_server as srv
+
+    importlib.reload(srv)
+    return srv
+
+
+def test_delegate_tool_covers_every_gate_required_field(
+    flow_module_pm: types.ModuleType,
+) -> None:
+    """Every TASK_AT_DELEGATE FieldRequirement is a delegate() parameter."""
+    params = set(inspect.signature(flow_module_pm.delegate).parameters)
+    required = {req.field for req in TASK_AT_DELEGATE.requires}
+    missing = required - params - _SERVER_RESOLVED
+    assert not missing, (
+        f"TASK_AT_DELEGATE demands fields the MCP delegate tool cannot send: "
+        f"{sorted(missing)}. A PM rejected with incomplete_input for these "
+        f"can NEVER comply — add them to flow_server.delegate and forward "
+        f"them in the payload."
+    )
+
+
+# Choreographer plan-depth gates hard-reject with `missing=[...]` naming these
+# fields (_pm_sub_tasks_gate for i_will_plan; the dev rich-plan gate for
+# i_will_work_on). The named tool must be able to send every one of them, or
+# the rejected agent can never comply — the delegate/intends_to_touch wall.
+_PLAN_GATE_FIELDS: dict[str, set[str]] = {
+    "i_will_plan": {"plan", "approach", "sub_tasks"},
+    "i_will_work_on": {"plan", "steps", "technical_considerations", "risks"},
+}
+
+
+def test_plan_gate_fields_are_tool_parameters(
+    flow_module_pm: types.ModuleType,
+) -> None:
+    """Every field a plan gate can demand exists on the corresponding tool."""
+    for verb, required in _PLAN_GATE_FIELDS.items():
+        params = set(inspect.signature(getattr(flow_module_pm, verb)).parameters)
+        missing = required - params
+        assert not missing, (
+            f"{verb} gate demands fields the MCP tool cannot send: "
+            f"{sorted(missing)} — same class as the delegate wall."
+        )
+
+
+def test_delegate_forwards_collision_surface_in_payload(
+    flow_module_pm: types.ModuleType,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The surface fields actually reach the POST body (not just the signature)."""
+    captured: dict[str, Any] = {}
+
+    def _client_factory(*_a: object, **_kw: object) -> MagicMock:
+        client = MagicMock()
+        client.__enter__ = MagicMock(return_value=client)
+        client.__exit__ = MagicMock(return_value=False)
+
+        def _post(url: str, **kwargs: object) -> MagicMock:
+            captured["url"] = url
+            captured["json"] = kwargs.get("json")
+            resp = MagicMock()
+            resp.status_code = 200
+            resp.json.return_value = {"status": "ok"}
+            return resp
+
+        client.post = _post
+        return client
+
+    monkeypatch.setattr(flow_module_pm.httpx, "Client", _client_factory)
+    flow_module_pm.delegate(
+        parent_task_id="00000000-0000-0000-0000-000000000001",
+        title="t",
+        description="a description well over twenty chars",
+        assigned_to="fe-dev-1",
+        team="frontend",
+        task_type="code",
+        nature="technical",
+        acceptance_criteria=["done"],
+        intends_to_touch=["frontend/src/components/behavioral-content.tsx"],
+        adds_migration=False,
+        touches_shared=True,
+        depends_on=["00000000-0000-0000-0000-000000000002"],
+    )
+    body = captured["json"]
+    assert body["intends_to_touch"] == [
+        "frontend/src/components/behavioral-content.tsx"
+    ]
+    assert body["adds_migration"] is False
+    assert body["touches_shared"] is True
+    assert body["depends_on"] == ["00000000-0000-0000-0000-000000000002"]

tests/unit/runtime/test_dispatch_qa_no_preclaim.py

@@ -0,0 +1,53 @@
+"""QA dispatch must not pre-claim the review task.
+
+Live 2026-07-02 (ba7b751c): the unassigned-QA branch claimed the task
+BEFORE spawning (awaiting_qa -> claimed via the transitioning claim), then
+spawned a QA agent whose own verbs demand awaiting_qa — claim_review bounced
+("cannot claim from 'claimed'"), pass_review bounced, and the agent gave up
+and unclaimed. The assigned-QA branch and the external-PR reviewer dispatch
+both already spawn WITHOUT pre-claiming (the agent claims itself via
+claim_review); the unassigned branch must match.
+"""
+
+from __future__ import annotations
+
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+def _orch() -> tuple[AgentOrchestrator, AsyncMock, AsyncMock]:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._pm_respawn_tracker = {}
+    orch._bg_tasks = set()
+    any_orch = cast("Any", orch)
+    any_orch._is_task_handled_this_tick = lambda _tid: False
+    any_orch._select_agent_for_cell = lambda _team, _role: "be-qa"
+    any_orch._is_agent_active = lambda _slug: False
+    any_orch._pm_respawn_should_gate = AsyncMock(return_value=False)
+    any_orch._build_qa_prompt = lambda _t: "review it"
+    any_orch._task_git_context = lambda _t: None
+    claim = AsyncMock(return_value=True)
+    spawn = AsyncMock()
+    any_orch._claim_task_for_agent = claim
+    any_orch.spawn_agent = spawn
+    return orch, claim, spawn
+
+
+@pytest.mark.asyncio
+async def test_unassigned_qa_dispatch_spawns_without_preclaim() -> None:
+    orch, claim, spawn = _orch()
+    task = {"id": str(uuid4()), "team": "backend", "assigned_to": None}
+    cast("Any", orch)._fetch_tasks = AsyncMock(return_value=[task])
+
+    await orch._dispatch_qa_work(MagicMock())
+
+    claim.assert_not_awaited()
+    spawn.assert_awaited_once()
+    spawn_call = spawn.await_args
+    assert spawn_call is not None
+    assert spawn_call.kwargs["task_id"] == task["id"]
+    assert spawn_call.kwargs["agent_id"] == "be-qa"

tests/unit/runtime/test_orchestrator_manifest.py

@@ -221,3 +221,40 @@ class TestBuildManifestForAgent:
         assert result is not None
         assert nested.exists()
         assert result.exists()
+
+
+class TestManifestWorkspacePath:
+    """workspace_path must be the TASK-resolved workspace, not the roboco default.
+
+    Live 2026-07-02: be-dev-2's manifest said /data/workspaces/roboco/... while
+    its task lived in guard-core-saas-backend — an agent trusting the manifest
+    hunts for its files in the wrong repository.
+    """
+
+    def test_workspace_override_reaches_manifest(self, tmp_path: Path) -> None:
+        worktree = (
+            "/data/workspaces/guard-core-saas-backend/backend/be-dev-1"
+            "/.worktrees/abc12345"
+        )
+        with patch("roboco.runtime.orchestrator.settings") as mock_settings:
+            mock_settings.manifest_host_dir = str(tmp_path)
+            mock_settings.workspaces_root = str(tmp_path / "workspaces")
+
+            result = _build_manifest_for_agent(
+                "be-dev-1", "claude-sonnet-5", workspace_path=worktree
+            )
+
+        assert result is not None
+        data = json.loads(result.read_text())
+        assert data["workspace_path"] == worktree
+
+    def test_no_override_keeps_roboco_default(self, tmp_path: Path) -> None:
+        with patch("roboco.runtime.orchestrator.settings") as mock_settings:
+            mock_settings.manifest_host_dir = str(tmp_path)
+            mock_settings.workspaces_root = str(tmp_path / "workspaces")
+
+            result = _build_manifest_for_agent("be-dev-1", "claude-sonnet-5")
+
+        assert result is not None
+        data = json.loads(result.read_text())
+        assert data["workspace_path"].endswith("workspaces/roboco/backend/be-dev-1")

tests/unit/runtime/test_respawn_gate_oscillation.py

@@ -0,0 +1,94 @@
+"""The respawn breaker must not be fooled by status ping-pong.
+
+Live 2026-07-02: a dev looped blocked -> in_progress -> blocked for two hours
+(8 spawns, 30 gateway rejections) and the breaker never tripped — every
+status CHANGE fully reset the strike counter, and an A<->B oscillation
+changes status on every spawn. A revisited status now gets a bounded reset
+budget (mirroring tracing_resets); genuinely new statuses keep the full
+reset so forward progress is never punished.
+"""
+
+from __future__ import annotations
+
+from typing import Any, cast
+from unittest.mock import AsyncMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+def _new_orchestrator() -> AgentOrchestrator:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._pm_respawn_tracker = {}
+    orch._bg_tasks = set()
+    cast("Any", orch)._schedule_respawn_persist = lambda *_a, **_k: None
+    return orch
+
+
+def _quiet_audit() -> AsyncMock:
+    audit = AsyncMock()
+    audit.has_recent_tracing_gap = AsyncMock(return_value=False)
+    return audit
+
+
+@pytest.mark.asyncio
+async def test_status_ping_pong_eventually_trips_the_gate() -> None:
+    """blocked <-> in_progress oscillation accrues strikes past the budget."""
+    orch = _new_orchestrator()
+    task_id = str(uuid4())
+    statuses = ["blocked", "in_progress"] * 6
+    results = []
+    with (
+        patch("roboco.services.audit.get_audit_service", return_value=_quiet_audit()),
+        patch(
+            "roboco.services.notification.NotificationService",
+            return_value=AsyncMock(),
+        ),
+    ):
+        for status in statuses:
+            results.append(
+                await orch._pm_respawn_should_gate(
+                    "be-dev-1", {"id": task_id, "status": status}
+                )
+            )
+    assert any(results), (
+        "an A<->B status oscillation never accumulated strikes — the exact "
+        "2026-07-02 two-hour loop the breaker exists to stop"
+    )
+
+
+@pytest.mark.asyncio
+async def test_forward_progress_through_new_statuses_never_gates() -> None:
+    orch = _new_orchestrator()
+    task_id = str(uuid4())
+    lifecycle = ["pending", "claimed", "in_progress", "verifying", "awaiting_qa"]
+    with (
+        patch("roboco.services.audit.get_audit_service", return_value=_quiet_audit()),
+        patch(
+            "roboco.services.notification.NotificationService",
+            return_value=AsyncMock(),
+        ),
+    ):
+        for status in lifecycle:
+            assert not await orch._pm_respawn_should_gate(
+                "be-dev-1", {"id": task_id, "status": status}
+            ), f"forward progress into {status} must not gate"
+
+
+@pytest.mark.asyncio
+async def test_single_revisit_within_budget_does_not_gate() -> None:
+    """A legitimate revision cycle (one revisit) stays under the budget."""
+    orch = _new_orchestrator()
+    task_id = str(uuid4())
+    with (
+        patch("roboco.services.audit.get_audit_service", return_value=_quiet_audit()),
+        patch(
+            "roboco.services.notification.NotificationService",
+            return_value=AsyncMock(),
+        ),
+    ):
+        for status in ["in_progress", "awaiting_qa", "in_progress", "awaiting_qa"]:
+            assert not await orch._pm_respawn_should_gate(
+                "be-dev-1", {"id": task_id, "status": status}
+            ), "one revision round-trip must not trip the breaker"

tests/unit/services/test_git_cherry_squash_merge_relief.py

@@ -0,0 +1,95 @@
+"""_cherry_unmerged_entry must not flag squash-merged children as missing.
+
+Live false positive (2026-07-02): three children of the S6 cell task were
+squash-merged (PRs #176/#185/#190) — their commits sat at the assembled
+branch tip, yet ``git cherry`` reported every individual child commit as
+unmerged (a squash rewrites N patches into one patch-id) and the assembly
+integrity guard refused every legitimate submit_up.
+
+Relief: every commit — including the squash commit — carries the
+``[taskid8]`` prefix, so a marker-bearing commit on the parent proves the
+child landed. A child with no marker on the parent stays flagged (the
+original incident #11 the guard exists for).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.git import GitService
+
+
+def _svc_with_git_responses(
+    responses: dict[str, SimpleNamespace],
+) -> tuple[GitService, list[list[str]]]:
+    """GitService with _run_git stubbed by subcommand name; records calls."""
+    svc = GitService.__new__(GitService)
+    calls: list[list[str]] = []
+
+    async def _run_git(
+        _workspace: Path, args: list[str], **_kw: Any
+    ) -> SimpleNamespace:
+        calls.append(args)
+        return responses[args[0]]
+
+    svc_any: Any = svc
+    svc_any._run_git = _run_git
+    return svc, calls
+
+
+def _child() -> MagicMock:
+    return MagicMock(
+        id=uuid4(), branch_name="feature/frontend/root--cell--child", title="t"
+    )
+
+
+@pytest.mark.asyncio
+async def test_squash_merged_child_with_task_marker_is_not_flagged() -> None:
+    """cherry says unmerged, but the [taskid8] squash commit is on the parent."""
+    svc, calls = _svc_with_git_responses(
+        {
+            "rev-parse": SimpleNamespace(returncode=0, stdout="abc\n"),
+            "cherry": SimpleNamespace(returncode=0, stdout="+ aaa\n+ bbb\n"),
+            "log": SimpleNamespace(
+                returncode=0, stdout="4771bd71 [deadbeef] title (#190)\n"
+            ),
+        }
+    )
+    entry = await svc._cherry_unmerged_entry(Path("/tmp"), "parent", _child())
+    assert entry is None
+    log_call = next(c for c in calls if c[0] == "log")
+    assert any("\\[" in arg for arg in log_call)  # grep pattern escapes the bracket
+
+
+@pytest.mark.asyncio
+async def test_genuinely_missing_child_stays_flagged() -> None:
+    """No marker commit on the parent → the original #11 catch still fires."""
+    child = _child()
+    svc, _calls = _svc_with_git_responses(
+        {
+            "rev-parse": SimpleNamespace(returncode=0, stdout="abc\n"),
+            "cherry": SimpleNamespace(returncode=0, stdout="+ aaa\n"),
+            "log": SimpleNamespace(returncode=0, stdout=""),
+        }
+    )
+    entry = await svc._cherry_unmerged_entry(Path("/tmp"), "parent", child)
+    assert entry == {"task_id": str(child.id)[:8], "title": "t", "unmerged": 1}
+
+
+@pytest.mark.asyncio
+async def test_cherry_clean_short_circuits_without_marker_probe() -> None:
+    """No + lines from cherry → merged; the log probe is never run."""
+    svc, calls = _svc_with_git_responses(
+        {
+            "rev-parse": SimpleNamespace(returncode=0, stdout="abc\n"),
+            "cherry": SimpleNamespace(returncode=0, stdout="- aaa\n"),
+        }
+    )
+    entry = await svc._cherry_unmerged_entry(Path("/tmp"), "parent", _child())
+    assert entry is None
+    assert not any(c[0] == "log" for c in calls)

tests/unit/services/test_git_resolve_head_ref_stale_local.py

@@ -0,0 +1,73 @@
+"""_resolve_head_ref must not diff off a stale local ref.
+
+Live incident (2026-07-02): the S6 cell branch advanced on ORIGIN as child
+PRs squash-merged on GitHub, but the assignee clone's local ref stayed
+parked pre-merge. ``diff()`` preferred the local ref, so the PR-gate
+reviewer's evidence diff re-flagged work that had already landed — two
+false ``pr_fail`` verdicts on a clean PR.
+
+Rule: when both refs exist and the local ref is STRICTLY BEHIND origin,
+use ``origin/<branch>``; a local ref that is ahead (unpushed commits) or
+diverged keeps priority, and single-ref cases are unchanged.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from roboco.services.git import GitService
+
+_BRANCH = "feature/frontend/root--cell"
+_ORIGIN = f"origin/{_BRANCH}"
+
+
+def _svc(*, refs: set[str], ancestor_rc: int) -> tuple[GitService, list[list[str]]]:
+    svc = GitService.__new__(GitService)
+    calls: list[list[str]] = []
+
+    async def _run_git(
+        _workspace: Path, args: list[str], **_kw: Any
+    ) -> SimpleNamespace:
+        calls.append(args)
+        if args[0] == "merge-base":
+            return SimpleNamespace(returncode=ancestor_rc, stdout="")
+        return SimpleNamespace(returncode=0, stdout="")
+
+    async def _ref_exists(_workspace: Path, ref: str) -> bool:
+        return ref in refs
+
+    svc_any: Any = svc
+    svc_any._run_git = _run_git
+    svc_any._ref_exists = _ref_exists
+    return svc, calls
+
+
+@pytest.mark.asyncio
+async def test_local_behind_origin_resolves_to_origin() -> None:
+    svc, calls = _svc(refs={_BRANCH, _ORIGIN}, ancestor_rc=0)
+    ref = await svc._resolve_head_ref(Path("/tmp"), _BRANCH)
+    assert ref == _ORIGIN
+    ancestor = next(c for c in calls if c[0] == "merge-base")
+    assert ancestor == ["merge-base", "--is-ancestor", _BRANCH, _ORIGIN]
+
+
+@pytest.mark.asyncio
+async def test_local_ahead_or_diverged_keeps_local() -> None:
+    svc, _calls = _svc(refs={_BRANCH, _ORIGIN}, ancestor_rc=1)
+    assert await svc._resolve_head_ref(Path("/tmp"), _BRANCH) == _BRANCH
+
+
+@pytest.mark.asyncio
+async def test_only_local_ref_unchanged() -> None:
+    svc, calls = _svc(refs={_BRANCH}, ancestor_rc=1)
+    assert await svc._resolve_head_ref(Path("/tmp"), _BRANCH) == _BRANCH
+    assert not any(c[0] == "merge-base" for c in calls)
+
+
+@pytest.mark.asyncio
+async def test_only_origin_ref_unchanged() -> None:
+    svc, _calls = _svc(refs={_ORIGIN}, ancestor_rc=1)
+    assert await svc._resolve_head_ref(Path("/tmp"), _BRANCH) == _ORIGIN