codegen-orchestrator-1705: Wake PR CI polling after PR review admission

docs/CHANGELOG.md

@@ -5,6 +5,8 @@ See the CHANGELOG rule in [AGENTS.md](../AGENTS.md).
 
 ## 2026-10-18
 
+- `POST /stories/{id}/pr_review` publishes a `pr_ci` wake hint after its commit, so the PR/CI sweep
+  reads a new PR review at once; GitHub CI completion still waits for the interval.
 - `PATCH /runs/{id}` publishes `qa_routing` and `temporary_access` wake hints after a QA Run's first
   terminal state commits, so the story verdict routes and access cleanup starts without the interval.
 - `PATCH /runs/{id}` publishes a `temporary_access` wake hint after committing a grant or revoke

docs/contracts/scheduler-wake-hints.md

@@ -11,8 +11,8 @@ path, and the awakened cycle decides from the same durable Story/Task/Run state
 - Message: the JSON body of `SchedulerWakeHint` in `shared/contracts/queues/scheduler_wake.py`,
   for example `{"target":"story_completion"}`. `target` is a `SchedulerWakeTarget`; no other
   field is accepted.
-- Targets: `story_completion`, `task_dispatcher`, `lifecycle_supervision`, `temporary_access` and
-  `qa_routing`; each wakes only the loop of that name.
+- Targets: `story_completion`, `task_dispatcher`, `lifecycle_supervision`, `temporary_access`,
+  `qa_routing` and `pr_ci`; each wakes only the loop of that name.
 - A hint carries routing only: no credentials, ids, correlation metadata or business state. It
   grants no permission to transition a Story, Task or Run and never invokes a business action.
 - Producers publish with `shared.redis.scheduler_wake.publish_scheduler_wake` after the durable
@@ -27,6 +27,7 @@ path, and the awakened cycle decides from the same durable Story/Task/Run state
 | `lifecycle_supervision` | api Run update `PATCH /runs/{id}` on the first typed success of an ordinary story deploy | scheduler-pipeline `lifecycle_supervision` loop |
 | `temporary_access` | api Run update `PATCH /runs/{id}` on the first readback proof of a temporary-access grant or revoke operation, and on a QA Run's first terminal state | scheduler-pipeline `temporary_access` loop |
 | `qa_routing` | api Run update `PATCH /runs/{id}` on a story-scoped QA Run's first terminal state | scheduler-pipeline `qa_routing` loop |
+| `pr_ci` | api Story transition `POST /stories/{id}/pr_review` | scheduler-pipeline `pr_ci` loop |
 
 ## Producers
 
@@ -166,6 +167,28 @@ path, and the awakened cycle decides from the same durable Story/Task/Run state
   its QA Run turned terminal, a queued QA handoff awaiting recovery, an invalid stored result, and
   every recovery, operator or direct-database write.
 
+### `pr_ci`
+
+- `POST /stories/{id}/pr_review` publishes one `pr_ci` hint for every committed entry into
+  `pr_review`, a later work cycle's entry included, on the initialized Redis client right after
+  its own `db.commit()` returns and before the response refresh. The transition rules refuse a
+  repeat from `pr_review` and every other invalid source status, so a refused request, a replay and
+  a failed commit publish nothing. No other Story update or transition publishes it.
+- The sender reads no GitHub state and no record beyond the locked Story. Story completion stores
+  the exact PR number before it calls this route, so the hint also covers a PR whose CI finished
+  before the entry committed.
+- The awakened cycle runs the unchanged `poll_pr_ci_once`: `poll_merged_prs` then
+  `poll_ci_failures`, each with its own error isolation. They still read the current PR, checks and
+  images from GitHub against the stored PR number and story branch, and own every merge, refusal,
+  deploy and CI-fix decision; the hint merges, deploys and routes nothing.
+- A Redis or socket error on publish is logged as `pr_ci_wake_failed` with the story id and the
+  error type only; the committed transition and its response are unchanged. Cancellation
+  propagates. There is no retry.
+- Every GitHub-side change still waits for the interval: CI completing with success or failure, a
+  PR becoming mergeable, behind or conflicted, a merge, and the published images of a merge
+  commit. No authenticated ingress for GitHub events exists, so this loop receives no hint from
+  GitHub.
+
 ## Listener
 
 scheduler-pipeline runs one `scheduler_wake` listener beside its loops under the same

docs/contracts/story-task-run-surfaces.md

@@ -32,7 +32,7 @@ composition models where listed. In API-exposure cells, `schemas/...` and
 
 | Surface / model family | Canonical source | API exposure / owner | Non-type invariant |
 |---|---|---|---|
-| Story create/update/status | `shared/contracts/dto/story.py` | `schemas/story.py`, `routers/stories.py`, `routers/_story_helpers.py`, `routers/_story_actions.py` | status, `waiting_on` and `status_entered_at` are written only by a transition, together on one locked row; `StoryUpdate` refuses all three; App-authenticated generated-product evidence, owner notifications and QA handoff are durable story lifecycle state; `unverified_decisions` is append-only |
+| Story create/update/status | `shared/contracts/dto/story.py` | `schemas/story.py`, `routers/stories.py`, `routers/_story_helpers.py`, `routers/_story_actions.py` | status, `waiting_on` and `status_entered_at` are written only by a transition, together on one locked row; `StoryUpdate` refuses all three; a committed entry into `pr_review` publishes the lossy `pr_ci` [wake hint](scheduler-wake-hints.md#pr_ci); App-authenticated generated-product evidence, owner notifications and QA handoff are durable story lifecycle state; `unverified_decisions` is append-only |
 | Task create/update/event/status | `shared/contracts/dto/task.py` | `schemas/task.py`, `routers/tasks.py` | scheduler dispatches only durable eligible task state |
 | Product Brief and requirement coverage | `shared/contracts/dto/product_brief.py` | `routers/product_briefs.py` | confirmed content is immutable; one live planning attempt; one idempotent admission releases that attempt's tasks |
 | Capability preview and stored plan | `shared/contracts/dto/capability_preview.py` | `routers/capability_previews.py`, `routers/product_briefs.py` | previews and plans are written and read by the platform only; the plan is derived from the stored preview at creation, derived again at confirmation, and never shown in the brief |

services/api/src/routers/runs.py

@@ -7,7 +7,6 @@
 from fastapi import APIRouter, Depends, Header, HTTPException, Query, status
 from fastapi.security import HTTPAuthorizationCredentials
 from pydantic import BaseModel, ConfigDict, ValidationError
-from redis.exceptions import RedisError
 from sqlalchemy import or_, select, tuple_
 from sqlalchemy.ext.asyncio import AsyncSession
 import structlog
@@ -41,7 +40,6 @@
 from shared.contracts.queues.scheduler_wake import SchedulerWakeTarget
 from shared.models import EngineeringAttemptLedger, EngineeringBudgetReservation, Project, Run, User
 from shared.redis.client import RedisStreamClient
-from shared.redis.scheduler_wake import publish_scheduler_wake
 
 from ..database import get_async_session
 from ..dependencies import (
@@ -57,6 +55,7 @@
 )
 from ..owner_notification_attempts import claim_attempt, refuse_superseded_write
 from ..owner_notification_settlement import preserve_po_settlement
+from ..scheduler_wake import publish_wake
 from ..schemas import RunCreate, RunRead, RunUpdate
 
 logger = structlog.get_logger()
@@ -677,7 +676,7 @@ async def _wake_on_first_deploy_proof(
     """Hint the owning sweep once a deploy Run's first success proof has committed."""
     target = _first_proof_wake(before, _deploy_proof(run))
     if target is not None:
-        await _publish_wake(target, run, redis)
+        await publish_wake(redis, target, run_id=run.id, story_id=run.story_id)
 
 
 def _holds_terminal_qa(run: Run) -> bool:
@@ -694,27 +693,11 @@ async def _wake_on_first_qa_terminal(
     if terminal_before or not _holds_terminal_qa(run):
         return
     if run.story_id is not None:
-        await _publish_wake(SchedulerWakeTarget.QA_ROUTING, run, redis)
-    await _publish_wake(SchedulerWakeTarget.TEMPORARY_ACCESS, run, redis)
-
-
-async def _publish_wake(target: SchedulerWakeTarget, run: Run, redis: RedisStreamClient) -> None:
-    """Publish one hint after the Run's commit; a lost hint waits for the sweep's interval."""
-    try:
-        receivers = await publish_scheduler_wake(redis, target)
-    except (RedisError, OSError) as exc:
-        logger.warning(
-            f"{target.value}_wake_failed",
-            run_id=run.id,
-            story_id=run.story_id,
-            error_type=type(exc).__name__,
+        await publish_wake(
+            redis, SchedulerWakeTarget.QA_ROUTING, run_id=run.id, story_id=run.story_id
         )
-        return
-    logger.info(
-        f"{target.value}_wake_published",
-        run_id=run.id,
-        story_id=run.story_id,
-        receivers=receivers,
+    await publish_wake(
+        redis, SchedulerWakeTarget.TEMPORARY_ACCESS, run_id=run.id, story_id=run.story_id
     )
 
 

services/api/src/routers/stories.py

@@ -29,6 +29,7 @@
 from shared.contracts.dto.story_planning import dispatch_owed_record
 from shared.contracts.queues.deploy import DeployAction, DeployMessage, DeployTrigger
 from shared.contracts.queues.qa import QAOutcome
+from shared.contracts.queues.scheduler_wake import SchedulerWakeTarget
 from shared.contracts.vocab import OwnerNotificationEvent
 from shared.models import WorkAdmissionAudit
 from shared.models.application import Application
@@ -51,6 +52,7 @@
 )
 from ..owner_notification_attempts import claim_attempt, refuse_superseded_write
 from ..owner_notification_settlement import preserve_po_settlement
+from ..scheduler_wake import publish_wake
 from ..schemas.actions import AdminAction
 from ..schemas.story import (
     StoryAccept,
@@ -1153,12 +1155,15 @@ async def pr_review_story(
     story_id: str,
     body: StoryTransition | None = None,
     db: AsyncSession = Depends(get_async_session),
+    redis: RedisStreamClient = Depends(get_redis_client),
 ) -> StoryRead:
+    """Move the Story to PR review; once that has committed, the PR/CI sweep is woken."""
     body = body or StoryTransition()
     story = await _get_story_for_update(story_id, db)
 
     _do_transition(story, StoryStatus.PR_REVIEW)
     await db.commit()
+    await publish_wake(redis, SchedulerWakeTarget.PR_CI, story_id=story.id)
     await db.refresh(story)
 
     logger.info("story_pr_review", story_id=story.id, actor=body.actor)

services/api/src/scheduler_wake.py

@@ -0,0 +1,25 @@
+"""Best-effort scheduler wake hints published by api routes after their own commit."""
+
+from redis.exceptions import RedisError
+import structlog
+
+from shared.contracts.queues.scheduler_wake import SchedulerWakeTarget
+from shared.redis.client import RedisStreamClient
+from shared.redis.scheduler_wake import publish_scheduler_wake
+
+logger = structlog.get_logger()
+
+
+async def publish_wake(
+    redis: RedisStreamClient, target: SchedulerWakeTarget, **subject: object
+) -> None:
+    """Publish one hint after the caller's commit; a lost hint waits for the sweep's interval.
+
+    `subject` identifies the committed row in the diagnostics, which name an error by type only.
+    """
+    try:
+        receivers = await publish_scheduler_wake(redis, target)
+    except (RedisError, OSError) as exc:
+        logger.warning(f"{target.value}_wake_failed", **subject, error_type=type(exc).__name__)
+        return
+    logger.info(f"{target.value}_wake_published", **subject, receivers=receivers)

services/api/tests/service/test_pr_review_wake.py

@@ -0,0 +1,58 @@
+"""A Story's committed entry into PR review publishes the PR/CI sweep's hint."""
+
+from _api_calls import call_api
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from shared.contracts.queues.scheduler_wake import SchedulerWakeTarget
+from shared.models import Story
+
+TASK_TEST_PROJECT_ID = "00000000-0000-0000-0000-000000000001"
+PR_NUMBER = 42
+
+
+async def _committed_review_state(db_engine, story_id: str) -> tuple[str, str, int | None]:
+    async with AsyncSession(db_engine) as session:
+        row = (
+            await session.execute(
+                select(Story.status, Story.waiting_on, Story.pr_number).where(Story.id == story_id)
+            )
+        ).one()
+        return row.status, row.waiting_on, row.pr_number
+
+
+@pytest.mark.asyncio
+async def test_a_committed_pr_review_entry_publishes_one_hint_and_a_repeat_none(
+    async_client, _tasks_project, wake_hints, db_engine
+):
+    story = await call_api(
+        async_client,
+        "POST",
+        "/api/stories/",
+        201,
+        json={"project_id": TASK_TEST_PROJECT_ID, "title": "PR review wake"},
+    )
+    story_id = story["id"]
+    await call_api(async_client, "POST", f"/api/stories/{story_id}/start", json={"actor": "test"})
+    await call_api(async_client, "PATCH", f"/api/stories/{story_id}", json={"pr_number": PR_NUMBER})
+    assert await wake_hints.published() == []
+
+    review = await call_api(
+        async_client, "POST", f"/api/stories/{story_id}/pr_review", json={"actor": "scheduler"}
+    )
+
+    assert review["status"] == "pr_review"
+    assert await wake_hints.published() == [SchedulerWakeTarget.PR_CI]
+    assert await _committed_review_state(db_engine, story_id) == ("pr_review", "ci", PR_NUMBER)
+
+    await call_api(
+        async_client,
+        "POST",
+        f"/api/stories/{story_id}/pr_review",
+        422,
+        json={"actor": "scheduler"},
+    )
+
+    assert await wake_hints.published() == []
+    assert await _committed_review_state(db_engine, story_id) == ("pr_review", "ci", PR_NUMBER)

services/api/tests/service/test_users_grant_intents.py

@@ -1,6 +1,7 @@
 """Service coverage for the durable intent lifecycle, separate from Runs."""
 
 from datetime import UTC, datetime
+from types import SimpleNamespace
 from unittest.mock import AsyncMock
 import uuid
 
@@ -89,8 +90,11 @@ async def test_public_caller_cannot_write_trusted_merged_deploy_observation(asyn
 
 
 class _PublishRedis:
+    """The initialized client's stream publish and its native Pub/Sub publish, unsubscribed."""
+
     def __init__(self, error: Exception | None = None) -> None:
         self.publish_message = AsyncMock(side_effect=error)
+        self.redis = SimpleNamespace(publish=AsyncMock(return_value=0))
 
 
 @pytest.fixture(autouse=True)