[805e525a] Extract repeated _require_ceo/_to_response/_status_value trio from 11 remaining route files (#785)

* [805e525a] refactor(api): extract _require_ceo/_to_response/_status_value from 11 route files

Move the per-file local _require_ceo wrapper, task->Response converters, and
status-value mappers out of board_programs.py, coroner.py, dogfood.py,
github_app.py, mirror.py, periscope.py, pest_control.py, scales.py,
sentinel.py, spackle.py, and telegram.py routes. _require_ceo call sites now
call the shared require_ceo_role directly; pure converters land in each
domain's roboco/api/schemas/*.py module (task_to_<domain>_response,
<domain>_status_value), mirroring the roadmap.py pattern; board_programs.py's
engine-backed _to_response becomes BoardProgramEngine.to_response, mirroring
release_proposal.py's task_to_proposal_response precedent. Placement-only,
no route/schema/behavior changes.

* [805e525a] docs(api): document Batch C route-helper relocation in api-routes-schemas map

---------

Co-authored-by: Backend Developer 2 <be-dev-2@roboco.tech>
Co-authored-by: Backend Documenter <be-doc@roboco.tech>

docs/map/api-routes-schemas.md

@@ -77,7 +77,7 @@ The FastAPI surface of RoboCo: every HTTP route under `roboco/api/routes/` (the
 | GET/POST | /api/roadmap/cycles, /cycles/{id}/items/{id}/{approve,reject} | roadmap.py | `require_ceo_role` (agent context) |
 | GET/POST | /api/telegram/credentials | telegram.py | `require_ceo_role` (agent context) |
 | GET/POST/DELETE | /api/providers/presets, /presets/{id}/apply, /presets/{id} | provider.py | agent context — save/apply/delete a named full routing snapshot (`docs/map/support-services.md`) |
-| GET/PUT/DELETE | /api/github-app/credentials ; GET /installations, /installations/{id}/repos | github_app.py | agent context — App id + private key CRUD, installation/repo listing for the panel's repo picker |
+| GET/PUT/DELETE | /api/github-app/credentials ; GET /installations, /installations/{id}/repos | github_app.py | `require_ceo_role` (agent context) — App id + private key CRUD, installation/repo listing for the panel's repo picker |
 | POST | /api/telegram/webapp-auth | telegram.py | public, pre-auth — Telegram `initData` HMAC validation; mounted only when `telegram_miniapp_enabled` AND `cloud_auth_enabled` |
 | GET | /api/telegram/today | telegram.py | `require_ceo_role` (agent context) + 30/60s rate limit — Mini App V4's "Today" brief, backed by `TgCockpitService.today()` (one DB round trip, see `docs/map/notification.md`) |
 | GET/POST | /api/auth/status (always), /auth/login, /auth/logout (mounted only when `cloud_auth_enabled`) | auth/routes.py | none (status) / FastAPI Users cookie login |
@@ -103,7 +103,9 @@ The FastAPI surface of RoboCo: every HTTP route under `roboco/api/routes/` (the
 | `CurrentAgentContext` | dep | api/deps.py:376 | Resolves agent from headers + HMAC, injects `AgentContext`. |
 | `require_orchestrator_ceo` | dep | api/deps.py:790 | Router-level CEO-HMAC guard on orchestrator control routes — relocated from a per-file `_require_ceo(agent)` wrapper that used to live in `routes/orchestrator.py`; `router = APIRouter(dependencies=[Depends(require_orchestrator_ceo)])`. |
 | `validate_agent_id_param` | fn | api/deps.py:839 | Path-injection guard (rejects empty/`.`/`..`/`/`/`\`/NUL) then normalizes via `_resolve_to_slug` — spawn/stop/status/resolve-wait/mark-waiting accept either a DB UUID or a slug and address the runtime container by the resolved slug; an unknown UUID passes through unchanged. Relocated from a route-local `_validated_agent_id` helper in `routes/orchestrator.py`. |
-| `require_ceo_role` / `require_pm_or_above` | fn | api/deps.py:627 / api/deps.py:618 | Shared role-check guards a2a.py/orchestrator.py/video.py/roadmap.py route handlers now call directly, replacing redundant per-file `_require_ceo(agent)` partial-application wrappers each of those route files used to define locally. |
+| `require_ceo_role` / `require_pm_or_above` | fn | api/deps.py:627 / api/deps.py:618 | Shared role-check guards a2a.py/orchestrator.py/video.py/roadmap.py route handlers now call directly, replacing redundant per-file `_require_ceo(agent)` partial-application wrappers each of those route files used to define locally; Batch C (task `805e525a`) extended the same direct-call pattern to board_programs.py/coroner.py/dogfood.py/github_app.py/mirror.py/periscope.py/pest_control.py/scales.py/sentinel.py/spackle.py/telegram.py. |
+| `task_to_postmortem_response` (coroner.py) / `task_to_dogfood_cycle_response` + `dogfood_status_value` / `task_to_mirror_cycle_response` + `mirror_status_value` / `task_to_market_brief_response` (periscope.py) / `task_to_sentinel_response`-equivalent (sentinel.py) / matching `task_to_<name>_cycle_response` + `<name>_status_value` pairs in pest_control.py/scales.py/spackle.py | fn | api/schemas/{coroner,dogfood,mirror,periscope,pest_control,scales,sentinel,spackle}.py | Per-domain `TaskTable` -> Response DTO converters + status-string mappers, relocated out of the paired route file's local `_to_response`/`_status_value` (Batch C, task `805e525a`), mirroring `task_to_response`. |
+| `BoardProgramEngine.to_response` | svc method | services/board_programs.py | board_programs.py's engine-backed `_to_response` (needs live DB reads via the engine, so it isn't a pure schema converter) relocated here instead of `api/schemas/board_programs.py` — mirrors the `release_proposal.py` precedent for an impure converter (Batch C, task `805e525a`). |
 | `require_auditor_or_ceo` | fn | api/deps.py:666 | Auditor-or-CEO 403 gate, added in the batch-B relocation (task `f8480831`) for dashboard.py's flag/report mutations and playbooks.py's curation endpoints — the two route files' identical inline role-check collapsed into one shared `deps.py` helper. |
 | `require_role_in` | fn | api/deps.py:652 | Generic "role must be a member of this set" 403 gate for an endpoint-specific role set with no standing named tier — added in the batch-B relocation (task `f8480831`), used by secretary.py's directive/state endpoints. |
 | `task_to_response` / `task_list_to_response` / `finding_to_response` | fn | api/schemas/tasks.py:889,964,969 | DTO conversion helpers (`TaskTable` -> `TaskResponse`/`TaskFindingResponse`); relocated out of `routes/tasks.py` into the schema module they convert to. |
@@ -261,6 +263,7 @@ roboco/api/
 > - ("panel-perf-p3-p4") adds `GET /api/dashboard/metrics/members` (batch scorecard fetch) — see `docs/map/metrics-observability.md`.
 > - (task `4baffaa3`, "Batch A: extract route helpers") placement-only refactor, no route/schema/behavior change: moves every non-`@router`-decorated top-level helper out of `tasks.py`, `a2a.py`, `orchestrator.py`, `video.py`, `v1/_role_dep.py`, `roadmap.py`, `prompter_live.py` (`journals.py` had none) per `.roboco/conventions.yml`'s `no_helpers_in_routes` rule — DB/side-effecting logic to the paired `roboco/services/*` module, DTO-conversion helpers to the matching `roboco/api/schemas/*.py` (e.g. `task_to_response`), and small HTTP-layer auth guards (`envelope_to_response`, `require_orchestrator_ceo`, `validate_agent_id_param`, `require_ceo_role`, `require_pm_or_above`) into `roboco/api/deps.py`, replacing several route-files' redundant local `_require_ceo(agent)` wrappers with direct calls to the shared `deps.py` guard. Two real regressions surfaced during the relocation's revision rounds and were fixed before merge: `envelope_to_response`'s "verb rejected" structlog event was dropped in the move (restored — see the Key Symbols row above), and `_auth_required()` was narrowed to a truthy-only check that silently dropped its unset-value production fallback, which would have accepted unauthenticated `X-Agent-Role: ceo` header spoofing on an unconfigured production deploy (GHSA-4f7g-w95g-5q2c) — the three-branch fallback logic was restored.
 > - (task `f8480831`, "Batch B: extract route helpers in remaining smaller-offender route files") placement-only refactor, no route/schema/behavior change: moved 28 non-`@router`-decorated helper functions out of 15 of the 24 batch-B route files (`optimal.py`, `project.py`, `release.py`, `dashboard.py`, `pitch.py`, `x.py`, `docs.py`, `git.py`, `playbooks.py`, `product.py`, `provider.py`, `research.py`, `secretary.py`, `system.py`, `work_session.py`) into their paired `roboco/services/*` module (DB/service-calling helpers), the route's own `roboco/api/schemas/*.py` as a converter (pure response/request shaping, mirroring `task_to_response`), or `roboco/utils/converters.py` (pure generic helpers); the other 9 files (`notifications.py`, `agents.py`, `cockpit.py`, `company_goals.py`, `kanban.py`, `secretary_live.py`, `settings.py`, `stream.py`, `usage.py`) had zero helpers by the precise `classify_python.py` classifier already. Added two small shared role-check helpers to `roboco/api/deps.py` (`require_auditor_or_ceo`, `require_role_in`) for endpoint-specific role gates that had no existing home.
+> - (task `805e525a`, "Batch C: extract route helpers from the 11 Board-Program route files Batch A/B never touched", PR #785) placement-only refactor, no route/schema/behavior change: relocated the remaining 26 helper-kind top-level defs (pr_gate finding `276ae32f`) out of `board_programs.py`, `coroner.py`, `dogfood.py`, `github_app.py`, `mirror.py`, `periscope.py`, `pest_control.py`, `scales.py`, `sentinel.py`, `spackle.py`, `telegram.py` — each file's local `_require_ceo(agent)` wrapper is gone, every call site now calls `require_ceo_role` directly (same `action=` strings preserved); each file's `_status_value`/`_to_response` pair moved into its matching `roboco/api/schemas/*.py` module as `<name>_status_value`/`task_to_<name>_response`, mirroring `task_to_response`, except `board_programs.py`'s engine-backed `_to_response` (needs live DB reads) which became `BoardProgramEngine.to_response` in `roboco/services/board_programs.py`, mirroring the `release_proposal.py` precedent for an impure converter. `telegram.py`'s `mount_telegram_miniapp_auth` (app-setup, not request-handling) was left in place, unflagged by the conventions checker. 9 of the 11 route files (`board_programs.py`, `coroner.py`, `dogfood.py`, `mirror.py`, `periscope.py`, `pest_control.py`, `scales.py`, `sentinel.py`, `spackle.py` — the Board Program registry route surface, see CLAUDE.md's "Board Program registry" section) aren't yet itemized in this doc's Files/Key-Endpoints tables above; only `github_app.py`/`telegram.py` were already present. That backfill is still owed and out of scope for this placement-only pass.
 
 ## Regression Risks
 

roboco/api/routes/board_programs.py

@@ -11,66 +11,24 @@ strategy-engine idle trigger uses.
 from __future__ import annotations
 
 from fastapi import APIRouter, HTTPException, status
-from pydantic import BaseModel
 
 from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
+from roboco.api.schemas.board_programs import BoardProgramResponse
 from roboco.foundation.policy.board_programs import PROGRAMS
 from roboco.security import guard_deco
-from roboco.services.board_programs import BoardProgramEngine, get_board_program_engine
+from roboco.services.board_programs import get_board_program_engine
 
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="view or act on Board Programs")
-
-
-class BoardProgramResponse(BaseModel):
-    """One registry entry's live status — the panel card + edit-project
-    dialog's opt-in controls both read this shape."""
-
-    key: str
-    title: str
-    description: str
-    role: str
-    trigger: str
-    scope: str
-    enabled: bool
-    opted_in_project_slugs: list[str]
-    last_opened_at: str | None
-    open_cycle: bool
-    last_cycle_summary: str | None
-
-
-async def _to_response(engine: BoardProgramEngine, key: str) -> BoardProgramResponse:
-    program = PROGRAMS[key]
-    enabled = await engine.enabled(key)
-    open_cycle, last_opened_at = await engine.cycle_state(key)
-    summary = await engine.prior_cycle_context(key, limit=1)
-    opted_in = await engine.opted_in_projects(program)
-    return BoardProgramResponse(
-        key=key,
-        title=program.title or key,
-        description=program.description,
-        role=program.role,
-        trigger=program.trigger.value,
-        scope=program.scope,
-        enabled=enabled,
-        opted_in_project_slugs=[p.slug for p in opted_in],
-        last_opened_at=last_opened_at.isoformat() if last_opened_at else None,
-        open_cycle=open_cycle,
-        last_cycle_summary=summary or None,
-    )
-
-
 @router.get("", response_model=list[BoardProgramResponse])
 async def list_board_programs(
     db: DbSession, agent: CurrentAgentContext
 ) -> list[BoardProgramResponse]:
     """Every registered Board Program's live status."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on Board Programs")
     engine = get_board_program_engine(db)
-    return [await _to_response(engine, key) for key in PROGRAMS]
+    return [await engine.to_response(key) for key in PROGRAMS]
 
 
 @router.post("/{key}/run-now", response_model=BoardProgramResponse)
@@ -86,7 +44,7 @@ async def run_program_now(
     — ``open_program_cycle`` collapses all three into the same None result,
     and the caller has no actionable distinction between them beyond retry.
     """
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on Board Programs")
     if key not in PROGRAMS:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail="Unknown Board Program"
@@ -103,4 +61,4 @@ async def run_program_now(
         )
     # Write route commits explicitly (get_db auto-commit is unreliable).
     await db.commit()
-    return await _to_response(engine, key)
+    return await engine.to_response(key)

roboco/api/routes/coroner.py

@@ -12,7 +12,6 @@ alone. CEO-only, mirroring every other Board Program surface.
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -22,62 +21,23 @@ from roboco.api.schemas.coroner import (
     PostmortemResponse,
     ProcessChangeActionResponse,
     ProcessChangeRejectRequest,
+    task_to_postmortem_response,
 )
-from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
-from roboco.services.coroner_service import PLAYBOOK_KIND, get_coroner_service
+from roboco.services.coroner_service import get_coroner_service
 from roboco.services.task import get_task_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="view or act on the Coroner postmortems list")
-
-
-def _to_response(task: TaskTable) -> PostmortemResponse:
-    incident = markers.get_coroner_incident(task) or {}
-    postmortem = markers.get_coroner_postmortem(task) or {}
-    process_change = postmortem.get("process_change") or {}
-    return PostmortemResponse(
-        task_id=str(task.id),
-        title=task.title,
-        completed_at=task.updated_at.isoformat() if task.updated_at else None,
-        incident_task_id=incident.get("incident_task_id"),
-        incident_kind=incident.get("kind"),
-        incident_title=incident.get("title"),
-        incident_summary=postmortem.get("incident_summary"),
-        root_cause=postmortem.get("root_cause"),
-        failed_stage=postmortem.get("failed_stage"),
-        process_change_kind=process_change.get("kind"),
-        process_change_description=process_change.get("description"),
-        playbook_id=postmortem.get("playbook_id"),
-        # A playbook-kind change already drafted into the playbook queue at
-        # propose time — there is nothing to decide, but the stored status
-        # stays "proposed", which left the panel rendering approve/dismiss
-        # buttons that both verbs refuse forever. Derive the terminal status
-        # the panel's contract expects instead.
-        process_change_status=(
-            "not_applicable"
-            if process_change.get("kind") == PLAYBOOK_KIND
-            else process_change.get("status", "proposed")
-        ),
-        process_change_reject_reason=process_change.get("reject_reason"),
-        process_change_materialized_task_id=process_change.get("materialized_task_id"),
-    )
-
-
 @router.get("/postmortems", response_model=list[PostmortemResponse])
 async def list_postmortems(
     db: DbSession, agent: CurrentAgentContext
 ) -> list[PostmortemResponse]:
     """Every completed Coroner postmortem, newest first."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the Coroner postmortems list")
     tasks = await get_task_service(db).list_completed_coroner_postmortems()
-    return [_to_response(t) for t in tasks]
+    return [task_to_postmortem_response(t) for t in tasks]
 
 
 @router.post(
@@ -93,7 +53,7 @@ async def approve_process_change(
 ) -> ProcessChangeActionResponse:
     """Materialize the postmortem's process change as a Main-PM-owned root
     task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the Coroner postmortems list")
     result = await get_coroner_service(db).approve_process_change(
         task_id, created_by=agent.agent_id
     )
@@ -125,7 +85,7 @@ async def reject_process_change(
     agent: CurrentAgentContext,
 ) -> ProcessChangeActionResponse:
     """Dismiss the postmortem's process change with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the Coroner postmortems list")
     result = await get_coroner_service(db).reject_process_change(task_id, data.reason)
     if result is None:
         raise HTTPException(

roboco/api/routes/dogfood.py

@@ -4,7 +4,6 @@ materializes it as a BACKLOG task; nothing here starts it — normal PM
 activation takes it from there. Mirrors ``roboco.api.routes.spackle``.
 """
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -13,39 +12,16 @@ from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
 from roboco.api.schemas.dogfood import (
     DogfoodCycleResponse,
     FrictionFixItemActionResponse,
-    FrictionFixItemResponse,
     FrictionFixRejectRequest,
+    task_to_dogfood_cycle_response,
 )
 from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
 from roboco.services.dogfood_service import get_dogfood_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="view or act on the dogfood queue")
-
-
-def _status_value(task: "TaskTable") -> str:
-    raw = task.status
-    return raw.value if hasattr(raw, "value") else str(raw)
-
-
-def _to_response(task: "TaskTable") -> DogfoodCycleResponse:
-    payload = markers.get_friction_fixes(task) or {}
-    items = [FrictionFixItemResponse(**item) for item in payload.get("items", [])]
-    return DogfoodCycleResponse(
-        task_id=str(task.id),
-        title=task.title,
-        status=_status_value(task),
-        items=items,
-    )
-
-
 @router.get("/cycles", response_model=list[DogfoodCycleResponse])
 async def list_dogfood_cycles(
     db: DbSession, agent: CurrentAgentContext
@@ -55,9 +31,13 @@ async def list_dogfood_cycles(
     A cycle the PO hasn't authored yet (no items drafted) is omitted — there
     is nothing for the CEO to review until ``propose_friction_fixes`` lands.
     """
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the dogfood queue")
     tasks = await get_dogfood_service(db).list_open_cycles()
-    return [_to_response(t) for t in tasks if markers.get_friction_fixes(t)]
+    return [
+        task_to_dogfood_cycle_response(t)
+        for t in tasks
+        if markers.get_friction_fixes(t)
+    ]
 
 
 @router.post(
@@ -73,7 +53,7 @@ async def approve_friction_fix_item(
     agent: CurrentAgentContext,
 ) -> FrictionFixItemActionResponse:
     """Materialize one proposed item as a BACKLOG task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the dogfood queue")
     result = await get_dogfood_service(db).approve_item(
         task_id, item_id, created_by=agent.agent_id
     )
@@ -107,7 +87,7 @@ async def reject_friction_fix_item(
     agent: CurrentAgentContext,
 ) -> FrictionFixItemActionResponse:
     """Reject one proposed item with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the dogfood queue")
     result = await get_dogfood_service(db).reject_item(task_id, item_id, data.reason)
     if result is None:
         raise HTTPException(

roboco/api/routes/github_app.py

@@ -34,16 +34,12 @@ from roboco.services.github_app_credentials import (
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="manage the GitHub App integration")
-
-
 @router.get("/credentials", response_model=GitHubAppCredentialsStatus)
 async def get_github_app_credentials(
     db: DbSession, agent: CurrentAgentContext
 ) -> GitHubAppCredentialsStatus:
     """Whether the App id + private key are stored. Never the key."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="manage the GitHub App integration")
     has_creds = await get_github_app_credentials_service(db).has_credentials()
     return GitHubAppCredentialsStatus(has_credentials=has_creds)
 
@@ -59,7 +55,7 @@ async def set_github_app_credentials(
     data: GitHubAppCredentialsSetRequest, db: DbSession, agent: CurrentAgentContext
 ) -> GitHubAppCredentialsStatus:
     """Set the App id + private key together (PEM paste)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="manage the GitHub App integration")
     svc = get_github_app_credentials_service(db)
     try:
         has_creds = await svc.set_credentials(
@@ -78,7 +74,7 @@ async def clear_github_app_credentials(
     db: DbSession, agent: CurrentAgentContext
 ) -> GitHubAppCredentialsStatus:
     """Clear the App id + private key."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="manage the GitHub App integration")
     has_creds = await get_github_app_credentials_service(db).set_credentials(
         app_id="", private_key=""
     )
@@ -93,7 +89,7 @@ async def get_installations(
     db: DbSession, agent: CurrentAgentContext
 ) -> list[InstallationResponse]:
     """List every installation of the configured App."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="manage the GitHub App integration")
     try:
         installations = await list_installations(db)
     except GitHubAppNotConfiguredError as e:
@@ -117,7 +113,7 @@ async def get_installation_repositories(
     installation_id: int, db: DbSession, agent: CurrentAgentContext
 ) -> list[InstallationRepositoryResponse]:
     """List every repository the given installation can access."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="manage the GitHub App integration")
     try:
         repos = await list_installation_repositories(db, installation_id)
     except GitHubAppNotConfiguredError as e:

roboco/api/routes/mirror.py

@@ -4,7 +4,6 @@ materializes it as a BACKLOG docs task; nothing here starts it — normal PM
 activation takes it from there. Mirrors ``roboco.api.routes.spackle``.
 """
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -12,40 +11,17 @@ from fastapi import APIRouter, HTTPException, status
 from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
 from roboco.api.schemas.mirror import (
     MessagingFixItemActionResponse,
-    MessagingFixItemResponse,
     MessagingFixRejectRequest,
     MirrorCycleResponse,
+    task_to_mirror_cycle_response,
 )
 from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
 from roboco.services.mirror_service import get_mirror_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="view or act on the mirror queue")
-
-
-def _status_value(task: "TaskTable") -> str:
-    raw = task.status
-    return raw.value if hasattr(raw, "value") else str(raw)
-
-
-def _to_response(task: "TaskTable") -> MirrorCycleResponse:
-    payload = markers.get_messaging_fixes(task) or {}
-    items = [MessagingFixItemResponse(**item) for item in payload.get("items", [])]
-    return MirrorCycleResponse(
-        task_id=str(task.id),
-        title=task.title,
-        status=_status_value(task),
-        items=items,
-    )
-
-
 @router.get("/cycles", response_model=list[MirrorCycleResponse])
 async def list_mirror_cycles(
     db: DbSession, agent: CurrentAgentContext
@@ -55,9 +31,13 @@ async def list_mirror_cycles(
     A cycle the HoM hasn't authored yet (no items drafted) is omitted — there
     is nothing for the CEO to review until ``propose_messaging_fixes`` lands.
     """
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the mirror queue")
     tasks = await get_mirror_service(db).list_open_cycles()
-    return [_to_response(t) for t in tasks if markers.get_messaging_fixes(t)]
+    return [
+        task_to_mirror_cycle_response(t)
+        for t in tasks
+        if markers.get_messaging_fixes(t)
+    ]
 
 
 @router.post(
@@ -73,7 +53,7 @@ async def approve_messaging_fix_item(
     agent: CurrentAgentContext,
 ) -> MessagingFixItemActionResponse:
     """Materialize one proposed item as a BACKLOG docs task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the mirror queue")
     result = await get_mirror_service(db).approve_item(
         task_id, item_id, created_by=agent.agent_id
     )
@@ -107,7 +87,7 @@ async def reject_messaging_fix_item(
     agent: CurrentAgentContext,
 ) -> MessagingFixItemActionResponse:
     """Reject one proposed item with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the mirror queue")
     result = await get_mirror_service(db).reject_item(task_id, item_id, data.reason)
     if result is None:
         raise HTTPException(

roboco/api/routes/periscope.py

@@ -5,7 +5,6 @@ propose time), but each finding carries its own per-item approve/reject,
 mirroring ``roboco.api.routes.roadmap``'s shape.
 """
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -14,43 +13,16 @@ from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
 from roboco.api.schemas.periscope import (
     MarketBriefFindingActionResponse,
     MarketBriefFindingRejectRequest,
-    MarketBriefFindingResponse,
     MarketBriefResponse,
+    task_to_market_brief_response,
 )
-from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
 from roboco.services.periscope_service import get_periscope_service
 from roboco.services.task import get_task_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(
-        agent.role, action="view or act on the Periscope market-briefs list"
-    )
-
-
-def _to_response(task: "TaskTable") -> MarketBriefResponse | None:
-    payload = markers.get_market_brief(task)
-    if payload is None:
-        return None
-    findings = [MarketBriefFindingResponse(**f) for f in payload.get("findings", [])]
-    return MarketBriefResponse(
-        task_id=str(task.id),
-        title=task.title,
-        completed_at=task.updated_at.isoformat() if task.updated_at else None,
-        headline=payload.get("headline", ""),
-        findings=findings,
-        threats=payload.get("threats", []),
-        opportunities=payload.get("opportunities", []),
-        positioning_note=payload.get("positioning_note", ""),
-    )
-
-
 @router.get("/briefs", response_model=list[MarketBriefResponse])
 async def list_market_briefs(
     db: DbSession, agent: CurrentAgentContext
@@ -58,9 +30,11 @@ async def list_market_briefs(
     """Recent filed market briefs, newest-first. A completed Periscope
     exploration without a marker (shouldn't happen — the verb always sets
     one before completing) is omitted rather than rendered blank."""
-    _require_ceo(agent)
+    require_ceo_role(
+        agent.role, action="view or act on the Periscope market-briefs list"
+    )
     tasks = await get_task_service(db).list_periscope_briefs()
-    return [r for t in tasks if (r := _to_response(t)) is not None]
+    return [r for t in tasks if (r := task_to_market_brief_response(t)) is not None]
 
 
 @router.post(
@@ -76,7 +50,9 @@ async def approve_market_brief_finding(
     agent: CurrentAgentContext,
 ) -> MarketBriefFindingActionResponse:
     """Materialize one finding as a Main-PM-owned root task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(
+        agent.role, action="view or act on the Periscope market-briefs list"
+    )
     result = await get_periscope_service(db).approve_finding(
         task_id, finding_id, created_by=agent.agent_id
     )
@@ -110,7 +86,9 @@ async def reject_market_brief_finding(
     agent: CurrentAgentContext,
 ) -> MarketBriefFindingActionResponse:
     """Dismiss one finding with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(
+        agent.role, action="view or act on the Periscope market-briefs list"
+    )
     result = await get_periscope_service(db).reject_finding(
         task_id, finding_id, data.reason
     )

roboco/api/routes/pest_control.py

@@ -4,7 +4,6 @@ materializes it as a BACKLOG task; nothing here starts it — normal PM
 activation takes it from there. Mirrors ``roboco.api.routes.roadmap``.
 """
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -13,39 +12,16 @@ from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
 from roboco.api.schemas.pest_control import (
     PestHuntCycleResponse,
     PestHuntItemActionResponse,
-    PestHuntItemResponse,
     PestHuntRejectRequest,
+    task_to_pest_hunt_cycle_response,
 )
 from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
 from roboco.services.pest_control_service import get_pest_control_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="view or act on the pest-control queue")
-
-
-def _status_value(task: "TaskTable") -> str:
-    raw = task.status
-    return raw.value if hasattr(raw, "value") else str(raw)
-
-
-def _to_response(task: "TaskTable") -> PestHuntCycleResponse:
-    payload = markers.get_pest_hunt(task) or {}
-    items = [PestHuntItemResponse(**item) for item in payload.get("items", [])]
-    return PestHuntCycleResponse(
-        task_id=str(task.id),
-        title=task.title,
-        status=_status_value(task),
-        items=items,
-    )
-
-
 @router.get("/cycles", response_model=list[PestHuntCycleResponse])
 async def list_pest_control_cycles(
     db: DbSession, agent: CurrentAgentContext
@@ -55,9 +31,11 @@ async def list_pest_control_cycles(
     A cycle the PO hasn't authored yet (no items drafted) is omitted — there
     is nothing for the CEO to review until ``propose_bug_hunt`` lands.
     """
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the pest-control queue")
     tasks = await get_pest_control_service(db).list_open_cycles()
-    return [_to_response(t) for t in tasks if markers.get_pest_hunt(t)]
+    return [
+        task_to_pest_hunt_cycle_response(t) for t in tasks if markers.get_pest_hunt(t)
+    ]
 
 
 @router.post(
@@ -73,7 +51,7 @@ async def approve_pest_hunt_item(
     agent: CurrentAgentContext,
 ) -> PestHuntItemActionResponse:
     """Materialize one proposed item as a BACKLOG task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the pest-control queue")
     result = await get_pest_control_service(db).approve_item(
         task_id, item_id, created_by=agent.agent_id
     )
@@ -107,7 +85,7 @@ async def reject_pest_hunt_item(
     agent: CurrentAgentContext,
 ) -> PestHuntItemActionResponse:
     """Reject one proposed item with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the pest-control queue")
     result = await get_pest_control_service(db).reject_item(
         task_id, item_id, data.reason
     )

roboco/api/routes/scales.py

@@ -5,7 +5,6 @@ cancel it) — nothing here creates a task. Mirrors
 ``roboco.api.routes.pest_control``.
 """
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -14,39 +13,16 @@ from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
 from roboco.api.schemas.scales import (
     RebalanceCycleResponse,
     RebalanceItemActionResponse,
-    RebalanceItemResponse,
     RebalanceRejectRequest,
+    task_to_rebalance_cycle_response,
 )
 from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
 from roboco.services.scales_service import get_scales_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="view or act on the Scales queue")
-
-
-def _status_value(task: "TaskTable") -> str:
-    raw = task.status
-    return raw.value if hasattr(raw, "value") else str(raw)
-
-
-def _to_response(task: "TaskTable") -> RebalanceCycleResponse:
-    payload = markers.get_rebalance_plan(task) or {}
-    items = [RebalanceItemResponse(**item) for item in payload.get("items", [])]
-    return RebalanceCycleResponse(
-        task_id=str(task.id),
-        title=task.title,
-        status=_status_value(task),
-        items=items,
-    )
-
-
 @router.get("/cycles", response_model=list[RebalanceCycleResponse])
 async def list_scales_cycles(
     db: DbSession, agent: CurrentAgentContext
@@ -56,9 +32,13 @@ async def list_scales_cycles(
     A cycle the PO hasn't authored yet (no items drafted) is omitted — there
     is nothing for the CEO to review until ``propose_rebalance`` lands.
     """
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the Scales queue")
     tasks = await get_scales_service(db).list_open_cycles()
-    return [_to_response(t) for t in tasks if markers.get_rebalance_plan(t)]
+    return [
+        task_to_rebalance_cycle_response(t)
+        for t in tasks
+        if markers.get_rebalance_plan(t)
+    ]
 
 
 @router.post(
@@ -74,7 +54,7 @@ async def approve_rebalance_item(
     agent: CurrentAgentContext,
 ) -> RebalanceItemActionResponse:
     """Execute one proposed item against its live target task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the Scales queue")
     result = await get_scales_service(db).approve_item(
         task_id, item_id, created_by=agent.agent_id
     )
@@ -108,7 +88,7 @@ async def reject_rebalance_item(
     agent: CurrentAgentContext,
 ) -> RebalanceItemActionResponse:
     """Reject one proposed item with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the Scales queue")
     result = await get_scales_service(db).reject_item(task_id, item_id, data.reason)
     if result is None:
         raise HTTPException(

roboco/api/routes/sentinel.py

@@ -5,7 +5,6 @@ propose time), but each item carries its own per-item approve/reject,
 mirroring ``roboco.api.routes.periscope``'s shape.
 """
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -14,41 +13,16 @@ from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
 from roboco.api.schemas.sentinel import (
     QualityReportItemActionResponse,
     QualityReportItemRejectRequest,
-    QualityReportItemResponse,
     QualityReportResponse,
+    task_to_quality_report_response,
 )
-from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
 from roboco.services.sentinel_service import get_sentinel_service
 from roboco.services.task import get_task_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(
-        agent.role, action="view or act on the Sentinel quality-reports list"
-    )
-
-
-def _to_response(task: "TaskTable") -> QualityReportResponse | None:
-    payload = markers.get_quality_report(task)
-    if payload is None:
-        return None
-    items = [QualityReportItemResponse(**i) for i in payload.get("items", [])]
-    return QualityReportResponse(
-        task_id=str(task.id),
-        title=task.title,
-        completed_at=task.updated_at.isoformat() if task.updated_at else None,
-        headline=payload.get("headline", ""),
-        items=items,
-        overall_assessment=payload.get("overall_assessment", ""),
-    )
-
-
 @router.get("/reports", response_model=list[QualityReportResponse])
 async def list_quality_reports(
     db: DbSession, agent: CurrentAgentContext
@@ -56,9 +30,11 @@ async def list_quality_reports(
     """Recent filed quality reports, newest-first. A completed Sentinel
     exploration without a marker (shouldn't happen — the verb always sets
     one before completing) is omitted rather than rendered blank."""
-    _require_ceo(agent)
+    require_ceo_role(
+        agent.role, action="view or act on the Sentinel quality-reports list"
+    )
     tasks = await get_task_service(db).list_sentinel_reports()
-    return [r for t in tasks if (r := _to_response(t)) is not None]
+    return [r for t in tasks if (r := task_to_quality_report_response(t)) is not None]
 
 
 @router.post(
@@ -74,7 +50,9 @@ async def approve_quality_report_item(
     agent: CurrentAgentContext,
 ) -> QualityReportItemActionResponse:
     """Materialize one drift item as a Main-PM-owned root task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(
+        agent.role, action="view or act on the Sentinel quality-reports list"
+    )
     result = await get_sentinel_service(db).approve_item(
         task_id, item_id, created_by=agent.agent_id
     )
@@ -108,7 +86,9 @@ async def reject_quality_report_item(
     agent: CurrentAgentContext,
 ) -> QualityReportItemActionResponse:
     """Dismiss one drift item with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(
+        agent.role, action="view or act on the Sentinel quality-reports list"
+    )
     result = await get_sentinel_service(db).reject_item(task_id, item_id, data.reason)
     if result is None:
         raise HTTPException(

roboco/api/routes/spackle.py

@@ -4,7 +4,6 @@ materializes it as a BACKLOG task; nothing here starts it — normal PM
 activation takes it from there. Mirrors ``roboco.api.routes.pest_control``.
 """
 
-from typing import TYPE_CHECKING
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, status
@@ -12,40 +11,17 @@ from fastapi import APIRouter, HTTPException, status
 from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
 from roboco.api.schemas.spackle import (
     GapFillItemActionResponse,
-    GapFillItemResponse,
     GapFillRejectRequest,
     SpackleCycleResponse,
+    task_to_spackle_cycle_response,
 )
 from roboco.foundation.policy.content import markers
 from roboco.security import guard_deco
 from roboco.services.spackle_service import get_spackle_service
 
-if TYPE_CHECKING:
-    from roboco.db.tables import TaskTable
-
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="view or act on the spackle queue")
-
-
-def _status_value(task: "TaskTable") -> str:
-    raw = task.status
-    return raw.value if hasattr(raw, "value") else str(raw)
-
-
-def _to_response(task: "TaskTable") -> SpackleCycleResponse:
-    payload = markers.get_gap_fill(task) or {}
-    items = [GapFillItemResponse(**item) for item in payload.get("items", [])]
-    return SpackleCycleResponse(
-        task_id=str(task.id),
-        title=task.title,
-        status=_status_value(task),
-        items=items,
-    )
-
-
 @router.get("/cycles", response_model=list[SpackleCycleResponse])
 async def list_spackle_cycles(
     db: DbSession, agent: CurrentAgentContext
@@ -55,9 +31,9 @@ async def list_spackle_cycles(
     A cycle the PO hasn't authored yet (no items drafted) is omitted — there
     is nothing for the CEO to review until ``propose_gap_fill`` lands.
     """
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the spackle queue")
     tasks = await get_spackle_service(db).list_open_cycles()
-    return [_to_response(t) for t in tasks if markers.get_gap_fill(t)]
+    return [task_to_spackle_cycle_response(t) for t in tasks if markers.get_gap_fill(t)]
 
 
 @router.post(
@@ -73,7 +49,7 @@ async def approve_gap_fill_item(
     agent: CurrentAgentContext,
 ) -> GapFillItemActionResponse:
     """Materialize one proposed item as a BACKLOG task (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the spackle queue")
     result = await get_spackle_service(db).approve_item(
         task_id, item_id, created_by=agent.agent_id
     )
@@ -107,7 +83,7 @@ async def reject_gap_fill_item(
     agent: CurrentAgentContext,
 ) -> GapFillItemActionResponse:
     """Reject one proposed item with a reason (idempotent)."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="view or act on the spackle queue")
     result = await get_spackle_service(db).reject_item(task_id, item_id, data.reason)
     if result is None:
         raise HTTPException(

roboco/api/routes/telegram.py

@@ -46,16 +46,12 @@ _logger = get_logger(__name__)
 router = APIRouter()
 
 
-def _require_ceo(agent: CurrentAgentContext) -> None:
-    require_ceo_role(agent.role, action="manage Telegram credentials")
-
-
 @router.get("/credentials", response_model=TelegramCredentialsStatus)
 async def get_telegram_credentials(
     db: DbSession, agent: CurrentAgentContext
 ) -> TelegramCredentialsStatus:
     """Whether the bot token + chat id are stored. Never the secrets."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="manage Telegram credentials")
     has_creds = await get_telegram_credentials_service(db).has_credentials()
     return TelegramCredentialsStatus(has_credentials=has_creds)
 
@@ -71,7 +67,7 @@ async def set_telegram_credentials(
     data: TelegramCredentialsSetRequest, db: DbSession, agent: CurrentAgentContext
 ) -> TelegramCredentialsStatus:
     """Set (or, passing both empty, clear) the bot token + chat id together."""
-    _require_ceo(agent)
+    require_ceo_role(agent.role, action="manage Telegram credentials")
     svc = get_telegram_credentials_service(db)
     try:
         has_creds = await svc.set_credentials(

roboco/api/schemas/board_programs.py

@@ -0,0 +1,23 @@
+"""Schemas for the Board Programs registry API — the CEO's status + off-
+schedule "run now" surface, mirroring ``roboco.api.schemas.roadmap``."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class BoardProgramResponse(BaseModel):
+    """One registry entry's live status — the panel card + edit-project
+    dialog's opt-in controls both read this shape."""
+
+    key: str
+    title: str
+    description: str
+    role: str
+    trigger: str
+    scope: str
+    enabled: bool
+    opted_in_project_slugs: list[str]
+    last_opened_at: str | None
+    open_cycle: bool
+    last_cycle_summary: str | None

roboco/api/schemas/coroner.py

@@ -12,8 +12,16 @@ action routes key on the task id alone.
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+from roboco.services.coroner_service import PLAYBOOK_KIND
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class PostmortemResponse(BaseModel):
     """One completed Coroner postmortem — the incident ref + the Auditor's
@@ -51,3 +59,35 @@ class ProcessChangeActionResponse(BaseModel):
     status: str
     materialized_task_id: str | None = None
     detail: str
+
+
+def task_to_postmortem_response(task: TaskTable) -> PostmortemResponse:
+    incident = markers.get_coroner_incident(task) or {}
+    postmortem = markers.get_coroner_postmortem(task) or {}
+    process_change = postmortem.get("process_change") or {}
+    return PostmortemResponse(
+        task_id=str(task.id),
+        title=task.title,
+        completed_at=task.updated_at.isoformat() if task.updated_at else None,
+        incident_task_id=incident.get("incident_task_id"),
+        incident_kind=incident.get("kind"),
+        incident_title=incident.get("title"),
+        incident_summary=postmortem.get("incident_summary"),
+        root_cause=postmortem.get("root_cause"),
+        failed_stage=postmortem.get("failed_stage"),
+        process_change_kind=process_change.get("kind"),
+        process_change_description=process_change.get("description"),
+        playbook_id=postmortem.get("playbook_id"),
+        # A playbook-kind change already drafted into the playbook queue at
+        # propose time — there is nothing to decide, but the stored status
+        # stays "proposed", which left the panel rendering approve/dismiss
+        # buttons that both verbs refuse forever. Derive the terminal status
+        # the panel's contract expects instead.
+        process_change_status=(
+            "not_applicable"
+            if process_change.get("kind") == PLAYBOOK_KIND
+            else process_change.get("status", "proposed")
+        ),
+        process_change_reject_reason=process_change.get("reject_reason"),
+        process_change_materialized_task_id=process_change.get("materialized_task_id"),
+    )

roboco/api/schemas/dogfood.py

@@ -3,8 +3,15 @@ Mirrors ``roboco.api.schemas.spackle`` exactly."""
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class FrictionFixItemResponse(BaseModel):
     """One evidence-backed UX-friction item draft within a Dogfood walk."""
@@ -44,3 +51,19 @@ class FrictionFixItemActionResponse(BaseModel):
     item_id: str
     materialized_task_id: str | None = None
     detail: str
+
+
+def dogfood_status_value(task: TaskTable) -> str:
+    raw = task.status
+    return raw.value if hasattr(raw, "value") else str(raw)
+
+
+def task_to_dogfood_cycle_response(task: TaskTable) -> DogfoodCycleResponse:
+    payload = markers.get_friction_fixes(task) or {}
+    items = [FrictionFixItemResponse(**item) for item in payload.get("items", [])]
+    return DogfoodCycleResponse(
+        task_id=str(task.id),
+        title=task.title,
+        status=dogfood_status_value(task),
+        items=items,
+    )

roboco/api/schemas/mirror.py

@@ -3,8 +3,15 @@ Mirrors ``roboco.api.schemas.spackle`` exactly."""
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class MessagingFixItemResponse(BaseModel):
     """One evidence-backed messaging-fix item draft within a Mirror audit."""
@@ -44,3 +51,19 @@ class MessagingFixItemActionResponse(BaseModel):
     item_id: str
     materialized_task_id: str | None = None
     detail: str
+
+
+def mirror_status_value(task: TaskTable) -> str:
+    raw = task.status
+    return raw.value if hasattr(raw, "value") else str(raw)
+
+
+def task_to_mirror_cycle_response(task: TaskTable) -> MirrorCycleResponse:
+    payload = markers.get_messaging_fixes(task) or {}
+    items = [MessagingFixItemResponse(**item) for item in payload.get("items", [])]
+    return MirrorCycleResponse(
+        task_id=str(task.id),
+        title=task.title,
+        status=mirror_status_value(task),
+        items=items,
+    )

roboco/api/schemas/periscope.py

@@ -9,8 +9,15 @@ that per-finding queue, mirrored on ``roboco.api.schemas.roadmap``."""
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class MarketBriefFindingResponse(BaseModel):
     """One cited finding within a Periscope market brief."""
@@ -52,3 +59,20 @@ class MarketBriefFindingActionResponse(BaseModel):
     finding_id: str
     materialized_task_id: str | None = None
     detail: str
+
+
+def task_to_market_brief_response(task: TaskTable) -> MarketBriefResponse | None:
+    payload = markers.get_market_brief(task)
+    if payload is None:
+        return None
+    findings = [MarketBriefFindingResponse(**f) for f in payload.get("findings", [])]
+    return MarketBriefResponse(
+        task_id=str(task.id),
+        title=task.title,
+        completed_at=task.updated_at.isoformat() if task.updated_at else None,
+        headline=payload.get("headline", ""),
+        findings=findings,
+        threats=payload.get("threats", []),
+        opportunities=payload.get("opportunities", []),
+        positioning_note=payload.get("positioning_note", ""),
+    )

roboco/api/schemas/pest_control.py

@@ -4,8 +4,15 @@ Mirrors ``roboco.api.schemas.roadmap`` — ``rationale`` becomes the required
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class PestHuntItemResponse(BaseModel):
     """One evidence-backed bug item draft within a Pest Control hunt."""
@@ -45,3 +52,19 @@ class PestHuntItemActionResponse(BaseModel):
     item_id: str
     materialized_task_id: str | None = None
     detail: str
+
+
+def pest_control_status_value(task: TaskTable) -> str:
+    raw = task.status
+    return raw.value if hasattr(raw, "value") else str(raw)
+
+
+def task_to_pest_hunt_cycle_response(task: TaskTable) -> PestHuntCycleResponse:
+    payload = markers.get_pest_hunt(task) or {}
+    items = [PestHuntItemResponse(**item) for item in payload.get("items", [])]
+    return PestHuntCycleResponse(
+        task_id=str(task.id),
+        title=task.title,
+        status=pest_control_status_value(task),
+        items=items,
+    )

roboco/api/schemas/scales.py

@@ -6,8 +6,15 @@ to the target task instead."""
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class RebalanceItemResponse(BaseModel):
     """One re-priority/cancellation item draft within a Scales rebalance plan."""
@@ -46,3 +53,19 @@ class RebalanceItemActionResponse(BaseModel):
     item_id: str
     executed_detail: str | None = None
     detail: str
+
+
+def scales_status_value(task: TaskTable) -> str:
+    raw = task.status
+    return raw.value if hasattr(raw, "value") else str(raw)
+
+
+def task_to_rebalance_cycle_response(task: TaskTable) -> RebalanceCycleResponse:
+    payload = markers.get_rebalance_plan(task) or {}
+    items = [RebalanceItemResponse(**item) for item in payload.get("items", [])]
+    return RebalanceCycleResponse(
+        task_id=str(task.id),
+        title=task.title,
+        status=scales_status_value(task),
+        items=items,
+    )

roboco/api/schemas/sentinel.py

@@ -7,8 +7,15 @@ CEO decides on afterward."""
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class QualityReportItemResponse(BaseModel):
     """One drift item within a Sentinel quality report."""
@@ -49,3 +56,18 @@ class QualityReportItemActionResponse(BaseModel):
     item_id: str
     materialized_task_id: str | None = None
     detail: str
+
+
+def task_to_quality_report_response(task: TaskTable) -> QualityReportResponse | None:
+    payload = markers.get_quality_report(task)
+    if payload is None:
+        return None
+    items = [QualityReportItemResponse(**i) for i in payload.get("items", [])]
+    return QualityReportResponse(
+        task_id=str(task.id),
+        title=task.title,
+        completed_at=task.updated_at.isoformat() if task.updated_at else None,
+        headline=payload.get("headline", ""),
+        items=items,
+        overall_assessment=payload.get("overall_assessment", ""),
+    )

roboco/api/schemas/spackle.py

@@ -3,8 +3,15 @@ Mirrors ``roboco.api.schemas.pest_control`` exactly."""
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel, Field
 
+from roboco.foundation.policy.content import markers
+
+if TYPE_CHECKING:
+    from roboco.db.tables import TaskTable
+
 
 class GapFillItemResponse(BaseModel):
     """One evidence-backed gap-fill item draft within a Spackle audit."""
@@ -44,3 +51,19 @@ class GapFillItemActionResponse(BaseModel):
     item_id: str
     materialized_task_id: str | None = None
     detail: str
+
+
+def spackle_status_value(task: TaskTable) -> str:
+    raw = task.status
+    return raw.value if hasattr(raw, "value") else str(raw)
+
+
+def task_to_spackle_cycle_response(task: TaskTable) -> SpackleCycleResponse:
+    payload = markers.get_gap_fill(task) or {}
+    items = [GapFillItemResponse(**item) for item in payload.get("items", [])]
+    return SpackleCycleResponse(
+        task_id=str(task.id),
+        title=task.title,
+        status=spackle_status_value(task),
+        items=items,
+    )

roboco/services/board_programs.py

@@ -28,6 +28,7 @@ from typing import TYPE_CHECKING, Any, cast
 
 from sqlalchemy import func, select
 
+from roboco.api.schemas.board_programs import BoardProgramResponse
 from roboco.config import settings
 from roboco.db.tables import BoardProgramCycleTable, ProjectTable, TaskTable
 from roboco.foundation.policy.board_programs import (
@@ -318,6 +319,29 @@ class BoardProgramEngine(BaseService):
         """Per-program settings-store override, else the legacy flag."""
         return await program_armed(self.session, key)
 
+    async def to_response(self, key: str) -> BoardProgramResponse:
+        """Render one registry entry's live status for the CEO surface —
+        the panel card + edit-project dialog's opt-in controls both read
+        this shape."""
+        program = PROGRAMS[key]
+        enabled = await self.enabled(key)
+        open_cycle, last_opened_at = await self.cycle_state(key)
+        summary = await self.prior_cycle_context(key, limit=1)
+        opted_in = await self.opted_in_projects(program)
+        return BoardProgramResponse(
+            key=key,
+            title=program.title or key,
+            description=program.description,
+            role=program.role,
+            trigger=program.trigger.value,
+            scope=program.scope,
+            enabled=enabled,
+            opted_in_project_slugs=[p.slug for p in opted_in],
+            last_opened_at=last_opened_at.isoformat() if last_opened_at else None,
+            open_cycle=open_cycle,
+            last_cycle_summary=summary or None,
+        )
+
     async def run_due_programs(self) -> list[str]:
         """Originate a cycle for every enabled, due CRON program, PLUS every
         program whose metric predicate (``_METRIC_PREDICATES``) fires this