diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 3e1be99b38..626ba6fd1c 100644 --- a/docs/environment-variables.md +++ b/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). diff --git a/src/aiperf/common/environment.py b/src/aiperf/common/environment.py index a7c6510188..03e6b56a6d 100644 --- a/src/aiperf/common/environment.py +++ b/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", diff --git a/src/aiperf/timing/strategies/agentic_replay.py b/src/aiperf/timing/strategies/agentic_replay.py index 462ef54cb9..acf84b45d2 100644 --- a/src/aiperf/timing/strategies/agentic_replay.py +++ b/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: diff --git a/tests/unit/common/test_environment.py b/tests/unit/common/test_environment.py index cafcd59ead..18f176e9c2 100644 --- a/tests/unit/common/test_environment.py +++ b/tests/unit/common/test_environment.py @@ -7,6 +7,7 @@ from pytest import param from aiperf.common.environment import ( + _AgenticSettings, _APIServerSettings, _CompressionSettings, _Environment, @@ -14,6 +15,28 @@ ) +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.""" diff --git a/tests/unit/timing/strategies/test_agentic_replay.py b/tests/unit/timing/strategies/test_agentic_replay.py index cab69e91da..f591be441f 100644 --- a/tests/unit/timing/strategies/test_agentic_replay.py +++ b/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