feat(planner): schedule batch work on native ticks

components/src/dynamo/planner/core/base.py

@@ -564,6 +564,7 @@ async def _gather_tick_input(self, tick: ScheduledTick) -> TickInput:
         traffic = None
         worker_counts = None
         fpm_obs = None
+        batch = None
 
         if tick.need_traffic_metrics:
             if tick.use_full_traffic_metrics:
@@ -576,11 +577,22 @@ async def _gather_tick_input(self, tick: ScheduledTick) -> TickInput:
             worker_counts = await self._collect_worker_counts()
         if tick.need_worker_fpm:
             fpm_obs = self._collect_fpm()
+        if tick.need_batch_scheduling:
+            try:
+                batch = await self.environment.collect_batch_scheduling()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception(
+                    "Batch scheduling observation failed; continuing fail-closed"
+                )
+            now = time.time()
         return TickInput(
             now_s=now,
             traffic=traffic,
             worker_counts=worker_counts,
             fpm_observations=fpm_obs,
+            batch=batch,
         )
 
     async def _observe_tick(
@@ -803,6 +815,7 @@ def _should_emit_tick_diagnostics(
             tick.run_load_scaling
             or tick.run_throughput_scaling
             or effects.scale_to is not None
+            or bool(effects.batch_drain_limits)
             or bool(diag.audit_events)
             or bool(diag.short_circuit_reason)
         )
@@ -822,7 +835,29 @@ async def _run_one_tick(
         # lock is released before connector rollouts that may take minutes.
         async with self._config_lock:
             effects = await engine.tick(tick, tick_input)
-        await self._apply_effects(effects)
+        batch_actuation_succeeded = True
+        if effects.batch_drain_limits and not self.config.advisory:
+            try:
+                # Apply the leased admission decision before any replica
+                # mutation. If Redis is unavailable, retain the current fleet;
+                # the previous lease expires fail-closed and the next tick
+                # retries without terminating the Planner loop.
+                await self.environment.apply_batch_drain_limits(
+                    effects.batch_drain_limits
+                )
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                batch_actuation_succeeded = False
+                logger.exception(
+                    "Batch drain-limit actuation failed; skipping replica "
+                    "actuation for this tick"
+                )
+                effects.diagnostics.audit_events.append(
+                    f"batch_drain_actuation_failed:{type(exc).__name__}"
+                )
+        if batch_actuation_succeeded:
+            await self._apply_effects(effects)
         emit_diagnostics = self._should_emit_tick_diagnostics(tick, effects)
         if emit_diagnostics:
             self._report_diagnostics(tick, effects.diagnostics)

components/src/dynamo/planner/plugins/builtins/observe.py

@@ -5,12 +5,18 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
+import time
+from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Protocol
 
 from dynamo.planner.core.types import ScheduledTick, TickInput, WorkerCounts
 from dynamo.planner.environment.interface import PlannerEnvironment
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass
 class ObserveStageRequest:
@@ -47,16 +53,19 @@ def __init__(
         *,
         require_prefill: bool,
         require_decode: bool,
+        clock: Callable[[], float] = time.time,
     ) -> None:
         self.environment = environment
         self.require_prefill = require_prefill
         self.require_decode = require_decode
+        self._clock = clock
 
     async def Observe(self, request: ObserveStageRequest) -> ObserveStageResponse:
         tick = request.scheduled_tick
         traffic = None
         worker_counts = None
         fpm_observations = None
+        batch = None
 
         if tick.need_traffic_metrics:
             if tick.use_full_traffic_metrics:
@@ -69,13 +78,36 @@ async def Observe(self, request: ObserveStageRequest) -> ObserveStageResponse:
             worker_counts = self._collect_worker_counts()
         if tick.need_worker_fpm:
             fpm_observations = self.environment.collect_fpm()
+        if tick.need_batch_scheduling:
+            try:
+                batch = await self.environment.collect_batch_scheduling()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                # Batch policy treats a missing snapshot as an explicit safety
+                # pause. Keep unrelated worker/traffic observations available
+                # and let the native tick renew a zero-rate lease.
+                logger.exception(
+                    "Batch scheduling observation failed; continuing fail-closed"
+                )
+
+        # Batch source timestamps are sampled during collection. Anchor policy
+        # freshness and lease expiry after that I/O so a valid observation can
+        # never appear to come from the future and the full lease remains for
+        # the decision/actuation phases.
+        tick_now_s = (
+            max(request.now_s, self._clock())
+            if tick.need_batch_scheduling
+            else request.now_s
+        )
 
         return ObserveStageResponse(
             tick_input=TickInput(
-                now_s=request.now_s,
+                now_s=tick_now_s,
                 traffic=traffic,
                 worker_counts=worker_counts,
                 fpm_observations=fpm_observations,
+                batch=batch,
             )
         )