MegaTask (#248)

* feat(batch): batch_id + collision descriptor columns

Sequenced batch intake ("Mega task") foundation: tasks.batch_id (indexed)
groups a batch of top-level tasks created together; intends_to_touch (text[]),
adds_migration and touches_shared (bool, NOT NULL default false) are the
per-task collision surface the SequencingService will read to wire dependency
waves. Mirrored on the Task model + TaskCreateRequest and wired through
TaskService.create. Migration 046 (real upgrade->downgrade->upgrade verified
vs a throwaway pgvector PG); a non-batch task declares no surface (defaults).

Task 1 of the 0.11.0 sequenced-batch-intake plan.

* feat(batch): flag + draft collision descriptors

Default-off ROBOCO_BATCH_INTAKE_ENABLED (config + FEATURE_FLAGS + panel card);
the propose_draft tool doc + the TS DraftProposal gain the per-task collision
surface intends_to_touch / adds_migration / touches_shared. The draft is a loose
dict so the descriptors ride it through the relay intact (test asserts the
forwarded payload); the analyzer (Task 3) reads them to wire dependency waves.

Task 2 of the 0.11.0 sequenced-batch-intake plan.

* feat(batch): deterministic collision-sequencing analyzer

SequencingService.analyze turns a batch's per-task collision surfaces into a
dependency DAG + execution waves — correctness in CODE, not agent judgment.
Rules in order: file overlap serializes (more-important first), migrations form
a serial chain (no concurrent Alembic heads), touches_shared runs last, cell
contention warns (never serializes); then dedupe, existence + cycle check, and
Kahn topological layering. Pure (no DB/services); SequencingError on a cycle or
out-of-range edge.

Golden test reproduces the CEO's hand-sequenced 4 waves of the 11-item
guard-core-app batch (the effort that deadlocked the Main PM): S6 alone last,
the R1/R3/R4 migration chain, R2/R3/S8 serialized on the shared threat service,
S1/S2/S7 in one parallel wave.

Task 3 of the 0.11.0 sequenced-batch-intake plan.

* chore(batch): brand the user-facing surfaces "MegaTask"

The user-facing name is MegaTask: the feature-flag label is "MegaTask intake",
the panel flag-card and the config description lead with MegaTask. Internal
names stay technical (batch_intake_enabled, batch_id, SequencingService).

* chore(batch): drop the feature flag — MegaTask is a core intake scope

MegaTask is additive and opt-in by its own nature (the Prompter proposes a
batch only when the CEO asks for several tasks; single-task intake is
unchanged), so there is no risk surface a flag protects — 'don't create a
MegaTask' is the off switch. Remove batch_intake_enabled from config, the
FEATURE_FLAGS registry, the panel flag card, and its tests. MegaTask will be
a third scope option in the Intake modal (single-cell / multi-project /
MegaTask), not a toggle.

* feat(batch): MegaTask identity predicate + orchestrator branchless recognition

The single source of truth for the umbrella's exemptions: pure
is_batch_umbrella / is_batch_root_subtask / is_branchless_coordination
(foundation/policy/batch.py) — an umbrella has a batch_id and is top-level; a
root-subtask shares the batch_id but is parented. The orchestrator's
_is_coordination_task now consults is_branchless_coordination, so a MegaTask
umbrella is recognized as doing no git of its own (git-exempt at spawn-readiness
/ stuck-detection) exactly like a product fan-out root. Non-batch behavior is
identical (the predicate reduces to the old no-project+product check; the
orchestrator coordination suite stays green), and the umbrella branch is inert
until the create path exists.

First slice of the MegaTask umbrella enforcement (branchless guard).

* feat(batch): branchless umbrella guard across the git-exemption sites

A MegaTask umbrella does no git of its own — every git-exemption site in
TaskService now consults the shared is_branchless_coordination predicate
instead of an inline product-only check, so the umbrella's exemptions
cannot drift between sites:

- the claimed->in_progress branch gate (GitContext.is_coordination) lets
  an unbranched umbrella reach in_progress and delegate;
- _ensure_branch_for_task short-circuits an umbrella to "" instead of the
  misconfigured raise (the claim path ignores the return, treating it as
  branchless);
- CEO-reject routing sends a rejected umbrella to the Main PM in PENDING
  (needs_revision is developer-claim-only and would deadlock it).

Covers both shapes via the predicate (product fan-out root OR umbrella);
a batch root-subtask keeps its own branch/PR. Adds orchestrator
recognition tests for the umbrella plus claim/branch/reject integration
tests.

* feat(batch): umbrella assembles no PR; completes branchless

submit_root now hard-rejects a MegaTask umbrella up front (a preflight
that also folds in the unknown-role refusal to stay within the
return-count budget): the umbrella spans many projects with no single
master, so each root-subtask opens and is reviewed on its own PR — the
umbrella never enters the in-path review gate. The Main PM completes it
directly once every root-subtask is terminal.

Umbrella completion needs no new code: it is branchless (no branch_name),
so _main_pm_complete_guard already accepts it from in_progress, checks
all_subtasks_terminal, and main_pm_complete walks it to awaiting_pm_review
and escalates to the CEO with no PR creation — exactly the product
fan-out root path. Adds the submit_root-reject and umbrella-completion
gateway tests; pins batch_id=None on the normal-root submit_root test
(a MagicMock auto-attr would otherwise read as an umbrella).

* feat(batch): MegaTask create path — umbrella + sequenced root-subtasks

PrompterService.confirm_live_batch turns N confirmed drafts into a real
MegaTask: it builds each draft's collision surface, runs the pure
SequencingService to get conflict-free waves, creates the branchless
umbrella (batch_id, no project/product), then one root-subtask per draft
(own project, parent=umbrella, sequence=wave index, descriptors), and
wires the analyzer's edges through add_dependency so the existing
dependency-gate runs the waves in order. The route picks the start path
like a single confirm: 'board' holds the root-subtasks in BACKLOG for the
batch review; 'main_pm' creates them PENDING so wave 0 dispatches at once.

create_task_from_draft gains a BatchPlacement (parent/batch/sequence/
team_override) and forwards the collision descriptors; the exactly-one-
target rule (here and the TaskService.create invariant) is relaxed for an
umbrella, which legitimately targets neither. New route
POST /live/{session}/confirm-batch + BatchConfirmRequest mirror the single
confirm. Adds the structural-invariant + board-hold + empty-batch tests.

* feat(batch): release MegaTask root-subtasks on CEO approval; board awareness

The board route holds a MegaTask's root-subtasks in BACKLOG so the work
waits for the batch review. approve_and_start (CEO gate #1, board->Main PM)
now releases them via _activate_batch_root_subtasks: each held child flips
BACKLOG -> PENDING + team=main_pm so the dependency-gate dispatches wave 0.
No-op for a non-umbrella; idempotent (children past BACKLOG untouched).

The Product Owner and Head of Marketing identity prompts gain a MegaTask
section so they review the whole batch + wave plan and adjust scope before
sign-off (they review drafts; the umbrella is their unit). Also extracts
the create() target invariant into _require_target_or_umbrella to keep the
method under the complexity gate after the umbrella exemption. Adds the
umbrella-approval activation test.

* feat(batch): multi-project intake scope for MegaTask

A MegaTask spans several possibly-unrelated repos, so the intake chat can
now be scoped to an explicit project list (not just one project or one
product). StartLiveRequest gains project_ids; /live/start threads it
through start/spawn_intake_session -> _spawn_intake_container ->
_clone_intake_scope. The multi-repo clone machinery already existed for
products; _intake_scope_slugs now also resolves an explicit project_ids
set (split into _slugs_for_project_ids / _slugs_for_product), cloning each
repo with the first as the primary cwd and the siblings readable. Scope
validation is now 'exactly one of project_slug / product_id / project_ids'
via the shared _require_one_intake_scope. Adds scope-resolution, spawn,
and route tests for the MegaTask path.

* feat(batch): propose_batch intake tool (MegaTask multi-draft hand-off)

The intake agent can now hand the panel a whole MegaTask in one tool call.
Both intake paths gain propose_batch alongside propose_draft:
- Claude (intake_driver): a propose_batch tool registered on the in-SDK
  MCP server + allowlisted; the driver intercepts the ToolUseBlock and
  emits ONE StreamChunk(kind="batch") carrying {drafts:[...], title}.
- grok (intake_server): a propose_batch tool that POSTs a "batch" relay
  event via the shared _post_event helper (post_draft/post_batch).

A batch carries N drafts, each the propose_draft shape PLUS its own
project_id (a MegaTask spans unrelated repos) and collision surface so the
analyzer sequences the waves. The prompter prompt documents the MegaTask
scope + when to call propose_batch. Adds Claude-normalize and grok-relay
tests for the batch path.

* feat(batch): MegaTask intake panel — third scope, batch review, waves

The panel now drives a MegaTask end to end. The intake modal gains a
third scope, 'MegaTask', beside Single cell and Board-led: a multi-project
checklist (a MegaTask spans several possibly-unrelated repos), validated
to at least two. start() sends project_ids; use-prompter accumulates the
agent's single propose_batch hand-off as a 'batch' SSE event into a
BatchProposal and lands in a new batch_preview state.

A new BatchReviewCard lists every proposed task with its target project +
collision-surface badges (migration / shared) and offers one start path
for the whole batch — Board review & Start or Approve & Start — wired to
confirmBatch → POST /confirm-batch. The success card shows the sequenced
result: N tasks in M waves (+ any advisory notes). prompter.ts gains the
DraftScale 'megatask' + the BatchConfirm payload/result types; the SSE
client allows the 'batch' kind. Panel typecheck + lint + 113 tests green.

* docs(batch): MegaTask across changelog, CLAUDE.md, site, and RAG

The four documentation obligations for the MegaTask feature:
- CHANGELOG: an Unreleased entry covering the umbrella model, sequencing,
  multi-project intake, propose_batch, and the create/approval path.
- CLAUDE.md: a MegaTask section (identity predicate, umbrella/root-subtask
  hierarchy, sequencing rules, intake + create path, board activation).
- Published site: a user-facing company/megatask.md (scopes, waves, the
  umbrella, the two start buttons) + nav entry; a pointer added to the
  intake chapter of the Tour.
- RAG corpus: workflows/megatask.md so the Main PM (and any agent) can
  retrieve the umbrella's branchless / no-PR / completion rules at runtime.

The runtime concurrent-migration guard is intentionally NOT added: the
analyzer already chains migration-adders into dependencies and the
dependency-gate serializes them, so a separate guard would be dead code.

* feat(batch): batch_id guardrail + wave preview + batch_id on TaskResponse

Guardrail (CEO): a batch_id is denied on any task that is not a well-formed
MegaTask member. is_valid_batch_shape permits batch_id only on an umbrella
(no parent → must target neither project nor product) or a root-subtask
(has a parent → exactly one target); TaskService.create enforces it AND
verifies a root-subtask's parent is the batch umbrella (same batch_id,
top-level). This closes a latent hole: is_batch_umbrella is true for a
batch_id + no-parent task even with a project, so a stray batch_id could
have spoofed the branchless branch-gate / no-PR exemption. (The public
task API never exposed batch_id for write; this guards the service layer.)

Wave preview: PrompterService.preview_batch + POST .../preview-batch
compute a MegaTask's waves from the proposed drafts WITHOUT creating
anything, so the panel can show the sequencing before confirm. Extracted
_sequence_drafts as the single source shared by preview and confirm, so
the previewed waves are exactly the ones wired.

TaskResponse now carries batch_id so the panel can badge the umbrella.

* feat(batch): MegaTask review — project editor, wave preview, persistence, badge

Closes the panel gaps in the MegaTask review experience:
- Per-task project editor: each proposed task gets an inline project
  Select (updateBatchDraftProject), so a task the agent put in the wrong
  or no repo can be fixed before launch — not only by re-chatting. Launch
  stays blocked until every task has a project.
- Wave preview: on a batch proposal the panel fetches POST .../preview-batch
  (no task created) and shows the conflict-free wave plan, so the human
  reviews the sequencing before confirming.
- Refresh durability: the MegaTask review (batch + waves + projectIds) is
  persisted, so a browser reload mid-review restores it like a single draft.
- MegaTask badge: TaskResponse exposes batch_id, the panel Task type
  carries it, and the task table badges the umbrella row 'MegaTask'.

Panel typecheck + lint + 113 tests green.

* test(batch): stub task carries batch_id for task_to_response

task_to_response now serializes batch_id (TaskResponse field), so the
_stub_task SimpleNamespace fixture must provide it — without it the reader
hit AttributeError, failing the 8 task-schema serialization/enrichment
tests. Test-only; the real TaskTable carries the column (migration 046).

* fix(batch): close MegaTask audit gaps — completion crash, analyzer cycle, guardrails

An adversarial multi-agent audit of the feature surfaced 20 verified gaps;
this closes the backend ones.

HIGH:
- Umbrella completion crashed. escalate_to_ceo hard-required a pr_number,
  which a branchless umbrella never has, so main_pm_complete dereferenced
  None. Both pr_number gates now waive a MegaTask umbrella (escalate_to_ceo
  + the awaiting_pm_review->awaiting_ceo_approval lifecycle gate via a new
  GitContext.is_umbrella), and main_pm_complete guards a None return. The
  completion test had mocked escalate_to_ceo, hiding it — now a real
  service test covers the waiver.
- The collision analyzer could fabricate a cycle (a touches_shared +
  adds_migration draft overlapping another migration draft) and raise
  SequencingError — a bare ValueError that escaped as an opaque 500. The
  migration chain is now shared-last-aware (never contradicts rule 3), and
  _sequence_drafts translates SequencingError to a clean 400.

MEDIUM:
- Collisions are now project-scoped: two repos can't collide on a
  coincidental path or serialize independent migrations (DraftSurface
  carries project_id; rules 1/2/3 respect it).
- The batch_id guardrail ran only at create. update() + the PATCH
  null-clear path now re-assert is_valid_batch_shape, so a mutation can't
  break a member's shape and spoof the branchless exemption.
- A draft missing title/acceptance_criteria now raises ValidationError
  (was a bare KeyError -> 500).
- confirm_live_batch re-asserts every draft targets a scoped project and
  the batch spans >=2 distinct projects (project_ids added to the request).
- Route-level tests for confirm-batch / preview-batch.

LOW: strict multi-repo clone (fail loud on any unresolvable project);
malformed/empty propose_batch surfaces an error chunk (Claude) / refuses
to POST (grok) instead of silently acking; dropped malformed drafts are
counted and surfaced; stale grok intake docstrings updated.

* fix(batch): MegaTask panel + doc audit gaps

Frontend half of the audit fixes:
- The confirm payload now carries project_ids (the schema requires it), and
  the panel re-checks every task targets one of the scoped repos before
  launching, naming the offending task.
- The Review-MegaTask project picker is filtered to the scoped repos and
  the per-task validity (border + launch gate) keys off scoped membership,
  so a task can only be (re)pointed at an in-scope project — also fixing the
  case where the agent emitted a non-UUID / unknown project.
- Dropped malformed drafts are surfaced as a chat error so the human knows
  the batch shrank instead of silently confirming fewer tasks.
- Doc wording: a wave releases on the previous wave's terminal state
  (normally a merge; a cancellation releases it too), not strictly 'merged'.

* test(batch): lock the CEO's EXACT 4-wave hand-sequencing as the golden bar

The golden test asserted the constraints (S6 last, the migration chain, the
shared-threats serialization, S1/S2/S7 parallel) but not the full wave
partition. The bar for MegaTask is 'reproduce my exact waves or it's not
done', so assert the exact 4-wave partition the analyzer produces for the
guard-core-app batch:
  wave 1: R1 R2 S1 S2 S3 S5 S7  ·  wave 2: R3  ·  wave 3: R4 S8  ·  wave 4: S6
Confirmed unchanged by the audit's analyzer fixes (no migration is shared;
single project).

* fix(batch): tolerate a stub task in assert_batch_shape_intact

The batch-shape re-validation read task.batch_id directly, but update()'s
partial-caller contract is exercised with a SimpleNamespace stub that has no
batch_id column → AttributeError. Use getattr(..., None) for batch_id and the
shape fields so the guard no-ops on any task lacking the column (a stub, or a
non-batch task) while still enforcing on a real batch member.

* fix(orchestrator): authenticate internal API self-calls with the system identity

The dispatcher httpx clients were built without an agent identity, so the
orchestrator's self-PATCHes to /api/tasks/{id} (auto-block, auto-resume,
auto-recover, SLA annotation) were rejected 401 "Missing X-Agent-ID" and
silently no-op'd. The auto-resume that lifts a PM's paused parent could never
write, so paused/blocked parents stayed wedged and stranded their dependents
(the fe-pm/be-pm respawn churn seen in prod).

Header propagation was inconsistent across the separate AsyncClient call-sites:
only the main dispatch client carried the system identity; the readiness and
sweep clients did not. Hoist the identity into a shared _SYSTEM_API_HEADERS
constant and apply it to every API-facing dispatcher client. The system role
holds TaskAction.ASSIGN, so it is authorized for the audited admin_set_status
path those write routes use. The external provider-recovery probe client is
intentionally left untouched.

---------

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

tests/integration/test_approve_and_start_service.py

@@ -148,3 +148,63 @@ async def test_succeeds_when_board_review_complete(start_setup: dict) -> None:
     assert out is not None
     assert out.assigned_to == start_setup["main_pm"].id
     assert out.status == TaskStatus.PENDING
+
+
+@pytest.mark.asyncio
+async def test_approving_megatask_umbrella_releases_backlog_root_subtasks(
+    start_setup: dict,
+) -> None:
+    """Approving a board-route MegaTask umbrella releases its held root-subtasks:
+    BACKLOG → PENDING and team → main_pm, so the dependency-gate dispatches them."""
+    db = start_setup["db"]
+    po = start_setup["po"]
+    project_id = start_setup["mk"]().project_id  # reuse the fixture's project
+    batch_id = uuid4()
+    umbrella = TaskTable(
+        id=uuid4(),
+        title="MegaTask: three things",
+        description="d",
+        acceptance_criteria=["ac"],
+        status=TaskStatus.PENDING,
+        priority=1,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=None,  # branchless umbrella: no project, no product
+        product_id=None,
+        batch_id=batch_id,
+        parent_task_id=None,
+        created_by=po.id,
+        team=Team.BOARD,
+        board_review_complete=True,
+        assigned_to=po.id,
+    )
+    db.add(umbrella)
+    await db.flush()
+    subs = [
+        TaskTable(
+            id=uuid4(),
+            title=f"Root-subtask {i}",
+            description="d",
+            acceptance_criteria=["ac"],
+            status=TaskStatus.BACKLOG,  # held until the batch review approves
+            priority=2,
+            task_type=TaskType.CODE,
+            nature=TaskNature.TECHNICAL,
+            project_id=project_id,
+            batch_id=batch_id,
+            parent_task_id=umbrella.id,
+            created_by=po.id,
+            team=Team.BOARD,
+        )
+        for i in range(2)
+    ]
+    db.add_all(subs)
+    await db.flush()
+
+    out = await start_setup["svc"].approve_and_start(umbrella.id, "x" * 25)
+    assert out is not None
+    assert out.assigned_to == start_setup["main_pm"].id
+    for sub in subs:
+        await db.refresh(sub)
+        assert sub.status == TaskStatus.PENDING
+        assert sub.team == Team.MAIN_PM

tests/integration/test_migration_batch_intake.py

@@ -0,0 +1,154 @@
+"""0.11.0 sequenced batch intake: tasks.batch_id + collision descriptor columns.
+
+Migration 046 adds ``tasks.batch_id`` (indexed ``ix_tasks_batch_id``) plus the
+per-task collision surface the SequencingService reads: ``intends_to_touch``
+(text[]), and ``adds_migration`` / ``touches_shared`` (bool, NOT NULL default
+false — a non-batch task declares no surface). The real upgrade/downgrade chain
+is verified separately against a throwaway Postgres; these assertions guard the
+resulting schema shape and a value round-trip.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, ProjectTable, TaskTable
+from roboco.models import AgentRole, AgentStatus, Team
+from roboco.models.base import TaskNature, TaskStatus, TaskType
+from sqlalchemy import text
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_backend_project(
+    db_session: AsyncSession,
+) -> tuple[AgentTable, ProjectTable]:
+    agent = AgentTable(
+        id=uuid4(),
+        name="Dev",
+        slug=f"be-dev-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(agent)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="B-Proj",
+        slug=f"b-proj-{uuid4().hex[:8]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=agent.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    return agent, project
+
+
+@pytest.mark.asyncio
+async def test_batch_columns_and_index_exist(db_session: AsyncSession) -> None:
+    rows = (
+        await db_session.execute(
+            text(
+                "SELECT column_name, is_nullable, column_default "
+                "FROM information_schema.columns "
+                "WHERE table_name = 'tasks' AND column_name IN "
+                "('batch_id', 'intends_to_touch', 'adds_migration', 'touches_shared')"
+            )
+        )
+    ).all()
+    by_name = {r[0]: (r[1], r[2]) for r in rows}
+    assert set(by_name) == {
+        "batch_id",
+        "intends_to_touch",
+        "adds_migration",
+        "touches_shared",
+    }
+    assert by_name["batch_id"][0] == "YES"  # nullable
+    assert by_name["intends_to_touch"][0] == "YES"  # nullable
+    assert by_name["adds_migration"][0] == "NO"  # NOT NULL
+    assert "false" in (by_name["adds_migration"][1] or "")  # default false
+    assert by_name["touches_shared"][0] == "NO"
+    assert "false" in (by_name["touches_shared"][1] or "")
+    idx = (
+        await db_session.execute(
+            text(
+                "SELECT indexname FROM pg_indexes "
+                "WHERE tablename = 'tasks' AND indexname = 'ix_tasks_batch_id'"
+            )
+        )
+    ).first()
+    assert idx is not None, "ix_tasks_batch_id must exist on tasks"
+
+
+@pytest.mark.asyncio
+async def test_batch_fields_round_trip(db_session: AsyncSession) -> None:
+    _, project = await _seed_backend_project(db_session)
+    batch = uuid4()
+    task = TaskTable(
+        id=uuid4(),
+        title="t",
+        description="d",
+        acceptance_criteria=["ac"],
+        status=TaskStatus.PENDING,
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=project.id,
+        created_by=project.created_by,
+        team=Team.BACKEND,
+        batch_id=batch,
+        intends_to_touch=["svc/x.py", "page/y.tsx"],
+        adds_migration=True,
+        touches_shared=True,
+    )
+    db_session.add(task)
+    await db_session.flush()
+    fetched = (
+        await db_session.execute(
+            text(
+                "SELECT batch_id, intends_to_touch, adds_migration, touches_shared "
+                "FROM tasks WHERE id = :id"
+            ),
+            {"id": task.id},
+        )
+    ).first()
+    assert fetched is not None
+    assert fetched[0] == batch
+    assert fetched[1] == ["svc/x.py", "page/y.tsx"]
+    assert fetched[2] is True
+    assert fetched[3] is True
+
+
+@pytest.mark.asyncio
+async def test_non_batch_task_gets_defaults(db_session: AsyncSession) -> None:
+    _, project = await _seed_backend_project(db_session)
+    task = TaskTable(
+        id=uuid4(),
+        title="t",
+        description="d",
+        acceptance_criteria=["ac"],
+        status=TaskStatus.PENDING,
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=project.id,
+        created_by=project.created_by,
+        team=Team.BACKEND,
+    )
+    db_session.add(task)
+    await db_session.flush()
+    await db_session.refresh(task)
+    assert task.batch_id is None
+    assert task.intends_to_touch is None
+    assert task.adds_migration is False
+    assert task.touches_shared is False

tests/integration/test_prompter_live_routes.py

@@ -138,6 +138,7 @@ class _FakeOrchestrator:
         *,
         project_slug: str | None = None,
         product_id: str | None = None,
+        project_ids: list[str] | None = None,
         initial_message: str | None = None,
     ) -> None:
         # The route is non-blocking now: it calls start_intake_session (returns
@@ -147,6 +148,7 @@ class _FakeOrchestrator:
                 "session_id": session_id,
                 "project_slug": project_slug,
                 "product_id": product_id,
+                "project_ids": project_ids,
                 "initial_message": initial_message,
             }
         )
@@ -193,11 +195,27 @@ async def test_start_product_scope_spawns_and_returns_session(
             "session_id": session_id,
             "project_slug": None,
             "product_id": product_id,
+            "project_ids": None,
             "initial_message": "build X",
         }
     ]
 
 
+@pytest.mark.asyncio
+async def test_start_megatask_scope_passes_project_ids(start_client: dict) -> None:
+    client, orch = start_client["client"], start_client["orch"]
+    ids = [str(uuid4()), str(uuid4())]
+
+    resp = await client.post(
+        "/api/prompter/live/start",
+        json={"project_ids": ids, "initial_message": "three repos"},
+    )
+    assert resp.status_code == HTTPStatus.CREATED
+    assert orch.spawned[0]["project_ids"] == ids
+    assert orch.spawned[0]["project_slug"] is None
+    assert orch.spawned[0]["product_id"] is None
+
+
 @pytest.mark.asyncio
 async def test_start_project_scope_resolves_slug(start_client: dict) -> None:
     client, orch = start_client["client"], start_client["orch"]
@@ -346,3 +364,99 @@ async def test_confirm_validation_error_is_translated_and_not_reaped(
         )
     assert resp.status_code == HTTPStatus.BAD_REQUEST
     assert orch.reaped == []  # a failed confirm must NOT reap the session
+
+
+# ---------------------------------------------------------------------------
+# MegaTask batch routes — confirm-batch (terminal, reaps) + preview-batch (pure).
+# ---------------------------------------------------------------------------
+
+
+def _batch_body() -> dict[str, Any]:
+    return {
+        "title": "MegaTask",
+        "drafts": [
+            {"title": "A", "acceptance_criteria": ["a"], "project_id": str(uuid4())},
+            {"title": "B", "acceptance_criteria": ["b"], "project_id": str(uuid4())},
+        ],
+        "project_ids": [str(uuid4()), str(uuid4())],
+        "route": "main_pm",
+    }
+
+
+@pytest.mark.asyncio
+async def test_confirm_batch_creates_and_reaps(confirm_client: dict) -> None:
+    client, orch = confirm_client["client"], confirm_client["orch"]
+    result = {
+        "umbrella_task_id": str(uuid4()),
+        "root_subtask_ids": [str(uuid4()), str(uuid4())],
+        "waves": [[0], [1]],
+        "warnings": [],
+    }
+
+    class _FakeService:
+        async def confirm_live_batch(self, *_a: Any, **_kw: Any) -> Any:
+            return result
+
+    with patch(
+        "roboco.api.routes.prompter_live.get_prompter_service",
+        lambda _db: _FakeService(),
+    ):
+        resp = await client.post(
+            "/api/prompter/live/s1/confirm-batch", json=_batch_body()
+        )
+    assert resp.status_code == HTTPStatus.CREATED
+    assert resp.json() == result
+    assert orch.reaped == ["s1"]  # confirm-batch is terminal → reap
+
+
+@pytest.mark.asyncio
+async def test_confirm_batch_validation_error_not_reaped(confirm_client: dict) -> None:
+    client, orch = confirm_client["client"], confirm_client["orch"]
+
+    class _FakeService:
+        async def confirm_live_batch(self, *_a: Any, **_kw: Any) -> Any:
+            raise ValidationError(message="bad batch", field="drafts")
+
+    with patch(
+        "roboco.api.routes.prompter_live.get_prompter_service",
+        lambda _db: _FakeService(),
+    ):
+        resp = await client.post(
+            "/api/prompter/live/s1/confirm-batch", json=_batch_body()
+        )
+    assert resp.status_code == HTTPStatus.BAD_REQUEST
+    assert orch.reaped == []  # a failed confirm must NOT reap
+
+
+@pytest.mark.asyncio
+async def test_confirm_batch_schema_rejects_too_few_projects(
+    confirm_client: dict,
+) -> None:
+    client = confirm_client["client"]
+    body = _batch_body()
+    body["project_ids"] = [str(uuid4())]  # < 2 → schema 422
+    resp = await client.post("/api/prompter/live/s1/confirm-batch", json=body)
+    assert resp.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+
+
+@pytest.mark.asyncio
+async def test_preview_batch_returns_waves_and_does_not_reap(
+    confirm_client: dict,
+) -> None:
+    client, orch = confirm_client["client"], confirm_client["orch"]
+
+    class _FakeService:
+        def preview_batch(self, _drafts: Any) -> Any:
+            return {"waves": [[0, 1]], "warnings": []}
+
+    with patch(
+        "roboco.api.routes.prompter_live.get_prompter_service",
+        lambda _db: _FakeService(),
+    ):
+        resp = await client.post(
+            "/api/prompter/live/s1/preview-batch",
+            json={"drafts": [{"title": "A"}, {"title": "B"}]},
+        )
+    assert resp.status_code == HTTPStatus.OK
+    assert resp.json() == {"waves": [[0, 1]], "warnings": []}
+    assert orch.reaped == []  # preview creates nothing and leaves the chat alive

tests/integration/test_task_service_misc.py

@@ -32,7 +32,7 @@ from uuid import UUID, uuid4
 
 import pytest
 import pytest_asyncio
-from roboco.db.tables import AgentTable, ProjectTable, WorkSessionTable
+from roboco.db.tables import AgentTable, ProjectTable, TaskTable, WorkSessionTable
 from roboco.enforcement import TaskLifecycleError
 from roboco.foundation.policy.content import markers
 from roboco.models import AgentRole, AgentStatus, Team
@@ -297,6 +297,81 @@ async def test_ensure_branch_no_project_id_raises(
         await svc._ensure_branch_for_task(task, task_setup["agent_id"])
 
 
+@pytest.mark.asyncio
+async def test_ensure_branch_batch_umbrella_returns_empty(
+    task_setup: dict,
+) -> None:
+    """A MegaTask umbrella (batch_id, top-level, no project/product) is branchless
+    by design — it must short-circuit to "" rather than hit the misconfigured
+    raise, so the claim path treats it as coordination, not a defect."""
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup))
+    task.project_id = None
+    task.product_id = None
+    task.batch_id = uuid4()
+    task.parent_task_id = None
+    out = await svc._ensure_branch_for_task(task, task_setup["agent_id"])
+    assert out == ""
+
+
+@pytest.mark.asyncio
+async def test_create_denies_stray_batch_id_on_targeted_top_level_task(
+    task_setup: dict,
+) -> None:
+    """Guardrail: a top-level task that targets a project must NOT carry a
+    batch_id — that umbrella-shaped-but-targeted task would otherwise spoof the
+    branchless exemption. create() refuses it."""
+    svc = task_setup["svc"]
+    # _req sets project_id; adding batch_id with no parent is the spoof shape.
+    with pytest.raises(ValueError, match="batch_id is only valid"):
+        await svc.create(_req(task_setup, batch_id=uuid4()))
+
+
+@pytest.mark.asyncio
+async def test_create_denies_batch_root_subtask_under_non_umbrella_parent(
+    task_setup: dict,
+) -> None:
+    """Guardrail: a batch root-subtask's parent must be the batch umbrella. A
+    child pointed at a normal (non-umbrella) parent, or a mismatched batch, is
+    refused."""
+    svc = task_setup["svc"]
+    parent = await svc.create(_req(task_setup))  # a normal task, no batch_id
+    with pytest.raises(ValueError, match="parent must be the batch umbrella"):
+        await svc.create(_req(task_setup, batch_id=uuid4(), parent_task_id=parent.id))
+
+
+@pytest.mark.asyncio
+async def test_update_rejects_breaking_batch_umbrella_shape(
+    task_setup: dict,
+) -> None:
+    """Guardrail completeness: update() re-validates the MegaTask shape, so a
+    PATCH that adds a project to a branchless umbrella (which would spoof the
+    branch-gate / no-PR exemption) is refused — the invariant the branchless
+    predicates trust is enforced on mutation, not only at create."""
+    db = task_setup["db"]
+    svc = task_setup["svc"]
+    umbrella = TaskTable(
+        id=uuid4(),
+        title="MegaTask",
+        description="d",
+        acceptance_criteria=["ac"],
+        status=TaskStatus.PENDING,
+        priority=1,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=None,  # a valid umbrella targets neither
+        product_id=None,
+        batch_id=uuid4(),
+        parent_task_id=None,
+        team=Team.MAIN_PM,
+        created_by=task_setup["agent_id"],
+    )
+    db.add(umbrella)
+    await db.flush()
+    with pytest.raises(ValueError, match="MegaTask shape"):
+        await svc.update(umbrella.id, project_id=task_setup["project_id"])
+
+
 @pytest.mark.asyncio
 async def test_auto_create_branch_no_project_raises(
     task_setup: dict,

tests/integration/test_task_service_transitions.py

@@ -145,6 +145,32 @@ async def test_start_with_plan_advances_to_in_progress(
     assert started.started_at is not None
 
 
+@pytest.mark.asyncio
+async def test_start_batch_umbrella_advances_without_branch(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """A MegaTask umbrella (batch_id, no project/product, no branch) is branchless
+    coordination — the claimed->in_progress branch gate must be skipped for it so
+    it can reach in_progress and delegate, exactly like a product fan-out root."""
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.CLAIMED
+    task.assigned_to = task_setup["agent_id"]
+    task.project_id = None
+    task.product_id = None
+    task.batch_id = uuid4()
+    task.parent_task_id = None
+    task.branch_name = None
+    task.plan = {"text": "sequence the waves"}
+    await db_session.flush()
+    started = await svc.start(
+        task.id, agent_id=task_setup["agent_id"], agent_role="developer"
+    )
+    assert started is not None
+    assert started.status == TaskStatus.IN_PROGRESS
+    assert started.branch_name is None
+
+
 @pytest.mark.asyncio
 async def test_start_returns_none_when_ownership_fails(
     task_setup: dict, db_session: AsyncSession
@@ -694,6 +720,49 @@ async def test_ceo_reject_routes_coordination_task_to_main_pm(
     assert rejected.claimed_by is None
 
 
+@pytest.mark.asyncio
+async def test_ceo_reject_routes_batch_umbrella_to_main_pm(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """A rejected MegaTask umbrella (batch_id, top-level, no project/product) is a
+    branchless coordination root too — it routes to the Main PM to re-plan, never
+    to a developer (needs_revision would deadlock the Main PM that owns it)."""
+    svc = task_setup["svc"]
+    main_pm_id = UUID(AGENT_UUIDS["main-pm"])
+    if await db_session.get(AgentTable, main_pm_id) is None:
+        db_session.add(
+            AgentTable(
+                id=main_pm_id,
+                name="Main PM",
+                slug="main-pm",
+                role=AgentRole.MAIN_PM,
+                team=Team.MAIN_PM,
+                status=AgentStatus.ACTIVE,
+                model_config={},
+                system_prompt="pm",
+                capabilities=[],
+                permissions={},
+                metrics={},
+            )
+        )
+        await db_session.flush()
+
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.AWAITING_CEO_APPROVAL
+    task.project_id = None  # umbrella: no project, no product — carries a batch_id
+    task.product_id = None
+    task.batch_id = uuid4()
+    task.parent_task_id = None
+    await db_session.flush()
+
+    rejected = await svc.ceo_reject(task.id, reason="re-sequence the waves")
+    assert rejected is not None
+    assert rejected.status == TaskStatus.PENDING
+    assert rejected.team == Team.MAIN_PM
+    assert rejected.assigned_to == main_pm_id
+    assert rejected.claimed_by is None
+
+
 @pytest.mark.asyncio
 async def test_ceo_reject_writes_handoff_journal(
     task_setup: dict, db_session: AsyncSession
@@ -786,6 +855,28 @@ async def test_escalate_to_ceo_returns_none_when_no_pr(
     assert out is None
 
 
+@pytest.mark.asyncio
+async def test_escalate_to_ceo_waives_pr_for_batch_umbrella(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """A MegaTask umbrella is branchless and assembles no PR — escalate_to_ceo
+    must NOT block it on a missing pr_number, or umbrella completion crashes."""
+    svc = task_setup["svc"]
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.AWAITING_PM_REVIEW
+    task.project_id = None  # umbrella: branchless, no repo, no PR
+    task.product_id = None
+    task.batch_id = uuid4()
+    task.parent_task_id = None
+    task.pr_number = None
+    await db_session.flush()
+    escalated = await svc.escalate_to_ceo(
+        task.id, agent_role="main_pm", notes="MegaTask ready for CEO sign-off"
+    )
+    assert escalated is not None
+    assert escalated.status == TaskStatus.AWAITING_CEO_APPROVAL
+
+
 @pytest.mark.asyncio
 async def test_escalate_to_ceo_advances_status_with_notes(
     task_setup: dict, db_session: AsyncSession