codegen-orchestrator-1697: Add scheduler wake hints with periodic backstop

ARCHITECTURE.md

@@ -74,7 +74,7 @@ taken from the default value.
 | `worker-broker` | The only service on both control-plane and worker networks. Authenticates per-worker credentials and brokers worker streams, sessions, status and Compose requests. |
 | `langgraph` | PO runtime (`po:input` consumer + ReactAgent). The same image also provides separate `engineering-worker`, `deploy-worker`, `qa-worker`, and `architect` container entrypoints; those roles are not the `langgraph` service itself. |
 | `architect` | Story→tasks LLM decomposition. Consumes `architect:queue`. A container of the `langgraph` image, not part of `scheduler` |
-| `scheduler-pipeline` | Ten loops with independent failure boundaries: `task_dispatcher` (engineering admission/dispatch), `scaffold`, `story_completion`, `lifecycle_supervision`, `qa_routing`, `pr_ci` (merged-PR handling and CI-failure routing), `worker_reconciliation`, `temporary_access` cleanup, `owner_notifications` recovery and `story_supervision` (state-age watchdog, stage notices). A container of the shared `scheduler` image |
+| `scheduler-pipeline` | Ten loops with independent failure boundaries: `task_dispatcher` (engineering admission/dispatch), `scaffold`, `story_completion`, `lifecycle_supervision`, `qa_routing`, `pr_ci` (merged-PR handling and CI-failure routing), `worker_reconciliation`, `temporary_access` cleanup, `owner_notifications` recovery and `story_supervision` (state-age watchdog, stage notices), plus the `scheduler_wake` Pub/Sub listener that wakes `story_completion` early. A container of the shared `scheduler` image |
 | `scheduler-infrastructure` | Fail-closed Time4VPS server sync, health checks, provisioner trigger and restart-safe result consumption. A container of the shared `scheduler` image |
 | `scheduler-maintenance` | GitHub project sync, analytics aggregation and queue cleanup. A container of the shared `scheduler` image |
 | `infra-service` | An Ansible runner and SSH operations |

docs/CHANGELOG.md

@@ -11,6 +11,8 @@ See the CHANGELOG rule in [AGENTS.md](../AGENTS.md).
   `USERS_GRANT_CAPABILITY` instead of redeploying the product, so QA access no longer costs a deploy.
 - A temporary-access operation takes its Run's dispatch claim before any external write, and every revoke
   re-derives the earlier-workflow stop from all the grant's Runs, so withdrawal or retry cannot skip either.
+- scheduler-pipeline's `story_completion` loop also wakes on a native Pub/Sub hint on `scheduler:wake`;
+  hints are lossy and coalesced, the interval stays the backstop, and nothing publishes one yet.
 
 ## 2026-10-17
 

docs/CONTRACTS.md

@@ -37,6 +37,7 @@ Feature-specific invariants live in focused guides. Read this index first, then
 | Generated-product template/package installation and QA | [Generated product kit and QA contracts](contracts/kit-template-and-qa.md) |
 | Runtime lifecycle/security invariants | [Lifecycle and security invariants](contracts/lifecycle-invariants.md) |
 | Managed target reconciliation/readiness | [Managed target readiness](contracts/managed-target-readiness.md) |
+| Scheduler loop wake hints over native Pub/Sub | [Scheduler wake hints](contracts/scheduler-wake-hints.md) |
 
 ## Canonical vocabularies
 
@@ -70,6 +71,10 @@ are repository-relative.
 | `po:response:{request_id}` | direct response | `queues/po.py` | PO consumer | telegram-bot |
 | `po:proactive` | `tg-bot-proactive` | `queues/po.py` | PO notification tools | telegram-bot |
 
+The native Pub/Sub channel `scheduler:wake` (`queues/scheduler_wake.py`) is not a queue: its
+lossy hints only shorten a scheduler-pipeline loop's wait, see
+[Scheduler wake hints](contracts/scheduler-wake-hints.md).
+
 ### Recipient and PO rules
 
 Telegram admits bounded concurrent updates through PTB's native update processor.

docs/contracts/scheduler-wake-hints.md

@@ -0,0 +1,50 @@
+# Scheduler wake hints
+
+A wake hint asks one scheduler loop to run its next cycle now instead of at the end of its
+interval. It is an optimisation of latency only: the loop's interval sweep stays the recovery
+path, and the awakened cycle decides from the same durable Story/Task/Run state it always reads.
+
+## Wire contract
+
+- Transport: native Redis Pub/Sub (`PUBLISH`/`SUBSCRIBE`), channel `scheduler:wake`. It is not a
+  Redis Stream: nothing is stored, acknowledged, retried or replayed.
+- 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` only.
+- 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
+  write the hint announces has committed. A publish failure loses only the hint.
+
+## Ownership
+
+| Target | Logical producer | Consumer |
+|---|---|---|
+| `story_completion` | none yet: no business transition publishes a hint | scheduler-pipeline `story_completion` loop |
+
+scheduler-pipeline runs one `scheduler_wake` listener beside its loops under the same
+supervisor. It must hold an event for every target in the vocabulary, or it refuses to start.
+
+## Delivery and coalescing
+
+- Delivery is lossy: a hint published while no listener is subscribed, during a reconnect, or
+  before a restart is gone. The loop's `scheduler.dispatch_interval_seconds` sweep covers it.
+- A hint sets the target loop's `asyncio.Event`. The loop clears the event when a cycle starts,
+  so any number of hints during a cycle, or between its end and the next wait, yield exactly one
+  prompt follow-up cycle. Cycles never overlap and hints queue no work or tasks.
+- After a cycle the loop waits for its event or its interval, whichever comes first; the
+  interval is reread from configuration for every wait. There is no minimum delay on the wake
+  path.
+
+## Rejection and failure
+
+- A payload that is not text, exceeds 256 characters, is not JSON, names an unknown target or
+  carries an extra field is rejected and logged as `scheduler_wake_hint_rejected` with a reason
+  and validation shape only, never the payload. It wakes nothing and is never mapped to another
+  target.
+- A Redis connect, subscribe or read failure, or an ended subscription, closes that Pub/Sub and
+  Redis client and resubscribes after one `scheduler.dispatch_interval_seconds`. The loops keep
+  their own interval deadlines and keep sweeping meanwhile.
+- Cancellation propagates through the listener and the loop wait; each closes its Pub/Sub and
+  owned Redis client once and leaves no helper task running.

services/scheduler/src/pipeline.py

@@ -1,9 +1,11 @@
 """Pipeline scheduler service entry point."""
 
 import asyncio
+from functools import partial
 
 import structlog
 
+from shared.contracts.queues.scheduler_wake import SchedulerWakeTarget
 from shared.log_config import setup_logging
 
 from . import runtime
@@ -13,6 +15,7 @@
 from .tasks.pr_ci_loop import pr_ci_loop
 from .tasks.qa_routing_loop import qa_routing_loop
 from .tasks.scaffold_loop import scaffold_loop
+from .tasks.scheduler_wake_listener import scheduler_wake_listener
 from .tasks.story_completion_loop import story_completion_loop
 from .tasks.story_supervision_loop import story_supervision_loop
 from .tasks.task_dispatcher import task_dispatcher_loop
@@ -26,11 +29,19 @@ async def main() -> None:
     setup_logging(service_name="scheduler-pipeline")
     logger.info("scheduler_pipeline_started")
     await runtime.initialize_configs(PIPELINE_REQUIRED_KEYS, service_name="scheduler-pipeline")
+    story_completion_wake = asyncio.Event()
     await runtime.run_workers(
         [
             ("task_dispatcher", task_dispatcher_loop),
             ("scaffold", scaffold_loop),
-            ("story_completion", story_completion_loop),
+            ("story_completion", partial(story_completion_loop, wake=story_completion_wake)),
+            (
+                "scheduler_wake",
+                partial(
+                    scheduler_wake_listener,
+                    {SchedulerWakeTarget.STORY_COMPLETION: story_completion_wake},
+                ),
+            ),
             ("lifecycle_supervision", lifecycle_supervision_loop),
             ("qa_routing", qa_routing_loop),
             ("pr_ci", pr_ci_loop),

services/scheduler/src/runtime.py

@@ -48,30 +48,51 @@ async def periodic_loop(
     stopped_event: str,
     redis_factory: Callable[[], RedisStreamClient] = RedisStreamClient,
     sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    wake: asyncio.Event | None = None,
 ) -> None:
     """Run one Redis-backed periodic worker with a stable lifecycle.
 
     Each cycle starts with the message context unbound, so its messages never inherit
-    a correlation id bound before it, at startup or by the previous cycle.
+    a correlation id bound before it, at startup or by the previous cycle. With `wake`,
+    the wait after a cycle also ends when the event is set; a set during a cycle or
+    before the wait still runs one prompt follow-up cycle.
     """
     redis_client = redis_factory()
     await redis_client.connect()
     logger.info(started_event, interval=interval())
     try:
         while True:
             unbind_message_context()
+            if wake is not None:
+                wake.clear()
             try:
                 fields = await cycle(redis_client)
                 if cycle_event is not None:
                     logger.info(cycle_event, **dict(fields or {}))
             except Exception:
                 logger.exception(error_event)
-            await sleep(interval())
+            if wake is None:
+                await sleep(interval())
+            else:
+                await _wait_for_wake(wake, sleep(interval()))
     finally:
         await redis_client.close()
         logger.info(stopped_event)
 
 
+async def _wait_for_wake(wake: asyncio.Event, timer: Awaitable[None]) -> None:
+    """Return when `wake` is set or `timer` finishes, cancelling and awaiting the other."""
+    waiters = {asyncio.ensure_future(wake.wait()), asyncio.ensure_future(timer)}
+    try:
+        done, _pending = await asyncio.wait(waiters, return_when=asyncio.FIRST_COMPLETED)
+    finally:
+        for waiter in waiters:
+            waiter.cancel()
+        await asyncio.gather(*waiters, return_exceptions=True)
+    for waiter in done:
+        waiter.result()
+
+
 async def run_workers(
     workers: Collection[tuple[str, Callable[[], Awaitable[None]]]],
     *,

services/scheduler/src/tasks/scheduler_wake_listener.py

@@ -0,0 +1,69 @@
+"""Route native Pub/Sub wake hints to the scheduler loops that accept them."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable, Mapping
+
+from redis.exceptions import RedisError
+import structlog
+
+from shared.contracts.queues.scheduler_wake import (
+    SCHEDULER_WAKE_CHANNEL,
+    SchedulerWakeHintRejected,
+    SchedulerWakeTarget,
+    parse_scheduler_wake_hint,
+)
+from shared.redis import RedisStreamClient
+
+from .. import startup
+
+logger = structlog.get_logger(__name__)
+
+
+def _reconnect_interval() -> int:
+    return startup.get_config().get_int("scheduler.dispatch_interval_seconds")
+
+
+async def scheduler_wake_listener(
+    wakes: Mapping[SchedulerWakeTarget, asyncio.Event],
+    *,
+    redis_factory: Callable[[], RedisStreamClient] = RedisStreamClient,
+    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+) -> None:
+    """Set the event of each valid hint's target; Redis failures reconnect after one interval.
+
+    Every target in the vocabulary needs an event, so a hint never lacks its receiver.
+    """
+    if set(wakes) != set(SchedulerWakeTarget):
+        raise ValueError(
+            f"scheduler wake targets without an event: {set(SchedulerWakeTarget) - set(wakes)}"
+        )
+    while True:
+        redis_client = redis_factory()
+        try:
+            await redis_client.connect()
+            pubsub = redis_client.redis.pubsub()
+            try:
+                await pubsub.subscribe(SCHEDULER_WAKE_CHANNEL)
+                logger.info("scheduler_wake_subscribed", channel=SCHEDULER_WAKE_CHANNEL)
+                async for message in pubsub.listen():
+                    if message["type"] == "message":
+                        _route(message["data"], wakes)
+                logger.warning("scheduler_wake_subscription_ended")
+            finally:
+                await pubsub.aclose()
+        except (RedisError, OSError) as exc:
+            logger.warning("scheduler_wake_listener_disconnected", error_type=type(exc).__name__)
+        finally:
+            await redis_client.close()
+        await sleep(_reconnect_interval())
+
+
+def _route(data: object, wakes: Mapping[SchedulerWakeTarget, asyncio.Event]) -> None:
+    try:
+        hint = parse_scheduler_wake_hint(data)
+    except SchedulerWakeHintRejected as rejected:
+        logger.warning("scheduler_wake_hint_rejected", **rejected.diagnostics)
+        return
+    wakes[hint.target].set()

services/scheduler/src/tasks/story_completion_loop.py

@@ -18,8 +18,8 @@ def _story_completion_interval() -> int:
     return startup.get_config().get_int("scheduler.dispatch_interval_seconds")
 
 
-async def story_completion_loop() -> None:
-    """Complete eligible stories from durable task/story state."""
+async def story_completion_loop(*, wake: asyncio.Event) -> None:
+    """Complete eligible stories from durable task/story state, early when `wake` is set."""
     from ..clients.api import api_client
 
     async def cycle(redis_client: RedisStreamClient) -> dict[str, object]:
@@ -35,4 +35,5 @@ async def cycle(redis_client: RedisStreamClient) -> dict[str, object]:
         stopped_event="story_completion_stopped",
         redis_factory=RedisStreamClient,
         sleep=asyncio.sleep,
+        wake=wake,
     )