sachalmalick · Oct 10, 2026 · Aug 31, 2026
@@ -3,9 +3,14 @@
 
 from dynamo.planner.core.state_machine import PlannerScalingState
 from dynamo.planner.core.types import (
+    BatchDispatcherFeedback,
+    BatchDrainLimitDecision,
+    BatchJobDemand,
+    BatchSchedulingObservation,
     EngineCapabilities,
     FpmObservations,
     PlannerEffects,
+    PoolTrafficDemand,
     ScalingDecision,
     ScheduledTick,
     TickInput,
@@ -15,10 +20,15 @@
 )
 
 __all__ = [
+    "BatchDispatcherFeedback",
+    "BatchDrainLimitDecision",
+    "BatchJobDemand",
+    "BatchSchedulingObservation",
     "EngineCapabilities",
     "FpmObservations",
     "PlannerEffects",
     "PlannerScalingState",
+    "PoolTrafficDemand",
     "ScalingDecision",
     "ScheduledTick",
     "TickInput",

@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import math
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Optional
 
@@ -88,6 +89,144 @@ class FpmObservations:
     decode: Optional[dict[tuple[str, int], ForwardPassMetrics]] = None
 
 
+@dataclass
+class BatchJobDemand:
+    """Batch job demand observed for one inference pool.
+
+    ``observed_at_s`` and ``deadline_at_s`` are absolute wall-clock
+    timestamps. ``deadline_at_s=None`` means the job has no SLA deadline.
+    ``work_class`` is an opaque request/workload class understood by the
+    capacity model; it is deliberately not tied to a Batch Gateway enum.
+
+    The raw counters and status are retained for auditability.
+    ``remaining_requests`` is derived so contradictory observations cannot
+    enter the planner contract.
+    """
+
+    observed_at_s: float
+    pool_id: str
+    job_id: str
+    status: str
+    total_requests: int
+    completed_requests: int
+    failed_requests: int
+    deadline_at_s: Optional[float]
+    work_class: str
+
+    def __post_init__(self) -> None:
+        _require_finite_non_negative("observed_at_s", self.observed_at_s)
+        _require_non_empty_string("pool_id", self.pool_id)
+        _require_non_empty_string("job_id", self.job_id)
+        _require_non_empty_string("status", self.status)
+        _require_non_empty_string("work_class", self.work_class)
+        counters = (
+            self.total_requests,
+            self.completed_requests,
+            self.failed_requests,
+        )
+        if any(
+            isinstance(count, bool) or not isinstance(count, int) or count < 0
+            for count in counters
+        ):
+            raise ValueError("batch request counters must be non-negative integers")
+        if self.completed_requests + self.failed_requests > self.total_requests:
+            raise ValueError(
+                "completed_requests + failed_requests must not exceed total_requests"
+            )
+        if self.deadline_at_s is not None:
+            _require_finite_non_negative("deadline_at_s", self.deadline_at_s)
+
+    @property
+    def remaining_requests(self) -> int:
+        """Requests not yet reported completed or failed."""
+
+        return self.total_requests - self.completed_requests - self.failed_requests
+
+
+@dataclass
+class PoolTrafficDemand:
+    """Online (non-batch) offered load for one inference pool.
+
+    This is intentionally separate from Batch Gateway state. The Planner
+    derives safe capacity from its existing worker and capability inputs.
+    ``observed_at_s`` is an absolute wall-clock timestamp.
+    """
+
+    observed_at_s: float
+    pool_id: str
+    online_offered_rps: float
+    # Current frontend concurrency for the selected pool. ``None`` preserves
+    # compatibility with observation producers that only report offered load;
+    # ``0`` is an explicit, observed idle frontend.
+    frontend_active_requests: Optional[int] = None
+
+    def __post_init__(self) -> None:
+        _require_finite_non_negative("observed_at_s", self.observed_at_s)
+        _require_non_empty_string("pool_id", self.pool_id)
+        _require_finite_non_negative("online_offered_rps", self.online_offered_rps)
+        if self.frontend_active_requests is not None and (
+            isinstance(self.frontend_active_requests, bool)
+            or not isinstance(self.frontend_active_requests, int)
+            or self.frontend_active_requests < 0
+        ):
+            raise ValueError(
+                "frontend_active_requests must be a non-negative integer or None"
+            )
+
+
+@dataclass
+class BatchDispatcherFeedback:
+    """Observed dispatcher state for one inference pool.
+
+    ``actual_dispatch_rps`` is the achieved batch admission rate over
+    ``observation_window_s``. ``applied_max_admission_rps=None`` means the
+    dispatcher did not report an applied Planner cap; ``0.0`` means an
+    explicit pause. ``observed_at_s`` is an absolute wall-clock timestamp.
+    """
+
+    observed_at_s: float
+    pool_id: str
+    observation_window_s: float
+    queued_requests: int
+    inflight_requests: int
+    actual_dispatch_rps: float
+    applied_max_admission_rps: Optional[float] = None
+
+    def __post_init__(self) -> None:
+        _require_finite_non_negative("observed_at_s", self.observed_at_s)
+        _require_non_empty_string("pool_id", self.pool_id)
+        _require_positive_finite("observation_window_s", self.observation_window_s)
+        if any(
+            isinstance(count, bool) or not isinstance(count, int) or count < 0
+            for count in (self.queued_requests, self.inflight_requests)
+        ):
+            raise ValueError("dispatcher request counts must be non-negative integers")
+        _require_finite_non_negative("actual_dispatch_rps", self.actual_dispatch_rps)
+        if self.applied_max_admission_rps is not None:
+            _require_finite_non_negative(
+                "applied_max_admission_rps", self.applied_max_admission_rps
+            )
+
+
+@dataclass
+class BatchSchedulingObservation:
+    """Batch scheduling inputs sampled independently by their adapters.
+
+    The durable job source is authoritative: a present observation means its
+    ``job_demands`` snapshot succeeded, including the valid empty-list case.
+    Traffic and dispatcher entries are independently optional per pool. An
+    absent pool entry means that source is unknown or unusable for this tick;
+    it must never be interpreted as a numeric zero. A known idle pool is
+    represented explicitly by ``PoolTrafficDemand.online_offered_rps == 0``;
+    frontend concurrency is independently unknown when
+    ``frontend_active_requests is None`` and explicitly idle when it is zero.
+    """
+
+    job_demands: list[BatchJobDemand] = field(default_factory=list)
+    pool_traffic: list[PoolTrafficDemand] = field(default_factory=list)
+    dispatcher_feedback: list[BatchDispatcherFeedback] = field(default_factory=list)
+
+
 @dataclass
 class TickInput:
     """What the adapter provides to the core on each tick.
@@ -100,6 +239,7 @@ class TickInput:
     traffic: Optional[TrafficObservation] = None
     worker_counts: Optional[WorkerCounts] = None
     fpm_observations: Optional[FpmObservations] = None
+    batch: Optional[BatchSchedulingObservation] = None
 
 
 @dataclass
@@ -112,6 +252,49 @@ class ScalingDecision:
     num_decode: Optional[int] = None
 
 
+@dataclass
+class BatchDrainLimitDecision:
+    """Leased batch admission limit for one inference pool.
+
+    ``max_admission_rps`` is the maximum rate at which new batch requests
+    may be admitted; ``0.0`` explicitly pauses admission. ``valid_until_s``
+    is an absolute wall-clock lease expiry. The dispatcher must stop using
+    the limit after expiry. ``decision_id`` identifies the decision for
+    idempotent application and audit correlation.
+    """
+
+    pool_id: str
+    max_admission_rps: float
+    valid_until_s: float
+    decision_id: str
+
+    def __post_init__(self) -> None:
+        _require_non_empty_string("pool_id", self.pool_id)
+        _require_non_empty_string("decision_id", self.decision_id)
+        _require_finite_non_negative("max_admission_rps", self.max_admission_rps)
+        _require_finite_non_negative("valid_until_s", self.valid_until_s)
+
+
+def _require_non_empty_string(name: str, value: object) -> None:
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"{name} must be a non-empty string")
+
+
+def _require_finite_non_negative(name: str, value: object) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError(f"{name} must be numeric")
+    normalized = float(value)
+    if not math.isfinite(normalized) or normalized < 0:
+        raise ValueError(f"{name} must be finite and non-negative")
+    return normalized
+
+
+def _require_positive_finite(name: str, value: object) -> None:
+    normalized = _require_finite_non_negative(name, value)
+    if normalized <= 0:
+        raise ValueError(f"{name} must be positive and finite")
+
+
 @dataclass
 class TickDiagnostics:
     """Intermediate decision data populated by the planner core for
@@ -201,6 +384,7 @@ class PlannerEffects:
     scale_to: Optional[ScalingDecision] = None
     next_tick: Optional[ScheduledTick] = None
     diagnostics: TickDiagnostics = field(default_factory=TickDiagnostics)
+    batch_drain_limits: list[BatchDrainLimitDecision] = field(default_factory=list)
 
 
 @dataclass

@@ -57,6 +57,10 @@
     pyd.TrafficMetrics: pb.TrafficMetrics,
     pyd.FpmData: pb.FpmData,
     pyd.WorkerState: pb.WorkerState,
+    pyd.BatchJobDemand: pb.BatchJobDemand,
+    pyd.PoolTrafficDemand: pb.PoolTrafficDemand,
+    pyd.BatchDispatcherFeedback: pb.BatchDispatcherFeedback,
+    pyd.BatchSchedulingData: pb.BatchSchedulingData,
     pyd.ObservationData: pb.ObservationData,
     pyd.PredictionData: pb.PredictionData,
     pyd.ComponentTarget: pb.ComponentTarget,

@@ -77,6 +77,7 @@
 )
 from dynamo.planner.core.state_machine import PlannerScalingState
 from dynamo.planner.core.types import (
+    BatchSchedulingObservation,
     FpmObservations,
     PlannerEffects,
     ScalingDecision,
@@ -106,15 +107,20 @@
 from dynamo.planner.plugins.scheduler import PluginScheduler
 from dynamo.planner.plugins.transport.config import make_transport_for_endpoint
 from dynamo.planner.plugins.types import (
+    BatchDispatcherFeedback as BatchDispatcherFeedbackData,
+)
+from dynamo.planner.plugins.types import BatchJobDemand as BatchJobDemandData
+from dynamo.planner.plugins.types import (
+    BatchSchedulingData,
     ComponentTarget,
     FpmData,
     ObservationData,
     OverrideResult,
     OverrideType,
     PipelineContext,
-    TrafficMetrics,
-    WorkerState,
 )
+from dynamo.planner.plugins.types import PoolTrafficDemand as PoolTrafficDemandData
+from dynamo.planner.plugins.types import TrafficMetrics, WorkerState
 
 log = logging.getLogger(__name__)
 
@@ -854,10 +860,62 @@ def _tick_input_to_context(self, ti: TickInput) -> PipelineContext:
         # and could not implement load-based decisions through
         # the public PipelineContext API.
         fpm = self._encode_fpm(ti.fpm_observations)
+        batch = self._encode_batch(ti.batch)
         return PipelineContext(
             request_id=f"tick-{ti.now_s}",
             decision_id=f"d-{ti.now_s}",
-            observations=ObservationData(traffic=traffic, fpm=fpm, workers=workers),
+            observations=ObservationData(
+                traffic=traffic,
+                fpm=fpm,
+                workers=workers,
+                batch=batch,
+            ),
+        )
+
+    @staticmethod
+    def _encode_batch(
+        obs: Optional[BatchSchedulingObservation],
+    ) -> Optional[BatchSchedulingData]:
+        """Project generic core batch observations onto the plugin contract."""
+
+        if obs is None:
+            return None
+        return BatchSchedulingData(
+            job_demands=[
+                BatchJobDemandData(
+                    observed_at_s=job.observed_at_s,
+                    pool_id=job.pool_id,
+                    job_id=job.job_id,
+                    status=job.status,
+                    total_requests=job.total_requests,
+                    completed_requests=job.completed_requests,
+                    failed_requests=job.failed_requests,
+                    deadline_at_s=job.deadline_at_s,
+                    work_class=job.work_class,
+                    remaining_requests=job.remaining_requests,
+                )
+                for job in obs.job_demands
+            ],
+            pool_traffic=[
+                PoolTrafficDemandData(
+                    observed_at_s=pool.observed_at_s,
+                    pool_id=pool.pool_id,
+                    online_offered_rps=pool.online_offered_rps,
+                )
+                for pool in obs.pool_traffic
+            ],
+            dispatcher_feedback=[
+                BatchDispatcherFeedbackData(
+                    observed_at_s=feedback.observed_at_s,
+                    pool_id=feedback.pool_id,
+                    observation_window_s=feedback.observation_window_s,
+                    queued_requests=feedback.queued_requests,
+                    inflight_requests=feedback.inflight_requests,
+                    actual_dispatch_rps=feedback.actual_dispatch_rps,
+                    applied_max_admission_rps=feedback.applied_max_admission_rps,
+                )
+                for feedback in obs.dispatcher_feedback
+            ],
         )
 
     @staticmethod

@@ -1,3 +1,8 @@
+<!--
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+-->
+
 # Plugin Proto v1
 
 Plugin RPC contract for the Dynamo Planner plugin framework.
@@ -38,7 +43,7 @@ Total: **6 services / 33 messages / 3 enums**
 ### Messages — by category
 
 - **PluginRegistry**: `RegisterRequest` / `RegisterResponse` / `HeartbeatRequest` / `HeartbeatResponse` / `UnregisterRequest` / `UnregisterResponse` / `ListPluginsRequest` / `ListPluginsResponse` / `PluginInfo`
-- **PipelineContext + observation**: `PipelineContext` / `ObservationData` / `TrafficMetrics` / `FpmData` / `WorkerState` / `PredictionData` / `ScalingProposal` / `ComponentTarget` / `OverrideResult` / `AcceptResult` / `RejectResult`
+- **PipelineContext + observation**: `PipelineContext` / `ObservationData` / `TrafficMetrics` / `FpmData` / `WorkerState` / `BatchSchedulingData` / `BatchJobDemand` / `PoolTrafficDemand` / `BatchDispatcherFeedback` / `PredictionData` / `ScalingProposal` / `ComponentTarget` / `OverrideResult` / `AcceptResult` / `RejectResult`
 - **Stage request/response**: `PredictStageRequest` / `PredictStageResponse` / `ProposeStageRequest` / `ProposeStageResponse` / `ProposeResult` / `ReconcileStageRequest` / `ReconcileStageResponse` / `ConstrainStageRequest` / `ConstrainStageResponse`
 - **PluginLifecycle**: `BootstrapRequest` / `BootstrapResponse` / `ResetRequest` / `ResetResponse`
 
@@ -235,6 +240,15 @@ Plugins that declare `needs=["observations.fpm"]` receive the per-engine
 maps when available; absence of the field means no FPM data was collected
 for that tick.
 
+## Batch scheduling observations
+
+`ObservationData.batch` carries flat, independently timestamped job demand,
+online pool traffic, and dispatcher feedback. Plugins can declare
+`needs=["observations.batch"]` and join records by `pool_id`. Job deadlines and
+all `observed_at_s` values are absolute wall-clock timestamps. Dispatcher
+`applied_max_admission_rps` preserves proto presence: unset means no applied cap
+was reported, while `0.0` means admission was explicitly paused.
+
 ## References
 
 - `dynamo/planner/plugins/types.py` — Pydantic v2 mirror