From c03e76c433c5260c4488346847eb4c404b9e071e Mon Sep 17 00:00:00 2001 From: Renn F Date: Fri, 26 Jun 2026 23:28:43 +0200 Subject: [PATCH] feat(megatask): per-cell project map root-subtasks (multi-project, multi-cell) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A MegaTask root-subtask can now target an ad-hoc per-cell project map — a third targeting shape that mirrors the existing product fan-out root. In RoboCo a project is per-cell (ProjectTable.assigned_cell); a monorepo is N per-cell projects sharing one git_url. So 'multi-cell' IS 'multi-project', and a task may mix per-cell projects across products or include OSS-library projects not in any product. Storage: migration 052 adds task_cell_projects (mirrors product_projects; unique per (task, team)). TaskTable gains a cascade-delete cell_projects relationship; TaskCreateRequest / TaskCreate / Task response carry the map. Policy: batch.is_branchless_coordination + is_valid_batch_shape gain a has_cell_projects param — a root-subtask targets exactly one of project / product / cell-map; the umbrella still targets none. TaskService passes has_cell_projects at every predicate call site and persists the rows in create(). _ensure_branch_for_task cuts feature/main_pm/{root} per distinct project in the map (via _distinct_projects_for_task); _require_target_or_umbrella and _validate_batch_membership accept the map shape. Fan-out: every distinct_project_ids site (task.py branch creation, routes _project_for_complete + _resolve_project_for_merge, orchestrator _ambient_projects_for_task, pr_review._project_slug_for, git._project_for_task) generalizes to first-distinct-project-of-map-or-product. Choreographer _resolve_subtask_project resolves a delegated subtask's cell from the parent's cell map. The product-scoped _slugs_for_product intake helper is unchanged. Intake: prompter._draft_cell_map extracts the per-cell map from the_work[]. _validate_batch_scope counts distinct projects across all drafts' cells (>=2 min stays; one 2-cell draft satisfies it). create_task_from_draft persists cell_projects for >=2-cell drafts (project_id/product_id None), collapses a 1-cell map to the single-project shape, and leaves single-cell top-level project_id drafts unchanged. _resolve_owning_team routes a multi-cell map to Main PM (coordination root, like a product root — a cell PM can't delegate cross-cell). propose_draft/propose_batch tool descriptions declare the per-cell project_id (both Claude SDK + grok runtimes). The umbrella stays branchless / pure-coordination / submit_root-rejected; the CEO-escalation pr_number gate is not widened (the map root is is_umbrella=False, mirroring a product root, so submit_root supplies it). Single-cell root-subtasks and everything below them are byte-for-byte unchanged. Un-run MegaTask waves (multi-cell drafts) become runnable. --- alembic/versions/052_task_cell_projects.py | 68 +++++++ roboco/agent_sdk/intake_driver.py | 16 +- roboco/api/routes/tasks.py | 39 +++- roboco/api/schemas/tasks.py | 23 +++ roboco/db/tables.py | 56 +++++ roboco/foundation/policy/batch.py | 28 ++- roboco/mcp/intake_server.py | 20 +- roboco/models/task.py | 36 +++- roboco/runtime/orchestrator.py | 59 ++++-- .../services/gateway/choreographer/_impl.py | 20 +- .../gateway/choreographer/pr_review.py | 30 ++- roboco/services/git.py | 27 ++- roboco/services/prompter.py | 139 +++++++++++-- roboco/services/task.py | 114 ++++++++--- tests/unit/api/test_schemas_tasks.py | 34 ++++ tests/unit/foundation/policy/test_batch.py | 76 ++++++- .../test_choreographer_subtask_project.py | 117 +++++++++++ tests/unit/services/test_prompter.py | 192 ++++++++++++++++++ tests/unit/services/test_task.py | 62 +++++- 19 files changed, 1022 insertions(+), 134 deletions(-) create mode 100644 alembic/versions/052_task_cell_projects.py create mode 100644 tests/unit/services/test_choreographer_subtask_project.py diff --git a/alembic/versions/052_task_cell_projects.py b/alembic/versions/052_task_cell_projects.py new file mode 100644 index 00000000..fc97a855 --- /dev/null +++ b/alembic/versions/052_task_cell_projects.py @@ -0,0 +1,68 @@ +"""Add the task_cell_projects table — ad-hoc per-cell project map for a task. + +A MegaTask root-subtask that spans multiple cells (and may mix per-cell projects +from different products / OSS libs) needs a per-cell routing map without standing +up a Product for it. ``task_cell_projects`` mirrors ``product_projects`` but is +owned by the task: one Project per cell per task (``UNIQUE (task_id, team)``). The +root-subtask then cuts ``feature/main_pm/{root}`` per repo and opens a root->master +PR per repo exactly like a Product fan-out root — only the map's source differs. +``team`` reuses the existing Postgres "team" enum (create_type=False). + +Revision ID: 052_task_cell_projects +Revises: 051_respawn_tracker +Create Date: 2026-06-26 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision = "052_task_cell_projects" +down_revision = "051_respawn_tracker" +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 +# tries to (re)create it. +_TEAM_ENUM = sa.Enum( + "backend", + "frontend", + "ux_ui", + "board", + "main_pm", + "fullstack", + "marketing", + "system", + name="team", + create_type=False, +) + + +def upgrade() -> None: + op.create_table( + "task_cell_projects", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column( + "task_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("tasks.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + sa.Column("team", _TEAM_ENUM, nullable=False), + sa.Column( + "project_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("projects.id", ondelete="RESTRICT"), + nullable=False, + index=True, + ), + sa.UniqueConstraint("task_id", "team", name="uq_task_cell_projects_task_team"), + ) + + +def downgrade() -> None: + op.drop_table("task_cell_projects") diff --git a/roboco/agent_sdk/intake_driver.py b/roboco/agent_sdk/intake_driver.py index c8e855e1..24c78892 100644 --- a/roboco/agent_sdk/intake_driver.py +++ b/roboco/agent_sdk/intake_driver.py @@ -391,9 +391,11 @@ def build_intake_options( "propose_draft", "Submit the finished task draft for the human to review and confirm. Call " "this once the spec is complete. Pass a JSON object: title, objective, " - "what_this_builds[], the_work[] ({team, summary, items}), notes[], " - "acceptance_criteria[], team, scale, task_type, nature, " - "estimated_complexity, priority.", + "what_this_builds[], the_work[] ({team, summary, items, project_id}), " + "notes[], acceptance_criteria[], team, scale, task_type, nature, " + "estimated_complexity, priority. A multi-cell task (be+fe, fe+uxui) puts " + "one entry per cell in the_work, each with its cell's project_id (the " + "per-cell repo); the system builds the cell->project map from them.", {"draft": dict}, ) async def _propose_draft(_args: dict[str, Any]) -> dict[str, Any]: @@ -411,9 +413,11 @@ def build_intake_options( "and confirm together. Use this (instead of propose_draft) when the CEO " "asked for multiple tasks across the scoped repos. Pass {drafts: [draft, " "...], title: '...'} where each draft has the same fields as propose_draft " - "PLUS its own project_id (which repo it targets) and collision surface " - "(intends_to_touch[], adds_migration, touches_shared) so the system can " - "sequence them into conflict-free waves.", + "PLUS a collision surface (intends_to_touch[], adds_migration, " + "touches_shared) so the system can sequence them into conflict-free waves. " + "Each draft targets its repos via the per-cell project_id on its the_work[] " + "entries (a multi-cell task has one project_id per cell; a single-cell task " + "may use one the_work entry or a top-level project_id).", {"drafts": list, "title": str}, ) async def _propose_batch(_args: dict[str, Any]) -> dict[str, Any]: diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index 7a2f2070..1b8511ff 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -185,6 +185,24 @@ async def _resolve_assigned_to_slug( return data.model_copy(update={"assigned_to": str(agent_row.id)}) +def _first_cell_map_project_id(task: Any) -> UUID | None: + """First distinct project_id from a task's ad-hoc per-cell map. + + Mirrors the product-root ``distinct_project_ids(...)[0]`` first-project + resolution: dedupes by project_id (a monorepo mapped across cells shares + one project), ordered by cell team for determinism. Returns None when the + task carries no cell map. + """ + cell_map = getattr(task, "cell_projects", None) or [] + seen: set[UUID] = set() + for mapping in sorted(cell_map, key=lambda m: m.team.value): + pid = UUID(str(mapping.project_id)) + if pid not in seen: + seen.add(pid) + return pid + return None + + async def _project_for_complete(task: Any, db: AsyncSession) -> Any: """Resolve the project for complete_task's pre-merge step. @@ -203,6 +221,9 @@ async def _project_for_complete(task: Any, db: AsyncSession) -> Any: pids = await product_service.distinct_project_ids(UUID(str(task.product_id))) if pids: return await project_service.get(pids[0]) + cell_pid = _first_cell_map_project_id(task) + if cell_pid is not None: + return await project_service.get(cell_pid) return None @@ -279,13 +300,17 @@ async def _resolve_project_for_merge(task: Any, db: AsyncSession) -> Any: ) resolved_id = project_ids[0] else: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=( - "NO_PROJECT: Task has neither project_id nor product_id; " - "cannot resolve workspace for merge. Set project_id on the task first." - ), - ) + cell_pid = _first_cell_map_project_id(task) + if cell_pid is None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=( + "NO_PROJECT: Task has neither project_id, product_id, nor a " + "cell->project map; cannot resolve workspace for merge. Set a " + "target on the task first." + ), + ) + resolved_id = cell_pid project = await project_service.get(resolved_id) if not project: raise HTTPException( diff --git a/roboco/api/schemas/tasks.py b/roboco/api/schemas/tasks.py index 25bfc1ce..88335df2 100644 --- a/roboco/api/schemas/tasks.py +++ b/roboco/api/schemas/tasks.py @@ -15,6 +15,7 @@ from sqlalchemy import select from roboco.db.tables import ProjectTable, TaskTable, WorkSessionTable from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team +from roboco.models.product import ProductCellMapping from roboco.models.session import SessionScope from roboco.utils.converters import require_uuid, to_python_uuid, to_python_uuid_list @@ -296,6 +297,9 @@ class TaskResponse(BaseModel): project_id: UUID | None = None # Repo this task targets (None for fan-out) project_slug: str | None = None # Project slug for MCP/git tool calls product_id: UUID | None = None + # Ad-hoc per-cell project map (a multi-cell MegaTask root-subtask); empty for + # a project/product-targeted task. + cell_projects: list[ProductCellMapping] = [] # Parallel Execution Tracking (for AWAITING_DOCUMENTATION phase) docs_complete: bool = False # Documenter has finished @@ -666,6 +670,24 @@ def convert_commits(commits_data: list | None) -> list[CommitRefResponse]: ] +def convert_cell_projects(task: "TaskTable") -> list[ProductCellMapping]: + """The task's ad-hoc per-cell project map, or ``[]``. + + Skips the access when the ``cell_projects`` relationship is unloaded (a + freshly-created task not yet re-queried) to avoid a sync lazy-load + ``MissingGreenlet`` — mirrors the ``project_slug`` guard. selectin loading + means a normally-queried task already carries the map. + """ + if "cell_projects" in sa_inspect(task).unloaded: + return [] + return [ + ProductCellMapping( + team=mapping.team, project_id=require_uuid(mapping.project_id) + ) + for mapping in task.cell_projects + ] + + def task_to_response(task: "TaskTable") -> TaskResponse: """Convert TaskTable to TaskResponse with proper UUID conversion.""" return TaskResponse( @@ -698,6 +720,7 @@ def task_to_response(task: "TaskTable") -> TaskResponse: dependency_ids=to_python_uuid_list(task.dependency_ids), blocker_ids=to_python_uuid_list(task.blocker_ids), batch_id=to_python_uuid(task.batch_id), + cell_projects=convert_cell_projects(task), created_at=task.created_at, updated_at=task.updated_at, claimed_at=task.claimed_at, diff --git a/roboco/db/tables.py b/roboco/db/tables.py index 6e7ea6c6..50422080 100644 --- a/roboco/db/tables.py +++ b/roboco/db/tables.py @@ -444,6 +444,18 @@ class TaskTable(Base): cascade="all, delete-orphan", passive_deletes=True, ) + # Per-cell project map for an ad-hoc (non-Product) coordination root — + # mirrors product_projects but is owned by the task itself, so a MegaTask + # root-subtask can target a per-cell map mixing projects from different + # products / OSS libs. selectin so the fan-out resolvers see the map on + # any task fetch; passive_deletes trusts the ON DELETE CASCADE FK. + cell_projects: Mapped[list["TaskCellProjectTable"]] = relationship( + "TaskCellProjectTable", + back_populates="task", + cascade="all, delete-orphan", + passive_deletes=True, + lazy="selectin", + ) __table_args__ = ( # Composite indexes for common queries @@ -617,6 +629,50 @@ class ProductProjectTable(Base): ) +class TaskCellProjectTable(Base): + """One Project per cell for an ad-hoc (non-Product) coordination root. + + Mirrors ``product_projects`` but the map is owned by the task, not a Product: + a MegaTask root-subtask that spans multiple cells (and may mix per-cell + projects from different products / OSS libs) carries its per-cell routing + here instead of a ``product_id``. The root itself does git per repo (it cuts + ``feature/main_pm/{root}`` and opens a root->master PR per repo exactly like a + Product fan-out root); the cell children each resolve their project from this + map via ``_resolve_subtask_project``. ``UNIQUE (task_id, team)`` enforces one + project per cell per task. + """ + + __tablename__ = "task_cell_projects" + + id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=uuid4 + ) + task_id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + team: Mapped[Team] = mapped_column(_str_enum(Team), nullable=False) + project_id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("projects.id", ondelete="RESTRICT"), + nullable=False, + index=True, + ) + + task: Mapped["TaskTable"] = relationship( + "TaskTable", back_populates="cell_projects" + ) + project: Mapped["ProjectTable"] = relationship( + "ProjectTable", foreign_keys=[project_id], lazy="joined" + ) + + __table_args__ = ( + UniqueConstraint("task_id", "team", name="uq_task_cell_projects_task_team"), + ) + + class PitchTable(Base): """A Board proposal the CEO approves to auto-provision a product. diff --git a/roboco/foundation/policy/batch.py b/roboco/foundation/policy/batch.py index 1a567925..c898bba2 100644 --- a/roboco/foundation/policy/batch.py +++ b/roboco/foundation/policy/batch.py @@ -41,20 +41,28 @@ def is_branchless_coordination( product_id: object | None, batch_id: object | None = None, parent_task_id: object | None = None, + has_cell_projects: bool = False, ) -> bool: """True for a task that does no git of its own (no branch, no PR). - Two shapes qualify: a product fan-out coordination root (no ``project_id``, - carries a ``product_id``), and a MegaTask umbrella (``batch_id`` set, - top-level). Both are Main-PM coordination points whose children do the git. + Three shapes qualify: a product fan-out coordination root (no ``project_id``, + carries a ``product_id``), an ad-hoc per-cell map coordination root (no + ``project_id``, no ``product_id``, carries a ``task_cell_projects`` map), and a + MegaTask umbrella (``batch_id`` set, top-level). All three are Main-PM + coordination points whose children do the git — the map/product root still + cuts ``feature/main_pm/{root}`` per repo and opens a root->master PR per repo, + but the *claim branch gate* skips the single-branch requirement for it exactly + as it does for a product root. Relies on the creation-time invariant ``is_valid_batch_shape`` that a - ``batch_id``-bearing top-level task carries no project/product — so a real + ``batch_id``-bearing top-level task carries no project/product/map — so a real umbrella is genuinely branchless and a normal task cannot spoof the exemption by attaching a ``batch_id``. """ if project_id is None and product_id is not None: return True + if project_id is None and product_id is None and has_cell_projects: + return True return is_batch_umbrella(batch_id=batch_id, parent_task_id=parent_task_id) @@ -64,16 +72,17 @@ def is_valid_batch_shape( parent_task_id: object | None, project_id: object | None, product_id: object | None, + has_cell_projects: bool = False, ) -> bool: """Guardrail: a ``batch_id`` is only valid on a well-formed MegaTask member. A ``batch_id`` is permitted on exactly two shapes: - an **umbrella** (no ``parent_task_id``) — which must target NEITHER a - project nor a product (it is branchless, grouping root-subtasks that each - carry their own repo); + project, a product, nor carry a cell map (it is branchless, grouping + root-subtasks that each carry their own repo); - a **root-subtask** (has a ``parent_task_id``) — which must target exactly - one of project / product (it does its own git). + one of project / product / ad-hoc cell map (it does its own git). A task without a ``batch_id`` is unconstrained here (the normal targeting rule applies). Denying every other ``batch_id`` shape stops a normal task @@ -83,7 +92,8 @@ def is_valid_batch_shape( """ if batch_id is None: return True + targets = bool(project_id) + bool(product_id) + bool(has_cell_projects) if parent_task_id is None: # umbrella - return project_id is None and product_id is None + return targets == 0 # root-subtask: exactly one target - return (project_id is None) != (product_id is None) + return targets == 1 diff --git a/roboco/mcp/intake_server.py b/roboco/mcp/intake_server.py index 6961904d..868f0acb 100644 --- a/roboco/mcp/intake_server.py +++ b/roboco/mcp/intake_server.py @@ -96,9 +96,11 @@ async def propose_draft(draft: dict[str, Any]) -> str: """Submit the finished task draft for the human to review and confirm. Call this once the spec is complete. Pass a JSON object: title, objective, - what_this_builds[], the_work[] ({team, summary, items}), notes[], + what_this_builds[], the_work[] ({team, summary, items, project_id}), notes[], acceptance_criteria[], team, scale, task_type, nature, estimated_complexity, - priority. + priority. A multi-cell task (be+fe, fe+uxui) puts one entry per cell in + the_work, each with its cell's project_id (the per-cell repo); the system + builds the cell->project map from them. Sequenced batch intake (when enabled): if the CEO asks for several tasks at once, propose one draft per item and set each item's collision surface so the @@ -127,12 +129,14 @@ async def propose_batch(drafts: list[dict[str, Any]], title: str = "") -> str: Use this instead of ``propose_draft`` when the CEO asked for multiple tasks across the scoped repos. ``drafts`` is a list where each item has the same - fields as a single draft PLUS its own ``project_id`` (which repo it targets) - and collision surface — ``intends_to_touch[]`` (files/dirs it will modify), - ``adds_migration`` (adds a DB migration?), ``touches_shared`` (edits a widely - shared component?). The system sequences them into conflict-free waves; - over-declaring a surface is safer than under-declaring. ``title`` names the - MegaTask. + fields as a single draft PLUS a collision surface — ``intends_to_touch[]`` + (files/dirs it will modify), ``adds_migration`` (adds a DB migration?), + ``touches_shared`` (edits a widely shared component?). The system sequences + them into conflict-free waves; over-declaring a surface is safer than + under-declaring. ``title`` names the MegaTask. Each draft targets its repos + via the per-cell ``project_id`` on its ``the_work[]`` entries (a multi-cell + task has one project_id per cell; a single-cell task may use one the_work + entry or a top-level ``project_id``). """ session_id = os.environ.get("ROBOCO_PROMPTER_SESSION_ID", "") if not session_id: diff --git a/roboco/models/task.py b/roboco/models/task.py index bf73ec00..e25c7d55 100644 --- a/roboco/models/task.py +++ b/roboco/models/task.py @@ -21,6 +21,7 @@ from roboco.models.base import ( Team, TimestampMixin, ) +from roboco.models.product import ProductCellMapping # ============================================================================= # SUPPORTING MODELS @@ -167,6 +168,13 @@ class Task(TimestampMixin): default=None, description="Product this task belongs to (additive; drives subtask routing)", ) + cell_projects: list[ProductCellMapping] = Field( + default_factory=list, + description=( + "Ad-hoc per-cell project map (a MegaTask root-subtask spanning " + "multiple cells); empty for a project/product-targeted task" + ), + ) branch_name: str | None = Field( default=None, description="Branch created for this task" ) @@ -373,21 +381,31 @@ class TaskCreate(RobocoBase): task_type: TaskType = Field(...) nature: TaskNature = Field(...) # A task targets a single repo (project_id) OR fans out across cells via a - # product (product_id, a cell->project map). Exactly one is needed; a - # board/coordination task uses product_id and has no project of its own. + # product (product_id, a cell->project map) OR carries an ad-hoc per-cell + # map (cell_projects — a MegaTask root-subtask spanning multiple cells). + # Exactly one is needed; a board/coordination task uses product_id and has + # no project of its own. project_id: UUID | None = None product_id: UUID | None = None + cell_projects: list[ProductCellMapping] = Field(default_factory=list) # Prompter origin tracking source: str = Field(default="manual") confirmed_by_human: bool = Field(default=False) @model_validator(mode="after") - def _project_or_product(self) -> "TaskCreate": - if self.project_id is None and self.product_id is None: + def _exactly_one_target(self) -> "TaskCreate": + targets = ( + self.project_id is not None, + self.product_id is not None, + bool(self.cell_projects), + ) + if sum(targets) != 1: raise ValueError( - "a task needs either a project_id (the repo it targets) or a " - "product_id (a cell->project map for a fan-out task)" + "a task needs exactly one target: a project_id (the repo it " + "targets), a product_id (a cell->project map for a fan-out " + "task), or cell_projects (an ad-hoc per-cell map for a " + "multi-cell coordination root)" ) return self @@ -456,6 +474,12 @@ class TaskCreateRequest: # cells' subtasks resolve their own project from it. project_id: UUID | None = None product_id: UUID | None = None + # Ad-hoc per-cell project map (mirrors a Product's cells but owned by the + # task): a MegaTask root-subtask spanning multiple cells (and possibly mixing + # per-cell projects from different products / OSS libs) sets this instead of + # project_id/product_id. The root cuts feature/main_pm/{root} per repo and + # opens a root->master PR per repo exactly like a Product fan-out root. + cell_projects: list[ProductCellMapping] = field(default_factory=list) # Ordering and dependencies sequence: int = 0 # Order within siblings (lower = first) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 5e0fde88..44035b3f 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -315,19 +315,23 @@ def _read_project_slug(task: dict[str, Any]) -> str | None: def _is_coordination_task(task: dict[str, Any]) -> bool: """True for a task that does no git of its own. - Two shapes qualify: a board/fan-out coordination root (carries a product, no - repo — its cell subtasks resolve a real project from the product's - cell->project map), and a MegaTask umbrella (carries a batch_id, top-level — - its root-subtasks each carry their own branch/PR). Such a task has no + Three shapes qualify: a board/fan-out coordination root (carries a product, + no repo — its cell subtasks resolve a real project from the product's + cell->project map), an ad-hoc per-cell map coordination root (carries a + ``cell_projects`` map but no project/product — a multi-cell MegaTask + root-subtask), and a MegaTask umbrella (carries a batch_id, top-level — its + root-subtasks each carry their own branch/PR). Such a task has no project_slug, branch_name, or git token, and must NOT be git-gated at the spawn-readiness or stuck-detection checks the way a code task is. A task with - none of project / product / batch is genuinely unroutable and stays gated. + none of project / product / cell-map / batch is genuinely unroutable and + stays gated. """ return is_branchless_coordination( project_id=task.get("project_id"), product_id=task.get("product_id"), batch_id=task.get("batch_id"), parent_task_id=task.get("parent_task_id"), + has_cell_projects=bool(task.get("cell_projects")), ) @@ -2640,11 +2644,14 @@ class AgentOrchestrator: task_id: str | None, product_id: str | None, ) -> list[Any]: - """The in-scope projects for the ambient block (single repo or product).""" - if product_id is None and task_id is not None: - product_id = await self._ambient_product_for_task(db, task_id) + """The in-scope projects for the ambient block (single repo, product, or + ad-hoc cell map).""" if product_id is not None: return await self._ambient_product_projects(db, product_id) + if task_id is not None: + projects = await self._ambient_projects_for_task(db, task_id) + if projects: + return projects if project_slug: from roboco.services.project import get_project_service @@ -2653,15 +2660,35 @@ class AgentOrchestrator: return [] @staticmethod - async def _ambient_product_for_task(db: Any, task_id: str) -> str | None: + async def _ambient_projects_for_task(db: Any, task_id: str) -> list[Any]: + """The in-scope projects for a task's ambient block, from its product OR + its ad-hoc ``cell_projects`` map. Empty for a plain project task (the + project_slug branch handles those) or a not-yet-mapped coordination root. + """ from uuid import UUID + from roboco.services.project import get_project_service from roboco.services.task import get_task_service task = await get_task_service(db).get(UUID(task_id)) - if task is not None and task.product_id is not None: - return str(task.product_id) - return None + if task is None: + return [] + project_service = get_project_service(db) + if task.product_id is not None: + from roboco.services.product import get_product_service + + ids = await get_product_service(db).distinct_project_ids( + UUID(str(task.product_id)) + ) + resolved = [await project_service.get(pid) for pid in ids] + return [p for p in resolved if p is not None] + # Ad-hoc per-cell map: resolve the distinct projects the map spans (de-dupe + # by project_id — a monorepo mapped across cells yields one project). + distinct_ids: dict[Any, None] = {} + for mapping in sorted(task.cell_projects, key=lambda m: m.team.value): + distinct_ids.setdefault(UUID(str(mapping.project_id)), None) + resolved = [await project_service.get(pid) for pid in distinct_ids] + return [p for p in resolved if p is not None] @staticmethod async def _ambient_product_projects(db: Any, product_id: str) -> list[Any]: @@ -6808,13 +6835,13 @@ Start by: return ( f"Task {task_id} has inadequate description ({len(description)} chars)" ) - # A coordination task carries a product instead of a repo; only a task - # with neither is genuinely unroutable. + # A coordination task carries a product or an ad-hoc cell map instead of a + # repo; only a task with neither is genuinely unroutable. if not task.get("project_id") and not _is_coordination_task(task): await self._auto_block_task( - client, task_id, "Task needs a project_id or product_id" + client, task_id, "Task needs a project_id, product_id, or cell map" ) - return f"Task {task_id} needs a project or product" + return f"Task {task_id} needs a project, product, or cell map" return None async def _check_dependencies_terminal( diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index af612712..83b04f06 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -4709,13 +4709,21 @@ class Choreographer: ) -> UUID: """Resolve the project a delegated subtask lands in. - Priority: explicit inputs.project_id -> the parent's Product map for - this cell -> the parent's own project. Raises TaskCompletenessError only - for a fan-out parent (product, no own project) whose product has no - mapping for this cell — i.e. the subtask would have no repo to land in. + Priority: explicit inputs.project_id -> the parent's ad-hoc cell_projects + map for this cell -> the parent's Product map for this cell -> the + parent's own project. Raises TaskCompletenessError only for a fan-out + parent (product or cell map, no own project) whose map has no mapping for + this cell — i.e. the subtask would have no repo to land in. """ if inputs.project_id is not None: return inputs.project_id + # Ad-hoc per-cell map (a multi-cell MegaTask root-subtask): mirror the + # product.project_for lookup but read the map off the parent task itself. + parent_cell_map = getattr(parent, "cell_projects", None) + if parent_cell_map: + for mapping in parent_cell_map: + if mapping.team == inputs.team: + return UUID(str(mapping.project_id)) parent_product_id = getattr(parent, "product_id", None) if self.product is not None and parent_product_id is not None: mapped = await self.product.project_for(parent_product_id, inputs.team) @@ -4730,8 +4738,8 @@ class Choreographer: field_hints={ "project_id": ( f"no project for team {inputs.team!r}: add a " - f"{inputs.team}->project mapping to the parent's product, or " - "pass an explicit project_id on delegate" + f"{inputs.team}->project mapping to the parent's cell map or " + "product, or pass an explicit project_id on delegate" ) }, message=f"cannot resolve a project for the {inputs.team} subtask", diff --git a/roboco/services/gateway/choreographer/pr_review.py b/roboco/services/gateway/choreographer/pr_review.py index af9bca81..f98da999 100644 --- a/roboco/services/gateway/choreographer/pr_review.py +++ b/roboco/services/gateway/choreographer/pr_review.py @@ -476,13 +476,25 @@ class PRReviewerMixin(_Base): project = await project_service.get(t.project_id) return project.slug if project is not None else None product_id = getattr(t, "product_id", None) - if product_id is None: - return None - from roboco.services.product import get_product_service + if product_id is not None: + from roboco.services.product import get_product_service - product_service = get_product_service(self.task.session) - project_ids = await product_service.distinct_project_ids(UUID(str(product_id))) - if not project_ids: - return None - project = await project_service.get(project_ids[0]) - return project.slug if project is not None else None + product_service = get_product_service(self.task.session) + project_ids = await product_service.distinct_project_ids( + UUID(str(product_id)) + ) + if not project_ids: + return None + project = await project_service.get(project_ids[0]) + return project.slug if project is not None else None + # Ad-hoc per-cell map root-subtask: mirror the product root's first-project + # resolution so the gate verdict reaches the PR in the mapped repo. + cell_map = getattr(t, "cell_projects", None) or [] + seen: set[UUID] = set() + for mapping in sorted(cell_map, key=lambda m: m.team.value): + pid = UUID(str(mapping.project_id)) + if pid not in seen: + seen.add(pid) + project = await project_service.get(pid) + return project.slug if project is not None else None + return None diff --git a/roboco/services/git.py b/roboco/services/git.py index 02b72f20..6b12f395 100644 --- a/roboco/services/git.py +++ b/roboco/services/git.py @@ -2965,15 +2965,26 @@ class GitService(BaseService): if task.project_id is not None: return await project_service.get(UUID(str(task.project_id))) product_id = getattr(task, "product_id", None) - if product_id is None: - return None - from roboco.services.product import get_product_service + if product_id is not None: + from roboco.services.product import get_product_service - product_service = get_product_service(self.session) - project_ids = await product_service.distinct_project_ids(UUID(str(product_id))) - if not project_ids: - return None - return await project_service.get(project_ids[0]) + product_service = get_product_service(self.session) + project_ids = await product_service.distinct_project_ids( + UUID(str(product_id)) + ) + if not project_ids: + return None + return await project_service.get(project_ids[0]) + # Ad-hoc per-cell map root-subtask: mirror the product root's first-project + # resolution so root-level git ops resolve a workspace for the mapped repo. + cell_map = getattr(task, "cell_projects", None) or [] + seen: set[UUID] = set() + for mapping in sorted(cell_map, key=lambda m: m.team.value): + pid = UUID(str(mapping.project_id)) + if pid not in seen: + seen.add(pid) + return await project_service.get(pid) + return None @staticmethod def _fast_gate_commands(project: Any) -> list[tuple[str, str]]: diff --git a/roboco/services/prompter.py b/roboco/services/prompter.py index 945cd61a..a64886f7 100644 --- a/roboco/services/prompter.py +++ b/roboco/services/prompter.py @@ -31,6 +31,7 @@ from roboco.models.base import ( TaskType, Team, ) +from roboco.models.product import ProductCellMapping from roboco.models.task import TaskCreateRequest from roboco.services.base import NotFoundError, ServiceError, ValidationError @@ -60,6 +61,11 @@ _CELL_CAPACITY: dict[str, int] = { # single-repo batch, which is just an ordinary (multi-)task, not a MegaTask. _MIN_MEGATASK_PROJECTS = 2 +# A draft whose per-cell map covers at least this many cells targets the ad-hoc +# multi-cell shape (a root-subtask with a cell->project map, no single project). +# Below it, a 1-cell map collapses to the single-project shape. +_MULTI_CELL_MIN = 2 + @dataclass class ReadinessTag: @@ -121,23 +127,33 @@ class PrompterService: product_id: UUID | None, *, is_umbrella: bool, + has_cell_projects: bool = False, ) -> None: - """A draft targets exactly one of project / product — or neither when it - is a MegaTask umbrella (branchless; its root-subtasks carry the projects). + """A draft targets exactly one of project / product / per-cell map — or + none when it is a MegaTask umbrella (branchless; its root-subtasks carry + the projects). The per-cell map is the multi-cell ad-hoc shape (a + root-subtask mixing per-cell projects from different products / OSS libs). """ - if project_id is None and product_id is None and not is_umbrella: + targets = bool(project_id) + bool(product_id) + bool(has_cell_projects) + if is_umbrella: + if targets != 0: + raise ValidationError( + message=( + "A MegaTask umbrella targets neither project nor product " + "(it is branchless); its root-subtasks carry the projects." + ), + field="project_id", + ) + return + if targets != 1: raise ValidationError( message=( - "The draft must target a project (single-cell) or a product " - "(board-led, multi-cell). Pick one in the confirm step." + "The draft must target exactly one of a project (single-cell), " + "a product (board-led, multi-cell), or a per-cell project map " + "(ad-hoc multi-cell). Pick one in the confirm step." ), field="project_id", ) - if project_id is not None and product_id is not None: - raise ValidationError( - message="Set exactly one of project_id or product_id, not both.", - field="product_id", - ) async def _resolve_owning_team( self, @@ -152,16 +168,22 @@ class PrompterService: ``team_override`` pins the team for a MegaTask batch (umbrella + every root-subtask share one owner). Otherwise: a project target is a - single-cell executable owned by the lead cell; a product target is a - board-led coordination root whose team follows the start mode (encoded in - the assignee) — the "Board review & Start" path assigns a board reviewer, - so it stays team=board until approved (else the CEO's Approve & Start - gate, which keys on team=board, never appears and the task strands). - "Approve & Start" (assignee main-pm) and the post-approval state are - team=main_pm. + single-cell executable owned by the lead cell; a product target OR an + ad-hoc per-cell map (≥2 cells in ``the_work``) is a multi-cell + coordination root owned by the Main PM — the cell map mirrors a product + fan-out root, so a cell PM (which can only delegate within its own cell) + must NOT own it (that would deadlock on the cross-cell fan-out). A + product's team follows the start mode (encoded in the assignee) — the + "Board review & Start" path assigns a board reviewer, so it stays + team=board until approved (else the CEO's Approve & Start gate, which + keys on team=board, never appears and the task strands). "Approve & + Start" (assignee main-pm) and the post-approval state are team=main_pm. """ if team_override is not None: return team_override + if len(_draft_cell_map(draft_data)) >= _MULTI_CELL_MIN: + # Ad-hoc multi-cell map → coordination root, like a product root. + return Team.MAIN_PM if resolved_product_id is None: return self._lead_cell_team(draft_data, default=default_lead) if resolved_assigned_to is not None and await self._assignee_is_board( @@ -219,12 +241,27 @@ class PrompterService: resolved_project_id = self._resolve_uuid_field(draft_data, "project_id") resolved_product_id = self._resolve_uuid_field(draft_data, "product_id") + # The ad-hoc per-cell map: ≥2 cells → multi-cell root-subtask (cell map + # shape, no project/product); 1 cell → single-project (use that project); + # 0 → fall back to the top-level project_id (single-cell legacy). + cell_map = _draft_cell_map(draft_data) + cell_projects: list[ProductCellMapping] = [] + if len(cell_map) >= _MULTI_CELL_MIN: + cell_projects = [ + ProductCellMapping(team=team, project_id=pid) for team, pid in cell_map + ] + resolved_project_id = None + resolved_product_id = None + elif len(cell_map) == 1: + resolved_project_id = cell_map[0][1] + resolved_product_id = None self._validate_draft_target( resolved_project_id, resolved_product_id, is_umbrella=is_batch_umbrella( batch_id=place.batch_id, parent_task_id=place.parent_task_id ), + has_cell_projects=bool(cell_projects), ) _lead, task_type, nature, complexity = self._coerce_draft_enums(draft_data) @@ -258,6 +295,7 @@ class PrompterService: assigned_to=resolved_assigned_to, project_id=resolved_project_id, product_id=resolved_product_id, + cell_projects=cell_projects, status=status, parent_task_id=place.parent_task_id, batch_id=place.batch_id, @@ -387,20 +425,41 @@ class PrompterService: """A MegaTask's drafts must each target one of the scoped repos (the only ones the intake agent read), and collectively span at least two distinct projects — otherwise it's a single-repo batch, not a MegaTask. + + Each draft targets its repos via its per-cell ``the_work[].project_id`` + map (a multi-cell draft) or, falling back, its top-level ``project_id`` + (a single-cell draft). Every targeted project must be in scope, and the + union across all drafts' cells must clear ``_MIN_MEGATASK_PROJECTS`` + (a single 2-cell draft already satisfies it). """ scope = {str(p) for p in project_ids} seen: set[str] = set() for idx, draft in enumerate(drafts): - pid = draft.get("project_id") - if not pid or str(pid) not in scope: + cell_map = _draft_cell_map(draft) + top_pid = draft.get("project_id") + if cell_map: + draft_pids = [str(pid) for _, pid in cell_map] + elif top_pid: + draft_pids = [str(top_pid)] + else: raise ValidationError( message=( - f"Task {idx + 1} targets a project outside this MegaTask's " - "selected repos. Point it at one of the scoped projects." + f"Task {idx + 1} has no project. Point each of its cells at " + "one of the scoped projects." ), field="drafts", ) - seen.add(str(pid)) + for pid in draft_pids: + if pid not in scope: + raise ValidationError( + message=( + f"Task {idx + 1} targets a project outside this " + "MegaTask's selected repos. Point it at one of the scoped " + "projects." + ), + field="drafts", + ) + seen.add(pid) if len(seen) < _MIN_MEGATASK_PROJECTS: raise ValidationError( message=( @@ -732,6 +791,42 @@ def _cell_teams(the_work: list[Any]) -> list[str]: return seen +def _draft_cell_map(draft: dict[str, Any]) -> list[tuple[Team, UUID]]: + """The draft's ad-hoc per-cell project map. + + One ``(team, project_id)`` per ``the_work`` entry whose ``team`` is a valid + cell AND that carries a ``project_id``, in ``the_work`` order, de-duped by + team (the first mapping for a cell wins — a ``task_cell_projects`` row is + unique per ``(task, team)``). Empty when no entry carries a project_id — the + draft then falls back to its top-level ``project_id`` (single-cell legacy). + + This is the multi-cell MegaTask root-subtask seam: a draft whose map has + ≥2 entries targets the ad-hoc cell-map shape (no project, no product), and + ``create_task_from_draft`` persists those rows on the root-subtask. + """ + cell_values = {t.value for t in CELL_TEAMS} + out: list[tuple[Team, UUID]] = [] + seen_teams: set[Team] = set() + for entry in draft.get("the_work") or []: + e = _as_work_entry(entry) + team_raw = str(e.get("team", "")) + if team_raw not in cell_values: + continue + team = Team(team_raw) + if team in seen_teams: + continue + pid_raw = e.get("project_id") + if not pid_raw: + continue + try: + pid = UUID(str(pid_raw)) + except (ValueError, TypeError): + continue + seen_teams.add(team) + out.append((team, pid)) + return out + + def derive_scale(the_work: list[Any]) -> str: """'multi' when more than one cell participates, else 'single'.""" return "multi" if len(_cell_teams(the_work)) > 1 else "single" diff --git a/roboco/services/task.py b/roboco/services/task.py index 21948a1a..ac629d44 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -20,6 +20,7 @@ from roboco.db.tables import ( JournalTable, ProjectTable, SessionTaskTable, + TaskCellProjectTable, TaskTable, WorkSessionTable, ) @@ -564,6 +565,7 @@ class TaskService(BaseService): product_id=task.product_id, batch_id=task.batch_id, parent_task_id=task.parent_task_id, + has_cell_projects=bool(task.cell_projects), ), # An external-PR review task reviews someone else's PR read-only — # no branch of its own — so it is branch-gate exempt. @@ -743,18 +745,22 @@ class TaskService(BaseService): def _require_target_or_umbrella(req: TaskCreateRequest) -> None: """Service-layer invariant (covers every create path — API, a2a, gateway). - A task targets a single repo (``project_id``) or fans out across cells via - a product (``product_id``) — it must have one or the other, EXCEPT a - MegaTask umbrella, which targets neither (it groups N root-subtasks that - each carry their own project) and is branchless. + A task targets a single repo (``project_id``), fans out across cells via a + product (``product_id``), or carries an ad-hoc per-cell map + (``cell_projects``) — it must have exactly one, EXCEPT a MegaTask umbrella, + which targets neither (it groups N root-subtasks that each carry their own + project) and is branchless. """ if req.project_id is not None or req.product_id is not None: return + if req.cell_projects: + return if is_batch_umbrella(batch_id=req.batch_id, parent_task_id=req.parent_task_id): return raise ValueError( - "task needs a project_id (the repo it targets) or a product_id " - "(a cell->project map for a fan-out task)" + "task needs a project_id (the repo it targets), a product_id " + "(a cell->project map for a fan-out task), or cell_projects " + "(an ad-hoc per-cell map for a multi-cell coordination root)" ) async def _validate_batch_membership(self, req: TaskCreateRequest) -> None: @@ -773,11 +779,12 @@ class TaskService(BaseService): parent_task_id=req.parent_task_id, project_id=req.project_id, product_id=req.product_id, + has_cell_projects=bool(req.cell_projects), ): raise ValueError( "batch_id is only valid on a MegaTask umbrella (targets neither " - "project nor product) or a root-subtask (targets exactly one); " - "refusing a stray batch_id on any other task." + "project, product, nor a cell map) or a root-subtask (targets " + "exactly one); refusing a stray batch_id on any other task." ) if req.parent_task_id is None: return # a well-formed umbrella @@ -843,6 +850,22 @@ class TaskService(BaseService): self.session.add(task) await self.session.flush() + # Persist the ad-hoc per-cell project map (a MegaTask root-subtask + # spanning multiple cells). Added as explicit rows with the flushed + # task id rather than via the `cell_projects` collection (which is + # unloaded on a freshly-built task — mutating it would trigger a lazy + # load). Unique (task_id, team) is enforced by the table; a caller + # passing duplicate teams raises IntegrityError here, which is the + # right failure for a malformed request. + for mapping in req.cell_projects: + self.session.add( + TaskCellProjectTable( + task_id=cast("UUID", task.id), + team=mapping.team, + project_id=mapping.project_id, + ) + ) + # Inherit parent task's primary session for subtasks if req.parent_task_id: await self._inherit_parent_session( @@ -1558,12 +1581,13 @@ class TaskService(BaseService): if not task.project_id: # A coordination/fan-out task carries a product (a cell->project - # map) but no repo of its own. Per the CEO-locked branch model it is - # the Main-PM integration point: it cuts feature/main_pm/{root} off - # master in EACH repo the product spans, so cells branch off it - # (not off master) and only the CEO merges the root into master. - # Only a task with neither project nor product is misconfigured. - if task.product_id: + # map) or an ad-hoc cell_projects map but no repo of its own. Per + # the CEO-locked branch model it is the Main-PM integration point: + # it cuts feature/main_pm/{root} off master in EACH repo the map + # 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: 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 — @@ -1574,9 +1598,9 @@ class TaskService(BaseService): ): return "" raise ValueError( - "Task requires a project_id (a repo) or a product_id (a " - "cell->project map) to create a branch. Assign one before " - "claiming." + "Task requires a project_id (a repo), a product_id, or a " + "cell_projects map (a cell->project map) to create a branch. " + "Assign one before claiming." ) return await self._auto_create_branch(task, agent_id) @@ -1762,33 +1786,55 @@ class TaskService(BaseService): ) return branch_name + async def _distinct_projects_for_task(self, task: TaskTable) -> list[UUID]: + """The distinct projects a coordination root's map spans — one + ``feature/main_pm/{root}`` integration branch each. + + A coordination root carries EITHER a product (``product_id`` → its + ``product_projects`` map) OR an ad-hoc per-cell map (``cell_projects``). + Both yield the same thing: the distinct project_ids the root spans, + de-duped (a monorepo mapped across cells yields one project per distinct + project, in team order). Empty when the root has neither (the umbrella, + which never reaches here, or a not-yet-mapped product — caller returns + ``""`` so delegation falls back to the parent's project). + """ + from roboco.services.product import get_product_service + + if task.product_id is not None: + return await get_product_service(self.session).distinct_project_ids( + UUID(str(task.product_id)) + ) + # Ad-hoc cell_projects map: de-dupe by project_id, in team order (mirrors + # ProductService.distinct_project_ids' ordering + dedup semantics). + seen: dict[UUID, None] = {} + for mapping in sorted(task.cell_projects, key=lambda m: m.team.value): + seen.setdefault(UUID(str(mapping.project_id)), None) + return list(seen) + async def _ensure_coordination_root_branches( self, task: TaskTable, agent_id: UUID, ) -> str: - """Cut the Main-PM integration branch in every repo the product spans. + """Cut the Main-PM integration branch in every repo the map spans. - The coordination root carries a product (a cell->repo map) but no - project of its own. Per the CEO-locked model, the Main-PM root branches - ``feature/main_pm/{root}`` OFF master in each distinct repo; cells then - branch off it (via the parent-branch resolution) instead of off master, - so cell work never targets master — only the CEO merges the root branch - into master, per repo. Monorepo => one branch; multi-repo => N. + The coordination root carries a product (a cell->repo map) or an ad-hoc + ``cell_projects`` map, but no project of its own. Per the CEO-locked model, + the Main-PM root branches ``feature/main_pm/{root}`` OFF master in each + distinct repo the map spans; cells then branch off it (via the + parent-branch resolution) instead of off master, so cell work never + targets master — only the CEO merges the root branch into master, per + repo. Monorepo => one branch; multi-repo => N. Returns the shared branch name (identical across repos), or ``""`` when - the product has no cell->repo map yet (delegation then falls back to the - parent's project per the routing spec, and the root stays branchless). + the map has no projects yet (delegation then falls back to the parent's + project per the routing spec, and the root stays branchless). """ - from roboco.services.product import get_product_service from roboco.services.project import get_project_service - product_service = get_product_service(self.session) project_service = get_project_service(self.session) - project_ids = await product_service.distinct_project_ids( - UUID(str(task.product_id)) - ) + project_ids = await self._distinct_projects_for_task(task) branch_name = "" for project_id in project_ids: project = await project_service.get(project_id) @@ -1840,11 +1886,12 @@ class TaskService(BaseService): parent_task_id=getattr(task, "parent_task_id", None), project_id=getattr(task, "project_id", None), product_id=getattr(task, "product_id", None), + has_cell_projects=bool(getattr(task, "cell_projects", None)), ): raise ValueError( "this update would break the task's MegaTask shape: a batch " - "member must stay an umbrella (targets neither project nor " - "product) or a root-subtask (exactly one target, parented)." + "member must stay an umbrella (targets neither project, product, " + "nor a cell map) or a root-subtask (exactly one target, parented)." ) async def update( @@ -5043,6 +5090,7 @@ class TaskService(BaseService): product_id=task.product_id, batch_id=task.batch_id, parent_task_id=task.parent_task_id, + has_cell_projects=bool(task.cell_projects), ) if not is_coordination_root: self._validate_and_set_status(task, TaskStatus.NEEDS_REVISION, "ceo") diff --git a/tests/unit/api/test_schemas_tasks.py b/tests/unit/api/test_schemas_tasks.py index d3f8603c..7e159495 100644 --- a/tests/unit/api/test_schemas_tasks.py +++ b/tests/unit/api/test_schemas_tasks.py @@ -30,6 +30,7 @@ from roboco.api.schemas.tasks import ( transform_update_data, ) from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType, Team +from roboco.models.product import ProductCellMapping _ORDER_DEFAULT = 0 @@ -291,6 +292,7 @@ def _stub_task(*, with_project: bool = False) -> SimpleNamespace: task_type=TaskType.CODE, project_id=uuid4(), product_id=None, + cell_projects=[], project=(SimpleNamespace(slug="proj-1") if with_project else None), docs_complete=False, pr_created=False, @@ -348,6 +350,38 @@ def test_task_to_response_includes_slug_when_project_loaded() -> None: assert resp.project_slug == "proj-1" +def test_task_to_response_serializes_cell_projects_when_loaded() -> None: + """An ad-hoc per-cell map (a multi-cell MegaTask root-subtask) round-trips + into the response when the relationship is loaded.""" + be_proj, fe_proj = uuid4(), uuid4() + stub = _stub_task() + stub.project_id = None + stub.product_id = None + stub.cell_projects = [ + SimpleNamespace(team=Team.BACKEND, project_id=be_proj), + SimpleNamespace(team=Team.FRONTEND, project_id=fe_proj), + ] + fake_inspector = MagicMock() + fake_inspector.unloaded = set() # cell_projects IS loaded + with patch("roboco.api.schemas.tasks.sa_inspect", return_value=fake_inspector): + resp = task_to_response(stub) # type: ignore[arg-type] + assert resp.cell_projects == [ + ProductCellMapping(team=Team.BACKEND, project_id=be_proj), + ProductCellMapping(team=Team.FRONTEND, project_id=fe_proj), + ] + + +def test_task_to_response_omits_cell_projects_when_unloaded() -> None: + """A freshly-created task whose cell_projects relationship is unloaded + serializes to [] rather than triggering a lazy load.""" + stub = _stub_task() + fake_inspector = MagicMock() + fake_inspector.unloaded = {"cell_projects"} + with patch("roboco.api.schemas.tasks.sa_inspect", return_value=fake_inspector): + resp = task_to_response(stub) # type: ignore[arg-type] + assert resp.cell_projects == [] + + def test_task_to_response_serializes_all_note_sections() -> None: """Regression: pr_reviewer_notes / doc_notes / notes_structured MUST be in the response. The builder previously omitted them, so the panel showed them blank diff --git a/tests/unit/foundation/policy/test_batch.py b/tests/unit/foundation/policy/test_batch.py index ba736a1e..d7ab48fd 100644 --- a/tests/unit/foundation/policy/test_batch.py +++ b/tests/unit/foundation/policy/test_batch.py @@ -46,10 +46,33 @@ def test_branchless_coordination_excludes_normal_and_root_subtasks() -> None: batch_id=uuid4(), parent_task_id=uuid4(), ) - # genuinely unroutable (none of project / product / batch) stays gated + # genuinely unroutable (none of project / product / batch / cell-map) stays gated assert not is_branchless_coordination(project_id=None, product_id=None) +def test_branchless_coordination_covers_ad_hoc_cell_map_root() -> None: + # An ad-hoc per-cell project map (no project_id, no product_id, carries a + # cell map) is a coordination root exactly like a Product fan-out root: it + # cuts feature/main_pm/{root} per repo and opens a root->master PR per repo, + # so the claim branch gate skips the single-branch requirement. Holds both + # for a MegaTask root-subtask and a standalone coordination root. + assert is_branchless_coordination( + project_id=None, product_id=None, has_cell_projects=True + ) + assert is_branchless_coordination( + project_id=None, + product_id=None, + batch_id=uuid4(), + parent_task_id=uuid4(), + has_cell_projects=True, + ) + # a cell map is NOT branchless if a project_id is also set (then it's a normal + # project task that happens to carry a stray map — gated, not exempt). + assert not is_branchless_coordination( + project_id=uuid4(), product_id=None, has_cell_projects=True + ) + + def test_valid_batch_shape_allows_umbrella_and_root_subtask() -> None: bid = uuid4() # umbrella: batch_id, no parent, NO target @@ -89,3 +112,54 @@ def test_valid_batch_shape_denies_stray_batch_id() -> None: assert not is_valid_batch_shape( batch_id=bid, parent_task_id=uuid4(), project_id=uuid4(), product_id=uuid4() ) + + +def test_valid_batch_shape_allows_ad_hoc_cell_map_root_subtask() -> None: + bid = uuid4() + # a root-subtask carrying an ad-hoc per-cell map (no project, no product) is a + # well-formed third targeting shape — exactly one of {project, product, map}. + assert is_valid_batch_shape( + batch_id=bid, + parent_task_id=uuid4(), + project_id=None, + product_id=None, + has_cell_projects=True, + ) + # a non-batch task carrying a cell map is unconstrained here (the normal + # targeting rule applies; the map is a coordination-root shape in its own + # right, not a batch-only construct). + assert is_valid_batch_shape( + batch_id=None, + parent_task_id=None, + project_id=None, + product_id=None, + has_cell_projects=True, + ) + + +def test_valid_batch_shape_denies_cell_map_alongside_another_target() -> None: + bid = uuid4() + # umbrella with a cell map: an umbrella must target NEITHER — a map is a target. + assert not is_valid_batch_shape( + batch_id=bid, + parent_task_id=None, + project_id=None, + product_id=None, + has_cell_projects=True, + ) + # root-subtask with BOTH a project and a cell map: two targets, malformed. + assert not is_valid_batch_shape( + batch_id=bid, + parent_task_id=uuid4(), + project_id=uuid4(), + product_id=None, + has_cell_projects=True, + ) + # root-subtask with BOTH a product and a cell map: two targets, malformed. + assert not is_valid_batch_shape( + batch_id=bid, + parent_task_id=uuid4(), + project_id=None, + product_id=uuid4(), + has_cell_projects=True, + ) diff --git a/tests/unit/services/test_choreographer_subtask_project.py b/tests/unit/services/test_choreographer_subtask_project.py new file mode 100644 index 00000000..f3e790de --- /dev/null +++ b/tests/unit/services/test_choreographer_subtask_project.py @@ -0,0 +1,117 @@ +"""Choreographer._resolve_subtask_project — fan-out resolution coverage. + +The method resolves which project a delegated subtask lands in. It has four +priority tiers, exercised here against a stub ``self`` (the cell-map match +path never touches ``self``, and the raise path only reads ``self.product``): + +1. explicit ``inputs.project_id`` wins outright. +2. the parent's ad-hoc ``cell_projects`` map → the mapping for ``inputs.team``. +3. the parent's Product map (delegated to ``self.product.project_for``). +4. the parent's own ``project_id``. +5. otherwise ``TaskCompletenessError`` (no repo to land in). + +The ad-hoc cell-map tier (2) is the multi-cell MegaTask root-subtask seam; the +other tiers are unchanged product-root / single-project behavior. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from uuid import UUID, uuid4 + +import pytest +from roboco.foundation.policy.task_completeness import TaskCompletenessError +from roboco.models.base import Team +from roboco.services.gateway.choreographer._impl import ( + Choreographer, + DelegateInputs, +) + + +def _inputs(*, team: Team, project_id: UUID | None = None) -> DelegateInputs: + return DelegateInputs( + title="t", + description="d", + assigned_to="be-dev-1", + team=team.value, + task_type="code", + nature="technical", + acceptance_criteria=["a"], + project_id=project_id, + ) + + +def _mapping(team: Team, project_id: UUID) -> SimpleNamespace: + return SimpleNamespace(team=team, project_id=project_id) + + +@pytest.mark.asyncio +async def test_explicit_inputs_project_id_wins_over_cell_map() -> None: + """Tier 1: an explicit project_id on delegate short-circuits the map.""" + be_proj, explicit = uuid4(), uuid4() + parent = SimpleNamespace( + cell_projects=[_mapping(Team.BACKEND, be_proj)], + product_id=None, + project_id=None, + ) + self_stub = SimpleNamespace(product=None) + resolved = await Choreographer._resolve_subtask_project( + self_stub, parent, _inputs(team=Team.BACKEND, project_id=explicit) + ) + assert resolved == explicit + + +@pytest.mark.asyncio +async def test_cell_map_resolves_project_for_matching_team() -> None: + """Tier 2: the parent's cell map yields the project for the delegated cell.""" + be_proj, fe_proj = uuid4(), uuid4() + parent = SimpleNamespace( + cell_projects=[ + _mapping(Team.FRONTEND, fe_proj), + _mapping(Team.BACKEND, be_proj), + ], + product_id=None, + project_id=None, + ) + self_stub = SimpleNamespace(product=None) + resolved = await Choreographer._resolve_subtask_project( + self_stub, parent, _inputs(team=Team.BACKEND) + ) + assert resolved == be_proj + + +@pytest.mark.asyncio +async def test_cell_map_missing_team_falls_through_to_parent_project() -> None: + """No mapping for the requested cell → fall through to the parent's own + project_id (tier 4), not raise — the parent may still carry a project.""" + be_proj, own = uuid4(), uuid4() + parent = SimpleNamespace( + cell_projects=[_mapping(Team.BACKEND, be_proj)], + product_id=None, + project_id=own, + ) + self_stub = 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) + ) + assert resolved == own + + +@pytest.mark.asyncio +async def test_cell_map_only_parent_with_no_match_raises_completeness() -> None: + """A fan-out parent (cell map, no own project, no product) with no mapping + for the requested cell raises TaskCompletenessError — the subtask has no + repo to land in.""" + be_proj = uuid4() + parent = SimpleNamespace( + cell_projects=[_mapping(Team.BACKEND, be_proj)], + product_id=None, + project_id=None, + ) + self_stub = SimpleNamespace(product=None) + with pytest.raises(TaskCompletenessError) as exc: + await Choreographer._resolve_subtask_project( + self_stub, parent, _inputs(team=Team.FRONTEND) + ) + assert "project_id" in exc.value.missing diff --git a/tests/unit/services/test_prompter.py b/tests/unit/services/test_prompter.py index 221309ba..5f2def8b 100644 --- a/tests/unit/services/test_prompter.py +++ b/tests/unit/services/test_prompter.py @@ -32,6 +32,7 @@ from roboco.services.base import ServiceError, ValidationError from roboco.services.prompter import ( PrompterService, _cell_teams, + _draft_cell_map, compose_description, derive_scale, get_prompter_service, @@ -102,6 +103,7 @@ def test_derive_scale_single_vs_multi() -> None: # "'str' object has no attribute 'get'"). # ----------------------------------------------------------------------------- + def test_cell_teams_tolerates_bare_string_entries() -> None: # The LLM emitted the_work as a list of team names, not objects. assert _cell_teams(["backend", "frontend", "backend"]) == ["backend", "frontend"] @@ -697,3 +699,193 @@ def test_preview_batch_tolerates_bare_string_the_work() -> None: result = service.preview_batch(drafts) assert isinstance(result["waves"], list) assert isinstance(result["warnings"], list) + + +# ============================================================================= +# Per-cell project map (multi-cell MegaTask root-subtask seam) — pure helpers +# ============================================================================= + + +def _work(team: str, project_id: UUID | None) -> dict[str, Any]: + entry: dict[str, Any] = {"team": team, "summary": "s", "items": ["x"]} + if project_id is not None: + entry["project_id"] = str(project_id) + return entry + + +def test_draft_cell_map_collects_per_cell_projects_in_order() -> None: + """A multi-cell draft yields one (team, project_id) per the_work entry, + in the_work order, de-duped by team.""" + be_proj, fe_proj = uuid4(), uuid4() + draft = { + "the_work": [ + _work("backend", be_proj), + _work("frontend", fe_proj), + ] + } + assert _draft_cell_map(draft) == [(Team.BACKEND, be_proj), (Team.FRONTEND, fe_proj)] + + +def test_draft_cell_map_dedupes_repeated_team_keeping_first() -> None: + """Two entries for the same cell (LLM noise) keep the first mapping — a + task_cell_projects row is unique per (task, team).""" + first, second = uuid4(), uuid4() + draft = { + "the_work": [ + _work("backend", first), + _work("backend", second), + ] + } + assert _draft_cell_map(draft) == [(Team.BACKEND, first)] + + +def test_draft_cell_map_skips_entries_without_project_id() -> None: + """An entry with no project_id (single-cell legacy or a bare team string) is + skipped — the draft then falls back to its top-level project_id.""" + be_proj = uuid4() + draft = { + "the_work": [ + _work("backend", be_proj), + {"team": "frontend", "summary": "s", "items": []}, # no project_id + ] + } + assert _draft_cell_map(draft) == [(Team.BACKEND, be_proj)] + + +def test_draft_cell_map_empty_when_no_entry_has_project_id() -> None: + """A legacy single-cell draft (top-level project_id, bare-string the_work) + yields an empty map — the caller falls back to the top-level project_id.""" + assert _draft_cell_map({"the_work": ["backend", "frontend"]}) == [] + assert _draft_cell_map({"the_work": [{"team": "backend"}]}) == [] + + +def test_draft_cell_map_ignores_off_enum_teams_and_bad_uuids() -> None: + """Off-enum team names and malformed project_ids are skipped, not crashed on + (the intake agent is an LLM).""" + good = uuid4() + draft = { + "the_work": [ + _work("backend", good), + {"team": "marketing", "project_id": str(uuid4())}, # not a cell + _work("frontend", None), # missing + {"team": "ux_ui", "project_id": "not-a-uuid"}, # bad uuid + ] + } + assert _draft_cell_map(draft) == [(Team.BACKEND, good)] + + +def test_validate_batch_scope_accepts_single_multi_cell_draft() -> None: + """One 2-cell draft already spans ≥2 distinct projects → valid MegaTask.""" + be_proj, fe_proj = uuid4(), uuid4() + drafts = [ + { + "title": "S1", + "acceptance_criteria": ["a"], + "the_work": [ + _work("backend", be_proj), + _work("frontend", fe_proj), + ], + } + ] + # Must not raise: 2 distinct projects across the one draft's cells. + PrompterService._validate_batch_scope(drafts, [be_proj, fe_proj]) + + +def test_validate_batch_scope_rejects_out_of_scope_per_cell_project() -> None: + """A per-cell project_id outside the scoped set is refused.""" + in_scope, out_of_scope = uuid4(), uuid4() + drafts = [ + { + "title": "S1", + "acceptance_criteria": ["a"], + "the_work": [ + _work("backend", in_scope), + _work("frontend", out_of_scope), + ], + } + ] + with pytest.raises(ValidationError, match="outside this MegaTask"): + PrompterService._validate_batch_scope(drafts, [in_scope, uuid4()]) + + +def test_validate_batch_scope_rejects_draft_with_no_project() -> None: + """A draft with neither a per-cell map nor a top-level project_id is refused.""" + drafts = [ + { + "title": "S1", + "acceptance_criteria": ["a"], + "the_work": [_work("backend", None), _work("frontend", None)], + } + ] + with pytest.raises(ValidationError, match="has no project"): + PrompterService._validate_batch_scope(drafts, [uuid4(), uuid4()]) + + +def test_validate_batch_scope_distinct_count_spans_all_cells() -> None: + """The ≥2 minimum counts distinct projects across ALL drafts' cells, not per + draft. Two single-cell drafts on the same project still fail (degenerate).""" + only = uuid4() + drafts = [ + { + "title": "A", + "acceptance_criteria": ["a"], + "the_work": [_work("backend", only)], + }, + { + "title": "B", + "acceptance_criteria": ["b"], + "the_work": [_work("frontend", only)], # same project, different cell + }, + ] + with pytest.raises(ValidationError, match="at least two distinct projects"): + PrompterService._validate_batch_scope(drafts, [only, uuid4()]) + + +def test_validate_batch_scope_legacy_single_cell_drafts_still_work() -> None: + """Back-compat: drafts using a top-level project_id (no the_work map) still + validate against the scope and the ≥2 distinct minimum.""" + p1, p2 = uuid4(), uuid4() + drafts = [ + {"title": "A", "acceptance_criteria": ["a"], "project_id": str(p1)}, + {"title": "B", "acceptance_criteria": ["b"], "project_id": str(p2)}, + ] + PrompterService._validate_batch_scope(drafts, [p1, p2]) + + +@pytest.mark.asyncio +async def test_resolve_owning_team_multi_cell_map_routes_to_main_pm() -> None: + """A multi-cell ad-hoc map is a coordination root (mirrors a product root), so + it routes to the Main PM — never the lead cell (a cell PM can't delegate + cross-cell; that would deadlock the fan-out). No DB access on this branch.""" + service = get_prompter_service() # no db — the cell-map branch never reads it + be_proj, fe_proj = uuid4(), uuid4() + draft = { + "the_work": [ + _work("backend", be_proj), + _work("frontend", fe_proj), + ] + } + team = await service._resolve_owning_team( + draft, + resolved_product_id=None, + resolved_assigned_to=None, + team_override=None, + default_lead=Team.BACKEND, + ) + assert team is Team.MAIN_PM + + +@pytest.mark.asyncio +async def test_resolve_owning_team_single_cell_still_routes_to_lead_cell() -> None: + """A single-cell project draft (no product, no multi-cell map) keeps its + legacy owner: the lead cell.""" + service = get_prompter_service() + draft = {"the_work": [_work("backend", uuid4())]} + team = await service._resolve_owning_team( + draft, + resolved_product_id=None, + resolved_assigned_to=None, + team_override=None, + default_lead=Team.BACKEND, + ) + assert team is Team.BACKEND diff --git a/tests/unit/services/test_task.py b/tests/unit/services/test_task.py index f2bb035d..6655c424 100644 --- a/tests/unit/services/test_task.py +++ b/tests/unit/services/test_task.py @@ -993,10 +993,66 @@ async def test_ensure_branch_coordination_root_no_cell_map_stays_branchless() -> @pytest.mark.asyncio -async def test_ensure_branch_raises_when_neither_project_nor_product() -> None: - """A task with neither a project nor a product is genuinely misconfigured.""" +async def test_ensure_branch_cell_map_root_cuts_integration_branch_per_project() -> ( + None +): + """An ad-hoc cell_projects root cuts feature/main_pm/{root} in each distinct + project the map spans — the product-root path with the map sourced from the + task instead of a Product.""" svc = TaskService(MagicMock()) - task = MagicMock(branch_name=None, project_id=None, product_id=None) + be_proj, fe_proj = uuid4(), uuid4() + cell_map = [ + SimpleNamespace(team=Team.BACKEND, project_id=be_proj), + SimpleNamespace(team=Team.FRONTEND, project_id=fe_proj), + ] + task = MagicMock( + branch_name=None, + project_id=None, + product_id=None, + batch_id=uuid4(), + parent_task_id=uuid4(), + cell_projects=cell_map, + ) + create_in_project = AsyncMock(return_value="feature/main_pm/root1234") + _bind(svc, "_create_branch_in_project", create_in_project) + project_svc = MagicMock(get=AsyncMock(return_value=MagicMock())) + with patch("roboco.services.project.get_project_service", return_value=project_svc): + result = await svc._ensure_branch_for_task(task, uuid4()) + assert result == "feature/main_pm/root1234" + # one integration branch per distinct project in the map (here 2 cells, 2 projects) + assert create_in_project.await_count == len(cell_map) + + +@pytest.mark.asyncio +async def test_distinct_projects_for_task_dedupes_cell_map_by_project_id() -> None: + """Two cells mapping at the same project (the monorepo case) yield ONE + integration branch, not two — mirroring product distinct_project_ids.""" + svc = TaskService(MagicMock()) + shared = uuid4() + task = MagicMock( + project_id=None, + product_id=None, + cell_projects=[ + SimpleNamespace(team=Team.FRONTEND, project_id=shared), + SimpleNamespace(team=Team.BACKEND, project_id=shared), + ], + ) + ids = await svc._distinct_projects_for_task(task) + assert ids == [shared] + + +@pytest.mark.asyncio +async def test_ensure_branch_raises_when_neither_project_nor_product() -> None: + """A task with neither a project, a product, nor a cell map is misconfigured.""" + svc = TaskService(MagicMock()) + task = MagicMock( + branch_name=None, + project_id=None, + product_id=None, + cell_projects=[], + batch_id=None, + parent_task_id=None, + ) with pytest.raises(ValueError, match="project_id"): await svc._ensure_branch_for_task(task, uuid4())