From b56d4c4287a030e8acfd63a074d4de7d86811464 Mon Sep 17 00:00:00 2001 From: q Date: Sun, 18 Oct 2026 18:56:15 +0000 Subject: [PATCH 1/2] Wake the PR/CI sweep after a committed Story entry into PR review SchedulerWakeTarget gains pr_ci. scheduler-pipeline binds one Event to pr_ci_loop and routes the listener's new target to it, so a hint runs the unchanged poll_pr_ci_once (poll_merged_prs then poll_ci_failures, each isolated) early through runtime.periodic_loop's wake path. POST /stories/{id}/pr_review publishes pr_ci right after its commit and before the refresh. Refused entries, replays and failed commits publish nothing; no other Story route gains the sender. The api best-effort publisher moves from the Run router into src/scheduler_wake.py and both routers use it. The scheduler wake service tests share the user/project/repository/Story setup in _scheduler_wake_service. --- docs/CHANGELOG.md | 2 + docs/contracts/scheduler-wake-hints.md | 27 ++- docs/contracts/story-task-run-surfaces.md | 2 +- services/api/src/routers/runs.py | 29 +-- services/api/src/routers/stories.py | 5 + services/api/src/scheduler_wake.py | 25 +++ .../api/tests/service/test_pr_review_wake.py | 58 +++++ .../api/tests/unit/test_pr_review_wake.py | 202 ++++++++++++++++++ services/scheduler/src/pipeline.py | 4 +- services/scheduler/src/tasks/pr_ci_loop.py | 4 +- .../tests/service/_scheduler_wake_service.py | 42 ++++ .../tests/service/test_pr_ci_wake.py | 127 +++++++++++ .../service/test_story_completion_wake.py | 41 +--- .../scheduler/tests/unit/test_pr_ci_loop.py | 4 +- .../tests/unit/test_scheduler_wake.py | 2 +- .../tests/unit/test_service_entrypoints.py | 56 ++--- shared/contracts/queues/scheduler_wake.py | 1 + shared/tests/test_scheduler_wake_contract.py | 4 +- 18 files changed, 541 insertions(+), 94 deletions(-) create mode 100644 services/api/src/scheduler_wake.py create mode 100644 services/api/tests/service/test_pr_review_wake.py create mode 100644 services/api/tests/unit/test_pr_review_wake.py create mode 100644 services/scheduler/tests/service/test_pr_ci_wake.py diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 668f59d65..2ed92d989 100644 --- a/docs/CHANGELOG.md +++ b/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 diff --git a/docs/contracts/scheduler-wake-hints.md b/docs/contracts/scheduler-wake-hints.md index 19c4c24dd..f9932b6c1 100644 --- a/docs/contracts/scheduler-wake-hints.md +++ b/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 diff --git a/docs/contracts/story-task-run-surfaces.md b/docs/contracts/story-task-run-surfaces.md index 92bbaa17f..0dafd4250 100644 --- a/docs/contracts/story-task-run-surfaces.md +++ b/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 | diff --git a/services/api/src/routers/runs.py b/services/api/src/routers/runs.py index 7e56396d2..8995998d4 100644 --- a/services/api/src/routers/runs.py +++ b/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 ) diff --git a/services/api/src/routers/stories.py b/services/api/src/routers/stories.py index 366fb4626..977736211 100644 --- a/services/api/src/routers/stories.py +++ b/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) diff --git a/services/api/src/scheduler_wake.py b/services/api/src/scheduler_wake.py new file mode 100644 index 000000000..ef96fc494 --- /dev/null +++ b/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) diff --git a/services/api/tests/service/test_pr_review_wake.py b/services/api/tests/service/test_pr_review_wake.py new file mode 100644 index 000000000..3fc7d9f31 --- /dev/null +++ b/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) diff --git a/services/api/tests/unit/test_pr_review_wake.py b/services/api/tests/unit/test_pr_review_wake.py new file mode 100644 index 000000000..e3c6bb91b --- /dev/null +++ b/services/api/tests/unit/test_pr_review_wake.py @@ -0,0 +1,202 @@ +"""A committed Story entry into PR review wakes the PR/CI sweep.""" + +import asyncio +from datetime import UTC, datetime +from types import SimpleNamespace +import uuid + +from fastapi import HTTPException +import pytest +from redis.exceptions import ConnectionError as RedisConnectionError +from sqlalchemy.exc import OperationalError +from structlog.testing import capture_logs +from wake_doubles import NativeRedis + +from shared.contracts.queues.scheduler_wake import SchedulerWakeTarget +from shared.models import Story +from src.routers.stories import deploy_story, pr_review_story + +STORY_ID = "story-wake" +PR_NUMBER = 42 +WAKE = ("wake", SchedulerWakeTarget.PR_CI) + + +def make_story(status: str) -> Story: + now = datetime.now(UTC) + return Story( + id=STORY_ID, + project_id=uuid.UUID("00000000-0000-0000-0000-000000000001"), + title="Wake", + type="product", + status=status, + waiting_on="none", + priority=0, + created_by="po", + unverified_decisions=[], + pr_number=PR_NUMBER, + created_at=now, + updated_at=now, + ) + + +class Session: + """Answers the Story lock and journals each commit and refresh that returns.""" + + def __init__(self, journal: list, status: str = "in_progress") -> None: + self.journal = journal + self.story = make_story(status) + self.hold: asyncio.Event | None = None + self.held = asyncio.Event() + self.failure: Exception | None = None + + async def execute(self, statement): + return SimpleNamespace(scalar_one_or_none=lambda: self.story) + + async def commit(self) -> None: + if self.hold is not None: + self.held.set() + await self.hold.wait() + if self.failure is not None: + raise self.failure + self.journal.append(("commit", self.story.status)) + + async def refresh(self, row) -> None: + self.journal.append(("refresh",)) + + +@pytest.fixture +def journal() -> list: + return [] + + +@pytest.fixture +def native(journal) -> NativeRedis: + return NativeRedis(journal) + + +@pytest.fixture +def redis(native) -> SimpleNamespace: + return SimpleNamespace(redis=native) + + +def review(session: Session, redis): + return pr_review_story(STORY_ID, None, db=session, redis=redis) + + +@pytest.mark.asyncio +async def test_an_entry_publishes_one_hint_after_its_commit_and_before_the_refresh(journal, redis): + read = await review(Session(journal), redis) + + assert read.status == "pr_review" + assert read.waiting_on == "ci" + assert read.pr_number == PR_NUMBER + assert journal == [("commit", "pr_review"), WAKE, ("refresh",)] + + +@pytest.mark.asyncio +async def test_each_later_work_cycle_entry_publishes_again(journal, redis): + session = Session(journal) + await review(session, redis) + # A CI-failure fix cycle returns the story to work before its next PR review. + session.story.status = "in_progress" + + await review(session, redis) + + assert journal == [("commit", "pr_review"), WAKE, ("refresh",)] * 2 + + +@pytest.mark.asyncio +async def test_nothing_is_published_while_the_commit_is_held(journal, redis): + session = Session(journal) + session.hold = asyncio.Event() + + request = asyncio.create_task(review(session, redis)) + await asyncio.wait_for(session.held.wait(), 1) + await asyncio.sleep(0) + assert journal == [] + + session.hold.set() + await request + assert journal == [("commit", "pr_review"), WAKE, ("refresh",)] + + +@pytest.mark.asyncio +async def test_a_failed_commit_publishes_nothing(journal, redis): + session = Session(journal) + session.failure = OperationalError("COMMIT", {}, Exception("connection lost")) + + with pytest.raises(OperationalError): + await review(session, redis) + + assert journal == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["pr_review", "deploying", "completed", "created"]) +async def test_a_refused_entry_publishes_nothing(journal, redis, status): + session = Session(journal, status) + + with pytest.raises(HTTPException) as refused: + await review(session, redis) + + assert refused.value.status_code == 422 + assert session.story.status == status + assert journal == [] + + +@pytest.mark.asyncio +async def test_another_transition_publishes_nothing(journal, redis): + session = Session(journal, "pr_review") + + read = await deploy_story(STORY_ID, None, db=session) + + assert read.status == "deploying" + assert journal == [("commit", "deploying"), ("refresh",)] + + +@pytest.mark.asyncio +async def test_zero_subscribers_is_a_published_hint(journal, redis, native): + native.receivers = 0 + + with capture_logs() as logs: + read = await review(Session(journal), redis) + + assert read.status == "pr_review" + assert journal == [("commit", "pr_review"), WAKE, ("refresh",)] + [published] = [log for log in logs if log["event"] == "pr_ci_wake_published"] + assert published["story_id"] == STORY_ID + assert published["receivers"] == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "failure", [RedisConnectionError("redis://:secret@redis:6379"), OSError("socket closed")] +) +async def test_a_failed_publish_loses_only_the_hint(journal, redis, native, failure): + native.failure = failure + + with capture_logs() as logs: + read = await review(Session(journal), redis) + + assert read.status == "pr_review" + assert journal == [("commit", "pr_review"), ("refresh",)] + [failed] = [log for log in logs if log["event"] == "pr_ci_wake_failed"] + assert failed == { + "event": "pr_ci_wake_failed", + "log_level": "warning", + "story_id": STORY_ID, + "error_type": type(failure).__name__, + } + + +@pytest.mark.asyncio +async def test_cancellation_during_publish_propagates_after_the_commit(journal, redis, native): + native.hold = asyncio.Event() + + request = asyncio.create_task(review(Session(journal), redis)) + await asyncio.wait_for(native.held.wait(), 1) + + request.cancel() + with pytest.raises(asyncio.CancelledError): + await request + assert journal == [("commit", "pr_review")] diff --git a/services/scheduler/src/pipeline.py b/services/scheduler/src/pipeline.py index c34380570..99154ef02 100644 --- a/services/scheduler/src/pipeline.py +++ b/services/scheduler/src/pipeline.py @@ -34,6 +34,7 @@ async def main() -> None: lifecycle_supervision_wake = asyncio.Event() temporary_access_wake = asyncio.Event() qa_routing_wake = asyncio.Event() + pr_ci_wake = asyncio.Event() await runtime.run_workers( [ ("task_dispatcher", partial(task_dispatcher_loop, wake=task_dispatcher_wake)), @@ -49,6 +50,7 @@ async def main() -> None: SchedulerWakeTarget.LIFECYCLE_SUPERVISION: lifecycle_supervision_wake, SchedulerWakeTarget.TEMPORARY_ACCESS: temporary_access_wake, SchedulerWakeTarget.QA_ROUTING: qa_routing_wake, + SchedulerWakeTarget.PR_CI: pr_ci_wake, }, ), ), @@ -57,7 +59,7 @@ async def main() -> None: partial(lifecycle_supervision_loop, wake=lifecycle_supervision_wake), ), ("qa_routing", partial(qa_routing_loop, wake=qa_routing_wake)), - ("pr_ci", pr_ci_loop), + ("pr_ci", partial(pr_ci_loop, wake=pr_ci_wake)), ("worker_reconciliation", worker_reconciliation_loop), ("temporary_access", partial(temporary_access_loop, wake=temporary_access_wake)), ("owner_notifications", owner_notification_loop), diff --git a/services/scheduler/src/tasks/pr_ci_loop.py b/services/scheduler/src/tasks/pr_ci_loop.py index 3f286dc4b..90cfc6558 100644 --- a/services/scheduler/src/tasks/pr_ci_loop.py +++ b/services/scheduler/src/tasks/pr_ci_loop.py @@ -32,7 +32,8 @@ async def poll_pr_ci_once(api_client, redis_client: RedisStreamClient) -> dict[s return {"prs_merged": merged, "ci_failures_routed": ci_failures} -async def pr_ci_loop() -> None: +async def pr_ci_loop(*, wake: asyncio.Event) -> None: + """Route merged PRs and CI failures on their own cadence; `wake` runs the sweep early.""" from ..clients.api import api_client async def cycle(redis_client: RedisStreamClient) -> dict[str, object]: @@ -48,4 +49,5 @@ async def cycle(redis_client: RedisStreamClient) -> dict[str, object]: stopped_event="pr_ci_stopped", redis_factory=RedisStreamClient, sleep=asyncio.sleep, + wake=wake, ) diff --git a/services/scheduler/tests/service/_scheduler_wake_service.py b/services/scheduler/tests/service/_scheduler_wake_service.py index 42bc148f3..540d75664 100644 --- a/services/scheduler/tests/service/_scheduler_wake_service.py +++ b/services/scheduler/tests/service/_scheduler_wake_service.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import uuid from shared.contracts.queues.scheduler_wake import SCHEDULER_WAKE_CHANNEL from shared.redis import RedisStreamClient @@ -19,6 +20,47 @@ async def call_api(api_client, method: str, path: str, **kwargs) -> dict: return response.json() +async def story_in_repository(api_client, repo_name: str, title: str) -> tuple[str, str]: + """A new user's project with one repository `test-org/` and one new Story. + + Returns the project and Story ids. + """ + telegram_id = uuid.uuid4().int % 1_000_000_000 + await call_api( + api_client, + "POST", + "users/", + json={"telegram_id": telegram_id, "username": f"wake-{telegram_id}"}, + ) + project_id = str(uuid.uuid4()) + await call_api( + api_client, + "POST", + "projects/", + json={ + "id": project_id, + "title": title, + "initiating_run_id": f"init-{uuid.uuid4().hex}", + "config": {}, + }, + headers={"X-Telegram-ID": str(telegram_id)}, + ) + await call_api( + api_client, + "POST", + "repositories/", + json={ + "project_id": project_id, + "name": repo_name, + "git_url": f"https://github.com/test-org/{repo_name}", + }, + ) + story = await call_api( + api_client, "POST", "stories/", json={"project_id": project_id, "title": "Wake"} + ) + return project_id, story["id"] + + async def wake_channel_subscribed(redis_client: RedisStreamClient) -> None: """Return once a listener is subscribed to `scheduler:wake`, within `WAIT_SECONDS`.""" async with asyncio.timeout(WAIT_SECONDS): diff --git a/services/scheduler/tests/service/test_pr_ci_wake.py b/services/scheduler/tests/service/test_pr_ci_wake.py new file mode 100644 index 000000000..73322aa36 --- /dev/null +++ b/services/scheduler/tests/service/test_pr_ci_wake.py @@ -0,0 +1,127 @@ +"""An API PR-review entry wakes the real-Redis PR/CI sweep long before its backstop.""" + +import asyncio +import os +import uuid + +from _scheduler_wake_service import ( + BACKSTOP_SECONDS, + WAIT_SECONDS, + call_api, + story_in_repository, + wake_channel_subscribed, +) +from _story_scope import OnlyStory +import pytest + +from shared.contracts.queues.scheduler_wake import SchedulerWakeTarget +from shared.redis import RedisStreamClient + +PR_NUMBER = 42 +HEAD_SHA = "0123456789abcdef0123456789abcdef01234567" + + +class _PendingGitHub: + """Reports this repository's PR and CI as still running; every other call is recorded.""" + + def __init__(self, repo_name: str) -> None: + self.repo_name = repo_name + self.reads: list[tuple] = [] + self.unexpected: list[str] = [] + + def __call__(self) -> "_PendingGitHub": + return self + + async def __aenter__(self) -> "_PendingGitHub": + return self + + async def __aexit__(self, *exc_info: object) -> bool: + return False + + def _own(self, owner: str, repo_name: str) -> None: + if (owner, repo_name) != ("test-org", self.repo_name): + raise RuntimeError("repository outside this test") + + async def get_pull_request(self, owner: str, repo_name: str, pr_number: int) -> dict: + self._own(owner, repo_name) + self.reads.append(("pull_request", pr_number)) + return { + "number": pr_number, + "node_id": "PR_wake", + "state": "open", + "merged_at": None, + "auto_merge": None, + "mergeable_state": "blocked", + "head": {"ref": "story", "sha": HEAD_SHA}, + } + + async def get_latest_workflow_run( + self, owner: str, repo_name: str, *, workflow_file: str, branch: str + ) -> dict: + self._own(owner, repo_name) + self.reads.append(("workflow_run", workflow_file, branch)) + return {"id": 1, "status": "in_progress", "conclusion": None, "head_sha": HEAD_SHA} + + def __getattr__(self, name: str): + self.unexpected.append(name) + raise AttributeError(name) + + +@pytest.mark.asyncio +async def test_an_api_pr_review_entry_reads_its_pr_and_ci_before_the_backstop( + api_client, monkeypatch +): + from src.tasks import pr_ci_loop, pr_poller, scheduler_wake_listener + + repo_name = f"wake-{uuid.uuid4().hex[:10]}" + github = _PendingGitHub(repo_name) + monkeypatch.setattr(pr_poller, "GitHubAppClient", github) + monkeypatch.setattr(pr_ci_loop, "_pr_ci_interval", lambda: BACKSTOP_SECONDS) + monkeypatch.setattr(scheduler_wake_listener, "_reconnect_interval", lambda: BACKSTOP_SECONDS) + _project_id, story_id = await story_in_repository(api_client, repo_name, "PR/CI wake") + await call_api(api_client, "POST", f"stories/{story_id}/start", json={"actor": "test"}) + await call_api(api_client, "PATCH", f"stories/{story_id}", json={"pr_number": PR_NUMBER}) + cycles: asyncio.Queue[dict[str, int]] = asyncio.Queue() + poll_pr_ci_once = pr_ci_loop.poll_pr_ci_once + + async def observed_cycle(api, redis_client): + counts = await poll_pr_ci_once(OnlyStory(api, story_id), redis_client) + cycles.put_nowait(counts) + return counts + + monkeypatch.setattr(pr_ci_loop, "poll_pr_ci_once", observed_cycle) + + wakes = {target: asyncio.Event() for target in SchedulerWakeTarget} + observer = RedisStreamClient(os.environ["REDIS_URL"]) + await observer.connect() + listener = asyncio.create_task(scheduler_wake_listener.scheduler_wake_listener(wakes)) + loop = None + try: + await wake_channel_subscribed(observer) + loop = asyncio.create_task(pr_ci_loop.pr_ci_loop(wake=wakes[SchedulerWakeTarget.PR_CI])) + assert await asyncio.wait_for(cycles.get(), WAIT_SECONDS) == { + "prs_merged": 0, + "ci_failures_routed": 0, + } + assert github.reads == [] + + await call_api(api_client, "POST", f"stories/{story_id}/pr_review", json={"actor": "test"}) + + assert await asyncio.wait_for(cycles.get(), WAIT_SECONDS) == { + "prs_merged": 0, + "ci_failures_routed": 0, + } + assert github.reads == [ + ("pull_request", PR_NUMBER), + ("workflow_run", "ci.yml", f"story/{story_id}"), + ] + assert github.unexpected == [] + story = await call_api(api_client, "GET", f"stories/{story_id}") + assert (story["status"], story["pr_number"]) == ("pr_review", PR_NUMBER) + assert await api_client.get_tasks_by_story(story_id) == [] + finally: + tasks = [task for task in (loop, listener) if task is not None] + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + await observer.close() diff --git a/services/scheduler/tests/service/test_story_completion_wake.py b/services/scheduler/tests/service/test_story_completion_wake.py index 36a17475b..bad95e4d1 100644 --- a/services/scheduler/tests/service/test_story_completion_wake.py +++ b/services/scheduler/tests/service/test_story_completion_wake.py @@ -8,6 +8,7 @@ BACKSTOP_SECONDS, WAIT_SECONDS, call_api, + story_in_repository, wake_channel_subscribed, ) import pytest @@ -45,51 +46,19 @@ async def create_pull_request(self, owner, repo_name, *, head, base, title, body async def _story_with_task_in_dev(api_client, repo_name: str) -> tuple[str, str]: - telegram_id = uuid.uuid4().int % 1_000_000_000 - await call_api( - api_client, - "POST", - "users/", - json={"telegram_id": telegram_id, "username": f"wake-{telegram_id}"}, - ) - project_id = str(uuid.uuid4()) - await call_api( - api_client, - "POST", - "projects/", - json={ - "id": project_id, - "title": "Completion wake", - "initiating_run_id": f"init-{uuid.uuid4().hex}", - "config": {}, - }, - headers={"X-Telegram-ID": str(telegram_id)}, - ) - await call_api( - api_client, - "POST", - "repositories/", - json={ - "project_id": project_id, - "name": repo_name, - "git_url": f"https://github.com/test-org/{repo_name}", - }, - ) - story = await call_api( - api_client, "POST", "stories/", json={"project_id": project_id, "title": "Wake"} - ) + project_id, story_id = await story_in_repository(api_client, repo_name, "Completion wake") task = await call_api( api_client, "POST", "tasks/", json={ "project_id": project_id, - "story_id": story["id"], + "story_id": story_id, "title": "Engineering", "status": "todo", }, ) - await call_api(api_client, "POST", f"stories/{story['id']}/start", json={"actor": "test"}) + await call_api(api_client, "POST", f"stories/{story_id}/start", json={"actor": "test"}) await call_api( api_client, "POST", @@ -97,7 +66,7 @@ async def _story_with_task_in_dev(api_client, repo_name: str) -> tuple[str, str] params={"to_status": "in_dev"}, json={"actor": "test"}, ) - return story["id"], task["id"] + return story_id, task["id"] @pytest.mark.asyncio diff --git a/services/scheduler/tests/unit/test_pr_ci_loop.py b/services/scheduler/tests/unit/test_pr_ci_loop.py index c9c762d2e..9992b357a 100644 --- a/services/scheduler/tests/unit/test_pr_ci_loop.py +++ b/services/scheduler/tests/unit/test_pr_ci_loop.py @@ -33,7 +33,7 @@ async def stop_after_one_cycle(_seconds): monkeypatch.setattr(pr_ci_loop, "_pr_ci_interval", lambda: 30) with pytest.raises(asyncio.CancelledError): - await pr_ci_loop.pr_ci_loop() + await pr_ci_loop.pr_ci_loop(wake=asyncio.Event()) redis.connect.assert_awaited_once() redis.close.assert_awaited_once() @@ -67,7 +67,7 @@ async def stop_after_one_cycle(_seconds): monkeypatch.setattr(pr_ci_loop, "_pr_ci_interval", lambda: 30) with pytest.raises(asyncio.CancelledError): - await pr_ci_loop.pr_ci_loop() + await pr_ci_loop.pr_ci_loop(wake=asyncio.Event()) merged.assert_awaited_once() failures.assert_awaited_once() diff --git a/services/scheduler/tests/unit/test_scheduler_wake.py b/services/scheduler/tests/unit/test_scheduler_wake.py index d9ce4e5f4..433609e2c 100644 --- a/services/scheduler/tests/unit/test_scheduler_wake.py +++ b/services/scheduler/tests/unit/test_scheduler_wake.py @@ -216,7 +216,7 @@ async def test_rejected_payloads_neither_wake_nor_stop_the_listener(monkeypatch) await broker.subscribed() for payload in ( - '{"target": "pr_ci"}', + '{"target": "worker_reconciliation"}', '{"target": "story_completion", "story_id": "story-1"}', "not json", b'{"target": "story_completion"}', diff --git a/services/scheduler/tests/unit/test_service_entrypoints.py b/services/scheduler/tests/unit/test_service_entrypoints.py index 7961595bf..329ec47df 100644 --- a/services/scheduler/tests/unit/test_service_entrypoints.py +++ b/services/scheduler/tests/unit/test_service_entrypoints.py @@ -330,7 +330,6 @@ async def test_pipeline_starts_without_loki(monkeypatch): "story_supervision", ] assert workers["scaffold"] is pipeline.scaffold_loop - assert workers["pr_ci"] is pipeline.pr_ci_loop assert workers["worker_reconciliation"] is pipeline.worker_reconciliation_loop assert workers["owner_notifications"] is pipeline.owner_notification_loop assert workers["story_supervision"] is pipeline.story_supervision_loop @@ -345,6 +344,7 @@ async def test_pipeline_routes_each_hint_only_to_its_own_loop(monkeypatch): from src import pipeline from src.tasks import ( lifecycle_supervision_loop, + pr_ci_loop, qa_routing_loop, scheduler_wake_listener, story_completion_loop, @@ -391,6 +391,10 @@ def record(step: str, result: object = None) -> AsyncMock: monkeypatch.setattr(qa_routing_loop, "RedisStreamClient", broker.client) monkeypatch.setattr(qa_routing_loop, "supervise_testing_stories", record("qa_routing", {})) monkeypatch.setattr(qa_routing_loop, "_qa_routing_interval", lambda: 3600) + monkeypatch.setattr(pr_ci_loop, "RedisStreamClient", broker.client) + monkeypatch.setattr(pr_ci_loop, "poll_merged_prs", record("merged_prs", 0)) + monkeypatch.setattr(pr_ci_loop, "poll_ci_failures", record("ci_failures", 0)) + monkeypatch.setattr(pr_ci_loop, "_pr_ci_interval", lambda: 3600) monkeypatch.setattr(scheduler_wake_listener, "_reconnect_interval", lambda: 3600) await pipeline.main() workers = dict(run_workers.await_args.args[0]) @@ -401,6 +405,7 @@ def record(step: str, result: object = None) -> AsyncMock: asyncio.create_task(workers["lifecycle_supervision"]()), asyncio.create_task(workers["temporary_access"]()), asyncio.create_task(workers["qa_routing"]()), + asyncio.create_task(workers["pr_ci"]()), asyncio.create_task( workers["scheduler_wake"](redis_factory=broker.client, sleep=Timers().sleep) ), @@ -408,35 +413,34 @@ def record(step: str, result: object = None) -> AsyncMock: await broker.subscribed() await settle() assert sorted(journal) == sorted( - ["complete", "dispatch", "recover", "temporary_access", "qa_routing", *lifecycle_sweeps] + [ + "complete", + "dispatch", + "recover", + "temporary_access", + "qa_routing", + "merged_prs", + "ci_failures", + *lifecycle_sweeps, + ] ) publisher = broker.client() await publisher.connect() - journal.clear() - await publish_scheduler_wake(publisher, SchedulerWakeTarget.TASK_DISPATCHER) - await settle() - assert journal == ["dispatch", "recover"] - - journal.clear() - await publish_scheduler_wake(publisher, SchedulerWakeTarget.STORY_COMPLETION) - await settle() - assert journal == ["complete"] - - journal.clear() - await publish_scheduler_wake(publisher, SchedulerWakeTarget.LIFECYCLE_SUPERVISION) - await settle() - assert journal == lifecycle_sweeps - - journal.clear() - await publish_scheduler_wake(publisher, SchedulerWakeTarget.TEMPORARY_ACCESS) - await settle() - assert journal == ["temporary_access"] - - journal.clear() - await publish_scheduler_wake(publisher, SchedulerWakeTarget.QA_ROUTING) - await settle() - assert journal == ["qa_routing"] + hinted_steps = { + SchedulerWakeTarget.TASK_DISPATCHER: ["dispatch", "recover"], + SchedulerWakeTarget.STORY_COMPLETION: ["complete"], + SchedulerWakeTarget.LIFECYCLE_SUPERVISION: lifecycle_sweeps, + SchedulerWakeTarget.TEMPORARY_ACCESS: ["temporary_access"], + SchedulerWakeTarget.QA_ROUTING: ["qa_routing"], + SchedulerWakeTarget.PR_CI: ["merged_prs", "ci_failures"], + } + assert set(hinted_steps) == set(SchedulerWakeTarget) + for target, steps in hinted_steps.items(): + journal.clear() + await publish_scheduler_wake(publisher, target) + await settle() + assert journal == steps, target for task in tasks: task.cancel() diff --git a/shared/contracts/queues/scheduler_wake.py b/shared/contracts/queues/scheduler_wake.py index caeb20c33..dba07d687 100644 --- a/shared/contracts/queues/scheduler_wake.py +++ b/shared/contracts/queues/scheduler_wake.py @@ -23,6 +23,7 @@ class SchedulerWakeTarget(StrEnum): LIFECYCLE_SUPERVISION = "lifecycle_supervision" TEMPORARY_ACCESS = "temporary_access" QA_ROUTING = "qa_routing" + PR_CI = "pr_ci" class SchedulerWakeHint(BaseModel): diff --git a/shared/tests/test_scheduler_wake_contract.py b/shared/tests/test_scheduler_wake_contract.py index 4571d0c9c..7959eb6c7 100644 --- a/shared/tests/test_scheduler_wake_contract.py +++ b/shared/tests/test_scheduler_wake_contract.py @@ -19,6 +19,7 @@ (SchedulerWakeTarget.LIFECYCLE_SUPERVISION, "lifecycle_supervision"), (SchedulerWakeTarget.TEMPORARY_ACCESS, "temporary_access"), (SchedulerWakeTarget.QA_ROUTING, "qa_routing"), + (SchedulerWakeTarget.PR_CI, "pr_ci"), ], ) def test_a_published_hint_parses_back_to_its_target(target, wire_target): @@ -31,12 +32,13 @@ def test_a_published_hint_parses_back_to_its_target(target, wire_target): @pytest.mark.parametrize( ("payload", "reason"), [ - ('{"target": "pr_ci"}', "invalid"), + ('{"target": "worker_reconciliation"}', "invalid"), ('{"target": "STORY_COMPLETION"}', "invalid"), ('{"target": "TASK_DISPATCHER"}', "invalid"), ('{"target": "LIFECYCLE_SUPERVISION"}', "invalid"), ('{"target": "TEMPORARY_ACCESS"}', "invalid"), ('{"target": "QA_ROUTING"}', "invalid"), + ('{"target": "PR_CI"}', "invalid"), ('{"target": 0}', "invalid"), ("{}", "invalid"), ('{"target": "story_completion", "story_id": "story-1"}', "invalid"), From 6b0b3332bb82a9d851257938dc67073a24919112 Mon Sep 17 00:00:00 2001 From: q Date: Sun, 18 Oct 2026 19:13:12 +0000 Subject: [PATCH 2/2] Give the grant-intent service fake Redis its native Pub/Sub publish The grant-intent service tests override the api Redis client with a fake that only had the stream publish. Their merged-replacement setup walks a Story through pr_review, which now also publishes a pr_ci wake hint on the client's native Pub/Sub, so the fake carries an unsubscribed publish. --- services/api/tests/service/test_users_grant_intents.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/services/api/tests/service/test_users_grant_intents.py b/services/api/tests/service/test_users_grant_intents.py index 38eadd15b..93f6b7926 100644 --- a/services/api/tests/service/test_users_grant_intents.py +++ b/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)