feat(board): materialize program items as Main-PM roots, make reports actionable (#711)

Two coupled gaps in the Board Program output path.

Approved items were created unowned and in BACKLOG. Nothing dispatches
BACKLOG, and once activated a cell PM claimed the parentless task as a root,
where _cell_pm_complete resolves its merge target through
resolve_parent_branch — which for a parentless task falls through to the
project head rung. The result was a cell branch merging straight into the
trunk, bypassing the Main-PM root, the root->master PR and the CEO gate
(live: PRs #703 and #704 both targeted slave directly).

All eight materializers now create a PENDING, main-pm-assigned root with
team=Team.MAIN_PM, matching what approve_and_start does for an intake draft.
The team is load-bearing, not cosmetic: _next_hint_pr_fail,
_deliver_pr_fail_to_owner, delegate's wave-chain dispatch and the PR layer
label all key on it, and a cell-teamed root drops the 'do NOT re-submit the
root' steer that exists because of PR #138's infinite pr_fail loop. The
item's own cell survives as a delegation hint in the description, which is
what the Main PM's briefing renders.

Periscope, Sentinel and Coroner produced artifacts with no way to act on
them — three panel surfaces carried explicit 'no approve/reject UI' comments
while each item already held a machine-readable suggested action. They now
have per-item approve and dismiss, modelled on the roadmap queue: idempotent
per item, CEO-gated, deep-copy-before-mutate so SQLAlchemy's dirty check
still fires, and every decision recorded through record_decision so it
reaches the next cycle's prompt. Approving materializes through the same
corrected Main-PM-owned path.

Target project resolves to each engine's own existing anchor — RoboCo's
project for Periscope and Sentinel, the incident's project for Coroner — and
fails with a clean invalid_state naming what is unresolvable rather than
guessing at a repo.

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

tests/unit/gateway/test_content_actions_coroner.py

@@ -200,6 +200,10 @@ async def test_propose_postmortem_completes_the_task_and_stamps_marker() -> None
     assert payload is not None
     assert payload["failed_stage"] == "awaiting_qa"
     assert payload["process_change"]["kind"] == "prompt_fix"
+    # A non-playbook process change stays "proposed" — the CEO's per-item
+    # approve/dismiss decision (CoronerService) is still open.
+    assert payload["process_change"]["status"] == "proposed"
+    assert payload["process_change"]["materialized_task_id"] is None
     assert payload["playbook_id"] is None
     assert task.status == TaskStatus.COMPLETED
 
@@ -244,6 +248,9 @@ async def test_propose_postmortem_drafts_playbook_when_kind_is_playbook() -> Non
     playbook_svc.draft.assert_awaited_once()
     payload = engine.complete_with_postmortem.await_args.args[1]
     assert payload["playbook_id"] == str(drafted.id)
+    # A "playbook" kind already routed into the curation queue above —
+    # CoronerService refuses to act on it (see its own test module).
+    assert payload["process_change"]["status"] == "not_applicable"
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_content_actions_periscope.py

@@ -424,6 +424,11 @@ async def test_propose_market_brief_persists_and_completes_the_exploration_task(
     assert payload["headline"] == "A rival tool shipped agentic PR review this week"
     assert len(payload["findings"]) == len(findings)
     assert payload["findings"][0]["id"] == "finding-0"
+    # Each finding still carries its own per-item CEO decision (Periscope
+    # Service.approve_finding/reject_finding) even though the exploration
+    # task completes here.
+    assert payload["findings"][0]["status"] == "proposed"
+    assert payload["findings"][0]["materialized_task_id"] is None
     assert payload["threats"] == ["Feature parity gap"]
     assert payload["opportunities"] == ["Lean into structured findings"]
     assert payload["positioning_note"] == "Emphasize the findings ledger in messaging"

tests/unit/gateway/test_content_actions_sentinel.py

@@ -393,6 +393,11 @@ async def test_propose_quality_report_persists_and_completes_the_exploration_tas
     assert len(payload["items"]) == len(items)
     assert payload["items"][0]["id"] == "item-0"
     assert payload["items"][0]["area"] == "waivers"
+    # Each item still carries its own per-item CEO decision (SentinelService.
+    # approve_item/reject_item) even though the exploration task completes
+    # here.
+    assert payload["items"][0]["status"] == "proposed"
+    assert payload["items"][0]["materialized_task_id"] is None
     assert (
         payload["overall_assessment"]
         == "Drift is concentrated in one hotspot, not systemic"

tests/unit/services/test_coroner_service.py

@@ -0,0 +1,516 @@
+"""CoronerService coverage: approve materializes the postmortem's ONE
+process change as a Main-PM-owned root task (idempotent), reject records a
+reason (idempotent). Unlike Periscope/Sentinel there is no item id — a
+postmortem is one process change, not a list — and the target project
+resolves against the INCIDENT task's own project (falling back to RoboCo's),
+not a per-item project_slug.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any, cast
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+from roboco.config import settings as cfg
+from roboco.db.tables import (
+    AgentTable,
+    BoardProgramCycleTable,
+    ProjectTable,
+    SystemSettingTable,
+    TaskTable,
+)
+from roboco.foundation import identity as _foundation
+from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
+from roboco.models.base import AgentRole, AgentStatus, Complexity, Team
+from roboco.models.base import TaskNature as TN
+from roboco.models.base import TaskStatus as TS
+from roboco.models.base import TaskType as TT
+from roboco.services import board_programs as bp_module
+from roboco.services.coroner_service import CoronerService, get_coroner_service
+from roboco.services.task import CORONER_ITEM_SOURCE, CORONER_SOURCE
+from sqlalchemy import delete, select, update
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+AUDITOR_UUID = _foundation.AGENTS["auditor"].uuid
+CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+ROBOCO_SLUG = "roboco-standin"
+INCIDENT_SLUG = "customer-app"
+ONE = 1
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _purge_board_program_pollution(db_session: AsyncSession) -> None:
+    """See test_board_program_engine.py's identical fixture."""
+    await db_session.execute(
+        delete(SystemSettingTable).where(SystemSettingTable.key.like("board_program.%"))
+    )
+    await db_session.execute(delete(BoardProgramCycleTable))
+    await db_session.execute(
+        update(TaskTable)
+        .where(
+            TaskTable.source == CORONER_SOURCE,
+            TaskTable.status.notin_([TS.COMPLETED, TS.CANCELLED]),
+        )
+        .values(status=TS.CANCELLED)
+    )
+    await db_session.commit()
+
+
+def _process_change(
+    *, kind: str = "conventions_rule", status: str | None = "proposed"
+) -> dict:
+    change: dict[str, Any] = {
+        "kind": kind,
+        "description": "Add a venv-freshness check to make quality",
+    }
+    if status is not None:
+        change["status"] = status
+        change["reject_reason"] = None
+        change["materialized_task_id"] = None
+    return change
+
+
+async def _seed_agents(session: AsyncSession) -> None:
+    for uuid, slug, role, team in (
+        (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
+        (AUDITOR_UUID, "auditor", AgentRole.AUDITOR, Team.BOARD),
+        (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
+    ):
+        if await session.get(AgentTable, uuid) is None:
+            session.add(
+                AgentTable(
+                    id=uuid,
+                    name=slug,
+                    slug=slug,
+                    role=role,
+                    team=team,
+                    status=AgentStatus.ACTIVE,
+                    model_config={},
+                    system_prompt="x",
+                    capabilities=[],
+                    permissions={},
+                    metrics={},
+                )
+            )
+    await session.flush()
+
+
+async def _seed_roboco_project(
+    session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> ProjectTable:
+    await _seed_agents(session)
+    project = ProjectTable(
+        id=uuid4(),
+        name="RoboCo",
+        slug=ROBOCO_SLUG,
+        git_url="https://example.com/roboco.git",
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+    )
+    session.add(project)
+    await session.flush()
+    monkeypatch.setattr(cfg, "self_heal_project_slug", ROBOCO_SLUG)
+    return project
+
+
+async def _seed_incident(session: AsyncSession, *, project: ProjectTable) -> TaskTable:
+    """A real incident task on its OWN project/team — distinct from any
+    RoboCo fallback project, so a test asserting the incident's project/team
+    wins can't accidentally pass via the fallback instead."""
+    await _seed_agents(session)
+    incident = TaskTable(
+        id=uuid4(),
+        title="Fix worktree venv rot",
+        description="x",
+        acceptance_criteria=["x"],
+        status=TS.NEEDS_REVISION,
+        priority=2,
+        task_type=TT.CODE,
+        nature=TN.TECHNICAL,
+        estimated_complexity=Complexity.MEDIUM,
+        created_by=SYSTEM_UUID,
+        team=Team.FRONTEND,
+        source="manual",
+        confirmed_by_human=True,
+        project_id=project.id,
+        revision_count=3,
+    )
+    session.add(incident)
+    await session.flush()
+    return incident
+
+
+async def _seed_incident_project(session: AsyncSession) -> ProjectTable:
+    await _seed_agents(session)
+    project = ProjectTable(
+        id=uuid4(),
+        name="Customer App",
+        slug=INCIDENT_SLUG,
+        git_url="https://example.com/customer-app.git",
+        assigned_cell=Team.FRONTEND,
+        created_by=SYSTEM_UUID,
+    )
+    session.add(project)
+    await session.flush()
+    return project
+
+
+async def _seed_postmortem(
+    session: AsyncSession,
+    *,
+    incident: TaskTable | None,
+    process_change: dict | None = None,
+    postmortem_project_id: object | None = None,
+) -> TaskTable:
+    await _seed_agents(session)
+    task = TaskTable(
+        id=uuid4(),
+        title="Coroner postmortem",
+        description="Autopsy the chronic task.",
+        acceptance_criteria=["propose_postmortem() called once"],
+        status=TS.COMPLETED,
+        priority=2,
+        task_type=TT.ADMINISTRATIVE,
+        nature=TN.NON_TECHNICAL,
+        estimated_complexity=Complexity.LOW,
+        created_by=SYSTEM_UUID,
+        assigned_to=AUDITOR_UUID,
+        team=Team.BOARD,
+        source=CORONER_SOURCE,
+        confirmed_by_human=False,
+        project_id=postmortem_project_id,
+    )
+    session.add(task)
+    await session.flush()
+    if incident is not None:
+        markers.set_coroner_incident(
+            task,
+            {
+                "incident_task_id": str(incident.id),
+                "kind": "bounced",
+                "revision_count": incident.revision_count or 0,
+                "title": incident.title,
+            },
+        )
+    markers.set_coroner_postmortem(
+        task,
+        {
+            "incident_summary": "the task bounced 3 times over a stale venv",
+            "root_cause": "the gate never verified the venv's dev extras",
+            "failed_stage": "awaiting_qa",
+            "process_change": process_change or _process_change(),
+            "playbook_id": None,
+        },
+    )
+    await session.flush()
+    return task
+
+
+def _svc(session: AsyncSession) -> CoronerService:
+    return get_coroner_service(session)
+
+
+def _id(task: TaskTable) -> UUID:
+    return cast("UUID", task.id)
+
+
+@pytest.mark.asyncio
+async def test_approve_materializes_main_pm_owned_task_on_incident_project(
+    db_session: AsyncSession,
+) -> None:
+    """The target project/team is the INCIDENT's own — not the postmortem
+    task's own project_id (left None here) and not a RoboCo fallback."""
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+
+    result = await _svc(db_session).approve_process_change(
+        _id(task), created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"
+    assert result.materialized_task_id is not None
+
+    materialized = await db_session.get(TaskTable, result.materialized_task_id)
+    assert materialized is not None
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
+    assert materialized.source == CORONER_ITEM_SOURCE
+    assert materialized.project_id == incident_project.id
+    # team is forced to Team.MAIN_PM (not the incident's own cell) — see
+    # test_roadmap_service.py's identical assertion for why: every "is this
+    # a coordination root" consumer keys on team, not assigned_to.
+    assert materialized.team == Team.MAIN_PM
+    # main_pm can never own a code task — see test_roadmap_service.py's
+    # identical assertion for the coercion rationale.
+    assert materialized.task_type == TT.PLANNING
+    # The incident's own cell (Team.FRONTEND) survives as a Notes delegation
+    # hint instead of the materialized task's team column.
+    assert "frontend cell" in (materialized.description or "")
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
+
+    await db_session.refresh(task)
+    payload = markers.get_coroner_postmortem(task)
+    assert payload is not None
+    assert payload["process_change"]["status"] == "approved"
+    assert payload["process_change"]["materialized_task_id"] == str(
+        result.materialized_task_id
+    )
+    assert task.status == TS.COMPLETED
+
+
+@pytest.mark.asyncio
+async def test_approve_falls_back_to_roboco_project_when_incident_gone(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    roboco_project = await _seed_roboco_project(db_session, monkeypatch)
+    # coroner_incident references an incident id that no longer resolves.
+    task = await _seed_postmortem(db_session, incident=None)
+    markers.set_coroner_incident(
+        task,
+        {
+            "incident_task_id": str(uuid4()),
+            "kind": "cancelled",
+            "revision_count": 0,
+            "title": "gone",
+        },
+    )
+    await db_session.flush()
+
+    result = await _svc(db_session).approve_process_change(
+        _id(task), created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"
+    materialized = await db_session.get(TaskTable, result.materialized_task_id)
+    assert materialized is not None
+    assert materialized.project_id == roboco_project.id
+    # team is forced to Team.MAIN_PM regardless of the fallback team
+    # (Team.BACKEND) _resolve_target reports when the incident is gone.
+    assert materialized.team == Team.MAIN_PM
+    assert "backend cell" in (materialized.description or "")
+
+
+@pytest.mark.asyncio
+async def test_approve_is_idempotent(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+    svc = _svc(db_session)
+    first = await svc.approve_process_change(_id(task), created_by=CEO_UUID)
+    second = await svc.approve_process_change(_id(task), created_by=CEO_UUID)
+    assert first is not None
+    assert second is not None
+    assert second.status == "already_approved"
+    assert second.materialized_task_id == first.materialized_task_id
+
+    result = await db_session.execute(
+        select(TaskTable).where(TaskTable.source == CORONER_ITEM_SOURCE)
+    )
+    assert len(result.scalars().all()) == ONE
+
+
+@pytest.mark.asyncio
+async def test_reject_records_reason(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+    result = await _svc(db_session).reject_process_change(
+        _id(task), "one-off incident, not worth a standing rule"
+    )
+    assert result is not None
+    assert result.status == "rejected"
+
+    await db_session.refresh(task)
+    payload = markers.get_coroner_postmortem(task)
+    assert payload is not None
+    assert payload["process_change"]["status"] == "rejected"
+    assert (
+        payload["process_change"]["reject_reason"]
+        == "one-off incident, not worth a standing rule"
+    )
+
+
+@pytest.mark.asyncio
+async def test_reject_is_idempotent(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+    svc = _svc(db_session)
+    await svc.reject_process_change(_id(task), "reason one")
+    second = await svc.reject_process_change(_id(task), "reason two")
+    assert second is not None
+    assert second.status == "already_rejected"
+
+
+@pytest.mark.asyncio
+async def test_cannot_reject_an_approved_change(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+    svc = _svc(db_session)
+    await svc.approve_process_change(_id(task), created_by=CEO_UUID)
+    result = await svc.reject_process_change(_id(task), "changed my mind")
+    assert result is not None
+    assert result.status == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_cannot_approve_a_rejected_change(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+    svc = _svc(db_session)
+    await svc.reject_process_change(_id(task), "not now")
+    result = await svc.approve_process_change(_id(task), created_by=CEO_UUID)
+    assert result is not None
+    assert result.status == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_playbook_kind_refuses_approve(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(
+        db_session,
+        incident=incident,
+        process_change=_process_change(kind="playbook", status="not_applicable"),
+    )
+    result = await _svc(db_session).approve_process_change(
+        _id(task), created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "invalid_state"
+    assert "already drafted as a playbook" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_playbook_kind_refuses_reject(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(
+        db_session,
+        incident=incident,
+        process_change=_process_change(kind="playbook", status="not_applicable"),
+    )
+    result = await _svc(db_session).reject_process_change(_id(task), "no thanks")
+    assert result is not None
+    assert result.status == "invalid_state"
+    assert "already drafted as a playbook" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_process_change_with_no_status_key_defaults_to_proposed(
+    db_session: AsyncSession,
+) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    legacy_change = _process_change(status=None)
+    assert "status" not in legacy_change
+    task = await _seed_postmortem(
+        db_session, incident=incident, process_change=legacy_change
+    )
+    result = await _svc(db_session).approve_process_change(
+        _id(task), created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"
+
+
+@pytest.mark.asyncio
+async def test_approve_unresolvable_project_is_invalid_state(
+    db_session: AsyncSession,
+) -> None:
+    """Incident gone AND no RoboCo project seeded — fails cleanly instead of
+    guessing a project."""
+    task = await _seed_postmortem(db_session, incident=None)
+    markers.set_coroner_incident(
+        task,
+        {
+            "incident_task_id": str(uuid4()),
+            "kind": "cancelled",
+            "revision_count": 0,
+            "title": "gone",
+        },
+    )
+    await db_session.flush()
+    result = await _svc(db_session).approve_process_change(
+        _id(task), created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "invalid_state"
+    assert "cannot anchor a materialized task" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_unknown_task_returns_none(db_session: AsyncSession) -> None:
+    result = await _svc(db_session).approve_process_change(uuid4(), created_by=CEO_UUID)
+    assert result is None
+
+
+async def _seed_cycle_ledger_row(session: AsyncSession, task: TaskTable) -> None:
+    session.add(
+        BoardProgramCycleTable(
+            program_key="coroner",
+            exploration_task_id=task.id,
+            opened_at=datetime.now(UTC),
+        )
+    )
+    await session.flush()
+
+
+@pytest.mark.asyncio
+async def test_approve_records_learn_decision(db_session: AsyncSession) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+    await _seed_cycle_ledger_row(db_session, task)
+    await _svc(db_session).approve_process_change(_id(task), created_by=CEO_UUID)
+
+    row = (
+        await db_session.execute(
+            select(BoardProgramCycleTable).where(
+                BoardProgramCycleTable.program_key == "coroner"
+            )
+        )
+    ).scalar_one()
+    assert row.items_approved == ONE
+    decision = row.decisions[0]
+    assert decision["verdict"] == "approved"
+    assert decision["item_ref"] == "Add a venv-freshness check to make quality"
+
+
+@pytest.mark.asyncio
+async def test_approve_survives_learn_recording_failure(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    incident_project = await _seed_incident_project(db_session)
+    incident = await _seed_incident(db_session, project=incident_project)
+    task = await _seed_postmortem(db_session, incident=incident)
+    await _seed_cycle_ledger_row(db_session, task)
+
+    async def _boom(_self: object, *_args: object, **_kwargs: object) -> None:
+        raise RuntimeError("learn boom")
+
+    monkeypatch.setattr(bp_module.BoardProgramEngine, "record_decision", _boom)
+    result = await _svc(db_session).approve_process_change(
+        _id(task), created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"

tests/unit/services/test_dogfood_service.py

@@ -25,6 +25,7 @@ from roboco.db.tables import (
 )
 from roboco.foundation import identity as _foundation
 from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
 from roboco.models.base import (
     AgentRole,
     AgentStatus,
@@ -56,6 +57,7 @@ if TYPE_CHECKING:
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 PO_UUID = _foundation.AGENTS["product-owner"].uuid
 CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
 ONE = 1
 TWO = 2
 
@@ -109,6 +111,7 @@ async def _seed_agents(session: AsyncSession) -> None:
         (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
         (PO_UUID, "product-owner", AgentRole.PRODUCT_OWNER, Team.BOARD),
         (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
     ):
         if await session.get(AgentTable, uuid) is None:
             session.add(
@@ -187,7 +190,12 @@ def _id(task: TaskTable) -> UUID:
 
 
 @pytest.mark.asyncio
-async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> None:
+async def test_approve_materializes_main_pm_owned_task(
+    db_session: AsyncSession,
+) -> None:
+    """Defect fix: mirrors test_roadmap_service.py's identical assertion
+    update — approval materializes PENDING + assigned_to=main-pm, never an
+    unowned BACKLOG task (see RoadmapService._materialize's docstring)."""
     await _seed_project(db_session, "frontend-app")
     task = await _seed_cycle(db_session, project_slug="frontend-app")
     result = await _svc(db_session).approve_item(
@@ -199,9 +207,24 @@ async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> No
 
     materialized = await db_session.get(TaskTable, result.materialized_task_id)
     assert materialized is not None
-    assert materialized.status == TS.BACKLOG
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
     assert materialized.source == DOGFOOD_ITEM_SOURCE
-    assert materialized.team == Team.FRONTEND
+    # team is forced to Team.MAIN_PM (not the item's own cell) — see
+    # test_roadmap_service.py's identical assertion for why: every "is this
+    # a coordination root" consumer keys on team, not assigned_to.
+    assert materialized.team == Team.MAIN_PM
+    # main_pm can never own a code task — see test_roadmap_service.py's
+    # identical assertion for the coercion rationale.
+    assert materialized.task_type == TT.PLANNING
+    # The item's own cell survives as a Notes delegation hint instead.
+    assert "frontend cell" in (materialized.description or "")
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
 
     await db_session.refresh(task)
     payload = markers.get_friction_fixes(task)

tests/unit/services/test_mirror_service.py

@@ -25,6 +25,7 @@ from roboco.db.tables import (
 )
 from roboco.foundation import identity as _foundation
 from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
 from roboco.models.base import (
     AgentRole,
     AgentStatus,
@@ -57,6 +58,7 @@ if TYPE_CHECKING:
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 HOM_UUID = _foundation.AGENTS["head-marketing"].uuid
 CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
 ONE = 1
 TWO = 2
 
@@ -111,6 +113,7 @@ async def _seed_agents(session: AsyncSession) -> None:
         (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
         (HOM_UUID, "head-marketing", AgentRole.HEAD_MARKETING, Team.BOARD),
         (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
     ):
         if await session.get(AgentTable, uuid) is None:
             session.add(
@@ -189,7 +192,12 @@ def _id(task: TaskTable) -> UUID:
 
 
 @pytest.mark.asyncio
-async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> None:
+async def test_approve_materializes_main_pm_owned_task(
+    db_session: AsyncSession,
+) -> None:
+    """Defect fix: mirrors test_roadmap_service.py's identical assertion
+    update — approval materializes PENDING + assigned_to=main-pm, never an
+    unowned BACKLOG task (see RoadmapService._materialize's docstring)."""
     await _seed_project(db_session, "backend-svc")
     task = await _seed_cycle(db_session, project_slug="backend-svc")
     result = await _svc(db_session).approve_item(
@@ -201,10 +209,22 @@ async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> No
 
     materialized = await db_session.get(TaskTable, result.materialized_task_id)
     assert materialized is not None
-    assert materialized.status == TS.BACKLOG
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
     assert materialized.source == MIRROR_ITEM_SOURCE
     assert materialized.task_type == TT.DOCUMENTATION
-    assert materialized.team == Team.BACKEND
+    # team is forced to Team.MAIN_PM (not the item's own cell) — see
+    # test_roadmap_service.py's identical assertion for why: every "is this
+    # a coordination root" consumer keys on team, not assigned_to.
+    assert materialized.team == Team.MAIN_PM
+    # The item's own cell survives as a Notes delegation hint instead.
+    assert "backend cell" in (materialized.description or "")
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
 
     await db_session.refresh(task)
     payload = markers.get_messaging_fixes(task)

tests/unit/services/test_periscope_service.py

@@ -0,0 +1,410 @@
+"""PeriscopeService coverage: per-finding approve materializes a Main-PM-
+owned root task (idempotent), reject records a reason (idempotent). Unlike
+RoadmapService the exploration task is ALREADY COMPLETED (complete-at-
+propose) — there is no cycle-completion transition to test, only the
+finding's own status.
+
+Mirrors test_roadmap_service.py's per-item shape, adapted for: no
+project_slug on the item (resolves against the RoboCo project instead), and
+a finding authored before this feature shipped carrying no status key at all
+(setdefault, not a hard requirement).
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any, cast
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+from roboco.config import settings as cfg
+from roboco.db.tables import (
+    AgentTable,
+    BoardProgramCycleTable,
+    ProjectTable,
+    SystemSettingTable,
+    TaskTable,
+)
+from roboco.foundation import identity as _foundation
+from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
+from roboco.models.base import AgentRole, AgentStatus, Complexity, Team
+from roboco.models.base import TaskNature as TN
+from roboco.models.base import TaskStatus as TS
+from roboco.models.base import TaskType as TT
+from roboco.services import board_programs as bp_module
+from roboco.services.periscope_service import PeriscopeService, get_periscope_service
+from roboco.services.task import PERISCOPE_ITEM_SOURCE, PERISCOPE_SOURCE
+from sqlalchemy import delete, select, update
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+HOM_UUID = _foundation.AGENTS["head-marketing"].uuid
+CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+ROBOCO_SLUG = "roboco-standin"
+ONE = 1
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _purge_board_program_pollution(db_session: AsyncSession) -> None:
+    """See test_board_program_engine.py's identical fixture."""
+    await db_session.execute(
+        delete(SystemSettingTable).where(SystemSettingTable.key.like("board_program.%"))
+    )
+    await db_session.execute(delete(BoardProgramCycleTable))
+    await db_session.execute(
+        update(TaskTable)
+        .where(
+            TaskTable.source == PERISCOPE_SOURCE,
+            TaskTable.status.notin_([TS.COMPLETED, TS.CANCELLED]),
+        )
+        .values(status=TS.CANCELLED)
+    )
+    await db_session.commit()
+
+
+def _finding(idx: int, *, status: str | None = "proposed") -> dict:
+    finding: dict[str, Any] = {
+        "id": f"finding-{idx}",
+        "claim": f"Competitor {idx} shipped an autonomous review agent",
+        "source_url": f"https://example.com/competitor-{idx}",
+        "relevance": f"Overlaps our pr_reviewer role, finding {idx}",
+    }
+    if status is not None:
+        finding["status"] = status
+        finding["reject_reason"] = None
+        finding["materialized_task_id"] = None
+    return finding
+
+
+async def _seed_agents(session: AsyncSession) -> None:
+    for uuid, slug, role, team in (
+        (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
+        (HOM_UUID, "head-marketing", AgentRole.HEAD_MARKETING, Team.BOARD),
+        (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
+    ):
+        if await session.get(AgentTable, uuid) is None:
+            session.add(
+                AgentTable(
+                    id=uuid,
+                    name=slug,
+                    slug=slug,
+                    role=role,
+                    team=team,
+                    status=AgentStatus.ACTIVE,
+                    model_config={},
+                    system_prompt="x",
+                    capabilities=[],
+                    permissions={},
+                    metrics={},
+                )
+            )
+    await session.flush()
+
+
+async def _seed_roboco_project(
+    session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> ProjectTable:
+    """The org's own project — the RoboCo project resolution anchor every
+    Periscope/Sentinel/Coroner materialization falls back to. Mirrors
+    test_periscope_engine.py's ``_seed``/``_arm`` shape."""
+    await _seed_agents(session)
+    project = ProjectTable(
+        id=uuid4(),
+        name="RoboCo",
+        slug=ROBOCO_SLUG,
+        git_url="https://example.com/roboco.git",
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+    )
+    session.add(project)
+    await session.flush()
+    monkeypatch.setattr(cfg, "self_heal_project_slug", ROBOCO_SLUG)
+    return project
+
+
+async def _seed_brief(
+    session: AsyncSession, *, findings: list[dict] | None = None
+) -> TaskTable:
+    await _seed_agents(session)
+    task = TaskTable(
+        id=uuid4(),
+        title="Periscope market-research cycle",
+        description="Research the market and file ONE brief.",
+        acceptance_criteria=["propose_market_brief() called once"],
+        status=TS.COMPLETED,
+        priority=2,
+        task_type=TT.ADMINISTRATIVE,
+        nature=TN.NON_TECHNICAL,
+        estimated_complexity=Complexity.LOW,
+        created_by=SYSTEM_UUID,
+        assigned_to=HOM_UUID,
+        team=Team.BOARD,
+        source=PERISCOPE_SOURCE,
+        confirmed_by_human=False,
+    )
+    session.add(task)
+    await session.flush()
+    findings = findings or [_finding(0), _finding(1)]
+    markers.set_market_brief(
+        task,
+        {
+            "headline": "A rival tool shipped agentic PR review",
+            "findings": findings,
+            "threats": [],
+            "opportunities": [],
+            "positioning_note": "",
+            "injection_hits": [],
+        },
+    )
+    await session.flush()
+    return task
+
+
+def _svc(session: AsyncSession) -> PeriscopeService:
+    return get_periscope_service(session)
+
+
+def _id(task: TaskTable) -> UUID:
+    return cast("UUID", task.id)
+
+
+@pytest.mark.asyncio
+async def test_approve_materializes_main_pm_owned_task(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    result = await _svc(db_session).approve_finding(
+        _id(task), "finding-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"
+    assert result.materialized_task_id is not None
+
+    materialized = await db_session.get(TaskTable, result.materialized_task_id)
+    assert materialized is not None
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
+    assert materialized.source == PERISCOPE_ITEM_SOURCE
+    # team is forced to Team.MAIN_PM — see test_roadmap_service.py's
+    # identical assertion for why: every "is this a coordination root"
+    # consumer keys on team, not assigned_to. A market signal has no natural
+    # owning cell (the prior Team.BACKEND was an arbitrary placeholder, not
+    # a real delegation hint), so there is no cell to preserve in Notes here.
+    assert materialized.team == Team.MAIN_PM
+    # main_pm can never own a code task (pm_cannot_own_code) — the intake
+    # coercion in create_task_from_draft retypes it to planning, the same
+    # shape a Main-PM coordination root always carries.
+    assert materialized.task_type == TT.PLANNING
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
+
+    await db_session.refresh(task)
+    payload = markers.get_market_brief(task)
+    assert payload is not None
+    finding0 = next(f for f in payload["findings"] if f["id"] == "finding-0")
+    assert finding0["status"] == "approved"
+    assert finding0["materialized_task_id"] == result.materialized_task_id
+    # The exploration task itself stays COMPLETED — approving a finding is
+    # orthogonal to the (already terminal) cycle.
+    assert task.status == TS.COMPLETED
+
+
+@pytest.mark.asyncio
+async def test_approve_is_idempotent(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    svc = _svc(db_session)
+    first = await svc.approve_finding(_id(task), "finding-0", created_by=CEO_UUID)
+    second = await svc.approve_finding(_id(task), "finding-0", created_by=CEO_UUID)
+    assert first is not None
+    assert second is not None
+    assert second.status == "already_approved"
+    assert second.materialized_task_id == first.materialized_task_id
+
+    result = await db_session.execute(
+        select(TaskTable).where(
+            TaskTable.source == PERISCOPE_ITEM_SOURCE,
+            TaskTable.title.like("Market signal:%"),
+        )
+    )
+    assert len(result.scalars().all()) == ONE
+
+
+@pytest.mark.asyncio
+async def test_reject_records_reason(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    result = await _svc(db_session).reject_finding(
+        _id(task), "finding-0", "not actionable this quarter"
+    )
+    assert result is not None
+    assert result.status == "rejected"
+
+    await db_session.refresh(task)
+    payload = markers.get_market_brief(task)
+    assert payload is not None
+    finding0 = next(f for f in payload["findings"] if f["id"] == "finding-0")
+    assert finding0["status"] == "rejected"
+    assert finding0["reject_reason"] == "not actionable this quarter"
+
+
+@pytest.mark.asyncio
+async def test_reject_is_idempotent(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    svc = _svc(db_session)
+    await svc.reject_finding(_id(task), "finding-0", "reason one")
+    second = await svc.reject_finding(_id(task), "finding-0", "reason two")
+    assert second is not None
+    assert second.status == "already_rejected"
+
+
+@pytest.mark.asyncio
+async def test_cannot_reject_an_approved_finding(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    svc = _svc(db_session)
+    await svc.approve_finding(_id(task), "finding-0", created_by=CEO_UUID)
+    result = await svc.reject_finding(_id(task), "finding-0", "changed my mind")
+    assert result is not None
+    assert result.status == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_cannot_approve_a_rejected_finding(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    svc = _svc(db_session)
+    await svc.reject_finding(_id(task), "finding-0", "not now")
+    result = await svc.approve_finding(_id(task), "finding-0", created_by=CEO_UUID)
+    assert result is not None
+    assert result.status == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_finding_with_no_status_key_defaults_to_proposed(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A finding authored before this feature shipped carries no status key
+    at all — setdefault treats it as proposed, not a crash."""
+    await _seed_roboco_project(db_session, monkeypatch)
+    legacy_finding = _finding(0, status=None)
+    assert "status" not in legacy_finding
+    task = await _seed_brief(db_session, findings=[legacy_finding])
+    result = await _svc(db_session).approve_finding(
+        _id(task), "finding-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"
+
+
+@pytest.mark.asyncio
+async def test_approve_unresolvable_project_is_invalid_state(
+    db_session: AsyncSession,
+) -> None:
+    """No RoboCo project seeded (and no monkeypatched slug pointing at one)
+    — the materialize fails cleanly instead of guessing a project."""
+    task = await _seed_brief(db_session)
+    result = await _svc(db_session).approve_finding(
+        _id(task), "finding-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "invalid_state"
+    assert "not resolvable" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_unknown_task_returns_none(db_session: AsyncSession) -> None:
+    result = await _svc(db_session).approve_finding(
+        uuid4(), "finding-0", created_by=CEO_UUID
+    )
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_unknown_finding_id_returns_none(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    result = await _svc(db_session).approve_finding(
+        _id(task), "finding-999", created_by=CEO_UUID
+    )
+    assert result is None
+
+
+async def _seed_cycle_ledger_row(session: AsyncSession, task: TaskTable) -> None:
+    session.add(
+        BoardProgramCycleTable(
+            program_key="periscope",
+            exploration_task_id=task.id,
+            opened_at=datetime.now(UTC),
+        )
+    )
+    await session.flush()
+
+
+@pytest.mark.asyncio
+async def test_approve_records_learn_decision(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    await _seed_cycle_ledger_row(db_session, task)
+    await _svc(db_session).approve_finding(_id(task), "finding-0", created_by=CEO_UUID)
+
+    row = (
+        await db_session.execute(
+            select(BoardProgramCycleTable).where(
+                BoardProgramCycleTable.program_key == "periscope"
+            )
+        )
+    ).scalar_one()
+    assert row.items_approved == ONE
+    # The ref is the finding's CLAIM (wrapped as learn_ref's "title" input) —
+    # a finding has no "title" field of its own.
+    decision = row.decisions[0]
+    assert decision["verdict"] == "approved"
+    assert decision["item_ref"] == "Competitor 0 shipped an autonomous review agent"
+
+
+@pytest.mark.asyncio
+async def test_approve_survives_learn_recording_failure(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_brief(db_session)
+    await _seed_cycle_ledger_row(db_session, task)
+
+    async def _boom(_self: object, *_args: object, **_kwargs: object) -> None:
+        raise RuntimeError("learn boom")
+
+    monkeypatch.setattr(bp_module.BoardProgramEngine, "record_decision", _boom)
+    result = await _svc(db_session).approve_finding(
+        _id(task), "finding-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"

tests/unit/services/test_pest_control_service.py

@@ -25,6 +25,7 @@ from roboco.db.tables import (
 )
 from roboco.foundation import identity as _foundation
 from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
 from roboco.models.base import (
     AgentRole,
     AgentStatus,
@@ -56,6 +57,7 @@ if TYPE_CHECKING:
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 PO_UUID = _foundation.AGENTS["product-owner"].uuid
 CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
 ONE = 1
 TWO = 2
 
@@ -106,6 +108,7 @@ async def _seed_agents(session: AsyncSession) -> None:
         (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
         (PO_UUID, "product-owner", AgentRole.PRODUCT_OWNER, Team.BOARD),
         (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
     ):
         if await session.get(AgentTable, uuid) is None:
             session.add(
@@ -184,7 +187,12 @@ def _id(task: TaskTable) -> UUID:
 
 
 @pytest.mark.asyncio
-async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> None:
+async def test_approve_materializes_main_pm_owned_task(
+    db_session: AsyncSession,
+) -> None:
+    """Defect fix: mirrors test_roadmap_service.py's identical assertion
+    update — approval materializes PENDING + assigned_to=main-pm, never an
+    unowned BACKLOG task (see RoadmapService._materialize's docstring)."""
     await _seed_project(db_session, "backend-svc")
     task = await _seed_cycle(db_session, project_slug="backend-svc")
     result = await _svc(db_session).approve_item(
@@ -196,9 +204,24 @@ async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> No
 
     materialized = await db_session.get(TaskTable, result.materialized_task_id)
     assert materialized is not None
-    assert materialized.status == TS.BACKLOG
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
     assert materialized.source == PEST_CONTROL_ITEM_SOURCE
-    assert materialized.team == Team.BACKEND
+    # team is forced to Team.MAIN_PM (not the item's own cell) — see
+    # test_roadmap_service.py's identical assertion for why: every "is this
+    # a coordination root" consumer keys on team, not assigned_to.
+    assert materialized.team == Team.MAIN_PM
+    # main_pm can never own a code task — see test_roadmap_service.py's
+    # identical assertion for the coercion rationale.
+    assert materialized.task_type == TT.PLANNING
+    # The item's own cell survives as a Notes delegation hint instead.
+    assert "backend cell" in (materialized.description or "")
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
 
     await db_session.refresh(task)
     payload = markers.get_pest_hunt(task)

tests/unit/services/test_roadmap_service.py

@@ -23,6 +23,7 @@ from roboco.db.tables import (
 )
 from roboco.foundation import identity as _foundation
 from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
 from roboco.models.base import (
     AgentRole,
     AgentStatus,
@@ -49,6 +50,7 @@ if TYPE_CHECKING:
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 PO_UUID = _foundation.AGENTS["product-owner"].uuid
 CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
 ONE = 1
 TWO = 2
 
@@ -98,6 +100,7 @@ async def _seed_agents(session: AsyncSession) -> None:
         (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
         (PO_UUID, "product-owner", AgentRole.PRODUCT_OWNER, Team.BOARD),
         (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
     ):
         if await session.get(AgentTable, uuid) is None:
             session.add(
@@ -175,7 +178,20 @@ def _id(task: TaskTable) -> UUID:
 
 
 @pytest.mark.asyncio
-async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> None:
+async def test_approve_materializes_main_pm_owned_task(
+    db_session: AsyncSession,
+) -> None:
+    """Defect fix (#703/#704): approval used to materialize an unowned
+    BACKLOG task — nothing dispatches BACKLOG, and once nudged to PENDING a
+    cell PM could claim + complete it as a bare root, merging straight to
+    the project's head rung and bypassing the Main-PM root / root->master PR
+    / CEO approval gate. It now materializes PENDING + assigned_to=main-pm
+    instead — team is forced to Team.MAIN_PM too (matching
+    TaskService.approve_and_start), since every "is this a coordination
+    root" consumer (pr_fail's next-hint, the PR-gate re-delegate steer,
+    delegate's wave-chain wiring, the PR labeler) keys on team, not
+    assigned_to. Leaving team on the item's own cell was itself the defect:
+    the root looked like a bare cell/dev task to all four."""
     await _seed_project(db_session, "backend-svc")
     task = await _seed_cycle(db_session, project_slug="backend-svc")
     result = await _svc(db_session).approve_item(
@@ -187,9 +203,29 @@ async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> No
 
     materialized = await db_session.get(TaskTable, result.materialized_task_id)
     assert materialized is not None
-    assert materialized.status == TS.BACKLOG
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
     assert materialized.source == ROADMAP_ITEM_SOURCE
-    assert materialized.team == Team.BACKEND
+    assert materialized.team == Team.MAIN_PM
+    # main_pm can never own a code task (pm_cannot_own_code) — the intake
+    # coercion in create_task_from_draft retypes it to planning, the same
+    # shape a Main-PM coordination root always carries.
+    assert materialized.task_type == TT.PLANNING
+    # The item's own cell ("backend") didn't just vanish — it survives as a
+    # Notes delegation hint in the composed description, which the Main PM's
+    # spawn briefing (_format_task_briefing_block) renders verbatim.
+    assert "backend cell" in (materialized.description or "")
+
+    # The predicate the four consumers key on: with team=Team.MAIN_PM and a
+    # branch (simulating a claimed root with its assembled PR), pr_fail's
+    # next-hint steers the Main PM to re-delegate rather than the nonsensical
+    # "dev will revise" hint a Main PM (no code-revise verb) can't act on.
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
 
     await db_session.refresh(task)
     payload = markers.get_roadmap_cycle(task)

tests/unit/services/test_sentinel_service.py

@@ -0,0 +1,406 @@
+"""SentinelService coverage: per-item approve materializes a Main-PM-owned
+root task (idempotent), reject records a reason (idempotent). Mirrors
+test_periscope_service.py exactly — the exploration task is ALREADY
+COMPLETED (complete-at-propose), so only the item's own status is under
+test, plus the docs-area task_type override (mirrors MirrorService).
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any, cast
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+from roboco.config import settings as cfg
+from roboco.db.tables import (
+    AgentTable,
+    BoardProgramCycleTable,
+    ProjectTable,
+    SystemSettingTable,
+    TaskTable,
+)
+from roboco.foundation import identity as _foundation
+from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
+from roboco.models.base import AgentRole, AgentStatus, Complexity, Team
+from roboco.models.base import TaskNature as TN
+from roboco.models.base import TaskStatus as TS
+from roboco.models.base import TaskType as TT
+from roboco.services import board_programs as bp_module
+from roboco.services.sentinel_service import SentinelService, get_sentinel_service
+from roboco.services.task import SENTINEL_ITEM_SOURCE, SENTINEL_SOURCE
+from sqlalchemy import delete, select, update
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+SYSTEM_UUID = _foundation.AGENTS["system"].uuid
+AUDITOR_UUID = _foundation.AGENTS["auditor"].uuid
+CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
+ROBOCO_SLUG = "roboco-standin"
+ONE = 1
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _purge_board_program_pollution(db_session: AsyncSession) -> None:
+    """See test_board_program_engine.py's identical fixture."""
+    await db_session.execute(
+        delete(SystemSettingTable).where(SystemSettingTable.key.like("board_program.%"))
+    )
+    await db_session.execute(delete(BoardProgramCycleTable))
+    await db_session.execute(
+        update(TaskTable)
+        .where(
+            TaskTable.source == SENTINEL_SOURCE,
+            TaskTable.status.notin_([TS.COMPLETED, TS.CANCELLED]),
+        )
+        .values(status=TS.CANCELLED)
+    )
+    await db_session.commit()
+
+
+def _item(idx: int, *, area: str = "waivers", status: str | None = "proposed") -> dict:
+    item: dict[str, Any] = {
+        "id": f"item-{idx}",
+        "area": area,
+        "observation": f"Minor findings keep getting waived, item {idx}",
+        "evidence": f"{idx + 3} waived-minor findings this week",
+        "suggested_action": f"Convert item {idx} to a Pest Control bug task",
+    }
+    if status is not None:
+        item["status"] = status
+        item["reject_reason"] = None
+        item["materialized_task_id"] = None
+    return item
+
+
+async def _seed_agents(session: AsyncSession) -> None:
+    for uuid, slug, role, team in (
+        (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
+        (AUDITOR_UUID, "auditor", AgentRole.AUDITOR, Team.BOARD),
+        (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
+    ):
+        if await session.get(AgentTable, uuid) is None:
+            session.add(
+                AgentTable(
+                    id=uuid,
+                    name=slug,
+                    slug=slug,
+                    role=role,
+                    team=team,
+                    status=AgentStatus.ACTIVE,
+                    model_config={},
+                    system_prompt="x",
+                    capabilities=[],
+                    permissions={},
+                    metrics={},
+                )
+            )
+    await session.flush()
+
+
+async def _seed_roboco_project(
+    session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> ProjectTable:
+    await _seed_agents(session)
+    project = ProjectTable(
+        id=uuid4(),
+        name="RoboCo",
+        slug=ROBOCO_SLUG,
+        git_url="https://example.com/roboco.git",
+        assigned_cell=Team.BACKEND,
+        created_by=SYSTEM_UUID,
+    )
+    session.add(project)
+    await session.flush()
+    monkeypatch.setattr(cfg, "self_heal_project_slug", ROBOCO_SLUG)
+    return project
+
+
+async def _seed_report(
+    session: AsyncSession, *, items: list[dict] | None = None
+) -> TaskTable:
+    await _seed_agents(session)
+    task = TaskTable(
+        id=uuid4(),
+        title="Sentinel drift-watch cycle",
+        description="Assess org-wide quality drift and file ONE report.",
+        acceptance_criteria=["propose_quality_report() called once"],
+        status=TS.COMPLETED,
+        priority=2,
+        task_type=TT.ADMINISTRATIVE,
+        nature=TN.NON_TECHNICAL,
+        estimated_complexity=Complexity.LOW,
+        created_by=SYSTEM_UUID,
+        assigned_to=AUDITOR_UUID,
+        team=Team.BOARD,
+        source=SENTINEL_SOURCE,
+        confirmed_by_human=False,
+    )
+    session.add(task)
+    await session.flush()
+    items = items or [_item(0), _item(1)]
+    markers.set_quality_report(
+        task,
+        {
+            "headline": "Waived findings climbed sharply this week",
+            "items": items,
+            "overall_assessment": "Drift is concentrated, not systemic",
+        },
+    )
+    await session.flush()
+    return task
+
+
+def _svc(session: AsyncSession) -> SentinelService:
+    return get_sentinel_service(session)
+
+
+def _id(task: TaskTable) -> UUID:
+    return cast("UUID", task.id)
+
+
+@pytest.mark.asyncio
+async def test_approve_materializes_main_pm_owned_task(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    result = await _svc(db_session).approve_item(
+        _id(task), "item-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"
+    assert result.materialized_task_id is not None
+
+    materialized = await db_session.get(TaskTable, result.materialized_task_id)
+    assert materialized is not None
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
+    assert materialized.source == SENTINEL_ITEM_SOURCE
+    # team is forced to Team.MAIN_PM — see test_roadmap_service.py's
+    # identical assertion for why: every "is this a coordination root"
+    # consumer keys on team, not assigned_to. A process/quality drift item
+    # has no natural owning cell (the prior Team.BACKEND was an arbitrary
+    # placeholder, not a real delegation hint), so there is no cell to
+    # preserve in Notes here.
+    assert materialized.team == Team.MAIN_PM
+    # main_pm can never own a code task (pm_cannot_own_code) — the intake
+    # coercion in create_task_from_draft retypes it to planning, the same
+    # shape a Main-PM coordination root always carries.
+    assert materialized.task_type == TT.PLANNING
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
+
+    await db_session.refresh(task)
+    payload = markers.get_quality_report(task)
+    assert payload is not None
+    item0 = next(i for i in payload["items"] if i["id"] == "item-0")
+    assert item0["status"] == "approved"
+    assert item0["materialized_task_id"] == result.materialized_task_id
+    assert task.status == TS.COMPLETED
+
+
+@pytest.mark.asyncio
+async def test_docs_area_materializes_documentation_task(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session, items=[_item(0, area="docs")])
+    result = await _svc(db_session).approve_item(
+        _id(task), "item-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    materialized = await db_session.get(TaskTable, result.materialized_task_id)
+    assert materialized is not None
+    assert materialized.task_type == TT.DOCUMENTATION
+
+
+@pytest.mark.asyncio
+async def test_approve_is_idempotent(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    svc = _svc(db_session)
+    first = await svc.approve_item(_id(task), "item-0", created_by=CEO_UUID)
+    second = await svc.approve_item(_id(task), "item-0", created_by=CEO_UUID)
+    assert first is not None
+    assert second is not None
+    assert second.status == "already_approved"
+    assert second.materialized_task_id == first.materialized_task_id
+
+    result = await db_session.execute(
+        select(TaskTable).where(
+            TaskTable.source == SENTINEL_ITEM_SOURCE,
+            TaskTable.title.like("Sentinel [waivers]:%"),
+        )
+    )
+    assert len(result.scalars().all()) == ONE
+
+
+@pytest.mark.asyncio
+async def test_reject_records_reason(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    result = await _svc(db_session).reject_item(
+        _id(task), "item-0", "already tracked elsewhere"
+    )
+    assert result is not None
+    assert result.status == "rejected"
+
+    await db_session.refresh(task)
+    payload = markers.get_quality_report(task)
+    assert payload is not None
+    item0 = next(i for i in payload["items"] if i["id"] == "item-0")
+    assert item0["status"] == "rejected"
+    assert item0["reject_reason"] == "already tracked elsewhere"
+
+
+@pytest.mark.asyncio
+async def test_reject_is_idempotent(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    svc = _svc(db_session)
+    await svc.reject_item(_id(task), "item-0", "reason one")
+    second = await svc.reject_item(_id(task), "item-0", "reason two")
+    assert second is not None
+    assert second.status == "already_rejected"
+
+
+@pytest.mark.asyncio
+async def test_cannot_reject_an_approved_item(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    svc = _svc(db_session)
+    await svc.approve_item(_id(task), "item-0", created_by=CEO_UUID)
+    result = await svc.reject_item(_id(task), "item-0", "changed my mind")
+    assert result is not None
+    assert result.status == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_cannot_approve_a_rejected_item(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    svc = _svc(db_session)
+    await svc.reject_item(_id(task), "item-0", "not now")
+    result = await svc.approve_item(_id(task), "item-0", created_by=CEO_UUID)
+    assert result is not None
+    assert result.status == "invalid_state"
+
+
+@pytest.mark.asyncio
+async def test_item_with_no_status_key_defaults_to_proposed(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    legacy_item = _item(0, status=None)
+    assert "status" not in legacy_item
+    task = await _seed_report(db_session, items=[legacy_item])
+    result = await _svc(db_session).approve_item(
+        _id(task), "item-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"
+
+
+@pytest.mark.asyncio
+async def test_approve_unresolvable_project_is_invalid_state(
+    db_session: AsyncSession,
+) -> None:
+    task = await _seed_report(db_session)
+    result = await _svc(db_session).approve_item(
+        _id(task), "item-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "invalid_state"
+    assert "not resolvable" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_unknown_task_returns_none(db_session: AsyncSession) -> None:
+    result = await _svc(db_session).approve_item(uuid4(), "item-0", created_by=CEO_UUID)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_unknown_item_id_returns_none(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    result = await _svc(db_session).approve_item(
+        _id(task), "item-999", created_by=CEO_UUID
+    )
+    assert result is None
+
+
+async def _seed_cycle_ledger_row(session: AsyncSession, task: TaskTable) -> None:
+    session.add(
+        BoardProgramCycleTable(
+            program_key="sentinel",
+            exploration_task_id=task.id,
+            opened_at=datetime.now(UTC),
+        )
+    )
+    await session.flush()
+
+
+@pytest.mark.asyncio
+async def test_approve_records_learn_decision(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    await _seed_cycle_ledger_row(db_session, task)
+    await _svc(db_session).approve_item(_id(task), "item-0", created_by=CEO_UUID)
+
+    row = (
+        await db_session.execute(
+            select(BoardProgramCycleTable).where(
+                BoardProgramCycleTable.program_key == "sentinel"
+            )
+        )
+    ).scalar_one()
+    assert row.items_approved == ONE
+    decision = row.decisions[0]
+    assert decision["verdict"] == "approved"
+    assert decision["item_ref"] == "Convert item 0 to a Pest Control bug task"
+
+
+@pytest.mark.asyncio
+async def test_approve_survives_learn_recording_failure(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    await _seed_roboco_project(db_session, monkeypatch)
+    task = await _seed_report(db_session)
+    await _seed_cycle_ledger_row(db_session, task)
+
+    async def _boom(_self: object, *_args: object, **_kwargs: object) -> None:
+        raise RuntimeError("learn boom")
+
+    monkeypatch.setattr(bp_module.BoardProgramEngine, "record_decision", _boom)
+    result = await _svc(db_session).approve_item(
+        _id(task), "item-0", created_by=CEO_UUID
+    )
+    assert result is not None
+    assert result.status == "approved"

tests/unit/services/test_spackle_service.py

@@ -25,6 +25,7 @@ from roboco.db.tables import (
 )
 from roboco.foundation import identity as _foundation
 from roboco.foundation.policy.content import markers
+from roboco.foundation.policy.lifecycle import _next_hint_pr_fail
 from roboco.models.base import (
     AgentRole,
     AgentStatus,
@@ -56,6 +57,7 @@ if TYPE_CHECKING:
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 PO_UUID = _foundation.AGENTS["product-owner"].uuid
 CEO_UUID = _foundation.AGENTS["ceo"].uuid
+MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid
 ONE = 1
 TWO = 2
 
@@ -109,6 +111,7 @@ async def _seed_agents(session: AsyncSession) -> None:
         (SYSTEM_UUID, "system", AgentRole.SYSTEM, None),
         (PO_UUID, "product-owner", AgentRole.PRODUCT_OWNER, Team.BOARD),
         (CEO_UUID, "ceo", AgentRole.CEO, None),
+        (MAIN_PM_UUID, "main-pm", AgentRole.MAIN_PM, Team.MAIN_PM),
     ):
         if await session.get(AgentTable, uuid) is None:
             session.add(
@@ -187,7 +190,12 @@ def _id(task: TaskTable) -> UUID:
 
 
 @pytest.mark.asyncio
-async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> None:
+async def test_approve_materializes_main_pm_owned_task(
+    db_session: AsyncSession,
+) -> None:
+    """Defect fix: mirrors test_roadmap_service.py's identical assertion
+    update — approval materializes PENDING + assigned_to=main-pm, never an
+    unowned BACKLOG task (see RoadmapService._materialize's docstring)."""
     await _seed_project(db_session, "backend-svc")
     task = await _seed_cycle(db_session, project_slug="backend-svc")
     result = await _svc(db_session).approve_item(
@@ -199,9 +207,24 @@ async def test_approve_materializes_backlog_task(db_session: AsyncSession) -> No
 
     materialized = await db_session.get(TaskTable, result.materialized_task_id)
     assert materialized is not None
-    assert materialized.status == TS.BACKLOG
+    assert materialized.status == TS.PENDING
+    assert materialized.assigned_to == MAIN_PM_UUID
+    assert materialized.parent_task_id is None
     assert materialized.source == SPACKLE_ITEM_SOURCE
-    assert materialized.team == Team.BACKEND
+    # team is forced to Team.MAIN_PM (not the item's own cell) — see
+    # test_roadmap_service.py's identical assertion for why: every "is this
+    # a coordination root" consumer keys on team, not assigned_to.
+    assert materialized.team == Team.MAIN_PM
+    # main_pm can never own a code task — see test_roadmap_service.py's
+    # identical assertion for the coercion rationale.
+    assert materialized.task_type == TT.PLANNING
+    # The item's own cell survives as a Notes delegation hint instead.
+    assert "backend cell" in (materialized.description or "")
+
+    materialized.branch_name = "feature/main_pm/deadbeef"
+    hint = _next_hint_pr_fail(materialized)
+    assert "re-delegate" in hint
+    assert "do NOT re-submit" in hint
 
     await db_session.refresh(task)
     payload = markers.get_gap_fill(task)