feat(agentic): replay api-time overlap barriers

docs/architecture.md

@@ -190,7 +190,7 @@ The `CreditIssuer` is timing-mode agnostic; the strategy in `src/aiperf/timing/s
 - **Fixed schedule** (`fixed_schedule.py`): replay trace timestamps from dataset metadata.
 - **Request-rate** (`request_rate.py`): issue at a target rate with constant / Poisson / gamma / concurrency-burst arrival patterns.
 - **User-centric rate** (`user_centric_rate.py`): each session is an independent user; turn gaps come from the trace.
-- **Agentic replay** (`agentic_replay.py`): scenario-driven DAG replay where children are dispatched on parent completion via the `BranchOrchestrator`.
+- **Agentic replay** (`agentic_replay.py`): scenario-driven DAG replay. Weka loaders persist interval-order predecessor frontiers derived from `[t, t + api_time]`; `ReplayBarrierCoordinator` enforces their fan-out/join barriers, while `BranchOrchestrator` starts branches from the parent send when their recorded intervals overlap and otherwise uses the normal completion path.
 
 #### Relationship to `--request-count`, `--num-conversations`, Concurrency
 

docs/benchmark-modes/timing-modes-reference.md

@@ -246,6 +246,23 @@ aiperf profile \
 
 **How it works:** The strategy picks `--concurrency` distinct conversations as *trajectories*, samples a per-trajectory starting turn `k_i` somewhere between 25% and 75% of each conversation (the default `--trajectory-start-min-ratio` / `--trajectory-start-max-ratio` window, clamped to leave at least one profile turn after warmup), and warms each trajectory by dispatching that one turn before profiling starts. During profiling, each trajectory resumes from `k_i + 1` and replays the remaining turns honoring the trace's recorded request-start schedule after applying the per-trace idle-gap rule. The default `--trace-idle-gap-cap-seconds` is `None` (no compression); the `inferencex-agentx-mvp` scenario locks it to `60` so coffee-break request-start gaps don't distort steady-state while preserving local subagent overlap.
 
+Weka replay also preserves the capture's fan-out/join shape. The loader compares
+request intervals `[t, t + api_time]` within each logical agent or subagent
+scope and records an explicit cross-stream completion frontier on every turn.
+Requests whose intervals overlap have no ordering edge and may execute in
+parallel. A later request waits until every request on its recorded predecessor
+frontier reaches a terminal outcome. Exact interval-boundary touches are
+sequential. Long transitive overlaps use this dependency frontier rather than
+an overlap connected-component, so a long request can overlap multiple
+sequential requests on another stream without launching all of them at once.
+Branches that began while their spawning request was in flight are scheduled
+from that request's send time; independent subagent scopes are not globally
+joined. The same barriers remain active during accelerated cache-pressure
+warmup, where idle delays are otherwise removed. When replay resumes at a
+sampled `t*` or crosses from accelerated warmup into profiling, the snapshot
+seeds each stream's exact completed prefix before any request can dispatch;
+request arrival order therefore cannot weaken or deadlock a join barrier.
+
 Concurrency is **per session tree**: each `--concurrency` lane holds one slot for a whole tree — the root conversation plus every subagent it spawns (children, subchildren, background `::fa:`/`::aux:` sidecars). A lane's slot is released, and the lane recycled into a fresh root, only once the entire tree drains (root terminal **and** all descendants returned) — so a background subagent that outlives its root does not free the lane early. Recycle then draws the next root from the dataset sampler (honoring the dataset's `sampling_strategy` — sequential / shuffle / random), starting it from turn 0. This keeps exactly `--concurrency` trees live at all times. The shared tree id (`root_correlation_id`) is persisted per record in `profile_export.jsonl`, so `aiperf analyze swim-lane` groups each tree under one lane and renders exactly `--concurrency` slots.
 
 **When to use:** A scenario-locked timing mode for multi-turn agentic-coding traces (currently WEKA), especially long runs where you want steady-state metrics rather than first-turn-only metrics. Pairs naturally with `--cache-bust first_turn_prefix` (auto-injected by the `inferencex-agentx-mvp` scenario) so recycled plays don't progressively warm the server's KV-cache prefix on identical content.

src/aiperf/common/models/__init__.py

@@ -20,6 +20,7 @@
     InputsFile,
     Media,
     MemoryMapClientMetadata,
+    ReplayTurnReference,
     SessionPayloads,
     Text,
     Turn,
@@ -200,6 +201,7 @@
     "Media",
     "MediaCounts",
     "MemoryMapClientMetadata",
+    "ReplayTurnReference",
     "MetricFamily",
     "MetricRecordInfo",
     "MetricRecordMetadata",

src/aiperf/common/models/dataset_models.py

@@ -111,6 +111,13 @@ class Video(Media):
     media_type: ClassVar[MediaTypeT] = MediaType.VIDEO
 
 
+class ReplayTurnReference(AIPerfBaseModel):
+    """Dataset-stable reference to one request in a replay dependency graph."""
+
+    conversation_id: str = Field(description="Referenced conversation ID.")
+    turn_index: int = Field(ge=0, description="Referenced turn index.")
+
+
 class TurnMetadata(AIPerfBaseModel):
     """Metadata of a turn."""
 
@@ -134,6 +141,13 @@ class TurnMetadata(AIPerfBaseModel):
             "without per-request timing."
         ),
     )
+    replay_predecessors: list["ReplayTurnReference"] = Field(
+        default_factory=list,
+        description=(
+            "Cross-stream requests that reached a recorded terminal outcome before "
+            "this request began and must complete before agentic replay may issue it."
+        ),
+    )
     branch_ids: list[str] = Field(
         default_factory=list,
         description="Branch IDs triggered after this turn completes (DAG projection).",
@@ -207,6 +221,14 @@ class Turn(AIPerfBaseModel):
             "duration (not warped). None for loaders without per-request timing."
         ),
     )
+    replay_predecessors: list["ReplayTurnReference"] = Field(
+        default_factory=list,
+        exclude=True,
+        description=(
+            "Explicit cross-stream completion frontier inferred from recorded "
+            "request intervals by trace-aware loaders."
+        ),
+    )
     max_tokens: int | None = Field(
         default=None, description="Maximum number of tokens to generate for this turn."
     )
@@ -309,6 +331,7 @@ def metadata(self) -> TurnMetadata:
             timestamp_ms=self.timestamp,
             delay_ms=self.delay,
             api_time_ms=self.api_time_ms,
+            replay_predecessors=self.replay_predecessors,
             branch_ids=self.branch_ids,
             prerequisites=self.prerequisites,
             raw_messages_count=None
@@ -372,6 +395,13 @@ class ConversationMetadata(AIPerfBaseModel):
         default=None,
         description="For DAG children: the parent conversation ID.",
     )
+    replay_scope_id: str | None = Field(
+        default=None,
+        description=(
+            "Logical agent/subagent scope whose request intervals participate in "
+            "one replay dependency graph. Independent scopes are never joined."
+        ),
+    )
     accuracy_ground_truth: str | None = Field(
         default=None,
         description="Ground-truth answer for this conversation (accuracy mode only). "
@@ -499,6 +529,13 @@ def _reject_unimplemented_context_mode(
         default=None,
         description="For DAG children: the parent conversation ID.",
     )
+    replay_scope_id: str | None = Field(
+        default=None,
+        exclude=True,
+        description=(
+            "Logical agent/subagent scope used to infer cross-stream replay barriers."
+        ),
+    )
     accuracy_ground_truth: str | None = Field(
         default=None,
         description="Ground-truth answer for this conversation (accuracy mode only). "
@@ -534,6 +571,7 @@ def metadata(self) -> ConversationMetadata:
             agent_depth=self.agent_depth,
             subagent_type=self.subagent_type,
             parent_conversation_id=self.parent_conversation_id,
+            replay_scope_id=self.replay_scope_id,
             accuracy_ground_truth=self.accuracy_ground_truth,
             accuracy_task=self.accuracy_task,
         )
@@ -554,6 +592,7 @@ def to_metadata(self) -> "ConversationMetadata":
             agent_depth=self.agent_depth,
             subagent_type=self.subagent_type,
             parent_conversation_id=self.parent_conversation_id,
+            replay_scope_id=self.replay_scope_id,
             accuracy_ground_truth=self.accuracy_ground_truth,
             accuracy_task=self.accuracy_task,
         )

src/aiperf/credit/issuer.py

@@ -22,6 +22,7 @@
 from aiperf.common.aiperf_logger import AIPerfLogger
 from aiperf.common.enums import CreditPhase
 from aiperf.credit.structs import Credit, TurnToSend
+from aiperf.timing.replay_dependencies import ReplayIssueGate
 from aiperf.timing.url_samplers import URLSelectionStrategyProtocol
 
 if TYPE_CHECKING:
@@ -32,6 +33,7 @@
     from aiperf.timing.phase.lifecycle import PhaseLifecycle
     from aiperf.timing.phase.progress_tracker import PhaseProgressTracker
     from aiperf.timing.phase.stop_conditions import StopConditionChecker
+    from aiperf.timing.replay_dependencies import ReplayBarrierCoordinator
     from aiperf.timing.request_cancellation import RequestCancellationSimulator
     from aiperf.timing.session_tree import SessionTreeRegistry
 
@@ -71,6 +73,7 @@ def __init__(
         url_selection_strategy: URLSelectionStrategyProtocol | None = None,
         session_tree_registry: SessionTreeRegistry | None = None,
         session_tree_registry_enabled: bool | None = None,
+        replay_barrier: ReplayBarrierCoordinator | None = None,
     ) -> None:
         """Initialize credit issuer.
 
@@ -108,6 +111,7 @@ def __init__(
             else None
         )
         self._issuing_stopped = False
+        self.replay_gate = ReplayIssueGate(replay_barrier)
         self._max_tokens_override: int | None = None
 
     def set_max_tokens_override(self, max_tokens: int | None) -> None:
@@ -222,6 +226,11 @@ async def issue_credit(self, turn: TurnToSend) -> bool:
             5. Create and send Credit
             6. If final credit: freeze counts + set event
         """
+        gate = getattr(self, "replay_gate", ReplayIssueGate(None))
+        return await gate.submit(turn, lambda: self._issue_credit_ready(turn))
+
+    async def _issue_credit_ready(self, turn: TurnToSend) -> bool:
+        """Issue a turn whose recorded predecessor frontier is complete."""
         if self._issuing_stopped:
             return False
 
@@ -327,11 +336,7 @@ async def try_issue_credit(self, turn: TurnToSend) -> bool | None:
         return await self._issue_credit_internal(turn)
 
     async def _issue_credit_internal(self, turn: TurnToSend) -> bool:
-        """Issue credit after slots are acquired. Mark as final if this was the final credit.
-
-        Returns:
-            True if more credits can be sent, False if this was the final credit.
-        """
+        """Issue credit after slots are acquired and mark the final credit."""
         if self._max_tokens_override is not None:
             turn = _struct_replace(turn, max_tokens_override=self._max_tokens_override)
         credit_index, is_final_credit = self._progress.increment_sent(turn)
@@ -376,6 +381,9 @@ async def _issue_credit_internal(self, turn: TurnToSend) -> bool:
         )
 
         await self._credit_router.send_credit(credit=credit)
+        replay_gate = getattr(self, "replay_gate", None)
+        if replay_gate is not None:
+            await replay_gate.observe_issued(credit)
         if is_final_credit:
             self._progress.freeze_sent_counts()
             self._progress.all_credits_sent_event.set()
@@ -414,6 +422,15 @@ async def dispatch_child_turn(self, turn: TurnToSend) -> bool:
         attempt would send the first sibling and permanently truncate every
         other sibling spawned in the same gather.
         """
+        gate = getattr(self, "replay_gate", ReplayIssueGate(None))
+        return await gate.submit(
+            turn,
+            lambda: self._dispatch_child_turn_ready(turn),
+            child_refusal_cleanup=True,
+        )
+
+    async def _dispatch_child_turn_ready(self, turn: TurnToSend) -> bool:
+        """Dispatch a child after its recorded predecessor frontier completes."""
         if self._issuing_stopped:
             return False
         can_proceed_fn = self._stop_checker.can_send_child_turn
@@ -474,5 +491,8 @@ async def dispatch_join_turn(self, pending: PendingBranchJoin) -> bool:
             cache_bust_marker=pending.parent_cache_bust_marker,
             cache_bust_target=pending.parent_cache_bust_target,
         )
-        result = await self.try_issue_credit(turn)
-        return result is True
+        replay_gate = getattr(self, "replay_gate", None)
+        if replay_gate is None or not replay_gate.enabled:
+            result = await self.try_issue_credit(turn)
+            return result is True
+        return await self.issue_credit(turn)

src/aiperf/dataset/loader/weka_trace.py

@@ -25,7 +25,7 @@
 from aiperf.common.enums import ConversationContextMode, TurnInputKind
 from aiperf.common.environment import Environment
 from aiperf.common.exceptions import DatasetLoaderError
-from aiperf.common.models import Conversation
+from aiperf.common.models import Conversation, ReplayTurnReference
 from aiperf.dataset.generator.prompt import PromptGenerator
 from aiperf.dataset.loader._delay_cap import DelayCapTracker
 from aiperf.dataset.loader.base_loader import BaseFileLoader
@@ -44,6 +44,56 @@
 _JOIN_EPSILON_SECONDS = 1e-6
 
 
+def _replay_scope_for_session(session_id: str, parent_trace_id: str) -> str:
+    """Keep each captured agent/subagent's interval graph independent."""
+    marker = "::sa:"
+    if marker not in session_id:
+        return parent_trace_id
+    trace_id, suffix = session_id.split(marker, 1)
+    agent_id = suffix
+    for worker_marker in (":aux:red:", ":aux:", ":fa:", ":wg:"):
+        if worker_marker in agent_id:
+            agent_id = agent_id.rsplit(worker_marker, 1)[0]
+    return f"{trace_id}{marker}{agent_id}"
+
+
+def _install_replay_dependencies(conversations: list[Conversation]) -> None:
+    """Persist cross-stream interval frontiers on reconstructed Weka turns."""
+    from aiperf.timing.replay_dependencies import (
+        RecordedTurnInterval,
+        ReplayTurnKey,
+        infer_cross_stream_predecessors,
+    )
+
+    by_scope: dict[str, list[RecordedTurnInterval]] = defaultdict(list)
+    turns_by_key = {}
+    for conversation in conversations:
+        scope_id = conversation.replay_scope_id
+        if scope_id is None:
+            continue
+        for turn_index, turn in enumerate(conversation.turns):
+            key = ReplayTurnKey(conversation.session_id, turn_index)
+            turns_by_key[key] = turn
+            by_scope[scope_id].append(
+                RecordedTurnInterval(
+                    key=key,
+                    stream_id=conversation.session_id,
+                    start_ms=turn.timestamp,
+                    api_time_ms=turn.api_time_ms,
+                )
+            )
+
+    for intervals in by_scope.values():
+        for key, predecessors in infer_cross_stream_predecessors(intervals).items():
+            turns_by_key[key].replay_predecessors = [
+                ReplayTurnReference(
+                    conversation_id=predecessor.conversation_id,
+                    turn_index=predecessor.turn_index,
+                )
+                for predecessor in predecessors
+            ]
+
+
 def _subagent_request_absolute_t(
     entry: WekaSubagentEntry, req: WekaNormalRequest
 ) -> float:
@@ -1595,6 +1645,8 @@ def convert_to_conversations(
             # that share the same PromptGenerator.
             self.prompt_generator._cache.clear()
 
+        _install_replay_dependencies(conversations)
+
         from aiperf.common.models import DatasetMetadata
         from aiperf.common.validators.orchestrator_v1 import (
             validate_for_orchestrator_v1,
@@ -1689,6 +1741,7 @@ def _reconstruct_serial(
             conv = Conversation(
                 session_id=plan.trace_id,
                 context_mode=self._resolved_context_mode(),
+                replay_scope_id=plan.trace_id,
             )
             recon = ConversationReconstructor(
                 block_size=plan.block_size,
@@ -2036,6 +2089,9 @@ def _reconstruct_serial(
                 is_root=False,
                 agent_depth=1,
                 parent_conversation_id=cp.parent_trace_id,
+                replay_scope_id=_replay_scope_for_session(
+                    cp.session_id, cp.parent_trace_id
+                ),
             )
             child_metric_values = metric_values_by_trace[cp.parent_trace_id]
             for k, creq in enumerate(cp.requests):
@@ -2138,6 +2194,7 @@ def _emit_flat_chain_conversation(
             is_root=False,
             agent_depth=1,
             parent_conversation_id=fp.parent_trace_id,
+            replay_scope_id=fp.parent_trace_id,
         )
         from aiperf.common.models import Turn
 
@@ -2625,6 +2682,7 @@ def _reconstruct_parallel(
             parent_conv = Conversation(
                 session_id=trace_id,
                 context_mode=self._resolved_context_mode(),
+                replay_scope_id=trace_id,
             )
             for t_dict in result["parent_turns"]:
                 parent_conv.turns.append(
@@ -2678,6 +2736,9 @@ def _reconstruct_parallel(
                     parent_conversation_id=child.get(
                         "parent_conversation_id", result["trace_id"]
                     ),
+                    replay_scope_id=_replay_scope_for_session(
+                        child["session_id"], result["trace_id"]
+                    ),
                 )
                 for t_dict in child["turns"]:
                     child_conv.turns.append(

src/aiperf/dataset/mmap_cache.py

@@ -75,6 +75,11 @@
 # Version 5 fixed the Conversation.metadata() projection of per-turn
 # theoretical prefix-cache block counts for realtime infinite-cache hit rate.
 MANIFEST_VERSION = (
+    # v21: Weka traces carry explicit api_time interval-frontier metadata in
+    # DatasetMetadata (replay_scope_id + per-turn replay_predecessors). Cached
+    # manifests produced before v21 deserialize with empty defaults, silently
+    # disabling fan-out/join barriers even though dataset.dat remains usable.
+    # Rebuild so the manifest sidecar contains the inferred dependency graph.
     # v20: DAG datasets (any FORK/SPAWN branch) are no longer preformatted into
     # the PAYLOAD_BYTES mmap fast path -- they are delta-compressed and
     # accumulate context across the tree (FORK children seed from the parent's
@@ -137,7 +142,7 @@
     # v10: merge of the flattened-agent-splitting lineage and the
     # tool-shaping lineage (boundary-cut overhang strip; shaping decided at
     # first emission so reset re-emits reproduce the first-sent shape).
-    20
+    21
 )
 MANIFEST_FILENAME = "manifest.json"
 INPUTS_JSON_FILENAME = "inputs.json"