Fix: dependency spawn gate and cell ownership (#73)

* Cleanup + Missing greenlet error

* fix(messaging): persist a group's active-session pointer so posts reuse it

create_session and create_session_with_access_check set group.active_session_id
from session.id BEFORE the flush that materializes it — the id is a flush-time
uuid4 default, so the pointer was written as NULL and every post opened a fresh
session, fragmenting one conversation across many. Flush first, then link, the
same ordering the seed path already uses.

Two tests fabricated "two distinct sessions" by calling create_session twice on
one group, which only differed because of this bug; switch them to two groups so
they keep testing their real intent. Add a regression guard that the pointer is
actually persisted and a second create reuses the live session.

* fix(orchestrator): gate spawns on dependencies and keep cell tasks in their cell

The cross-task dependency check ran only on the dev dispatch path, so cell-PM,
Main-PM and board agents were spawned onto dependency-blocked tasks and flailed
unblock / escalate / notify against an unfinished upstream — climbing ownership
of cell work up to the board, which cannot drive it, and deadlocking the task.

- Move the dependency gate into the shared spawn readiness check so it covers
  every role, and auto-block the task so it leaves the pending pool until the
  upstream reaches a terminal state (then the existing auto-unblock revives it).
- Cell-ownership invariant: a backend/frontend/ux_ui task may only be worked or
  owned by its own cell. The readiness gate refuses a board or Main-PM spawn
  onto a cell task; reassign refuses and clears such an owner; and on
  dependency-clear a mis-owned cell task is re-homed to its cell's pending pool
  instead of reviving under an owner that cannot progress it.
- A dependency block is never a CEO signal: notify(target=ceo) is refused while
  the task is waiting on an unfinished upstream, with a remediate to idle and
  wait — the block clears on its own.

* Uploading images + Fixing pyproject.toml

* ++

* revert(orchestrator): drop the cell-ownership block pending a tooling audit

The cell-ownership invariant added earlier — a board / Main-PM role may never be
spawned onto or reassigned to a cell task, plus re-homing a mis-owned cell task
on dependency-clear — was too absolute. It forbids a higher role from stepping
in when something genuinely deeper is going on, and contradicts the existing
rule that main_pm may hold a task at awaiting_pm_review. The dependency spawn
gate already prevents the cascade that handed the board cell tasks; the deadlock
it guarded against will be addressed with a return-path approach after auditing
what tools the cell PMs actually need. Keeps the dependency gate and the CEO
dependency-block notify guard.

* docs(prompts): a dependency wait is wait-and-idle, not escalate

The cell-PM and Main-PM prompts told agents to escalate_up / retry unblock on a
blocked task without distinguishing a dependency wait (which auto-clears the
moment the upstream completes) from a real wedge — the source of the
escalate/unblock flail and the CEO-notification spam. Split the blocked-state
guidance: a cross-cell dependency wait = note + i_am_idle (do not escalate,
unblock, or notify the CEO); escalate only a genuinely broken upstream. Fix two
stale references to i_am_blocked, a developer-only verb the PMs do not have,
to escalate_up.

Correct the CLAUDE.md verb-surface table, which understated every role: it
listed 4 cell_pm verbs while the flow manifest derives the full set (11,
including unclaim and i_am_idle) from lifecycle.spec.intents_for_role.

* feat(gateway): cell_pm reassign verb — intra-cell developer hand-off

A cell PM can now hand a claimed/in_progress task to another developer in its
own cell without unclaim (which drops the work back to the pool and loses the
assignee). The branch is keyed to the task, so the work-in-progress is
preserved; the new dev is respawned to continue. Intra-cell only: the task must
be in the caller's cell and new_assignee must be a developer of that same cell.

Wired through every layer: the reassign IntentSpec (composes=(), cell_pm-only),
the choreographer verb + intra-cell guard, a reaper-safe
TaskService.reassign_active_claim (reseeds the claim heartbeat so the new dev
is not immediately reaped), the ReassignRequest schema, the cell_pm flow route,
and the MCP flow-server tool. Tracing-waived like unclaim (mechanical hand-off).
Regenerated lifecycle/verb artifacts; prompt + CLAUDE.md updated.

---------

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

tests/integration/test_messaging_service.py

@@ -278,16 +278,35 @@ async def test_create_session_missing_group_raises(msg_setup: dict) -> None:
 
 
 @pytest.mark.asyncio
-async def test_create_session_replaces_active(msg_setup: dict) -> None:
-    """Second create_session against same group still produces an ACTIVE session."""
+async def test_create_session_reuses_active(msg_setup: dict) -> None:
+    """A group has ONE live session: a second create reuses it, never a new one."""
     svc = msg_setup["svc"]
     ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
     grp = await svc.create_group(GroupCreateRequest(name="g1", channel_id=ch.id))
-    await svc.create_session(SessionCreateRequest(group_id=grp.id))
+    first = await svc.create_session(SessionCreateRequest(group_id=grp.id))
     second = await svc.create_session(SessionCreateRequest(group_id=grp.id))
+    assert second.id == first.id
     assert second.status == SessionStatus.ACTIVE
 
 
+@pytest.mark.asyncio
+async def test_create_session_persists_group_active_pointer(msg_setup: dict) -> None:
+    """create_session must persist group.active_session_id to the DB.
+
+    The pointer is what every post keys off to find the live session; if it is
+    left NULL (e.g. assigned before the session id is flushed), the group opens a
+    brand-new session on each post and one conversation fragments across many.
+    """
+    svc = msg_setup["svc"]
+    ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
+    grp = await svc.create_group(GroupCreateRequest(name="g1", channel_id=ch.id))
+    sess = await svc.create_session(SessionCreateRequest(group_id=grp.id))
+    # Re-read the pointer from the DB (async-safe) to prove it actually persisted,
+    # not just the in-memory object.
+    await svc.session.refresh(grp, ["active_session_id"])
+    assert grp.active_session_id == sess.id
+
+
 @pytest.mark.asyncio
 async def test_get_session_returns_none(msg_setup: dict) -> None:
     svc = msg_setup["svc"]
@@ -423,14 +442,15 @@ async def test_get_or_create_active_session_returns_active(
 async def test_get_or_create_active_session_returns_existing(
     msg_setup: dict,
 ) -> None:
-    """Lines 802-804: returns the existing active session when one is registered."""
+    """Returns the existing active session instead of opening a new one.
+
+    No manual pointer-setting: the first call must itself register
+    group.active_session_id so the second call finds and reuses it.
+    """
     svc = msg_setup["svc"]
     ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
     grp = await svc.create_group(GroupCreateRequest(name="g1", channel_id=ch.id))
-    # First call creates and registers active_session_id on the group.
     first = await svc.get_or_create_active_session(grp.id)
-    # Make sure DB sees the active_session_id set.
-    grp.active_session_id = first.id
     second = await svc.get_or_create_active_session(grp.id)
     assert second.id == first.id
 
@@ -1157,7 +1177,10 @@ async def test_validate_reply_target_wrong_session_raises(
     msg = await svc.send_message(
         MessageCreateRequest(agent_id=aid, session_id=sess1.id, content="msg")
     )
-    sess2 = await svc.create_session(SessionCreateRequest(group_id=grp.id))
+    # A genuinely different session: a group holds ONE live session, so use a
+    # second group. The reply target must be rejected as not belonging to it.
+    grp2 = await svc.create_group(GroupCreateRequest(name="g2", channel_id=ch.id))
+    sess2 = await svc.create_session(SessionCreateRequest(group_id=grp2.id))
     with pytest.raises(ValueError, match="not found in this session"):
         await svc._validate_reply_target(msg.id, sess2.id)
 
@@ -1808,9 +1831,11 @@ async def test_link_session_to_task_primary_conflict(
     aid = msg_setup["agent_id"]
     tid = msg_setup["task_id"]
     ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
+    # Two distinct live sessions: a group holds ONE live session, so use two groups.
     grp = await svc.create_group(GroupCreateRequest(name="g1", channel_id=ch.id))
+    grp2 = await svc.create_group(GroupCreateRequest(name="g2", channel_id=ch.id))
     sess1 = await svc.create_session(SessionCreateRequest(group_id=grp.id))
-    sess2 = await svc.create_session(SessionCreateRequest(group_id=grp.id))
+    sess2 = await svc.create_session(SessionCreateRequest(group_id=grp2.id))
     await svc.link_session_to_task(sess1.id, tid, aid, is_primary=True)
     with pytest.raises(ConflictError):
         await svc.link_session_to_task(sess2.id, tid, aid, is_primary=True)

tests/unit/gateway/test_notify.py

@@ -306,3 +306,75 @@ async def test_notify_priority_high_passed_through() -> None:
     assert body["error"] is None
     call_kwargs = notif_svc.send_ack_notification.call_args.kwargs
     assert call_kwargs["priority"] == "high"
+
+
+# ---------------------------------------------------------------------------
+# A dependency block is never a CEO signal — notify(target="ceo") is refused
+# while the related task is waiting on an unfinished upstream.
+# ---------------------------------------------------------------------------
+
+
+def _ca_for_notify(
+    role: str, task: object, unmet: list[object]
+) -> tuple[ContentActions, AsyncMock]:
+    task_svc = AsyncMock()
+    task_svc.agent_for.return_value = MagicMock(role=role)
+    task_svc.get.return_value = task
+    task_svc.unmet_dependency_ids.return_value = unmet
+    notif_svc = AsyncMock()
+    ca = ContentActions(_make_deps(task=task_svc, notifications=notif_svc))
+    # Ownership is exercised elsewhere; isolate the dependency-block gate.
+    object.__setattr__(
+        ca, "_verify_explicit_task_ownership", AsyncMock(return_value=None)
+    )
+    return ca, notif_svc
+
+
+@pytest.mark.asyncio
+async def test_notify_ceo_about_dependency_block_refused() -> None:
+    """PO cannot page the CEO about a task that is just waiting on an upstream."""
+    dep_id = uuid4()
+    task = MagicMock(id=uuid4(), dependency_ids=[dep_id])
+    ca, notif_svc = _ca_for_notify("product_owner", task, unmet=[dep_id])
+    env = await ca.notify(
+        agent_id=uuid4(),
+        target="ceo",
+        text="URGENT: relax the backend dependency so the cell can resume.",
+        priority="urgent",
+        task_id=task.id,
+    )
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert "dependency block" in body["message"]
+    notif_svc.send_ack_notification.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notify_ceo_about_unblocked_task_allowed() -> None:
+    """A CEO notification about a task with no open dependency still goes through."""
+    task = MagicMock(id=uuid4(), dependency_ids=[])
+    ca, notif_svc = _ca_for_notify("product_owner", task, unmet=[])
+    env = await ca.notify(
+        agent_id=uuid4(),
+        target="ceo",
+        text="Product review complete — ready for your go/no-go.",
+        task_id=task.id,
+    )
+    assert env.as_dict()["error"] is None
+    notif_svc.send_ack_notification.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_notify_noncel_target_about_blocked_task_allowed() -> None:
+    """The gate is CEO-scoped: notifying another PM about a block is unaffected."""
+    dep_id = uuid4()
+    task = MagicMock(id=uuid4(), dependency_ids=[dep_id])
+    ca, notif_svc = _ca_for_notify("main_pm", task, unmet=[dep_id])
+    env = await ca.notify(
+        agent_id=uuid4(),
+        target="be-pm",
+        text="Heads up: this task is waiting on the UX design.",
+        task_id=task.id,
+    )
+    assert env.as_dict()["error"] is None
+    notif_svc.send_ack_notification.assert_awaited_once()

tests/unit/gateway/test_reassign_verb.py

@@ -0,0 +1,125 @@
+"""The cell_pm `reassign` verb — hand a claimed/in_progress task to another
+developer in the caller's OWN cell, preserving the branch.
+
+Covers the intra-cell guard (`Choreographer._validate_reassign`, using real
+agents_config data) and the reaper-safe service write
+(`TaskService.reassign_active_claim`).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+from uuid import UUID, uuid4
+
+import pytest
+from roboco.models.base import TaskStatus
+from roboco.seeds.initial_data import AGENT_UUIDS
+from roboco.services.gateway.choreographer._impl import Choreographer
+from roboco.services.task import TaskService
+
+_BE_PM = UUID(AGENT_UUIDS["be-pm"])
+
+
+def _task(team: str = "backend", status: str = "in_progress") -> MagicMock:
+    return MagicMock(team=MagicMock(value=team), status=MagicMock(value=status))
+
+
+# ---------------------------------------------------------------------------
+# _validate_reassign — intra-cell guard
+# ---------------------------------------------------------------------------
+
+
+def test_allows_same_cell_developer() -> None:
+    assert Choreographer._validate_reassign(_task(), _BE_PM, "be-dev-2") is None
+
+
+def test_rejects_cross_cell_developer() -> None:
+    # fe-dev-1 is a frontend dev; a backend PM may not reassign to it.
+    env = Choreographer._validate_reassign(_task("backend"), _BE_PM, "fe-dev-1")
+    assert env is not None
+    assert env.error == "not_authorized"
+
+
+def test_rejects_non_developer_target() -> None:
+    # be-qa is in the cell but is not a developer.
+    env = Choreographer._validate_reassign(_task("backend"), _BE_PM, "be-qa")
+    assert env is not None
+    assert env.error == "not_authorized"
+
+
+def test_rejects_task_outside_callers_cell() -> None:
+    env = Choreographer._validate_reassign(_task("frontend"), _BE_PM, "be-dev-2")
+    assert env is not None
+    assert env.error == "not_authorized"
+
+
+def test_rejects_non_active_status() -> None:
+    env = Choreographer._validate_reassign(
+        _task("backend", "awaiting_qa"), _BE_PM, "be-dev-2"
+    )
+    assert env is not None
+    assert env.error == "invalid_state"
+
+
+def test_rejects_unknown_slug() -> None:
+    env = Choreographer._validate_reassign(_task("backend"), _BE_PM, "be-dev-99")
+    assert env is not None
+    assert env.error == "invalid_state"
+
+
+def test_allows_claimed_status() -> None:
+    assert (
+        Choreographer._validate_reassign(
+            _task("backend", "claimed"), _BE_PM, "be-dev-1"
+        )
+        is None
+    )
+
+
+# ---------------------------------------------------------------------------
+# reassign_active_claim — reaper-safe service write
+# ---------------------------------------------------------------------------
+
+
+def _build_task(**over: object) -> MagicMock:
+    base: dict[str, object] = {
+        "id": uuid4(),
+        "status": TaskStatus.IN_PROGRESS,
+        "assigned_to": None,
+        "claimed_by": None,
+        "claimed_at": None,
+        "last_heartbeat_at": None,
+        "active_claimant_id": None,
+    }
+    base.update(over)
+    return MagicMock(**base)
+
+
+def _service() -> TaskService:
+    session = MagicMock()
+    session.flush = AsyncMock()
+    return TaskService(session)
+
+
+@pytest.mark.asyncio
+async def test_reassign_active_claim_seeds_a_fresh_claim() -> None:
+    task = _build_task(status=TaskStatus.IN_PROGRESS)
+    svc = _service()
+    object.__setattr__(svc, "get", AsyncMock(return_value=task))
+    new_id = uuid4()
+    result = await svc.reassign_active_claim(task.id, new_id)
+    assert result is task
+    assert task.assigned_to == new_id
+    assert task.claimed_by == new_id
+    assert task.active_claimant_id == new_id
+    # Fresh claim window so the reaper doesn't treat the new dev as stale.
+    assert task.claimed_at is not None
+    assert task.last_heartbeat_at is not None
+
+
+@pytest.mark.asyncio
+async def test_reassign_active_claim_refuses_non_active_status() -> None:
+    task = _build_task(status=TaskStatus.AWAITING_QA)
+    svc = _service()
+    object.__setattr__(svc, "get", AsyncMock(return_value=task))
+    assert await svc.reassign_active_claim(task.id, uuid4()) is None

tests/unit/runtime/test_readiness_dependency_gate.py

@@ -0,0 +1,228 @@
+"""The spawn readiness gate refuses ANY role onto a dependency-blocked task.
+
+Previously the cross-task dependency check lived only on the dev dispatch
+path, so cell-PM, Main-PM and board agents were spawned onto tasks whose
+upstream (e.g. the UX/UI design a frontend task waits on) was still open.
+Those agents then flailed unblock / escalate / notify against an unfinished
+dependency. The gate now lives in ``_readiness_gate`` — the single pre-flight
+every ``spawn_agent`` call funnels through — so it covers every role.
+"""
+
+from __future__ import annotations
+
+from http import HTTPStatus
+from typing import Any
+
+import pytest
+from roboco.runtime import orchestrator as orchestrator_module
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+_TASK_ID = "11111111-1111-1111-1111-111111111111"
+_DEP_ID = "22222222-2222-2222-2222-222222222222"
+
+
+class _FakeResp:
+    def __init__(self, status_code: int = 200, payload: dict[str, Any] | None = None):
+        self.status_code = status_code
+        self._payload = payload or {}
+
+    @property
+    def is_success(self) -> bool:
+        return self.status_code == HTTPStatus.OK
+
+    def json(self) -> dict[str, Any]:
+        return self._payload
+
+
+class _FakeClient:
+    """Async-context HTTP stub routing by URL substring."""
+
+    def __init__(self, routes: dict[str, _FakeResp]):
+        self._routes = routes
+        self.patches: list[tuple[str, dict[str, Any] | None]] = []
+
+    async def __aenter__(self) -> _FakeClient:
+        return self
+
+    async def __aexit__(self, *_: object) -> bool:
+        return False
+
+    async def get(self, url: str) -> _FakeResp:
+        for key, resp in self._routes.items():
+            if key in url:
+                return resp
+        return _FakeResp(404, {})
+
+    async def patch(self, url: str, json: dict[str, Any] | None = None) -> _FakeResp:
+        # The dependency gate auto-blocks via PATCH; record + accept it.
+        self.patches.append((url, json))
+        return _FakeResp(200, {})
+
+
+def _orch_with_routes(
+    monkeypatch: pytest.MonkeyPatch, routes: dict[str, _FakeResp]
+) -> tuple[AgentOrchestrator, _FakeClient]:
+    orch = object.__new__(AgentOrchestrator)
+    client = _FakeClient(routes)
+    # `_api_url` is a property (reads settings); routes match by `/tasks/<id>`
+    # substring so the resolved base URL is irrelevant.
+    monkeypatch.setattr(
+        orchestrator_module.httpx,
+        "AsyncClient",
+        lambda *_a, **_k: client,
+    )
+    return orch, client
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_refused_on_nonterminal_dependency(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A cell PM must NOT be spawned while a cross-cell dependency is open."""
+    routes = {
+        f"/tasks/{_TASK_ID}": _FakeResp(
+            200,
+            {
+                "id": _TASK_ID,
+                "status": "pending",
+                "dependency_ids": [_DEP_ID],
+                "acceptance_criteria": ["x"],
+                "project_id": "r1",
+                "project_slug": "roboco",
+            },
+        ),
+        f"/tasks/{_DEP_ID}": _FakeResp(200, {"id": _DEP_ID, "status": "in_progress"}),
+    }
+    orch, _ = _orch_with_routes(monkeypatch, routes)
+    reason = await orch._readiness_gate("be-pm", _TASK_ID)
+    assert reason is not None
+    assert _DEP_ID in reason
+
+
+@pytest.mark.asyncio
+async def test_board_refused_on_nonterminal_dependency(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A board role (product-owner) is gated the same as everyone else."""
+    routes = {
+        f"/tasks/{_TASK_ID}": _FakeResp(
+            200,
+            {
+                "id": _TASK_ID,
+                "status": "pending",
+                "dependency_ids": [_DEP_ID],
+                "acceptance_criteria": ["x"],
+                "product_id": "p1",
+                "project_id": None,
+            },
+        ),
+        f"/tasks/{_DEP_ID}": _FakeResp(200, {"id": _DEP_ID, "status": "paused"}),
+    }
+    orch, _ = _orch_with_routes(monkeypatch, routes)
+    reason = await orch._readiness_gate("product-owner", _TASK_ID)
+    assert reason is not None
+    assert _DEP_ID in reason
+
+
+@pytest.mark.asyncio
+async def test_unreadable_dependency_fails_closed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A dependency we cannot read is treated as unmet — never spawn ahead."""
+    routes = {
+        f"/tasks/{_TASK_ID}": _FakeResp(
+            200,
+            {
+                "id": _TASK_ID,
+                "status": "pending",
+                "dependency_ids": [_DEP_ID],
+                "acceptance_criteria": ["x"],
+                "product_id": "p1",
+                "project_id": None,
+            },
+        ),
+        # no route for the dependency → 404 → unreadable
+    }
+    orch, _ = _orch_with_routes(monkeypatch, routes)
+    reason = await orch._readiness_gate("be-pm", _TASK_ID)
+    assert reason is not None
+
+
+@pytest.mark.asyncio
+async def test_allowed_when_dependency_terminal(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Once the dependency is completed, the dependency gate no longer refuses."""
+    routes = {
+        f"/tasks/{_TASK_ID}": _FakeResp(
+            200,
+            {
+                "id": _TASK_ID,
+                "status": "pending",
+                "dependency_ids": [_DEP_ID],
+                "acceptance_criteria": ["x"],
+                # coordination task → project/branch/git-token gates skipped, so a
+                # passing dependency gate yields a clean None.
+                "product_id": "p1",
+                "project_id": None,
+            },
+        ),
+        f"/tasks/{_DEP_ID}": _FakeResp(200, {"id": _DEP_ID, "status": "completed"}),
+    }
+    orch, _ = _orch_with_routes(monkeypatch, routes)
+    reason = await orch._readiness_gate("product-owner", _TASK_ID)
+    assert reason is None
+
+
+@pytest.mark.asyncio
+async def test_no_dependencies_passes(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A task with no dependencies is never refused by this gate."""
+    routes = {
+        f"/tasks/{_TASK_ID}": _FakeResp(
+            200,
+            {
+                "id": _TASK_ID,
+                "status": "pending",
+                "dependency_ids": [],
+                "acceptance_criteria": ["x"],
+                "product_id": "p1",
+                "project_id": None,
+            },
+        ),
+    }
+    orch, _ = _orch_with_routes(monkeypatch, routes)
+    reason = await orch._readiness_gate("main-pm", _TASK_ID)
+    assert reason is None
+
+
+@pytest.mark.asyncio
+async def test_dependency_block_auto_blocks_task(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The refusal auto-blocks the task so it leaves the pending pool.
+
+    A transient (no-block) refusal would re-raise every tick and starve sibling
+    tasks in the same dispatcher loop; auto-blocking removes it from the pool
+    until the upstream completes.
+    """
+    routes = {
+        f"/tasks/{_TASK_ID}": _FakeResp(
+            200,
+            {
+                "id": _TASK_ID,
+                "status": "pending",
+                "dependency_ids": [_DEP_ID],
+                "acceptance_criteria": ["x"],
+                "product_id": "p1",
+                "project_id": None,
+            },
+        ),
+        f"/tasks/{_DEP_ID}": _FakeResp(200, {"id": _DEP_ID, "status": "in_progress"}),
+    }
+    orch, client = _orch_with_routes(monkeypatch, routes)
+    reason = await orch._readiness_gate("be-pm", _TASK_ID)
+    assert reason is not None
+    assert any(
+        _TASK_ID in url and (body or {}).get("status") == "blocked"
+        for url, body in client.patches
+    )