feat(replay): stabilize deterministic agentic runtime contract

crates/core/src/replay/README.md

@@ -45,6 +45,49 @@ the importer against two revision-pinned rows from the public SemiAnalysis
 `cc-traces-weka-062126-256k` dataset. The rows are held in a temporary directory
 and deleted when the check exits; the complete 570 MB corpus is not downloaded.
 
+### Agentic driver/runtime contract
+
+M1 execution consumes one completely preloaded, immutable
+`ValidatedAgenticGraph`; neither the runtime nor an engine adapter polls a
+client or extends the graph dynamically. The replay runtime is the sole owner
+of logical time. At each timestamp it collects engine feedback and applies it
+as one `AgenticRuntimeFeedback`, ordered by immutable graph ordinal: output
+progress first, causal terminals second, and resource quiescence last.
+
+A causal terminal resolves a request's client-visible outcome; successful
+completion releases completion-triggered graph edges. Client lanes limit whole
+plays, including background requests. A successful play releases its lane only
+after all authored nodes complete, including delayed or blocked nodes. Under the
+current failure policy, a failed play skips undispatched nodes and releases its
+lane after every already-dispatched request becomes terminal. Merely finishing
+the root or a blocking join does not finish still-running background requests.
+
+Quiescence means the engine/router/handoff state owned by a request has settled.
+It controls resource settlement and final drain, independently of client lane
+reuse. The next play can therefore submit while an earlier play retains P/D
+source holds or has pending cancellation actions; engine admission still queues
+requests when those resources are unavailable. Late cleanup records the earlier
+play's settlement without releasing the lane again. Request quiescence does not
+require flushing reusable prefix-cache entries or ending an agent conversation.
+
+In this in-process timing model, final decode or an observed request failure
+stands in for the client response terminal. HTTP/SSE delivery and client task
+teardown latency are not modeled. The strict failed-play policy above is an
+explicit replay contract, not a claim of full AIPerf error-policy parity.
+
+Equal-time event phases are engine pass completion, worker ready,
+transfer completion, admission, telemetry, then scaling. Within a phase,
+stable worker/pass/handoff identities replace insertion order as the primary
+tie-breaker.
+
+Both aggregated and disaggregated runtimes expose an internal `step()` seam.
+It returns only after a semantic timestamp reaches a fixed point and preserves
+all engine, placement, router, handoff, and KV state, so resuming does not
+rebuild the simulation. Agentic requests carry a stable identity envelope
+(request, play, conversation, and optional lane/tree/cache identities) across
+the workload-to-runtime boundary. The driver can emit a canonical lifecycle
+JSONL transcript and domain-separated digest for conformance tests.
+
 ## File Map
 
 - `src/replay/replayer.rs`

crates/core/src/replay/agg.rs

@@ -14,9 +14,10 @@ use super::evidence::{
 };
 use super::progress::ReplayProgress;
 use super::runtime_utils::{
-    next_non_telemetry_event_ms, next_timestamp as choose_next_timestamp, pop_ready_scaling_tick,
-    pop_ready_telemetry_tick, pop_ready_worker_completions, pop_ready_worker_ready,
-    push_scaling_tick, push_telemetry_tick, push_worker_completions, push_worker_ready,
+    ReplayStepOutcome, next_non_telemetry_event_ms, next_timestamp as choose_next_timestamp,
+    pop_ready_scaling_tick, pop_ready_telemetry_tick, pop_ready_worker_completions,
+    pop_ready_worker_ready, push_scaling_tick, push_telemetry_tick, push_worker_completions,
+    push_worker_ready,
 };
 use super::scaling::{LatestFpmBuffer, ReplayScalingPolicy, ReplayScalingSnapshot};
 use super::telemetry::{
@@ -99,6 +100,8 @@ where
     /// Set between the halves of that delta cycle: the next step commits the
     /// admissions and worker drives the previous step deferred.
     drive_pending: bool,
+    drive_started: bool,
+    drive_finalized: bool,
 }
 
 impl<PlacementPolicyImpl, Observation, Metadata>
@@ -161,6 +164,8 @@ where
             step_freed_slot: false,
             defer_drive: false,
             drive_pending: false,
+            drive_started: false,
+            drive_finalized: false,
         })
     }
 
@@ -479,7 +484,7 @@ where
     /// Consume one output signal, updating router state, collector state, and completion counts.
     fn process_output_signal(&mut self, signal: OutputSignal) -> anyhow::Result<()> {
         if let Some(token_id) = signal.token_id {
-            CoreAdmissionSource::on_output_token(&mut self.admission, signal.uuid, token_id)?;
+            self.admission.defer_output_token(signal.uuid, token_id)?;
             self.collector.on_token(signal.uuid, self.now_ms);
             if self.defer_drive {
                 self.step_tokens.push((signal.uuid, token_id));
@@ -521,12 +526,8 @@ where
                     latencies,
                 );
             }
-            CoreAdmissionSource::on_terminal(
-                &mut self.admission,
-                signal.uuid,
-                self.now_ms,
-                status,
-            )?;
+            self.admission
+                .defer_terminal(signal.uuid, self.now_ms, status)?;
             self.progress.inc_completed();
             self.dispatch_placements(placements)?;
             return Ok(());
@@ -840,6 +841,7 @@ where
                 changed |= self.settle_internal_work(&mut consecutive_internal_steps)?;
             }
             changed |= self.apply_worker_ready_events()?;
+            changed |= self.admission.flush_agentic_runtime_feedback(self.now_ms)?;
             changed |= self.release_ready_arrivals()?;
             if self.defer_drive && self.step_freed_slot {
                 self.drive_pending = true;
@@ -1290,23 +1292,46 @@ where
         );
     }
 
-    /// Run the aggregated offline replay until all arrivals and worker work are exhausted.
-    /// If `max_sim_time_ms` is set, exits gracefully when the next scheduled
-    /// timestamp would exceed that cap; in-flight requests at that point are
-    /// reported as incomplete.
-    pub(crate) fn run(mut self) -> anyhow::Result<(TraceCollector, AggRuntimeStats)> {
+    fn ensure_drive_started(&mut self) -> anyhow::Result<bool> {
+        if self.drive_started {
+            return Ok(false);
+        }
         if let Some(cap_ms) = self.max_sim_time_ms
             && (!cap_ms.is_finite() || cap_ms < 0.0)
         {
             bail!("max_sim_time_ms must be a finite, non-negative value; got {cap_ms}");
         }
         self.drain_current_timestamp()?;
         self.seed_first_telemetry_tick()?;
-        // With a planner attached, seed the recurring heartbeat; ticks then fire as
-        // events inside drain_current_timestamp.
         self.seed_first_scaling_tick()?;
+        // Keep the baseline before the first scaling decision, but settle any
+        // tick seeded at this instant before exposing a settled step boundary.
+        if !self.is_done()
+            && self
+                .events
+                .peek()
+                .is_some_and(|event| event.at_ms <= self.now_ms)
+        {
+            self.drain_current_timestamp()?;
+        }
+        self.drive_started = true;
+        Ok(true)
+    }
+
+    /// Advance to the next fully settled semantic timestamp without rebuilding
+    /// engine, router, placement, or KV state.
+    pub(crate) fn step(&mut self) -> anyhow::Result<ReplayStepOutcome> {
+        let just_started = self.ensure_drive_started()?;
+        if self.is_done() {
+            return Ok(ReplayStepOutcome::Complete);
+        }
+        if just_started {
+            return Ok(ReplayStepOutcome::Settled {
+                now_ms: self.now_ms,
+            });
+        }
 
-        while !self.is_done() {
+        loop {
             let (next_timestamp_ms, canonical_timestamp_ms) = self.next_timestamps();
             let Some(canonical_timestamp_ms) = canonical_timestamp_ms else {
                 // Aggregated workers have no external handoff dependency. If
@@ -1328,7 +1353,9 @@ where
             if let Some(cap_ms) = self.max_sim_time_ms
                 && canonical_timestamp_ms > cap_ms
             {
-                break;
+                return Ok(ReplayStepOutcome::TimeLimitReached {
+                    now_ms: self.now_ms,
+                });
             }
             let next_timestamp_ms = next_timestamp_ms
                 .expect("canonical replay activity must have a next scheduled timestamp");
@@ -1338,9 +1365,31 @@ where
             }
             self.advance_now_ms(next_timestamp_ms);
             self.drain_current_timestamp()?;
+            return Ok(if self.is_done() {
+                ReplayStepOutcome::Complete
+            } else {
+                ReplayStepOutcome::Settled {
+                    now_ms: self.now_ms,
+                }
+            });
         }
+    }
 
-        self.publish_final_telemetry_sample()?;
+    fn run_to_completion(&mut self) -> anyhow::Result<()> {
+        while let ReplayStepOutcome::Settled { .. } = self.step()? {}
+        if !self.drive_finalized {
+            self.publish_final_telemetry_sample()?;
+            self.drive_finalized = true;
+        }
+        Ok(())
+    }
+
+    /// Run the aggregated offline replay until all arrivals and worker work are exhausted.
+    /// If `max_sim_time_ms` is set, exits gracefully when the next scheduled
+    /// timestamp would exceed that cap; in-flight requests at that point are
+    /// reported as incomplete.
+    pub(crate) fn run(mut self) -> anyhow::Result<(TraceCollector, AggRuntimeStats)> {
+        self.run_to_completion()?;
 
         self.progress.finish();
         if let Some(snapshot) = self.admission.agentic_trajectory_snapshot() {
@@ -1350,6 +1399,9 @@ where
             self.collector.set_agentic_graph(identity);
         }
         self.collector.g3_offload = self.engine.g3_stats();
+        if let Some(transcript) = self.admission.agentic_lifecycle_transcript() {
+            self.collector.set_agentic_lifecycle(transcript);
+        }
         self.collector.set_runtime_evidence(self.evidence.finish());
         Ok((self.collector, self.stats))
     }

crates/core/src/replay/agg_tests.rs

@@ -9,6 +9,7 @@ use crate::replay::components::NoReplayMetadata;
 use crate::replay::core::NoEngineEvents;
 use crate::replay::core::round_robin::AggregatedRoundRobinPlacement;
 use crate::replay::engine::{ReplayEngineConfig, ReplayEngineFactory};
+use crate::replay::scaling::ReplayScalingDecision;
 use crate::replay::{
     ReplayArtifactKvEventVisibility, ReplaySpec, ReplayTelemetryObserver, ReplayTelemetrySnapshot,
     Replayer, WorkerStage, run_engine_replay,
@@ -37,6 +38,13 @@ fn request(uuid: u128, arrival_ms: f64) -> DirectRequest {
 }
 
 fn runtime(pending: VecDeque<DirectRequest>) -> RoundRobinAggRuntime {
+    runtime_with_startup_time(pending, None)
+}
+
+fn runtime_with_startup_time(
+    pending: VecDeque<DirectRequest>,
+    startup_time_ms: Option<f64>,
+) -> RoundRobinAggRuntime {
     let config = ReplayEngineConfig {
         rank: EngineConfig {
             timing_model: TimingModelConfig::Fixed {
@@ -54,7 +62,7 @@ fn runtime(pending: VecDeque<DirectRequest>) -> RoundRobinAggRuntime {
         role_factory,
         AdmissionQueue::new_requests(pending, ReplayMode::Trace),
         1,
-        None,
+        startup_time_ms,
         |dp_size, topology| Ok(AggregatedRoundRobinPlacement::new(dp_size, topology)),
     )
     .unwrap()
@@ -718,3 +726,70 @@ fn g3_summary_serializes_g3_counters_only_when_enabled() {
             .is_none()
     );
 }
+
+#[test]
+fn agg_settled_steps_resume_without_changing_the_result() {
+    let pending = VecDeque::from([
+        DirectRequest {
+            max_output_tokens: 4,
+            ..request(1, 0.0)
+        },
+        request(2, 3.0),
+    ]);
+    let (continuous, _) = runtime(pending.clone()).run().unwrap();
+
+    let mut stepped = runtime(pending);
+    for _ in 0..2 {
+        match stepped.step().unwrap() {
+            ReplayStepOutcome::Settled { .. } => {}
+            ReplayStepOutcome::Complete => panic!("replay completed before resume point"),
+            ReplayStepOutcome::TimeLimitReached { .. } => panic!("unexpected time limit"),
+        }
+    }
+    assert!(
+        stepped.cluster_in_flight() > 0,
+        "resume must retain live work"
+    );
+    let (resumed, _) = stepped.run().unwrap();
+    let resumed = resumed.finish();
+
+    assert_eq!(resumed.request_counts.completed_requests, 2);
+    assert_eq!(
+        serde_json::to_value(continuous.finish()).unwrap(),
+        serde_json::to_value(resumed).unwrap()
+    );
+}
+
+#[test]
+fn agg_first_step_settles_initial_scaling_and_zero_delay_worker_startup() {
+    struct ScaleAtZero;
+
+    impl ReplayScalingPolicy for ScaleAtZero {
+        fn initial_tick_ms(&mut self) -> anyhow::Result<f64> {
+            Ok(0.0)
+        }
+
+        fn on_tick(
+            &mut self,
+            snapshot: ReplayScalingSnapshot,
+        ) -> anyhow::Result<ReplayScalingDecision> {
+            assert_eq!(snapshot.now_ms, 0.0);
+            Ok(ReplayScalingDecision {
+                target_decode: Some(2),
+                ..Default::default()
+            })
+        }
+    }
+
+    let mut stepped = runtime_with_startup_time(VecDeque::from([request(1, 0.0)]), Some(0.0))
+        .with_scaling_policy(Box::new(ScaleAtZero));
+
+    assert_eq!(
+        stepped.step().unwrap(),
+        ReplayStepOutcome::Settled { now_ms: 0.0 }
+    );
+    assert_eq!(stepped.engine.active_group_ids(), vec![0, 1]);
+    assert!(stepped.engine.starting_group_ids().is_empty());
+    assert!(stepped.events.iter().all(|event| event.at_ms > 0.0));
+    assert!(stepped.next_timestamps().1.unwrap() > 0.0);
+}