diff --git a/crates/core/src/replay/README.md b/crates/core/src/replay/README.md index 908b6786c..2cc5a736d 100644 --- a/crates/core/src/replay/README.md +++ b/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` diff --git a/crates/core/src/replay/agg.rs b/crates/core/src/replay/agg.rs index c5d7b19fd..79cd86c9b 100644 --- a/crates/core/src/replay/agg.rs +++ b/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 @@ -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,11 +1292,10 @@ 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 { + 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) { @@ -1302,11 +1303,35 @@ where } 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 { + 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)) } diff --git a/crates/core/src/replay/agg_tests.rs b/crates/core/src/replay/agg_tests.rs index 8a37c8988..225b7b9d2 100644 --- a/crates/core/src/replay/agg_tests.rs +++ b/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) -> RoundRobinAggRuntime { + runtime_with_startup_time(pending, None) +} + +fn runtime_with_startup_time( + pending: VecDeque, + startup_time_ms: Option, +) -> RoundRobinAggRuntime { let config = ReplayEngineConfig { rank: EngineConfig { timing_model: TimingModelConfig::Fixed { @@ -54,7 +62,7 @@ fn runtime(pending: VecDeque) -> 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 { + Ok(0.0) + } + + fn on_tick( + &mut self, + snapshot: ReplayScalingSnapshot, + ) -> anyhow::Result { + 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); +} diff --git a/crates/core/src/replay/components/admission.rs b/crates/core/src/replay/components/admission.rs index af73e0acd..454e1e1fc 100644 --- a/crates/core/src/replay/components/admission.rs +++ b/crates/core/src/replay/components/admission.rs @@ -8,9 +8,11 @@ use anyhow::Result; use uuid::Uuid; use super::ReplayMode; +use crate::replay::ReplayTerminalStatus; use crate::replay::core::{AdmissionSource as CoreAdmissionSource, ReadyArrival}; use crate::replay::loadgen::{ - GeneratedRequests, ReplayRequestHashes, ReplayRequestPayload, WorkloadDriver, + AgenticOutputFeedback, AgenticRuntimeFeedback, AgenticTerminalFeedback, GeneratedRequests, + ReplayRequestHashes, ReplayRequestPayload, WorkloadDriver, }; use crate::replay::protocol::DirectRequest; @@ -77,7 +79,36 @@ impl ReplayReadyArrival { enum AdmissionSource { Requests(VecDeque), GeneratedRequests(GeneratedRequests), - Workload(WorkloadDriver), + Workload { + driver: WorkloadDriver, + pending_agentic_runtime_feedback: PendingAgenticRuntimeFeedback, + }, +} + +/// Runtime feedback collected beside the workload driver until the current +/// logical timestamp is flushed as one deterministic batch. +#[derive(Default)] +struct PendingAgenticRuntimeFeedback { + output_tokens: Vec, + causal_terminals: Vec, + quiescent_requests: Vec, +} + +impl PendingAgenticRuntimeFeedback { + fn is_empty(&self) -> bool { + self.output_tokens.is_empty() + && self.causal_terminals.is_empty() + && self.quiescent_requests.is_empty() + } + + fn take(&mut self, at_ms: f64) -> AgenticRuntimeFeedback { + AgenticRuntimeFeedback { + at_ms, + output_tokens: std::mem::take(&mut self.output_tokens), + causal_terminals: std::mem::take(&mut self.causal_terminals), + quiescent_requests: std::mem::take(&mut self.quiescent_requests), + } + } } pub(crate) struct AdmissionQueue { @@ -105,7 +136,10 @@ impl AdmissionQueue { pub(crate) fn new_workload(driver: WorkloadDriver, mode: ReplayMode) -> Self { Self { - source: AdmissionSource::Workload(driver), + source: AdmissionSource::Workload { + driver, + pending_agentic_runtime_feedback: PendingAgenticRuntimeFeedback::default(), + }, mode, metadata: PhantomData, } @@ -120,10 +154,12 @@ impl AdmissionQueue { (ReplayMode::Trace, AdmissionSource::Requests(pending)) => pending .front() .and_then(|request| request.arrival_timestamp_ms), - (ReplayMode::Trace, AdmissionSource::Workload(driver)) => driver.next_ready_time_ms(), + (ReplayMode::Trace, AdmissionSource::Workload { driver, .. }) => { + driver.next_ready_time_ms() + } // Concurrency: the driver owns the session cap and gates admission, so defer to // it directly (no in-flight clamp needed here). - (ReplayMode::Concurrency { .. }, AdmissionSource::Workload(driver)) => { + (ReplayMode::Concurrency { .. }, AdmissionSource::Workload { driver, .. }) => { driver.next_ready_time_ms() } (ReplayMode::Concurrency { .. }, AdmissionSource::Requests(_)) @@ -187,7 +223,7 @@ impl AdmissionQueue { } Ok(ready) } - (ReplayMode::Trace, AdmissionSource::Workload(driver)) => Ok(driver + (ReplayMode::Trace, AdmissionSource::Workload { driver, .. }) => Ok(driver .pop_ready_compact(now_ms, usize::MAX) .into_iter() .map(|ready| { @@ -235,7 +271,7 @@ impl AdmissionQueue { (ReplayMode::Trace, AdmissionSource::GeneratedRequests(_)) => { anyhow::bail!("generated requests require concurrency admission") } - (ReplayMode::Concurrency { .. }, AdmissionSource::Workload(driver)) => { + (ReplayMode::Concurrency { .. }, AdmissionSource::Workload { driver, .. }) => { // The driver owns the session cap and only ever holds active sessions' // turns in its heap, so drain everything ready in heap (i.e. limit=usize MAX). Ok(driver @@ -310,7 +346,7 @@ impl AdmissionQueue { now_ms: f64, status: crate::replay::ReplayTerminalStatus, ) -> Result<()> { - let AdmissionSource::Workload(driver) = &mut self.source else { + let AdmissionSource::Workload { driver, .. } = &mut self.source else { return Ok(()); }; driver.on_terminal(uuid, now_ms, status) @@ -322,51 +358,168 @@ impl AdmissionQueue { now_ms: f64, status: crate::replay::ReplayTerminalStatus, ) -> Result<()> { - let AdmissionSource::Workload(driver) = &mut self.source else { + let AdmissionSource::Workload { driver, .. } = &mut self.source else { return Ok(()); }; driver.on_causal_terminal(uuid, now_ms, status) } pub(crate) fn on_request_quiescent(&mut self, uuid: Uuid, now_ms: f64) -> Result<()> { - let AdmissionSource::Workload(driver) = &mut self.source else { + let AdmissionSource::Workload { driver, .. } = &mut self.source else { return Ok(()); }; driver.on_quiescent(uuid, now_ms) } pub(crate) fn on_output_token(&mut self, uuid: Uuid, token_id: u32) -> Result<()> { - let AdmissionSource::Workload(driver) = &mut self.source else { + let AdmissionSource::Workload { driver, .. } = &mut self.source else { return Ok(()); }; driver.on_output_token(uuid, token_id) } + fn is_agentic(&self) -> bool { + matches!(&self.source, AdmissionSource::Workload { driver, .. } if driver.is_agentic()) + } + + pub(crate) fn defer_output_token(&mut self, uuid: Uuid, token_id: u32) -> Result<()> { + if !self.is_agentic() { + return CoreAdmissionSource::on_output_token(self, uuid, token_id); + } + let AdmissionSource::Workload { + pending_agentic_runtime_feedback, + .. + } = &mut self.source + else { + unreachable!("only an agentic workload can buffer agentic feedback"); + }; + if let Some(existing) = pending_agentic_runtime_feedback + .output_tokens + .iter_mut() + .find(|feedback| feedback.request_uuid == uuid) + { + existing.token_ids.push(token_id); + } else { + pending_agentic_runtime_feedback + .output_tokens + .push(AgenticOutputFeedback { + request_uuid: uuid, + token_ids: vec![token_id], + }); + } + Ok(()) + } + + pub(crate) fn defer_causal_terminal( + &mut self, + uuid: Uuid, + now_ms: f64, + status: ReplayTerminalStatus, + ) -> Result<()> { + if !self.is_agentic() { + return self.on_request_causal_terminal(uuid, now_ms, status); + } + let AdmissionSource::Workload { + pending_agentic_runtime_feedback, + .. + } = &mut self.source + else { + unreachable!("only an agentic workload can buffer agentic feedback"); + }; + pending_agentic_runtime_feedback + .causal_terminals + .push(AgenticTerminalFeedback { + request_uuid: uuid, + status, + }); + Ok(()) + } + + pub(crate) fn defer_quiescent(&mut self, uuid: Uuid, now_ms: f64) -> Result<()> { + if !self.is_agentic() { + return self.on_request_quiescent(uuid, now_ms); + } + let AdmissionSource::Workload { + pending_agentic_runtime_feedback, + .. + } = &mut self.source + else { + unreachable!("only an agentic workload can buffer agentic feedback"); + }; + pending_agentic_runtime_feedback + .quiescent_requests + .push(uuid); + Ok(()) + } + + pub(crate) fn defer_terminal( + &mut self, + uuid: Uuid, + now_ms: f64, + status: ReplayTerminalStatus, + ) -> Result<()> { + if !self.is_agentic() { + return CoreAdmissionSource::on_terminal(self, uuid, now_ms, status); + } + let AdmissionSource::Workload { + pending_agentic_runtime_feedback, + .. + } = &mut self.source + else { + unreachable!("only an agentic workload can buffer agentic feedback"); + }; + pending_agentic_runtime_feedback + .causal_terminals + .push(AgenticTerminalFeedback { + request_uuid: uuid, + status, + }); + pending_agentic_runtime_feedback + .quiescent_requests + .push(uuid); + Ok(()) + } + + pub(crate) fn flush_agentic_runtime_feedback(&mut self, now_ms: f64) -> Result { + let AdmissionSource::Workload { + driver, + pending_agentic_runtime_feedback, + } = &mut self.source + else { + return Ok(false); + }; + if pending_agentic_runtime_feedback.is_empty() { + return Ok(false); + } + driver.apply_agentic_runtime_feedback(pending_agentic_runtime_feedback.take(now_ms))?; + Ok(true) + } + pub(crate) fn is_drained(&self) -> bool { match &self.source { AdmissionSource::Requests(pending) => pending.is_empty(), AdmissionSource::GeneratedRequests(source) => source.remaining() == 0, - AdmissionSource::Workload(driver) => driver.is_drained(), + AdmissionSource::Workload { driver, .. } => driver.is_drained(), } } #[cfg(test)] pub(crate) fn is_workload(&self) -> bool { - matches!(self.source, AdmissionSource::Workload(_)) + matches!(self.source, AdmissionSource::Workload { .. }) } pub(crate) fn total_requests(&self) -> usize { match &self.source { AdmissionSource::Requests(pending) => pending.len(), AdmissionSource::GeneratedRequests(source) => source.remaining(), - AdmissionSource::Workload(driver) => driver.total_turns(), + AdmissionSource::Workload { driver, .. } => driver.total_turns(), } } pub(crate) fn agentic_trajectory_snapshot( &self, ) -> Option { - let AdmissionSource::Workload(driver) = &self.source else { + let AdmissionSource::Workload { driver, .. } = &self.source else { return None; }; driver.agentic_trajectory_snapshot() @@ -375,11 +528,20 @@ impl AdmissionQueue { pub(crate) fn agentic_graph_identity( &self, ) -> Option { - let AdmissionSource::Workload(driver) = &self.source else { + let AdmissionSource::Workload { driver, .. } = &self.source else { return None; }; driver.agentic_graph_identity() } + + pub(crate) fn agentic_lifecycle_transcript( + &self, + ) -> Option { + let AdmissionSource::Workload { driver, .. } = &self.source else { + return None; + }; + driver.agentic_lifecycle_transcript() + } } impl CoreAdmissionSource for AdmissionQueue { diff --git a/crates/core/src/replay/disagg.rs b/crates/core/src/replay/disagg.rs index 722c4e28c..9c5443a00 100644 --- a/crates/core/src/replay/disagg.rs +++ b/crates/core/src/replay/disagg.rs @@ -32,10 +32,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_transfer_complete, pop_ready_worker_completions, - pop_ready_worker_ready, push_scaling_tick, push_telemetry_tick, push_transfer_complete, - 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_transfer_complete, + pop_ready_worker_completions, pop_ready_worker_ready, push_scaling_tick, push_telemetry_tick, + push_transfer_complete, push_worker_completions, push_worker_ready, }; use super::scaling::{LatestFpmBuffer, ReplayScalingPolicy, ReplayScalingSnapshot}; #[cfg(test)] @@ -274,6 +274,7 @@ impl DisaggActionQueues { struct DisaggFlowState { requests: HashMap, requests_by_handoff: HashMap, + next_handoff_ordinal: u128, handoff_order: HandoffOrder, handoff_latency_ms: f64, action_queues: DisaggActionQueues, @@ -302,6 +303,7 @@ impl DisaggFlowState { Self { requests: HashMap::new(), requests_by_handoff: HashMap::new(), + next_handoff_ordinal: 1, handoff_order, handoff_latency_ms, action_queues: DisaggActionQueues::default(), @@ -630,7 +632,14 @@ impl DisaggFlowState { if self.requests.contains_key(&uuid) { bail!("offline disagg replay request {uuid} is already active"); } - let handoff_id = HandoffId::new(Uuid::new_v4()); + // Handoff identities are replay-local. Allocate in arrival order so + // simultaneous transfers have stable priority even with random request + // UUIDs, and never reuse an identity while stale events may remain. + let handoff_id = HandoffId::new(Uuid::from_u128(self.next_handoff_ordinal)); + self.next_handoff_ordinal = self + .next_handoff_ordinal + .checked_add(1) + .context("offline disagg replay handoff ordinal overflow")?; let mut state = DisaggRequestState::new( request, arrival_time_ms, @@ -937,6 +946,8 @@ where /// above. Only the planner consumes them, so the plain `run()` path leaves this /// `false`. collect_fpm: bool, + drive_started: bool, + drive_finalized: bool, } #[cfg(test)] @@ -1072,6 +1083,8 @@ where scaling_policy: None, telemetry: None, collect_fpm: false, + drive_started: false, + drive_finalized: false, }) } @@ -1232,11 +1245,11 @@ where return Ok(()); }; self.admission - .on_request_causal_terminal(uuid, self.now_ms, status) + .defer_causal_terminal(uuid, self.now_ms, status) } fn notify_quiescent(&mut self, uuid: Uuid) -> Result<()> { - self.admission.on_request_quiescent(uuid, self.now_ms) + self.admission.defer_quiescent(uuid, self.now_ms) } /// Submit a coordinator-owned prefill onto a selected worker. @@ -2009,7 +2022,7 @@ where /// Process one decode output signal, including decode router frees and request completion. fn process_decode_signal(&mut self, signal: OutputSignal) -> 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)?; // Generalized-engine completion effects become visible at the // attention-DP group boundary. Recording the token here therefore // preserves the old EngineComponent::align_pass_token_times @@ -2451,8 +2464,9 @@ where changed |= self.apply_worker_completions()?; changed |= self.apply_worker_ready_events()?; changed |= self.apply_transfer_completions()?; - changed |= self.release_ready_arrivals()?; changed |= self.drive_pending_actions()?; + changed |= self.admission.flush_agentic_runtime_feedback(self.now_ms)?; + changed |= self.release_ready_arrivals()?; changed |= self.drive_prefill_workers()?; changed |= self.drive_decode_workers()?; let removed_prefill = self @@ -3223,7 +3237,10 @@ where self.decode_fpm_buffer.take() } - fn run_to_completion(&mut self) -> Result<()> { + fn ensure_drive_started(&mut self) -> Result { + 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) { @@ -3231,11 +3248,35 @@ where } 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) + } - while !self.is_done() { + /// Advance to the next fully settled semantic timestamp while preserving + /// the live engine, router, handoff, placement, and KV state. + pub(crate) fn step(&mut self) -> Result { + 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, + }); + } + + loop { let (next_timestamp_ms, canonical_timestamp_ms) = self.next_timestamps(); let Some(canonical_timestamp_ms) = canonical_timestamp_ms else { if self.prefill_engine.has_runnable_worker() @@ -3256,7 +3297,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"); @@ -3266,9 +3309,22 @@ 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) -> Result<()> { + while let ReplayStepOutcome::Settled { .. } = self.step()? {} + if !self.drive_finalized { + self.publish_final_telemetry_sample()?; + self.drive_finalized = true; + } Ok(()) } @@ -3288,6 +3344,9 @@ where if let Some(identity) = self.admission.agentic_graph_identity() { self.collector.set_agentic_graph(identity); } + 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)) } diff --git a/crates/core/src/replay/disagg_tests.rs b/crates/core/src/replay/disagg_tests.rs index d92663082..3112a0d62 100644 --- a/crates/core/src/replay/disagg_tests.rs +++ b/crates/core/src/replay/disagg_tests.rs @@ -46,6 +46,7 @@ impl ReplayScalingPolicy for CaptureOncePolicy { } struct CaptureAndScaleOncePolicy { + at_ms: f64, captured: Rc>>, } @@ -187,7 +188,7 @@ impl PlacementPolicy for QueueUntilWorkerPlacement { impl ReplayScalingPolicy for CaptureAndScaleOncePolicy { fn initial_tick_ms(&mut self) -> anyhow::Result { - Ok(0.1) + Ok(self.at_ms) } fn on_tick( @@ -246,6 +247,14 @@ struct TestDisaggConfig { impl TestDisaggConfig { fn runtime_config(&self, emit_kv_events: bool) -> anyhow::Result { + self.runtime_config_with_factory(emit_kv_events, ReplayEngineFactory::new()) + } + + fn runtime_config_with_factory( + &self, + emit_kv_events: bool, + factory: ReplayEngineFactory, + ) -> anyhow::Result { let engine = ReplayEngineConfig { dp_size: 1, tensor_parallel_size: 1, @@ -264,7 +273,6 @@ impl TestDisaggConfig { rank: self.decode_args.clone(), }), }; - let factory = ReplayEngineFactory::new(); let prefill_factory = factory.role_factory(&engine, crate::replay::WorkerStage::Prefill, emit_kv_events)?; let decode_factory = @@ -444,6 +452,42 @@ fn request( } } +#[test] +fn simultaneous_handoffs_follow_arrival_order_across_replays() { + let request_ids = [3, 1, 2].map(Uuid::from_u128); + let run = || { + let mut flow = DisaggFlowState::new(HandoffOrder::SourceFirst, 0.0, false); + let mut collector = TraceCollector::default(); + let handoffs = request_ids.map(|uuid| { + flow.on_external_arrival( + ReplayRequestPayload::materialized(request(uuid.as_u128(), 64, 1, 0.0)), + 0.0, + None, + None, + &mut collector, + ) + .unwrap(); + flow.state(uuid).unwrap().handoff_id + }); + + // Completion callback order and request UUID order must not change + // which request reaches decode first when transfers complete together. + let mut events: BinaryHeap> = BinaryHeap::new(); + let mut next_event_seq = 0; + for handoff_id in handoffs.into_iter().rev() { + push_transfer_complete(&mut events, &mut next_event_seq, 10.0, handoff_id); + } + let mut released = Vec::new(); + while let Some(handoff_id) = pop_ready_transfer_complete(&mut events, 10.0) { + released.push(flow.uuid_for_handoff(handoff_id).unwrap()); + } + assert_eq!(released, request_ids); + handoffs + }; + + assert_eq!(run(), run()); +} + fn agentic_row( request_id: &str, play_id: &str, @@ -565,6 +609,89 @@ fn run_trace_collect( .unwrap() } +#[test] +fn disagg_settled_steps_resume_without_changing_the_result() { + let config = disagg_config(); + let pending = VecDeque::from([request(1, 64, 4, 0.0), request(2, 64, 1, 3.0)]); + let (continuous, _) = + DisaggRuntime::from_requests(&config, None, None, pending.clone(), ReplayMode::Trace) + .unwrap() + .run() + .unwrap(); + + let mut stepped = + DisaggRuntime::from_requests(&config, None, None, pending, ReplayMode::Trace).unwrap(); + 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 disagg_first_step_settles_initial_scaling_and_zero_delay_worker_startup() { + let mut config = disagg_config().runtime_config(false).unwrap(); + config.num_prefill_workers = 1; + config.num_decode_workers = 1; + config.prefill_startup_time_ms = Some(0.0); + config.decode_startup_time_ms = Some(0.0); + let captured = Rc::new(RefCell::new(None)); + let telemetry_samples = Arc::new(Mutex::new(Vec::new())); + let mut stepped = RoundRobinDisaggRuntime::new_composed( + &config, + AdmissionQueue::new_requests(VecDeque::from([request(1, 64, 2, 0.0)]), ReplayMode::Trace), + false, + |_, prefill_topology, _, decode_topology| { + Ok(( + PoolRoundRobinPlacement::new(prefill_topology), + PoolRoundRobinPlacement::new(decode_topology), + )) + }, + ) + .unwrap() + .with_telemetry_observer( + 1.0, + Box::new(CaptureTelemetryObserver { + samples: Arc::clone(&telemetry_samples), + }), + ) + .with_scaling_policy(Box::new(CaptureAndScaleOncePolicy { + at_ms: 0.0, + captured: Rc::clone(&captured), + })); + + assert_eq!( + stepped.step().unwrap(), + ReplayStepOutcome::Settled { now_ms: 0.0 } + ); + assert_eq!(captured.borrow().as_ref().unwrap().now_ms, 0.0); + assert_eq!(stepped.prefill_engine.active_group_ids(), vec![0, 1]); + assert_eq!(stepped.decode_engine.active_group_ids(), vec![0, 1]); + assert!(stepped.prefill_engine.starting_group_ids().is_empty()); + assert!(stepped.decode_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); + let samples = telemetry_samples.lock().unwrap(); + assert_eq!(samples.len(), 1); + assert_eq!(samples[0].kind, ReplayTelemetrySampleKind::Baseline); + assert_eq!(samples[0].active_prefill_ids, vec![0]); + assert_eq!(samples[0].active_decode_ids, vec![0]); +} + fn run_concurrency_collect( config: &TestDisaggConfig, requests: Vec, @@ -733,6 +860,7 @@ fn telemetry_samples_router_queues_before_a_coincident_scale_up() { crate::replay::normalize_trace_requests(vec![request(9_303, 64, 1, 0.0)], 1.0).unwrap(); let captured = Rc::new(RefCell::new(None)); let policy = CaptureAndScaleOncePolicy { + at_ms: 0.1, captured: Rc::clone(&captured), }; let telemetry_samples = Arc::new(Mutex::new(Vec::new())); @@ -1225,7 +1353,7 @@ fn agentic_pd_edges_use_emission_and_final_decode_boundaries(#[case] engine_type #[rstest::rstest] #[case(EngineType::Vllm)] #[case(EngineType::Sglang)] -fn agentic_lane_recycles_only_after_pd_quiescence(#[case] engine_type: EngineType) { +fn agentic_lane_waits_for_background_child_terminal(#[case] engine_type: EngineType) { let mut config = cleanup_overtake_config(engine_type); config.num_prefill_workers = 1; config.num_decode_workers = 1; @@ -1278,14 +1406,252 @@ fn agentic_lane_recycles_only_after_pd_quiescence(#[case] engine_type: EngineTyp for request_id in ["first-root", "long-prefill-child"] { let prior = by_id[request_id]; assert!(next_dispatch >= prior.terminal_time_ms); - assert!(next_dispatch >= prior.source_released_ms.unwrap()); } + assert_eq!( + next_dispatch, + by_id["first-root"] + .terminal_time_ms + .max(by_id["long-prefill-child"].terminal_time_ms) + ); let trajectories = report.trajectories.unwrap(); assert_eq!(trajectories.total, 2); assert_eq!(trajectories.completed, 2); assert_eq!(trajectories.incomplete, 0); } +#[rstest::rstest] +#[case(EngineType::Vllm)] +#[case(EngineType::Sglang)] +fn agentic_lane_submission_is_independent_of_pd_source_cleanup(#[case] engine_type: EngineType) { + struct BusyPrefillTiming { + large_prefill_ms: f64, + } + + impl crate::engine::TimingModel for BusyPrefillTiming { + fn predict_prefill_ms( + &self, + batch_size: usize, + mean_isl: usize, + _mean_prefix: usize, + ) -> anyhow::Result { + Ok(if batch_size == 0 { + 0.0 + } else if mean_isl >= 1_024 { + self.large_prefill_ms + } else { + 1.0 + }) + } + + fn predict_decode_ms( + &self, + batch_size: usize, + _active_kv_tokens: usize, + _mean_context_length: usize, + _total_kv_tokens: usize, + ) -> anyhow::Result { + Ok(if batch_size == 0 { 0.0 } else { 1.0 }) + } + } + + let mut config = match engine_type { + EngineType::Vllm => disagg_config(), + EngineType::Sglang => sglang_disagg_config(), + EngineType::Trtllm => unreachable!(), + }; + config.num_prefill_workers = 1; + config.num_decode_workers = 1; + for args in [&mut config.prefill_args, &mut config.decode_args] { + args.speedup_ratio = 1.0; + args.decode_speedup_ratio = 1.0; + } + let block_size = config.prefill_args.block_size; + // Play IDs sort into lane 0's first-play then next-play. An independent play on lane 1 + // starts a long prefill at t=2, after both short roots finish prefill but + // before their transfers finish at t=3. Its busy source worker delays + // ReleaseSource without changing first-root's final decode terminal. + let trace = agentic_trace( + block_size, + vec![ + agentic_row( + "first-root", + "00-first-play", + 4, + 2, + block_size, + 10_000, + vec![], + ), + agentic_row( + "other-root", + "01-other-play", + 4, + 1, + block_size, + 20_000, + vec![], + ), + agentic_row( + "busy-prefill", + "01-other-play", + 4_096, + 1, + block_size, + 30_000, + vec![AgenticDependency { + request_id: "other-root".to_string(), + trigger: AgenticDependencyTrigger::Dispatch, + delay_ms: 2.0, + relation: AgenticDependencyRelation::Spawn, + }], + ), + agentic_row( + "next-root", + "02-next-play", + 4, + 1, + block_size, + 40_000, + vec![], + ), + ], + ); + + let run = |large_prefill_ms, stepped| { + let factory = ReplayEngineFactory::with_timing_model(Arc::new(BusyPrefillTiming { + large_prefill_ms, + })); + let mut runtime_config = config.runtime_config_with_factory(false, factory).unwrap(); + runtime_config.handoff_latency_ms = 2.0; + let driver = + WorkloadDriver::new_agentic_trace_with_lanes(trace.clone(), block_size, 2).unwrap(); + let mut runtime = RoundRobinDisaggRuntime::new_round_robin_workload( + &runtime_config, + driver, + ReplayMode::Trace, + ) + .unwrap() + .with_per_request_records(true); + + if stepped { + let mut observed_terminal_with_source_held = false; + while !runtime.is_done() { + runtime.step().unwrap(); + let first = runtime + .collector + .per_request_records() + .into_iter() + .find(|record| record.request_id.as_deref() == Some("first-root")); + if let Some(first) = first + && !observed_terminal_with_source_held + { + assert_eq!(first.terminal_status, ReplayTerminalStatus::Completed); + assert!(first.source_released_ms.is_none()); + let uuid = Uuid::parse_str(&first.uuid).unwrap(); + let state = runtime.state(uuid).unwrap(); + assert_eq!(state.phase, DisaggPhase::CleanupPending); + assert!(!state.counted_in_flight); + assert!(!state.coordinator.is_complete()); + assert!( + runtime + .flow + .requests_by_handoff + .contains_key(&state.handoff_id) + ); + assert!(runtime.flow.action_queues.contains(uuid)); + assert!(runtime.prefill_engine.worker_is_busy(0).unwrap()); + assert!(!runtime.prefill_engine.is_drained()); + assert!(!CoreAdmissionSource::is_drained(&runtime.admission)); + let transcript = runtime.admission.agentic_lifecycle_transcript().unwrap(); + assert!( + transcript.events.iter().any(|event| { + event.request_id.as_deref() == Some("next-root") + && event.event + == crate::replay::loadgen::AgenticLifecycleEventKind::Dispatch + && event.at_ms == first.terminal_time_ms + }), + "the next play must be submitted while source cleanup is still pending" + ); + observed_terminal_with_source_held = true; + } + } + assert!(observed_terminal_with_source_held); + } else { + runtime.run_to_completion().unwrap(); + } + + assert!(runtime.is_done()); + assert!(runtime.prefill_engine.is_drained()); + assert!(runtime.decode_engine.is_drained()); + assert!(runtime.flow.action_queues.is_empty()); + assert!(runtime.flow.requests_by_handoff.is_empty()); + assert!(CoreAdmissionSource::is_drained(&runtime.admission)); + let drained_at_ms = runtime.now_ms; + let (collector, stats) = runtime.run().unwrap(); + assert!( + stats + .request_snapshots + .values() + .all(|state| state.phase == DisaggPhase::Done) + ); + let report = collector.finish(); + assert_eq!(report.request_counts.completed_requests, 4); + assert_eq!(report.request_counts.total_output_tokens, 5); + (report, drained_at_ms) + }; + + let mut observations = Vec::new(); + for large_prefill_ms in [20.0, 40.0] { + let (report, drained_at_ms) = run(large_prefill_ms, false); + let (stepped, stepped_drained_at_ms) = run(large_prefill_ms, true); + assert_eq!(drained_at_ms, stepped_drained_at_ms); + assert_eq!(report.agentic_lifecycle, stepped.agentic_lifecycle); + assert_eq!( + serde_json::to_vec(&report.per_request).unwrap(), + serde_json::to_vec(&stepped.per_request).unwrap() + ); + let by_id = report + .per_request + .iter() + .map(|record| (record.request_id.as_deref().unwrap(), record)) + .collect::>(); + let first = by_id["first-root"]; + let next = by_id["next-root"]; + let source_released_ms = first.source_released_ms.unwrap(); + let next_dispatch_ms = next.dispatched_at_ms.unwrap(); + let next_admit_ms = next.first_admit_ms.unwrap(); + assert_eq!(next_dispatch_ms, first.terminal_time_ms); + assert!(source_released_ms > first.terminal_time_ms); + assert!(next_admit_ms >= source_released_ms); + assert!(next_admit_ms > next_dispatch_ms); + assert!(drained_at_ms > source_released_ms); + observations.push(( + first.terminal_time_ms, + next_dispatch_ms, + source_released_ms, + next_admit_ms, + drained_at_ms, + )); + } + let [short, long] = observations.as_slice() else { + unreachable!() + }; + assert_eq!(short.0, long.0, "client terminal must stay fixed"); + assert_eq!(short.1, long.1, "client lane timing must stay fixed"); + assert!( + short.2 < long.2, + "source ownership must reflect the cleanup delay" + ); + assert!( + short.3 < long.3, + "backend admission must respect the occupied worker" + ); + assert!( + short.4 < long.4, + "final drain must still wait for all server work" + ); +} + #[test] fn agentic_prefill_rejection_skips_descendants_and_releases_the_lane() { let mut config = disagg_config(); diff --git a/crates/core/src/replay/event.rs b/crates/core/src/replay/event.rs index 398cf7d3a..538f7f1d1 100644 --- a/crates/core/src/replay/event.rs +++ b/crates/core/src/replay/event.rs @@ -11,7 +11,7 @@ use super::core::{EngineEventBatch, EngineProgress}; use crate::engine::HandoffId; use crate::replay::protocol::{ForwardPassSnapshot, OutputSignal}; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub(crate) enum SimulationWorkerStage { Aggregated, Prefill, @@ -88,15 +88,44 @@ pub(crate) enum SimulationEventKind { } impl SimulationEventKind { - /// Tie-breaker among events at the *same* `at_ms`: telemetry first observes - /// fully settled workload state, then scaling makes a decision from that - /// timestamp. `seq_no` is globally unique, so this only reorders control - /// events relative to same-timestamp work. + /// Canonical phase order among events at the same logical timestamp. + /// Runtime drains each phase to a fixed point before telemetry observes the + /// settled state and scaling makes a decision. fn ordering_rank(&self) -> u8 { match self { - SimulationEventKind::TelemetryTick => 1, - SimulationEventKind::ScalingTick => 2, - _ => 0, + SimulationEventKind::EnginePassCompletion(_) => 0, + SimulationEventKind::WorkerReady { .. } => 1, + SimulationEventKind::TransferComplete { .. } => 2, + SimulationEventKind::TelemetryTick => 3, + SimulationEventKind::ScalingTick => 4, + } + } + + fn semantic_cmp(&self, other: &Self) -> Ordering { + match (self, other) { + ( + SimulationEventKind::EnginePassCompletion(left), + SimulationEventKind::EnginePassCompletion(right), + ) => (right.stage, right.worker_id, right.pass_id.get()).cmp(&( + left.stage, + left.worker_id, + left.pass_id.get(), + )), + ( + SimulationEventKind::WorkerReady { + stage: left_stage, + worker_id: left_worker, + }, + SimulationEventKind::WorkerReady { + stage: right_stage, + worker_id: right_worker, + }, + ) => (*right_stage, *right_worker).cmp(&(*left_stage, *left_worker)), + ( + SimulationEventKind::TransferComplete { handoff_id: left }, + SimulationEventKind::TransferComplete { handoff_id: right }, + ) => right.cmp(left), + _ => Ordering::Equal, } } } @@ -133,6 +162,7 @@ impl Ord for SimulationEvent { .at_ms .total_cmp(&self.at_ms) .then_with(|| other.kind.ordering_rank().cmp(&self.kind.ordering_rank())) + .then_with(|| self.kind.semantic_cmp(&other.kind)) .then_with(|| other.seq_no.cmp(&self.seq_no)) } } diff --git a/crates/core/src/replay/loadgen/driver.rs b/crates/core/src/replay/loadgen/driver.rs index aa6331d96..a5d3e0ca6 100644 --- a/crates/core/src/replay/loadgen/driver.rs +++ b/crates/core/src/replay/loadgen/driver.rs @@ -7,23 +7,116 @@ use std::collections::BinaryHeap; use anyhow::{Context, Result, anyhow, bail}; use rand::SeedableRng; use rand::rngs::StdRng; -use rustc_hash::FxHashMap; +use rustc_hash::{FxHashMap, FxHashSet}; +use serde::Serialize; use uuid::Uuid; use super::trace::validate_synthesizable_prompt; use super::types::{ - AgenticDependencyTrigger, AgenticGraphIdentity, AgenticTrace, AgenticTrajectorySnapshot, - CompactReadyTurn, ReadyTurn, ReplayRequestHashes, ReplayRequestPayload, Trace, + AgenticDependencyRelation, AgenticDependencyTrigger, AgenticGraphIdentity, AgenticTrace, + AgenticTrajectorySnapshot, CompactReadyTurn, ReadyTurn, ReplayRequestHashes, + ReplayRequestPayload, Trace, }; use super::{SYNTHETIC_OUTPUT_SEED, planned_output_token_ids}; use crate::replay::ReplayTerminalStatus; -use crate::replay::protocol::DirectRequest; +use crate::replay::protocol::{ + AgenticRuntimeIdentity, DirectRequest, ReplayPromptTokenSource, ReplayRequestContext, +}; + +pub const AGENTIC_LIFECYCLE_SCHEMA_V1: &str = "aisimulate.agentic.lifecycle.v1"; +const AGENTIC_LIFECYCLE_DIGEST_DOMAIN_V1: &[u8] = b"aisimulate-agentic-lifecycle-v1\0"; + +/// Output progress for one request inside a same-timestamp feedback batch. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AgenticOutputFeedback { + pub request_uuid: Uuid, + pub token_ids: Vec, +} + +/// Causal terminal state for one request inside a same-timestamp batch. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AgenticTerminalFeedback { + pub request_uuid: Uuid, + pub status: ReplayTerminalStatus, +} + +/// Agentic workload feedback produced by the runtime at one logical timestamp. +/// +/// The runtime owns `at_ms`. The driver canonicalizes request order by graph +/// ordinal and applies output progress, causal terminals, then quiescence. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct AgenticRuntimeFeedback { + pub at_ms: f64, + pub output_tokens: Vec, + pub causal_terminals: Vec, + pub quiescent_requests: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum AgenticLifecycleEventKind { + Dispatch, + CausalTerminal, + Quiescent, + Skipped, + PlayQuiescent, +} + +/// Canonical workload-level lifecycle record produced by the agentic driver. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct AgenticLifecycleEvent { + pub schema: &'static str, + pub ordinal: u64, + pub at_ms: f64, + pub play_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub request_id: Option, + pub event: AgenticLifecycleEventKind, + #[serde(skip_serializing_if = "Option::is_none")] + pub status: Option, +} + +/// Byte-stable lifecycle evidence for one preloaded graph execution. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct AgenticLifecycleTranscript { + pub events: Vec, +} + +impl AgenticLifecycleTranscript { + pub fn to_jsonl(&self) -> Result> { + let mut bytes = Vec::new(); + for event in &self.events { + serde_json::to_writer(&mut bytes, event) + .context("serializing agentic lifecycle event")?; + bytes.push(b'\n'); + } + Ok(bytes) + } + + pub fn digest(&self) -> Result { + let bytes = self.to_jsonl()?; + let mut hasher = blake3::Hasher::new(); + hasher.update(AGENTIC_LIFECYCLE_DIGEST_DOMAIN_V1); + hasher.update(&bytes); + Ok(hasher.finalize().to_hex().to_string()) + } +} + +#[derive(Debug, Clone, Copy)] +struct AgenticLifecycleRecord { + ordinal: u64, + at_ms: f64, + play_index: usize, + node_index: Option, + event: AgenticLifecycleEventKind, + status: Option, +} #[derive(Debug)] enum SchedulingPolicy { Trace, Concurrency(ConcurrencyState), - Agentic(AgenticState), + Agentic(Box), } #[derive(Debug)] @@ -35,6 +128,7 @@ struct ConcurrencyState { #[derive(Debug)] struct AgenticState { + identities: Vec, node_states: Vec, remaining_dependencies: Vec, authored_not_before_ms: Vec, @@ -44,6 +138,9 @@ struct AgenticState { node_to_play: Vec, plays: Vec, lanes: Vec, + lifecycle: Vec, + next_lifecycle_ordinal: u64, + last_runtime_feedback_at_ms: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -64,12 +161,15 @@ struct AgenticDependentEdge { #[derive(Debug)] struct AgenticPlayState { + play_id: String, nodes: Vec, root_nodes: Vec, lane_index: Option, + pending_terminals: usize, emitted_in_flight: usize, completed_nodes: usize, failed: bool, + client_finished: bool, quiescent: bool, root_dispatch_ms: Option, max_terminal_ms: Option, @@ -290,6 +390,29 @@ impl ConcurrencyState { } impl AgenticState { + fn record_lifecycle( + &mut self, + at_ms: f64, + play_index: usize, + node_index: Option, + event: AgenticLifecycleEventKind, + status: Option, + ) { + let ordinal = self.next_lifecycle_ordinal; + self.next_lifecycle_ordinal = self + .next_lifecycle_ordinal + .checked_add(1) + .expect("agentic lifecycle ordinal overflow"); + self.lifecycle.push(AgenticLifecycleRecord { + ordinal, + at_ms, + play_index, + node_index, + event, + status, + }); + } + fn activate_play( &mut self, play_index: usize, @@ -414,6 +537,7 @@ impl AgenticState { debug_assert_eq!(self.node_states[node_index], AgenticNodeState::Ready); self.node_states[node_index] = AgenticNodeState::Emitted; let play = &mut self.plays[self.node_to_play[node_index]]; + play.pending_terminals += 1; play.emitted_in_flight += 1; if play.root_nodes.contains(&node_index) { play.root_dispatch_ms = Some( @@ -421,6 +545,13 @@ impl AgenticState { .map_or(now_ms, |seen| seen.min(now_ms)), ); } + self.record_lifecycle( + now_ms, + self.node_to_play[node_index], + Some(node_index), + AgenticLifecycleEventKind::Dispatch, + None, + ); } fn on_node_terminal( @@ -432,9 +563,26 @@ impl AgenticState { outcome: TurnOutcome, ) { let play_index = self.node_to_play[node_index]; + let status = match outcome { + TurnOutcome::Completed => ReplayTerminalStatus::Completed, + TurnOutcome::Rejected => ReplayTerminalStatus::Rejected, + TurnOutcome::Cancelled => ReplayTerminalStatus::Canceled, + TurnOutcome::Failed => ReplayTerminalStatus::Failed, + }; + self.record_lifecycle( + now_ms, + play_index, + Some(node_index), + AgenticLifecycleEventKind::CausalTerminal, + Some(status), + ); let was_failed = self.plays[play_index].failed; { let play = &mut self.plays[play_index]; + play.pending_terminals = play + .pending_terminals + .checked_sub(1) + .expect("a dispatched agentic node becomes terminal exactly once"); play.max_terminal_ms = Some(play.max_terminal_ms.map_or(now_ms, |seen| seen.max(now_ms))); } @@ -453,7 +601,8 @@ impl AgenticState { self.node_states[node_index] = AgenticNodeState::Failed; self.plays[play_index].failed = true; if !was_failed { - for &pending_node in &self.plays[play_index].nodes { + let play_nodes = self.plays[play_index].nodes.clone(); + for pending_node in play_nodes { if matches!( self.node_states[pending_node], AgenticNodeState::Blocked | AgenticNodeState::Ready @@ -462,31 +611,52 @@ impl AgenticState { let session = &mut sessions[pending_node]; session.next_ready_at_ms = None; session.next_turn_index = session.turns.len(); + self.record_lifecycle( + now_ms, + play_index, + Some(pending_node), + AgenticLifecycleEventKind::Skipped, + None, + ); } } } } } + + let play = &self.plays[play_index]; + let all_completed = !play.failed && play.completed_nodes == play.nodes.len(); + // A failed play skips undispatched work, but every dispatched request, + // including background children, still owns client work until terminal. + // Successful plays must also finish delayed and blocked authored nodes. + if (all_completed || play.failed) && play.pending_terminals == 0 { + self.release_lane(play_index, now_ms, sessions, ready_sessions); + } } - fn on_node_quiescent( - &mut self, - sessions: &mut [SessionRuntime], - ready_sessions: &mut BinaryHeap, - node_index: usize, - now_ms: f64, - ) -> Result<()> { + fn on_node_quiescent(&mut self, node_index: usize, now_ms: f64) -> Result<()> { let play_index = self.node_to_play[node_index]; + self.record_lifecycle( + now_ms, + play_index, + Some(node_index), + AgenticLifecycleEventKind::Quiescent, + None, + ); let play = &mut self.plays[play_index]; play.emitted_in_flight = play .emitted_in_flight .checked_sub(1) .context("agentic play settlement count underflow")?; - let play = &self.plays[play_index]; - let all_completed = !play.failed && play.completed_nodes == play.nodes.len(); - let failed_and_settled = play.failed && play.emitted_in_flight == 0; - if (all_completed || failed_and_settled) && play.emitted_in_flight == 0 { - self.release_lane(play_index, now_ms, sessions, ready_sessions); + if !play.quiescent && play.client_finished && play.emitted_in_flight == 0 { + play.quiescent = true; + self.record_lifecycle( + now_ms, + play_index, + None, + AgenticLifecycleEventKind::PlayQuiescent, + None, + ); } Ok(()) } @@ -498,10 +668,10 @@ impl AgenticState { sessions: &mut [SessionRuntime], ready_sessions: &mut BinaryHeap, ) { - if self.plays[play_index].quiescent { + if self.plays[play_index].client_finished { return; } - self.plays[play_index].quiescent = true; + self.plays[play_index].client_finished = true; let Some(lane_index) = self.plays[play_index].lane_index else { return; }; @@ -550,6 +720,10 @@ pub struct WorkloadDriver { } impl WorkloadDriver { + pub fn is_agentic(&self) -> bool { + matches!(self.policy, SchedulingPolicy::Agentic(_)) + } + pub fn new_trace(trace: Trace, engine_block_size: usize) -> Result { Self::new( trace, @@ -687,6 +861,36 @@ impl WorkloadDriver { for (node_index, node) in trace.nodes.iter().enumerate() { index_by_id.insert(node.request_id.clone(), node_index); } + let root_id_by_play = trace + .plays + .iter() + .map(|play| { + let root_id = (play.root_nodes.len() == 1) + .then(|| trace.nodes[play.root_nodes[0]].request_id.clone()); + (play.play_id.clone(), root_id) + }) + .collect::>(); + let mut identities = trace + .nodes + .iter() + .map(|node| { + let parent_id = node + .dependencies + .iter() + .filter(|dependency| dependency.relation == AgenticDependencyRelation::Spawn) + .min_by_key(|dependency| index_by_id[&dependency.request_id]) + .map(|dependency| dependency.request_id.clone()); + AgenticRuntimeIdentity { + request_id: node.request_id.clone(), + play_id: node.play_id.clone(), + conversation_id: node.session_id.clone(), + lane_id: None, + root_id: root_id_by_play.get(&node.play_id).cloned().flatten(), + parent_id, + cache_id: None, + } + }) + .collect::>(); let mut dispatch_dependents = vec![Vec::new(); trace.nodes.len()]; let mut completion_dependents = vec![Vec::new(); trace.nodes.len()]; let mut remaining_dependencies = Vec::with_capacity(trace.nodes.len()); @@ -779,12 +983,15 @@ impl WorkloadDriver { node_to_play[node_index] = play_index; } AgenticPlayState { + play_id: play.play_id, nodes: play.nodes, root_nodes: play.root_nodes, lane_index: None, + pending_terminals: 0, emitted_in_flight: 0, completed_nodes: 0, failed: false, + client_finished: false, quiescent: false, root_dispatch_ms: None, max_terminal_ms: None, @@ -806,9 +1013,16 @@ impl WorkloadDriver { plays[play_index].lane_index = Some(lane_index); lanes[lane_index].plays.push(play_index); } + for (node_index, identity) in identities.iter_mut().enumerate() { + identity.lane_id = Some(format!( + "lane:{}", + plays[node_to_play[node_index]].lane_index.unwrap() + )); + } } let mut state = AgenticState { + identities, node_states: vec![AgenticNodeState::Blocked; sessions.len()], remaining_dependencies, ready_after_ms: authored_not_before_ms.clone(), @@ -818,6 +1032,9 @@ impl WorkloadDriver { node_to_play, plays, lanes, + lifecycle: Vec::new(), + next_lifecycle_ordinal: 0, + last_runtime_feedback_at_ms: None, }; let mut ready_sessions = BinaryHeap::new(); if state.lanes.is_empty() { @@ -844,7 +1061,7 @@ impl WorkloadDriver { } Ok(Self { - policy: SchedulingPolicy::Agentic(state), + policy: SchedulingPolicy::Agentic(Box::new(state)), prompt_mode: PromptMode::Full, emit_session_metadata: true, trace_block_size, @@ -1066,8 +1283,22 @@ impl WorkloadDriver { continue; }; let request_uuid = self.request_uuid(session_index, turn_index); + let agentic_identity = match &self.policy { + SchedulingPolicy::Agentic(state) => Some(state.identities[session_index].clone()), + SchedulingPolicy::Trace | SchedulingPolicy::Concurrency(_) => None, + }; let session = &mut self.sessions[session_index]; let turn = &mut session.turns[turn_index]; + let replay_context = turn.request_id.as_ref().zip(turn.play_id.as_ref()).map( + |(request_id, _play_id)| ReplayRequestContext { + authored_id: request_id.clone(), + session_id: Some(session.session_id.clone()), + turn_index: None, + metadata: serde_json::Value::Null, + prompt_token_source: ReplayPromptTokenSource::Materialized, + agentic: agentic_identity, + }, + ); let arrival_timestamp_ms = self.policy.arrival_timestamp_ms(scheduled_ready_at_ms); let (request, replay_hashes) = match self.prompt_mode { PromptMode::Full => { @@ -1084,7 +1315,7 @@ impl WorkloadDriver { priority: turn.priority, strict_priority: turn.strict_priority, policy_class: turn.policy_class.clone(), - replay_context: None, + replay_context: replay_context.clone(), }; let request = ReplayRequestPayload::deferred( request_metadata, @@ -1127,7 +1358,7 @@ impl WorkloadDriver { priority: turn.priority, strict_priority: turn.strict_priority, policy_class: turn.policy_class.clone(), - replay_context: None, + replay_context, }); (request, replay_hashes) } @@ -1210,6 +1441,145 @@ impl WorkloadDriver { Ok(()) } + fn agentic_node_ordinal(&self, request_uuid: Uuid) -> Result { + self.in_flight + .get(&request_uuid) + .map(|turn| turn.session_index) + .or_else(|| self.agentic_settling.get(&request_uuid).copied()) + .ok_or_else(|| anyhow!("unknown agentic workload request {request_uuid}")) + } + + /// Apply every workload-visible effect at one runtime-owned timestamp. + /// + /// Callers may collect effects in engine-specific order. This boundary + /// makes that order unobservable by sorting requests by immutable graph + /// ordinal, then applying output progress, causal terminals, and finally + /// resource quiescence. Batch timestamps must be nondecreasing; a rejected + /// batch leaves request state, dependencies, lifecycle, and clock unchanged. + pub fn apply_agentic_runtime_feedback( + &mut self, + mut feedback: AgenticRuntimeFeedback, + ) -> Result<()> { + if !feedback.at_ms.is_finite() || feedback.at_ms < 0.0 { + bail!( + "agentic runtime feedback timestamp must be finite and non-negative; got {}", + feedback.at_ms + ); + } + let SchedulingPolicy::Agentic(state) = &self.policy else { + bail!("agentic runtime feedback requires an agentic workload driver"); + }; + if let Some(last_at_ms) = state.last_runtime_feedback_at_ms + && feedback.at_ms < last_at_ms + { + bail!( + "agentic runtime feedback timestamp regressed from {last_at_ms} ms to {} ms", + feedback.at_ms + ); + } + + let ordinal_by_uuid = feedback + .output_tokens + .iter() + .map(|output| output.request_uuid) + .chain( + feedback + .causal_terminals + .iter() + .map(|terminal| terminal.request_uuid), + ) + .chain(feedback.quiescent_requests.iter().copied()) + .map(|uuid| Ok((uuid, self.agentic_node_ordinal(uuid)?))) + .collect::>>()?; + feedback + .output_tokens + .sort_by_key(|output| ordinal_by_uuid[&output.request_uuid]); + feedback + .causal_terminals + .sort_by_key(|terminal| ordinal_by_uuid[&terminal.request_uuid]); + feedback + .quiescent_requests + .sort_by_key(|uuid| ordinal_by_uuid[uuid]); + + for pair in feedback.output_tokens.windows(2) { + if pair[0].request_uuid == pair[1].request_uuid { + bail!( + "agentic request {} has duplicate output groups at {} ms", + pair[0].request_uuid, + feedback.at_ms + ); + } + } + for pair in feedback.causal_terminals.windows(2) { + if pair[0].request_uuid == pair[1].request_uuid { + bail!( + "agentic request {} has duplicate causal terminals at {} ms", + pair[0].request_uuid, + feedback.at_ms + ); + } + } + for pair in feedback.quiescent_requests.windows(2) { + if pair[0] == pair[1] { + bail!( + "agentic request {} has duplicate quiescence at {} ms", + pair[0], + feedback.at_ms + ); + } + } + + // Agentic graphs use full prompts, for which output feedback is a + // no-op. Validate every fallible lifecycle transition before applying + // any terminal: a later invalid transition must not commit earlier + // completions, release dependencies, or consume pending cleanup. + debug_assert_eq!(self.prompt_mode, PromptMode::Full); + let mut becoming_terminal = FxHashSet::default(); + for terminal in &feedback.causal_terminals { + let outcome = match terminal.status { + ReplayTerminalStatus::Completed => TurnOutcome::Completed, + ReplayTerminalStatus::Rejected => TurnOutcome::Rejected, + ReplayTerminalStatus::Canceled => TurnOutcome::Cancelled, + ReplayTerminalStatus::Failed => TurnOutcome::Failed, + }; + if self.agentic_settling.contains_key(&terminal.request_uuid) + || self + .validate_turn_resolution(terminal.request_uuid, outcome)? + .is_none() + { + bail!( + "agentic request {} received duplicate causal terminal", + terminal.request_uuid + ); + } + becoming_terminal.insert(terminal.request_uuid); + } + for request_uuid in &feedback.quiescent_requests { + if !self.agentic_settling.contains_key(request_uuid) + && !becoming_terminal.contains(request_uuid) + { + bail!("agentic request {request_uuid} became quiescent before its causal terminal"); + } + } + + for output in feedback.output_tokens { + for token_id in output.token_ids { + self.on_output_token(output.request_uuid, token_id)?; + } + } + for terminal in feedback.causal_terminals { + self.on_causal_terminal(terminal.request_uuid, feedback.at_ms, terminal.status)?; + } + for request_uuid in feedback.quiescent_requests { + self.on_quiescent(request_uuid, feedback.at_ms)?; + } + let SchedulingPolicy::Agentic(state) = &mut self.policy else { + unreachable!("feedback application retains the agentic scheduling policy"); + }; + state.last_runtime_feedback_at_ms = Some(feedback.at_ms); + Ok(()) + } + pub fn on_complete(&mut self, request_uuid: Uuid, now_ms: f64) -> Result<()> { self.on_terminal(request_uuid, now_ms, ReplayTerminalStatus::Completed) } @@ -1224,11 +1594,12 @@ impl WorkloadDriver { self.on_quiescent(request_uuid, now_ms) } - /// Resolve the logical request outcome and update causal successors. + /// Resolve the logical request outcome and update client scheduling. /// /// Disaggregated replay calls this as soon as final decode succeeds or a - /// failure is observed, then calls [`Self::on_quiescent`] only after all - /// P/D resources and coordinator actions settle. Aggregated replay uses + /// failure is observed. A play releases its client lane once all required + /// requests are terminal, independently of the later [`Self::on_quiescent`] + /// callbacks for P/D resources and coordinator actions. Aggregated replay uses /// [`Self::on_terminal`], where both events coincide. #[allow(clippy::collapsible_if)] // Keep agentic-only duplicate detection explicit. pub fn on_causal_terminal( @@ -1271,20 +1642,14 @@ impl WorkloadDriver { let Some(node_index) = self.agentic_settling.remove(&request_uuid) else { bail!("agentic request {request_uuid} became quiescent before its causal terminal"); }; - state.on_node_quiescent( - &mut self.sessions, - &mut self.ready_sessions, - node_index, - now_ms, - ) + state.on_node_quiescent(node_index, now_ms) } - fn resolve_turn( - &mut self, + fn validate_turn_resolution( + &self, request_uuid: Uuid, - now_ms: f64, outcome: TurnOutcome, - ) -> Result> { + ) -> Result> { let Some(in_flight) = self.in_flight.get(&request_uuid).copied() else { return match outcome { TurnOutcome::Completed | TurnOutcome::Rejected | TurnOutcome::Failed => Err( @@ -1297,7 +1662,7 @@ impl WorkloadDriver { .sessions .get(in_flight.session_index) .ok_or_else(|| anyhow!("unknown workload session {}", in_flight.session_index))?; - let turn = session.turns.get(in_flight.turn_index).ok_or_else(|| { + session.turns.get(in_flight.turn_index).ok_or_else(|| { anyhow!( "unknown workload turn {} for session {}", in_flight.turn_index, @@ -1327,6 +1692,26 @@ impl WorkloadDriver { in_flight.emitted_output_tokens ); } + if matches!(outcome, TurnOutcome::Completed | TurnOutcome::Rejected) { + in_flight + .turn_index + .checked_add(1) + .context("workload turn index overflow")?; + } + Ok(Some(in_flight)) + } + + fn resolve_turn( + &mut self, + request_uuid: Uuid, + now_ms: f64, + outcome: TurnOutcome, + ) -> Result> { + let Some(in_flight) = self.validate_turn_resolution(request_uuid, outcome)? else { + return Ok(None); + }; + let session = &self.sessions[in_flight.session_index]; + let turn = &session.turns[in_flight.turn_index]; let completed_output_tokens = (outcome == TurnOutcome::Completed && self.prompt_mode == PromptMode::DeltaCumulative) .then(|| { @@ -1442,6 +1827,32 @@ impl WorkloadDriver { pub fn agentic_graph_identity(&self) -> Option { self.agentic_graph_identity.clone() } + + pub fn agentic_lifecycle_transcript(&self) -> Option { + let SchedulingPolicy::Agentic(state) = &self.policy else { + return None; + }; + Some(AgenticLifecycleTranscript { + events: state + .lifecycle + .iter() + .map(|record| AgenticLifecycleEvent { + schema: AGENTIC_LIFECYCLE_SCHEMA_V1, + ordinal: record.ordinal, + at_ms: record.at_ms, + play_id: state.plays[record.play_index].play_id.clone(), + request_id: record.node_index.map(|node_index| { + self.sessions[node_index].turns[0] + .request_id + .clone() + .expect("agentic node must retain its authored request ID") + }), + event: record.event, + status: record.status, + }) + .collect(), + }) + } } #[cfg(test)] @@ -2290,6 +2701,276 @@ mod tests { assert_eq!(ids[0], vec![Uuid::from_u128(1), Uuid::from_u128(2)]); } + #[test] + fn agentic_dispatch_carries_the_stable_identity_envelope() { + let trace = agentic_trace(vec![agentic_node("root", "play-a", 0.0, Vec::new())]); + let mut driver = WorkloadDriver::new_agentic_trace_with_lanes(trace, 1, 1).unwrap(); + + let ready = driver.pop_ready(0.0, 1).pop().unwrap(); + let context = ready.request.replay_context.as_ref().unwrap(); + let identity = context.agentic.as_ref().unwrap(); + + assert_eq!(context.authored_id, "root"); + assert_eq!(identity.request_id, "root"); + assert_eq!(identity.play_id, "play-a"); + assert_eq!(identity.conversation_id, "play-a"); + assert_eq!(identity.lane_id.as_deref(), Some("lane:0")); + assert_eq!(identity.root_id.as_deref(), Some("root")); + assert_eq!(identity.parent_id, None); + } + + #[test] + fn same_timestamp_feedback_is_canonicalized_by_graph_ordinal() { + let trace = agentic_trace(vec![ + agentic_node("a", "play", 0.0, Vec::new()), + agentic_node("b", "play", 0.0, Vec::new()), + ]); + + let run = |reverse: bool| { + let mut driver = WorkloadDriver::new_agentic_trace(trace.clone(), 1).unwrap(); + let ready = driver.pop_ready(0.0, usize::MAX); + let mut terminals = vec![ + AgenticTerminalFeedback { + request_uuid: ready[0].request_uuid, + status: ReplayTerminalStatus::Completed, + }, + AgenticTerminalFeedback { + request_uuid: ready[1].request_uuid, + status: ReplayTerminalStatus::Failed, + }, + ]; + let mut quiescent = ready + .iter() + .map(|turn| turn.request_uuid) + .collect::>(); + if reverse { + terminals.reverse(); + quiescent.reverse(); + } + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 10.0, + output_tokens: Vec::new(), + causal_terminals: terminals, + quiescent_requests: quiescent, + }) + .unwrap(); + driver.agentic_lifecycle_transcript().unwrap() + }; + + let authored = run(false); + let reversed = run(true); + assert_eq!(authored.to_jsonl().unwrap(), reversed.to_jsonl().unwrap()); + assert_eq!(authored.digest().unwrap(), reversed.digest().unwrap()); + assert_eq!( + authored + .events + .iter() + .filter(|event| event.event == AgenticLifecycleEventKind::CausalTerminal) + .map(|event| event.request_id.as_deref().unwrap()) + .collect::>(), + vec!["a", "b"] + ); + } + + #[test] + fn agentic_feedback_rejects_time_regression_without_advancing_dependencies_or_clock() { + let trace = agentic_trace(vec![ + agentic_node("clock", "play", 0.0, Vec::new()), + agentic_node("parent", "play", 0.0, Vec::new()), + agentic_node( + "child", + "play", + 0.0, + vec![dependency( + "parent", + AgenticDependencyTrigger::Completion, + 2.0, + AgenticDependencyRelation::Sequence, + )], + ), + ]); + let mut driver = WorkloadDriver::new_agentic_trace(trace, 1).unwrap(); + let ready = driver.pop_ready(0.0, usize::MAX); + assert_eq!(ready.len(), 2); + let clock = ready[0].request_uuid; + let parent = ready[1].request_uuid; + let completed = |request_uuid| AgenticTerminalFeedback { + request_uuid, + status: ReplayTerminalStatus::Completed, + }; + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 10.0, + causal_terminals: vec![completed(clock)], + quiescent_requests: vec![clock], + ..Default::default() + }) + .unwrap(); + let transcript_before = driver.agentic_lifecycle_transcript().unwrap(); + + let error = driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 5.0, + causal_terminals: vec![completed(parent)], + quiescent_requests: vec![parent], + ..Default::default() + }) + .unwrap_err(); + assert!(error.to_string().contains("regressed from 10 ms to 5 ms")); + assert_eq!( + driver.agentic_lifecycle_transcript().unwrap(), + transcript_before + ); + assert_eq!(driver.next_ready_time_ms(), None); + assert!(driver.in_flight.contains_key(&parent)); + assert!(!driver.agentic_settling.contains_key(&parent)); + + // A known request with invalid lifecycle feedback reaches application + // validation, but must not make the next valid t=10 batch look stale. + let error = driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 20.0, + quiescent_requests: vec![parent], + ..Default::default() + }) + .unwrap_err(); + assert!( + error + .to_string() + .contains("became quiescent before its causal terminal") + ); + assert_eq!( + driver.agentic_lifecycle_transcript().unwrap(), + transcript_before + ); + assert_eq!(driver.next_ready_time_ms(), None); + + // A second successful batch at the same timestamp remains legal. + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 10.0, + causal_terminals: vec![completed(parent)], + ..Default::default() + }) + .unwrap(); + assert_eq!(driver.next_ready_time_ms(), Some(12.0)); + assert!(driver.pop_ready(11.0, usize::MAX).is_empty()); + let child = driver.pop_ready(12.0, usize::MAX); + assert_eq!(child.len(), 1); + assert_eq!(child[0].authored_request_id.as_deref(), Some("child")); + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 12.0, + causal_terminals: vec![completed(child[0].request_uuid)], + quiescent_requests: vec![child[0].request_uuid], + ..Default::default() + }) + .unwrap(); + assert!(!driver.is_drained(), "parent cleanup is still pending"); + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 15.0, + quiescent_requests: vec![parent], + ..Default::default() + }) + .unwrap(); + assert!(driver.is_drained()); + assert!( + driver + .agentic_lifecycle_transcript() + .unwrap() + .events + .windows(2) + .all(|events| events[0].at_ms <= events[1].at_ms) + ); + } + + #[rstest::rstest] + #[case::premature_quiescence(false)] + #[case::duplicate_terminal(true)] + fn agentic_rejected_mixed_feedback_preserves_state(#[case] duplicate_terminal: bool) { + let trace = agentic_trace(vec![ + agentic_node("a-parent", "play", 0.0, Vec::new()), + agentic_node("b-stale", "play", 0.0, Vec::new()), + agentic_node("c-peer", "play", 0.0, Vec::new()), + agentic_node( + "d-child", + "play", + 0.0, + vec![dependency( + "a-parent", + AgenticDependencyTrigger::Completion, + 2.0, + AgenticDependencyRelation::Sequence, + )], + ), + ]); + let mut driver = WorkloadDriver::new_agentic_trace_with_lanes(trace, 1, 1).unwrap(); + let ready = driver.pop_ready(0.0, usize::MAX); + let parent = ready[0].request_uuid; + let stale = ready[1].request_uuid; + let peer = ready[2].request_uuid; + let completed = |request_uuid| AgenticTerminalFeedback { + request_uuid, + status: ReplayTerminalStatus::Completed, + }; + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 10.0, + causal_terminals: vec![completed(stale)], + ..Default::default() + }) + .unwrap(); + let state_before = format!("{driver:?}"); + let mut invalid = AgenticRuntimeFeedback { + at_ms: 20.0, + causal_terminals: vec![completed(parent)], + quiescent_requests: vec![parent, stale], + ..Default::default() + }; + if duplicate_terminal { + invalid.causal_terminals.push(completed(stale)); + } else { + invalid.quiescent_requests.push(peer); + } + + assert!(driver.apply_agentic_runtime_feedback(invalid).is_err()); + // Include request/session state, dependency counters, lanes, cleanup, + // output progress, lifecycle records, and the successful batch clock. + assert_eq!(format!("{driver:?}"), state_before); + assert_eq!(driver.next_ready_time_ms(), None); + + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 10.0, + causal_terminals: vec![completed(parent), completed(peer)], + quiescent_requests: vec![parent, stale, peer], + ..Default::default() + }) + .unwrap(); + assert_eq!(driver.next_ready_time_ms(), Some(12.0)); + let child = driver.pop_ready(12.0, 1).pop().unwrap(); + assert_eq!(child.authored_request_id.as_deref(), Some("d-child")); + driver + .apply_agentic_runtime_feedback(AgenticRuntimeFeedback { + at_ms: 12.0, + causal_terminals: vec![completed(child.request_uuid)], + quiescent_requests: vec![child.request_uuid], + ..Default::default() + }) + .unwrap(); + assert!(driver.is_drained()); + assert!( + driver + .agentic_lifecycle_transcript() + .unwrap() + .events + .windows(2) + .all(|events| events[0].at_ms <= events[1].at_ms) + ); + } + #[test] fn agentic_failed_parent_skips_unemitted_descendants() { let trace = agentic_trace(vec![ @@ -2320,26 +3001,149 @@ mod tests { } #[test] - fn agentic_lane_waits_for_quiescence_after_causal_completion() { + fn agentic_lane_recycles_at_client_terminal_but_drain_waits_for_cleanup() { + for status in [ + ReplayTerminalStatus::Completed, + ReplayTerminalStatus::Rejected, + ReplayTerminalStatus::Canceled, + ReplayTerminalStatus::Failed, + ] { + for cleanup_at_ms in [20.0, 40.0] { + let trace = agentic_trace(vec![ + agentic_node("a", "play-a", 0.0, Vec::new()), + agentic_node("b", "play-b", 0.0, Vec::new()), + ]); + let mut driver = WorkloadDriver::new_agentic_trace_with_lanes(trace, 1, 1).unwrap(); + let first = driver.pop_ready(0.0, 1).pop().unwrap(); + driver + .on_causal_terminal(first.request_uuid, 10.0, status) + .unwrap(); + + assert_eq!(driver.next_ready_time_ms(), Some(10.0)); + let second = driver.pop_ready(10.0, 1).pop().unwrap(); + assert_eq!(second.authored_request_id.as_deref(), Some("b")); + assert_eq!(second.dispatched_at_ms, 10.0); + driver.on_complete(second.request_uuid, 12.0).unwrap(); + assert!(!driver.is_drained(), "play-a still owns server cleanup"); + + driver + .on_quiescent(first.request_uuid, cleanup_at_ms) + .unwrap(); + assert!(driver.is_drained()); + let transcript = driver.agentic_lifecycle_transcript().unwrap(); + let settlements = transcript + .events + .iter() + .filter(|event| event.event == AgenticLifecycleEventKind::PlayQuiescent) + .map(|event| (event.play_id.as_str(), event.at_ms)) + .collect::>(); + assert_eq!( + settlements, + vec![("play-b", 12.0), ("play-a", cleanup_at_ms)] + ); + } + } + } + + #[test] + fn late_quiescence_cannot_release_a_lane_owned_by_the_next_play() { let trace = agentic_trace(vec![ agentic_node("a", "play-a", 0.0, Vec::new()), agentic_node("b", "play-b", 0.0, Vec::new()), + agentic_node("c", "play-c", 0.0, Vec::new()), ]); let mut driver = WorkloadDriver::new_agentic_trace_with_lanes(trace, 1, 1).unwrap(); - - let first = driver.pop_ready(0.0, usize::MAX); - assert_eq!(first.len(), 1); - assert_eq!(first[0].authored_request_id.as_deref(), Some("a")); + let first = driver.pop_ready(0.0, 1).pop().unwrap(); driver - .on_causal_terminal(first[0].request_uuid, 10.0, ReplayTerminalStatus::Completed) + .on_causal_terminal(first.request_uuid, 10.0, ReplayTerminalStatus::Completed) .unwrap(); + let second = driver.pop_ready(10.0, 1).pop().unwrap(); + + driver.on_quiescent(first.request_uuid, 20.0).unwrap(); assert!(driver.pop_ready(20.0, usize::MAX).is_empty()); + driver + .on_causal_terminal(second.request_uuid, 30.0, ReplayTerminalStatus::Completed) + .unwrap(); + let third = driver.pop_ready(30.0, 1).pop().unwrap(); + assert_eq!(third.authored_request_id.as_deref(), Some("c")); + driver.on_complete(third.request_uuid, 31.0).unwrap(); + assert!(!driver.is_drained()); + driver.on_quiescent(second.request_uuid, 35.0).unwrap(); + assert!(driver.is_drained()); + } - driver.on_quiescent(first[0].request_uuid, 20.0).unwrap(); - let second = driver.pop_ready(20.0, usize::MAX); - assert_eq!(second.len(), 1); - assert_eq!(second[0].authored_request_id.as_deref(), Some("b")); - assert_eq!(second[0].dispatched_at_ms, 20.0); + #[test] + fn agentic_lane_waits_for_background_terminals_even_after_root_failure() { + for status in [ + ReplayTerminalStatus::Completed, + ReplayTerminalStatus::Rejected, + ReplayTerminalStatus::Canceled, + ReplayTerminalStatus::Failed, + ] { + let trace = agentic_trace(vec![ + agentic_node("root", "play-a", 0.0, Vec::new()), + agentic_node( + "background", + "play-a", + 0.0, + vec![dependency( + "root", + AgenticDependencyTrigger::Dispatch, + 0.0, + AgenticDependencyRelation::Spawn, + )], + ), + agentic_node("next", "play-b", 0.0, Vec::new()), + ]); + let mut driver = WorkloadDriver::new_agentic_trace_with_lanes(trace, 1, 1).unwrap(); + let ready = driver.pop_ready(0.0, usize::MAX); + assert_eq!(ready.len(), 2); + driver + .on_terminal(ready[0].request_uuid, 10.0, status) + .unwrap(); + assert!(driver.pop_ready(10.0, usize::MAX).is_empty()); + + driver + .on_causal_terminal(ready[1].request_uuid, 15.0, ReplayTerminalStatus::Completed) + .unwrap(); + let next = driver.pop_ready(15.0, 1).pop().unwrap(); + assert_eq!(next.authored_request_id.as_deref(), Some("next")); + driver.on_complete(next.request_uuid, 16.0).unwrap(); + assert!(!driver.is_drained()); + driver.on_quiescent(ready[1].request_uuid, 20.0).unwrap(); + assert!(driver.is_drained()); + } + } + + #[test] + fn agentic_lane_waits_for_delayed_authored_requests() { + let trace = agentic_trace(vec![ + agentic_node("root", "play-a", 0.0, Vec::new()), + agentic_node( + "delayed", + "play-a", + 0.0, + vec![dependency( + "root", + AgenticDependencyTrigger::Completion, + 5.0, + AgenticDependencyRelation::Sequence, + )], + ), + agentic_node("next", "play-b", 0.0, Vec::new()), + ]); + let mut driver = WorkloadDriver::new_agentic_trace_with_lanes(trace, 1, 1).unwrap(); + let root = driver.pop_ready(0.0, 1).pop().unwrap(); + driver.on_complete(root.request_uuid, 10.0).unwrap(); + assert!(driver.pop_ready(14.0, usize::MAX).is_empty()); + let delayed = driver.pop_ready(15.0, usize::MAX); + assert_eq!(delayed.len(), 1); + assert_eq!(delayed[0].authored_request_id.as_deref(), Some("delayed")); + driver.on_complete(delayed[0].request_uuid, 20.0).unwrap(); + let next = driver.pop_ready(20.0, 1).pop().unwrap(); + assert_eq!(next.authored_request_id.as_deref(), Some("next")); + driver.on_complete(next.request_uuid, 21.0).unwrap(); + assert!(driver.is_drained()); } #[test] @@ -2365,21 +3169,21 @@ mod tests { driver .on_causal_terminal(first[0].request_uuid, 10.0, ReplayTerminalStatus::Completed) .unwrap(); - driver.on_quiescent(first[0].request_uuid, 20.0).unwrap(); - - let second_root = driver.pop_ready(20.0, usize::MAX); + let second_root = driver.pop_ready(10.0, usize::MAX); assert_eq!(second_root.len(), 1); assert_eq!( second_root[0].authored_request_id.as_deref(), Some("second-root") ); - assert!(driver.pop_ready(24.9, usize::MAX).is_empty()); - let second_child = driver.pop_ready(25.0, usize::MAX); + assert_eq!(second_root[0].scheduled_ready_at_ms, 10.0); + assert!(driver.pop_ready(14.9, usize::MAX).is_empty()); + let second_child = driver.pop_ready(15.0, usize::MAX); assert_eq!(second_child.len(), 1); assert_eq!( second_child[0].authored_request_id.as_deref(), Some("second-child") ); + driver.on_quiescent(first[0].request_uuid, 20.0).unwrap(); } #[test] diff --git a/crates/core/src/replay/loadgen/mod.rs b/crates/core/src/replay/loadgen/mod.rs index bf946226c..24fc64695 100644 --- a/crates/core/src/replay/loadgen/mod.rs +++ b/crates/core/src/replay/loadgen/mod.rs @@ -13,7 +13,11 @@ mod weka; use rand::Rng; use rand::rngs::StdRng; -pub use driver::WorkloadDriver; +pub use driver::{ + AGENTIC_LIFECYCLE_SCHEMA_V1, AgenticLifecycleEvent, AgenticLifecycleEventKind, + AgenticLifecycleTranscript, AgenticOutputFeedback, AgenticRuntimeFeedback, + AgenticTerminalFeedback, WorkloadDriver, +}; pub use dynamo::DynamoRequestTrace; pub use generated::GeneratedRequests; pub use steppable::{EngineEvent, StepOutcome, SteppableAgg, SteppableEngine, SteppableReplay}; diff --git a/crates/core/src/replay/mod.rs b/crates/core/src/replay/mod.rs index 41b34c91b..93cb30e21 100644 --- a/crates/core/src/replay/mod.rs +++ b/crates/core/src/replay/mod.rs @@ -153,7 +153,9 @@ pub use handoff::{ }; pub use protocol::ForwardPassSnapshot; #[doc(hidden)] -pub use protocol::{DirectRequest, ReplayPromptTokenSource, ReplayRequestContext}; +pub use protocol::{ + AgenticRuntimeIdentity, DirectRequest, ReplayPromptTokenSource, ReplayRequestContext, +}; #[doc(hidden)] pub use replayer::ReplayRuntimeInput; pub use replayer::{ReplayComposition, Replayer, RoundRobinComposition}; diff --git a/crates/core/src/replay/protocol.rs b/crates/core/src/replay/protocol.rs index 6d03af498..63e617385 100644 --- a/crates/core/src/replay/protocol.rs +++ b/crates/core/src/replay/protocol.rs @@ -8,6 +8,26 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use uuid::Uuid; +/// Stable Agentic correlation data carried across replay-owned boundaries. +/// +/// The static M1 graph already owns request, play, and conversation identity. +/// Later replay phases may populate lane, tree, parent, and cache identities +/// without changing the placement or engine request contracts again. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AgenticRuntimeIdentity { + pub request_id: String, + pub play_id: String, + pub conversation_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lane_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub root_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cache_id: Option, +} + /// Where the prompt token identities carried by a lowered request came from. /// /// Length-only specifications still need deterministic token identities for @@ -39,6 +59,8 @@ pub struct ReplayRequestContext { pub metadata: Value, #[serde(default)] pub prompt_token_source: ReplayPromptTokenSource, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agentic: Option, } /// One materialized replay request before it enters a generalized engine. diff --git a/crates/core/src/replay/replayer.rs b/crates/core/src/replay/replayer.rs index e0194da79..2b4b1660e 100644 --- a/crates/core/src/replay/replayer.rs +++ b/crates/core/src/replay/replayer.rs @@ -642,6 +642,7 @@ fn lower_requests( turn_index: request.turn_index, metadata: request.metadata.clone(), prompt_token_source, + agentic: None, }), }) }) diff --git a/crates/core/src/replay/report.rs b/crates/core/src/replay/report.rs index 0de84eca2..753c1f7cf 100644 --- a/crates/core/src/replay/report.rs +++ b/crates/core/src/replay/report.rs @@ -11,7 +11,9 @@ use uuid::Uuid; use crate::engine::CacheTierAttribution; use crate::replay::PlacementCacheSample; -use crate::replay::loadgen::{AgenticGraphIdentity, AgenticTrajectorySnapshot}; +use crate::replay::loadgen::{ + AgenticGraphIdentity, AgenticLifecycleTranscript, AgenticTrajectorySnapshot, +}; // 0.1% relative quantile error. The enlarged store covers latency/rate values // spanning roughly 10^28 within one sign while remaining bounded (~512 KiB for @@ -30,6 +32,9 @@ pub struct ReplayReport { pub latency: TraceLatencyStats, pub trajectories: Option, pub agentic_graph: Option, + /// Canonical driver lifecycle evidence. The compact JSON report publishes + /// only its digest and event count; conformance tests can inspect all events. + pub agentic_lifecycle: Option, /// SLA-goodput stats. `Some` only when an SLA was supplied to the collector /// (via `set_sla_thresholds`); `None` otherwise — goodput is undefined /// without an SLA, so the `goodput_*` keys are omitted from the report. @@ -320,6 +325,13 @@ impl Serialize for ReplayReport { if let Some(agentic_graph) = &self.agentic_graph { map.serialize_entry("agentic_graph", agentic_graph)?; } + if let Some(lifecycle) = &self.agentic_lifecycle { + map.serialize_entry("agentic_lifecycle_event_count", &lifecycle.events.len())?; + map.serialize_entry( + "agentic_lifecycle_digest", + &lifecycle.digest().map_err(serde::ser::Error::custom)?, + )?; + } serialize_distribution(&mut map, "e2e_latency", &self.latency.e2e)?; serialize_rate_distribution( &mut map, @@ -786,6 +798,7 @@ pub struct TraceCollector { runtime_evidence: crate::replay::OfflineRuntimeEvidence, agentic_trajectory: Option, agentic_graph: Option, + agentic_lifecycle: Option, } impl TraceRequestStats { @@ -999,6 +1012,10 @@ impl TraceCollector { self.agentic_graph = Some(identity); } + pub fn set_agentic_lifecycle(&mut self, transcript: AgenticLifecycleTranscript) { + self.agentic_lifecycle = Some(transcript); + } + /// Retain the ReplaySpec correlation fields before the request crosses /// placement and engine boundaries. pub fn on_request_context( @@ -1431,6 +1448,7 @@ impl TraceCollector { let decode_gpus_per_worker = self.decode_gpus_per_worker; let runtime_evidence = self.runtime_evidence; let agentic_graph = self.agentic_graph; + let agentic_lifecycle = self.agentic_lifecycle; let trajectories = self .agentic_trajectory .map(|snapshot| TraceTrajectoryStats { @@ -1585,6 +1603,7 @@ impl TraceCollector { }, trajectories, agentic_graph, + agentic_lifecycle, goodput, per_request, runtime_evidence, diff --git a/crates/core/src/replay/runtime_utils.rs b/crates/core/src/replay/runtime_utils.rs index 5c63ef19a..2d1152b1c 100644 --- a/crates/core/src/replay/runtime_utils.rs +++ b/crates/core/src/replay/runtime_utils.rs @@ -14,6 +14,14 @@ use crate::engine::HandoffId; #[cfg(test)] use crate::replay::protocol::DirectRequest; +/// Result of advancing a replay runtime to its next settled semantic boundary. +#[derive(Debug, Clone, Copy, PartialEq)] +pub(crate) enum ReplayStepOutcome { + Settled { now_ms: f64 }, + Complete, + TimeLimitReached { now_ms: f64 }, +} + pub(super) fn next_timestamp( next_arrival_ms: Option, next_event_ms: Option, @@ -253,6 +261,8 @@ pub(super) fn pop_ready_telemetry_tick( #[cfg(test)] mod tests { use super::*; + use crate::engine::generalized::PassId; + use crate::replay::components::ScheduledEngineCompletion; use crate::replay::events::SimulationWorkerStage; use uuid::Uuid; @@ -378,4 +388,54 @@ mod tests { // pop_ready_worker_ready should succeed. assert!(pop_ready_worker_ready(&mut events, 10.0).is_some()); } + + #[test] + fn same_timestamp_events_follow_semantic_phase_and_identity_order() { + let mut events: BinaryHeap> = BinaryHeap::new(); + let mut next_event_seq = 0; + push_transfer_complete( + &mut events, + &mut next_event_seq, + 10.0, + HandoffId::new(Uuid::from_u128(2)), + ); + push_worker_ready( + &mut events, + &mut next_event_seq, + 10.0, + SimulationWorkerStage::Decode, + 3, + ); + push_worker_completions( + &mut events, + &mut next_event_seq, + ScheduledEngineCompletion { + at_ms: 10.0, + completion: EnginePassCompletion::new(SimulationWorkerStage::Decode, 4, PassId(9)), + }, + ); + push_worker_completions( + &mut events, + &mut next_event_seq, + ScheduledEngineCompletion { + at_ms: 10.0, + completion: EnginePassCompletion::new(SimulationWorkerStage::Prefill, 2, PassId(3)), + }, + ); + + let first = pop_ready_worker_completions(&mut events, 10.0).unwrap(); + assert_eq!(first.stage, SimulationWorkerStage::Prefill); + assert_eq!(first.worker_id, 2); + let second = pop_ready_worker_completions(&mut events, 10.0).unwrap(); + assert_eq!(second.stage, SimulationWorkerStage::Decode); + assert_eq!(second.worker_id, 4); + assert_eq!( + pop_ready_worker_ready(&mut events, 10.0), + Some((SimulationWorkerStage::Decode, 3)) + ); + assert_eq!( + pop_ready_transfer_complete(&mut events, 10.0), + Some(HandoffId::new(Uuid::from_u128(2))) + ); + } } diff --git a/docs/cli/user-guide.md b/docs/cli/user-guide.md index 8f45199ee..e7c5c3e6b 100644 --- a/docs/cli/user-guide.md +++ b/docs/cli/user-guide.md @@ -766,9 +766,13 @@ models are not supported yet. The lowering records a zero-based `source_play_ordinal` on every v2 row so materialized graphs retain deterministic directory and JSONL order; missing ordinals remain valid for older v2 inputs, but an ordered graph must provide one unique contiguous ordinal for every play. -An explicit `agentic_lanes: N` assigns plays round-robin to N lanes and starts the next play in a lane -only after the current play becomes quiescent. Omitting the field preserves authored timestamp -behavior; corpus wrapping and fixed-duration lane orchestration are unsupported. +An explicit `agentic_lanes: N` assigns plays round-robin to N client lanes. The next play starts when +the current play's client work ends: all authored requests complete on success, or all dispatched +requests become terminal after a failure skips undispatched work. Background requests remain part +of their play even without a parent join. P/D source holds and other server cleanup may outlive this +boundary; they still constrain engine admission and final drain, but do not delay client submission. +Omitting the field preserves authored timestamp behavior; corpus wrapping and fixed-duration lane +orchestration are outside the version 1 contract. ### Mooncake and Mooncake Delta JSONL