fix(agentic): pace snapshot warmup dispatches

docs/environment-variables.md

@@ -22,6 +22,14 @@ export AIPERF_ZMQ_RCVTIMEO=600000
 > Environment variable names, default values, and definitions are subject to change.
 > These settings may be modified, renamed, or removed in future releases.
 
+## AGENTIC
+
+Settings for agentic replay timing behavior.
+
+| Environment Variable | Default | Constraints | Description |
+|----------------------|---------|-------------|-------------|
+| `AIPERF_AGENTIC_SNAPSHOT_WARMUP_MIN_INTERVAL_S` | `0.0` | ≥ 0.0 | Minimum interval in seconds between snapshot warmup request dispatches. Values greater than zero extend the timestamp-derived warmup ramp when needed, reducing synchronized connection bursts at high concurrency. The default 0 preserves trace-derived dispatch timing. |
+
 ## AGENTX
 
 Settings for the InferenceX AgentX scenario family. Controls runtime detection knobs for the agentx scenario, currently the substring allowlist used to classify a server response as a context-overflow error (RFC 2026-04-26 §7).

src/aiperf/common/environment.py

@@ -7,6 +7,7 @@
 All settings can be configured via environment variables with the AIPERF_ prefix.
 
 Structure:
+    Environment.AGENTIC.*        - Agentic replay settings
     Environment.AGENTX.*         - InferenceX AgentX scenario settings
     Environment.API_SERVER.*     - API server settings
     Environment.COMPRESSION.*    - Compression settings for streaming file transfers
@@ -90,6 +91,23 @@ class _APIServerSettings(BaseSettings):
     )
 
 
+class _AgenticSettings(BaseSettings):
+    """Settings for agentic replay timing behavior."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="AIPERF_AGENTIC_",
+    )
+
+    SNAPSHOT_WARMUP_MIN_INTERVAL_S: float = Field(
+        ge=0.0,
+        default=0.0,
+        description="Minimum interval in seconds between snapshot warmup request "
+        "dispatches. Values greater than zero extend the timestamp-derived warmup "
+        "ramp when needed, reducing synchronized connection bursts at high "
+        "concurrency. The default 0 preserves trace-derived dispatch timing.",
+    )
+
+
 class _AgentXSettings(BaseSettings):
     """Settings for the InferenceX AgentX scenario family.
 
@@ -1323,6 +1341,10 @@ class _Environment(BaseSettings):
     )
 
     # Nested subsystem settings (alphabetically ordered)
+    AGENTIC: _AgenticSettings = Field(
+        default_factory=_AgenticSettings,
+        description="Agentic replay settings",
+    )
     AGENTX: _AgentXSettings = Field(
         default_factory=_AgentXSettings,
         description="InferenceX AgentX scenario settings",

src/aiperf/timing/strategies/agentic_replay.py

@@ -533,19 +533,14 @@ async def _execute_warmup(self) -> None:
         # the same instant (warmup-time ``max_lead``); the furthest-before-t*
         # request fires at 0. Do NOT mark sending complete -- the count path
         # finalizes once the last scheduled dispatch fires.
-        leads = [d for _, d in prepared if d is not None]
-        max_lead_ms = max(leads, default=0.0)
-        spread_s = (max_lead_ms - min(leads, default=0.0)) / MILLIS_PER_SECOND
+        dispatches, aligned_request_count = self._warmup_dispatches(prepared)
+        spread_s = max((offset_s for offset_s, _ in dispatches), default=0.0)
         self.info(
-            f"WARMUP spread: {spread_s:.1f}s ramp aligning {len(leads)} request(s) "
+            f"WARMUP spread: {spread_s:.1f}s ramp aligning "
+            f"{aligned_request_count} request(s) "
             f"on t* (earliest fires at 0, last at {spread_s:.1f}s)"
         )
-        for turn, lead_ms in prepared:
-            offset_s = (
-                (max_lead_ms - lead_ms) / MILLIS_PER_SECOND
-                if lead_ms is not None
-                else 0.0
-            )
+        for offset_s, turn in dispatches:
             if offset_s > 0:
                 self.scheduler.schedule_later(
                     offset_s, self.credit_issuer.issue_credit(turn)
@@ -554,6 +549,34 @@ async def _execute_warmup(self) -> None:
                 await self.credit_issuer.issue_credit(turn)
         await self._start_accelerated_warmup_if_empty(prepared)
 
+    @staticmethod
+    def _warmup_dispatches(
+        prepared: list[tuple[TurnToSend, float | None]],
+    ) -> tuple[list[tuple[float, TurnToSend]], int]:
+        """Build chronological warmup offsets and apply optional throttling."""
+        leads = [lead_ms for _, lead_ms in prepared if lead_ms is not None]
+        max_lead_ms = max(leads, default=0.0)
+        dispatches = [
+            (
+                (max_lead_ms - lead_ms) / MILLIS_PER_SECOND
+                if lead_ms is not None
+                else 0.0,
+                turn,
+            )
+            for turn, lead_ms in prepared
+        ]
+        min_interval_s = Environment.AGENTIC.SNAPSHOT_WARMUP_MIN_INTERVAL_S
+        if min_interval_s <= 0:
+            return dispatches, len(leads)
+
+        next_offset_s = 0.0
+        throttled_dispatches: list[tuple[float, TurnToSend]] = []
+        for offset_s, turn in sorted(dispatches, key=lambda item: item[0]):
+            offset_s = max(offset_s, next_offset_s)
+            throttled_dispatches.append((offset_s, turn))
+            next_offset_s = offset_s + min_interval_s
+        return throttled_dispatches, len(leads)
+
     async def _finish_initial_warmup_dispatch(
         self, prepared: list[tuple[TurnToSend, float | None]]
     ) -> None:

tests/unit/common/test_environment.py

@@ -7,13 +7,36 @@
 from pytest import param
 
 from aiperf.common.environment import (
+    _AgenticSettings,
     _APIServerSettings,
     _CompressionSettings,
     _Environment,
     _ServiceSettings,
 )
 
 
+class TestAgenticSettings:
+    """Test agentic replay environment settings."""
+
+    def test_snapshot_warmup_min_interval_default(self) -> None:
+        settings = _AgenticSettings()
+        assert settings.SNAPSHOT_WARMUP_MIN_INTERVAL_S == 0.0
+
+    def test_snapshot_warmup_min_interval_env_override(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("AIPERF_AGENTIC_SNAPSHOT_WARMUP_MIN_INTERVAL_S", "0.05")
+        settings = _AgenticSettings()
+        assert settings.SNAPSHOT_WARMUP_MIN_INTERVAL_S == 0.05
+
+    def test_snapshot_warmup_min_interval_rejects_negative_value(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("AIPERF_AGENTIC_SNAPSHOT_WARMUP_MIN_INTERVAL_S", "-0.01")
+        with pytest.raises(ValueError):
+            _AgenticSettings()
+
+
 class TestServiceSettingsUvloopWindows:
     """Test suite for automatic uvloop disabling on Windows."""
 

tests/unit/timing/strategies/test_agentic_replay.py

@@ -17,6 +17,7 @@
 import pytest
 
 from aiperf.common.enums import CacheBustTarget, ConversationBranchMode, CreditPhase
+from aiperf.common.environment import Environment
 from aiperf.common.models import (
     ConversationMetadata,
     DatasetMetadata,
@@ -666,6 +667,28 @@ async def capture(turn):
     lifecycle.mark_sending_complete.assert_not_called()
 
 
+def test_warmup_min_interval_extends_dispatch_ramp(monkeypatch):
+    """Throttling follows chronological offsets, not trajectory list order."""
+    turns = [
+        MagicMock(spec=TurnToSend, x_correlation_id="r_a"),
+        MagicMock(spec=TurnToSend, x_correlation_id="r_b"),
+        MagicMock(spec=TurnToSend, x_correlation_id="r_c"),
+    ]
+    prepared = [
+        (turns[0], 5_000.0),
+        (turns[1], 15_000.0),
+        (turns[2], 10_000.0),
+    ]
+    monkeypatch.setattr(Environment.AGENTIC, "SNAPSHOT_WARMUP_MIN_INTERVAL_S", 6.0)
+
+    dispatches, aligned_request_count = AgenticReplayStrategy._warmup_dispatches(
+        prepared
+    )
+
+    assert aligned_request_count == 3
+    assert dispatches == [(0.0, turns[1]), (6.0, turns[2]), (12.0, turns[0])]
+
+
 @pytest.mark.asyncio
 async def test_warmup_lead_clamped_to_idle_gap_cap():
     """A per-conversation idle far exceeding the idle-gap cap is clamped so the