[fix] MegaTask verification: migration 052 enum + async cell-map read

Two real bugs surfaced running the full gate against a containerized
Postgres (and the orchestrator boot log):

1. Migration 052 crashed a real orchestrator boot with
   'type "team" already exists'. The generic sa.Enum(create_type=False)
   does NOT set the postgres enum's create_type attribute, so op.create_table
   (checkfirst=False) emitted a redundant CREATE TYPE against the pre-existing
   team enum. Switched to postgresql.ENUM(create_type=False) — the postgres-
   native enum whose create_type _check_for_name_in_memos actually reads, so
   the CREATE TYPE is suppressed. Verified: 051->052 upgrade against a DB where
   the team enum pre-existed (the exact path that crashed) now succeeds;
   downgrade 052->051 drops the table and preserves the shared enum; fresh
   upgrade head clean. (Migration 016 has the same latent sa.Enum pattern but
   never re-runs in prod, so it's noted, not touched here.)

2. _ensure_branch_for_task read task.cell_projects (lazy=selectin to-many)
   directly, tripping MissingGreenlet on a freshly-created/unqueried task —
   which then poisoned the async session (PendingRollbackError). Replaced with
   _task_has_cell_map: peeks InstanceState.unloaded (no IO) and reads the
   already-loaded map, falling back to an awaited count query only when the
   relationship is genuinely unloaded. Non-ORM stubs route to the plain
   attribute. Fixes 2 integration tests; the 6 cell-map unit tests still pass.

Also: typed the self stub as Any in test_choreographer_subtask_project
(mypy tests/ wants Choreographer, not SimpleNamespace) — the codebase idiom.

Gate: ruff format/check clean; mypy roboco/ + tests/ clean; full pytest
10371 passed / 388 skipped against containerized pgvector:pg16; vulture clean.
Pre-existing xenon C-rank on reassign (from prior commit 19a474d3, not this
feature) still blocks make quality — surfaced separately.

alembic/versions/052_task_cell_projects.py

@@ -25,9 +25,13 @@ branch_labels = None
 depends_on = None
 
 # Reuse the existing Postgres "team" enum in place (created in 001_initial_schema,
-# widened since by later migrations); create_type=False so this migration never
+# widened since by later migrations). ``create_type=False`` MUST be set on the
-# tries to (re)create it.
+# postgres-native ``postgresql.ENUM``: it's that class's ``create_type`` attribute
-_TEAM_ENUM = sa.Enum(
+# that ``_check_for_name_in_memos`` reads to suppress the redundant ``CREATE TYPE``
+# on ``op.create_table`` (checkfirst=False, so the has_type probe is skipped). On
+# the generic ``sa.Enum`` the kwarg is silently dropped, so the CREATE TYPE fires
+# and crashes a real orchestrator boot ("type 'team' already exists").
+_TEAM_ENUM = postgresql.ENUM(
     "backend",
     "frontend",
     "ux_ui",

roboco/services/task.py

@@ -12,7 +12,9 @@ from typing import TYPE_CHECKING, Any, ClassVar, cast
 from uuid import UUID, uuid4
 
 from sqlalchemy import and_, func, or_, select, update
+from sqlalchemy import inspect as sa_inspect
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import InstanceState
 
 from roboco.db.tables import (
     AgentTable,
@@ -1555,6 +1557,37 @@ class TaskService(BaseService):
         )
         return task
 
+    async def _task_has_cell_map(self, task: TaskTable) -> bool:
+        """True iff ``task`` carries an ad-hoc per-cell project map.
+
+        The ``cell_projects`` relationship is ``lazy="selectin"`` — it loads on
+        a task *query*, not on a freshly created/flushed instance. Reading the
+        attribute on an unloaded instance fires a greenlet-less lazy SELECT that
+        rolls the async transaction back (``MissingGreenlet`` then
+        ``PendingRollbackError``), so we must NOT touch it blindly. We peek the
+        instance state (no IO): if the map is already loaded (the claim path
+        queries the task, so selectin fired) we read it directly; if it is
+        genuinely unloaded we ask the DB with an awaited count instead. A
+        non-ORM stub (unit-test MagicMock) isn't a real ``InstanceState``, so we
+        fall back to its plain ``cell_projects`` attribute.
+        """
+        # ``sa_inspect`` is typed to return ``InstanceState`` for a mapped
+        # ``TaskTable``, but unit-test stubs pass a ``MagicMock`` whose fake
+        # inspector is NOT a real ``InstanceState`` — annotate ``object`` so the
+        # non-InstanceState fallback stays reachable (and routes the stub to its
+        # plain ``cell_projects`` attribute).
+        state: object = sa_inspect(task)
+        if isinstance(state, InstanceState):
+            if "cell_projects" not in state.unloaded:
+                return bool(task.cell_projects)
+            stmt = (
+                select(func.count())
+                .select_from(TaskCellProjectTable)
+                .where(TaskCellProjectTable.task_id == task.id)
+            )
+            return (await self.session.scalar(stmt) or 0) > 0
+        return bool(task.cell_projects)
+
     async def _ensure_branch_for_task(
         self,
         task: TaskTable,
@@ -1587,7 +1620,7 @@ class TaskService(BaseService):
             # spans, so cells branch off it (not off master) and only the CEO
             # merges the root into master. Only a task with neither project,
             # product, nor a cell map is misconfigured.
-            if task.product_id or task.cell_projects:
+            if task.product_id or await self._task_has_cell_map(task):
                 return await self._ensure_coordination_root_branches(task, agent_id)
             # A MegaTask umbrella is branchless by design: it spans many projects
             # (no single master to branch off) and assembles no PR of its own —

tests/unit/services/test_choreographer_subtask_project.py

@@ -17,6 +17,7 @@ other tiers are unchanged product-root / single-project behavior.
 from __future__ import annotations
 
 from types import SimpleNamespace
+from typing import Any
 from uuid import UUID, uuid4
 
 import pytest
@@ -54,7 +55,7 @@ async def test_explicit_inputs_project_id_wins_over_cell_map() -> None:
         product_id=None,
         project_id=None,
     )
-    self_stub = SimpleNamespace(product=None)
+    self_stub: Any = SimpleNamespace(product=None)
     resolved = await Choreographer._resolve_subtask_project(
         self_stub, parent, _inputs(team=Team.BACKEND, project_id=explicit)
     )
@@ -73,7 +74,7 @@ async def test_cell_map_resolves_project_for_matching_team() -> None:
         product_id=None,
         project_id=None,
     )
-    self_stub = SimpleNamespace(product=None)
+    self_stub: Any = SimpleNamespace(product=None)
     resolved = await Choreographer._resolve_subtask_project(
         self_stub, parent, _inputs(team=Team.BACKEND)
     )
@@ -90,7 +91,7 @@ async def test_cell_map_missing_team_falls_through_to_parent_project() -> None:
         product_id=None,
         project_id=own,
     )
-    self_stub = SimpleNamespace(product=None)
+    self_stub: Any = SimpleNamespace(product=None)
     # Frontend subtask but the map only covers backend → fall to parent.project_id.
     resolved = await Choreographer._resolve_subtask_project(
         self_stub, parent, _inputs(team=Team.FRONTEND)
@@ -109,7 +110,7 @@ async def test_cell_map_only_parent_with_no_match_raises_completeness() -> None:
         product_id=None,
         project_id=None,
     )
-    self_stub = SimpleNamespace(product=None)
+    self_stub: Any = SimpleNamespace(product=None)
     with pytest.raises(TaskCompletenessError) as exc:
         await Choreographer._resolve_subtask_project(
             self_stub, parent, _inputs(team=Team.FRONTEND)