1mod replay;
2
3use arc_swap::ArcSwap;
4use std::sync::Arc;
5use tokio::sync::Mutex;
6
7use super::{LashRuntime, ProcessHandleGrantEntry, ProcessHandleSummary, ProcessRegistry};
8
9pub use replay::{
10 InMemoryLiveReplayStore, InMemoryLiveReplayStoreConfig, LiveReplayGap, LiveReplayGapReason,
11 LiveReplayResult, LiveReplayStore, LiveReplayStoreError, LiveReplaySubscribeResult,
12 LiveReplaySubscription, SessionCursor, SessionCursorError, SessionObservation,
13 SessionObservationEvent, SessionObservationEventPayload, SessionObservationSubscription,
14 SessionProcessEventKind, SessionQueueEventKind, SessionResume, SessionRevision,
15};
16
17#[derive(Clone)]
18pub struct RuntimeObservation {
19 pub session_id: Arc<str>,
20 pub revision: SessionRevision,
21 pub cursor: SessionCursor,
22 pub policy: crate::SessionPolicy,
23 pub read_view: crate::SessionReadView,
24 pub persisted_state: super::RuntimeSessionState,
25 pub usage_report: super::SessionUsageReport,
26 pub tool_state: Option<crate::ToolState>,
27 pub tool_catalog: Arc<Vec<serde_json::Value>>,
28 pub tool_catalog_error: Option<String>,
29 pub plugin_session: Option<Arc<crate::PluginSession>>,
30 pub session_read_service: Option<Arc<dyn crate::plugin::SessionReadService>>,
31 pub process_read_service: Option<Arc<dyn crate::plugin::ProcessReadService>>,
32 pub process_registry: Option<Arc<dyn ProcessRegistry>>,
33 pub queue_store: Option<Arc<dyn crate::RuntimePersistence>>,
34 pub queued_work_driver: Option<super::QueuedWorkDriver>,
35}
36
37impl RuntimeObservation {
38 fn from_runtime(
39 runtime: &LashRuntime,
40 cursor: SessionCursor,
41 previous: Option<&RuntimeObservation>,
42 revision: SessionRevision,
43 read_view: crate::SessionReadView,
44 persisted_state: super::RuntimeSessionState,
45 usage_report: super::SessionUsageReport,
46 ) -> Self {
47 let (tool_catalog, tool_catalog_error) = match runtime.active_tool_catalog_shared() {
48 Ok(catalog) => (catalog, None),
49 Err(err) => (Arc::new(Vec::new()), Some(err.to_string())),
50 };
51 let tool_state_generation = runtime
52 .session
53 .as_ref()
54 .map(|session| session.plugins().tool_registry().generation());
55 let tool_state = match (
56 tool_state_generation,
57 previous.and_then(|observation| observation.tool_state.as_ref()),
58 ) {
59 (Some(generation), Some(snapshot)) if snapshot.generation() == generation => {
60 Some(snapshot.clone())
61 }
62 (Some(_), _) => match runtime.tool_state() {
63 Ok(state) => Some(state),
64 Err(err) => {
65 tracing::warn!(
66 session_id = %runtime.session_id(),
67 error = %err,
68 "failed to capture tool state for observation; omitting the snapshot",
69 );
70 None
71 }
72 },
73 (None, _) => None,
74 };
75 let (plugin_session, session_read_service, process_read_service) =
76 match (runtime.session.as_ref(), runtime.runtime_session_services()) {
77 (Some(session), Ok(services)) => (
78 Some(Arc::clone(session.plugins())),
79 Some(services.read_service()),
80 Some(services.process_read_service()),
81 ),
82 (_, Err(err)) => {
83 tracing::warn!(
84 session_id = %runtime.session_id(),
85 error = %err,
86 "failed to capture plugin query services for observation",
87 );
88 (None, None, None)
89 }
90 (None, _) => (None, None, None),
91 };
92 Self {
93 session_id: Arc::from(runtime.session_id()),
94 revision,
95 cursor,
96 policy: read_view.policy().clone(),
97 read_view,
98 persisted_state,
99 usage_report,
100 tool_state,
101 tool_catalog,
102 tool_catalog_error,
103 plugin_session,
104 session_read_service,
105 process_read_service,
106 process_registry: runtime.host.process_registry.clone(),
107 queue_store: runtime
108 .session
109 .as_ref()
110 .and_then(|session| session.history_store()),
111 queued_work_driver: runtime.host.queued_work_driver.clone(),
112 }
113 }
114
115 pub fn session_id(&self) -> &str {
116 &self.session_id
117 }
118
119 pub fn session_revision(&self) -> SessionRevision {
120 self.revision
121 }
122
123 pub fn cursor(&self) -> &SessionCursor {
124 &self.cursor
125 }
126
127 pub fn session_observation(&self) -> SessionObservation {
128 SessionObservation {
129 read_view: self.read_view.clone(),
130 cursor: self.cursor.clone(),
131 }
132 }
133
134 pub fn process_scope(&self) -> crate::SessionScope {
135 crate::SessionScope::new(self.session_id.as_ref())
136 }
137
138 pub fn process_scope_id(&self) -> crate::SessionScopeId {
139 self.process_scope().id()
140 }
141
142 pub async fn query_plugin(
143 &self,
144 name: &str,
145 args: serde_json::Value,
146 session_id: Option<String>,
147 ) -> Result<(String, serde_json::Value), crate::PluginOperationInvokeError> {
148 let Some(plugin_session) = self.plugin_session.as_ref().cloned() else {
149 return Err(crate::PluginOperationInvokeError::Unknown(
150 "runtime session not available".to_string(),
151 ));
152 };
153 let Some(session_read_service) = self.session_read_service.as_ref().cloned() else {
154 return Err(crate::PluginOperationInvokeError::Unknown(
155 "runtime session read service not available".to_string(),
156 ));
157 };
158 let Some(process_read_service) = self.process_read_service.as_ref().cloned() else {
159 return Err(crate::PluginOperationInvokeError::Unknown(
160 "runtime process read service not available".to_string(),
161 ));
162 };
163 plugin_session
164 .query_plugin(
165 name,
166 args,
167 session_id,
168 true,
169 session_read_service,
170 process_read_service,
171 )
172 .await
173 }
174
175 pub async fn list_process_handles(&self) -> Vec<ProcessHandleSummary> {
176 let Some(executor) = self.process_registry.as_ref() else {
177 return Vec::new();
178 };
179 self.list_process_handles_with_mode(executor, crate::ProcessListMode::Live)
180 .await
181 }
182
183 pub async fn list_all_process_handles(&self) -> Vec<ProcessHandleSummary> {
184 let Some(executor) = self.process_registry.as_ref() else {
185 return Vec::new();
186 };
187 self.list_process_handles_with_mode(executor, crate::ProcessListMode::All)
188 .await
189 }
190
191 async fn list_process_handles_with_mode(
192 &self,
193 executor: &Arc<dyn crate::ProcessRegistry>,
194 mode: crate::ProcessListMode,
195 ) -> Vec<ProcessHandleSummary> {
196 let root_scope = self.process_scope();
197 let mut entries = list_scope_process_handles(executor, &root_scope, mode).await;
198 let agent_frame_id = self.persisted_state.current_agent_frame_id.as_str();
199 if !agent_frame_id.is_empty() {
200 let frame_scope =
201 crate::SessionScope::for_agent_frame(self.session_id.as_ref(), agent_frame_id);
202 if frame_scope.id() != root_scope.id() {
203 entries.extend(list_scope_process_handles(executor, &frame_scope, mode).await);
204 entries.sort_by(|(left, _), (right, _)| left.process_id.cmp(&right.process_id));
205 entries.dedup_by(|(left, _), (right, _)| left.process_id == right.process_id);
206 }
207 }
208 entries
209 .into_iter()
210 .map(ProcessHandleSummary::from)
211 .collect()
212 }
213}
214
215fn export_observation_state(
216 runtime: &LashRuntime,
217) -> (
218 super::RuntimeSessionState,
219 crate::SessionReadView,
220 super::SessionUsageReport,
221) {
222 let mut state = runtime.export_persisted_state();
223 let read_view = runtime.read_view();
224 let shared_ledger = runtime
225 .shared_token_ledger
226 .lock()
227 .expect("token ledger lock");
228 for entry in shared_ledger.iter().cloned() {
229 super::merge_ledger_entry(&mut state.token_ledger, entry);
230 }
231 let usage_report = super::SessionUsageReport::from_entries(&state.token_ledger);
232 (state, read_view, usage_report)
233}
234
235async fn list_scope_process_handles(
236 executor: &Arc<dyn crate::ProcessRegistry>,
237 scope: &crate::SessionScope,
238 mode: crate::ProcessListMode,
239) -> Vec<ProcessHandleGrantEntry> {
240 match mode {
241 crate::ProcessListMode::Live => executor.list_live_handle_grants(scope).await,
242 crate::ProcessListMode::All => executor.list_handle_grants(scope).await,
243 }
244 .unwrap_or_default()
245}
246
247#[derive(Clone)]
248pub struct RuntimeHandle {
249 pub(in crate::runtime) runtime: Arc<Mutex<LashRuntime>>,
250 observation: Arc<ArcSwap<RuntimeObservation>>,
251 live_replay_store: Arc<dyn LiveReplayStore>,
252}
253
254impl RuntimeHandle {
255 pub fn new(runtime: LashRuntime) -> Self {
256 Self::with_live_replay_store(runtime, Arc::new(InMemoryLiveReplayStore::default()))
257 }
258
259 pub fn with_live_replay_store(
260 runtime: LashRuntime,
261 live_replay_store: Arc<dyn LiveReplayStore>,
262 ) -> Self {
263 let revision = SessionRevision::from_runtime(&runtime);
264 let cursor = live_replay_store.current_cursor(runtime.session_id(), revision);
265 let (state, read_view, usage_report) = export_observation_state(&runtime);
266 let observation = RuntimeObservation::from_runtime(
267 &runtime,
268 cursor,
269 None,
270 revision,
271 read_view,
272 state,
273 usage_report,
274 );
275 Self {
276 runtime: Arc::new(Mutex::new(runtime)),
277 observation: Arc::new(ArcSwap::from_pointee(observation)),
278 live_replay_store,
279 }
280 }
281
282 pub fn writer(&self) -> Arc<Mutex<LashRuntime>> {
283 Arc::clone(&self.runtime)
284 }
285
286 pub fn observe(&self) -> Arc<RuntimeObservation> {
287 self.observation.load_full()
288 }
289
290 pub fn publish_from(&self, runtime: &LashRuntime) {
291 let revision = SessionRevision::from_runtime(runtime);
292 let previous = self.observation.load_full();
293 let (state, read_view, usage_report) = export_observation_state(runtime);
294 if previous.persisted_state.current_agent_frame_id != state.current_agent_frame_id
295 && !state.current_agent_frame_id.is_empty()
296 && let Err(err) = self.live_replay_store.append(
297 runtime.session_id(),
298 revision,
299 SessionObservationEventPayload::AgentFrameSwitched {
300 frame_id: state.current_agent_frame_id.clone(),
301 },
302 )
303 {
304 tracing::warn!(
305 session_id = %runtime.session_id(),
306 error = %err,
307 "failed to append agent-frame observation event; reconnect may require gap recovery",
308 );
309 }
310 let cursor = match self.live_replay_store.append(
311 runtime.session_id(),
312 revision,
313 SessionObservationEventPayload::Committed {
314 read_view: read_view.clone(),
315 },
316 ) {
317 Ok(event) => event.cursor.clone(),
318 Err(err) => {
319 tracing::warn!(
320 session_id = %runtime.session_id(),
321 error = %err,
322 "failed to append session observation commit event; reconnect will fall back to gap recovery",
323 );
324 self.live_replay_store
325 .current_cursor(runtime.session_id(), revision)
326 }
327 };
328 self.observation
329 .store(Arc::new(RuntimeObservation::from_runtime(
330 runtime,
331 cursor,
332 Some(previous.as_ref()),
333 revision,
334 read_view,
335 state,
336 usage_report,
337 )));
338 }
339
340 pub fn record_turn_activity(&self, activity: crate::TurnActivity) {
341 let observation = self.observe();
342 if let Err(err) = self.live_replay_store.append(
343 observation.session_id(),
344 observation.session_revision(),
345 SessionObservationEventPayload::TurnActivity(activity),
346 ) {
347 tracing::warn!(
348 session_id = %observation.session_id(),
349 error = %err,
350 "failed to append live turn activity to session observation replay; reconnect may require gap recovery",
351 );
352 }
353 }
354
355 pub fn record_queue_changed(&self, kind: SessionQueueEventKind, batch_ids: Vec<String>) {
356 let observation = self.observe();
357 if let Err(err) = self.live_replay_store.append(
358 observation.session_id(),
359 observation.session_revision(),
360 SessionObservationEventPayload::QueueChanged { kind, batch_ids },
361 ) {
362 tracing::warn!(
363 session_id = %observation.session_id(),
364 error = %err,
365 "failed to append queue observation event; reconnect may require gap recovery",
366 );
367 }
368 }
369
370 pub fn record_process_changed(&self, kind: SessionProcessEventKind, process_ids: Vec<String>) {
371 let observation = self.observe();
372 if let Err(err) = self.live_replay_store.append(
373 observation.session_id(),
374 observation.session_revision(),
375 SessionObservationEventPayload::ProcessChanged { kind, process_ids },
376 ) {
377 tracing::warn!(
378 session_id = %observation.session_id(),
379 error = %err,
380 "failed to append process observation event; reconnect may require gap recovery",
381 );
382 }
383 }
384
385 pub fn current_session_observation(&self) -> SessionObservation {
386 let observation = self.observe();
387 self.session_observation_from(observation.as_ref())
388 }
389
390 pub fn resume_session_observation(
391 &self,
392 cursor: &SessionCursor,
393 ) -> Result<SessionResume, LiveReplayStoreError> {
394 let observation = self.observe();
395 cursor.parse_for_session(observation.session_id())?;
396 match self.live_replay_store.replay_after_cursor(cursor)? {
397 LiveReplayResult::Replayed(events) => Ok(SessionResume::Replayed { events }),
398 LiveReplayResult::Gap(reason) => Ok(SessionResume::Gap {
399 gap: self.live_replay_gap(cursor, reason, observation.as_ref()),
400 observation: self.session_observation_from(observation.as_ref()),
401 }),
402 }
403 }
404
405 pub fn subscribe_session_observation(
406 &self,
407 cursor: &SessionCursor,
408 ) -> Result<SessionObservationSubscription, LiveReplayStoreError> {
409 let observation = self.observe();
410 cursor.parse_for_session(observation.session_id())?;
411 match self.live_replay_store.subscribe_after_cursor(cursor)? {
412 LiveReplaySubscribeResult::Subscribed(subscription) => {
413 Ok(SessionObservationSubscription::Subscribed(subscription))
414 }
415 LiveReplaySubscribeResult::Gap(reason) => Ok(SessionObservationSubscription::Gap {
416 gap: self.live_replay_gap(cursor, reason, observation.as_ref()),
417 observation: self.session_observation_from(observation.as_ref()),
418 }),
419 }
420 }
421
422 fn session_observation_from(&self, observation: &RuntimeObservation) -> SessionObservation {
423 SessionObservation {
424 read_view: observation.read_view.clone(),
425 cursor: self
426 .live_replay_store
427 .current_cursor(observation.session_id(), observation.session_revision()),
428 }
429 }
430
431 fn live_replay_gap(
432 &self,
433 requested_cursor: &SessionCursor,
434 reason: LiveReplayGapReason,
435 observation: &RuntimeObservation,
436 ) -> LiveReplayGap {
437 let latest_cursor = self
438 .live_replay_store
439 .current_cursor(observation.session_id(), observation.session_revision());
440 LiveReplayGap {
441 session_id: observation.session_id().to_string(),
442 requested_cursor: requested_cursor.clone(),
443 latest_cursor,
444 latest_revision: observation.session_revision(),
445 reason,
446 }
447 }
448
449 pub async fn enqueue_turn_input(
450 &self,
451 input: crate::TurnInput,
452 ingress: crate::TurnInputIngress,
453 source_key: Option<String>,
454 ) -> Result<crate::PendingTurnInput, crate::RuntimeError> {
455 let observation = self.observe();
456 let store = observation
457 .queue_store
458 .clone()
459 .ok_or_else(super::session_api::queued_turn_input_store_required)?;
460 let is_next_turn = matches!(ingress, crate::TurnInputIngress::NextTurn);
461 super::session_api::enqueue_turn_input_to_store(
462 observation.session_id.as_ref().to_string(),
463 store,
464 observation.queued_work_driver.clone(),
465 input,
466 ingress,
467 source_key,
468 )
469 .await
470 .inspect(|input| {
471 self.record_queue_changed(
472 SessionQueueEventKind::Enqueued,
473 if is_next_turn {
474 vec![input.input_id.clone()]
475 } else {
476 Vec::new()
477 },
478 );
479 })
480 }
481
482 pub async fn cancel_pending_turn_input(
483 &self,
484 session_id: &str,
485 input_id: &str,
486 ) -> Result<crate::PendingTurnInputCancelOutcome, crate::RuntimeError> {
487 let observation = self.observe();
488 let store = observation
489 .queue_store
490 .clone()
491 .ok_or_else(super::session_api::queued_turn_input_store_required)?;
492 store
493 .cancel_pending_turn_input(session_id, input_id)
494 .await
495 .map_err(|err| {
496 crate::RuntimeError::new(
497 crate::RuntimeErrorCode::StoreCommitFailed,
498 err.to_string(),
499 )
500 })
501 .inspect(|outcome| {
502 if outcome.is_cancelled() {
503 self.record_queue_changed(
504 SessionQueueEventKind::Cancelled,
505 vec![input_id.to_string()],
506 );
507 }
508 })
509 }
510
511 pub async fn cancel_pending_turn_inputs(
512 &self,
513 session_id: &str,
514 targets: &[crate::PendingTurnInputCancelTarget],
515 ) -> Result<Vec<crate::PendingTurnInputCancelResult>, crate::RuntimeError> {
516 let observation = self.observe();
517 let store = observation
518 .queue_store
519 .clone()
520 .ok_or_else(super::session_api::queued_turn_input_store_required)?;
521 store
522 .cancel_pending_turn_inputs(session_id, targets)
523 .await
524 .map_err(|err| {
525 crate::RuntimeError::new(
526 crate::RuntimeErrorCode::StoreCommitFailed,
527 err.to_string(),
528 )
529 })
530 .inspect(|results| {
531 let cancelled_ids = results
532 .iter()
533 .filter_map(|result| match &result.outcome {
534 crate::PendingTurnInputCancelOutcome::Cancelled(input) => {
535 Some(input.input_id.clone())
536 }
537 _ => None,
538 })
539 .collect::<Vec<_>>();
540 if !cancelled_ids.is_empty() {
541 self.record_queue_changed(SessionQueueEventKind::Cancelled, cancelled_ids);
542 }
543 })
544 }
545
546 pub async fn cancel_pending_turn_input_suffix(
547 &self,
548 session_id: &str,
549 anchor: &crate::PendingTurnInputCancelTarget,
550 ) -> Result<crate::PendingTurnInputSuffixCancelOutcome, crate::RuntimeError> {
551 let observation = self.observe();
552 let store = observation
553 .queue_store
554 .clone()
555 .ok_or_else(super::session_api::queued_turn_input_store_required)?;
556 store
557 .cancel_pending_turn_input_suffix(session_id, anchor)
558 .await
559 .map_err(|err| {
560 crate::RuntimeError::new(
561 crate::RuntimeErrorCode::StoreCommitFailed,
562 err.to_string(),
563 )
564 })
565 .inspect(|outcome| {
566 let crate::PendingTurnInputSuffixCancelOutcome::Outcomes { outcomes, .. } = outcome
567 else {
568 return;
569 };
570 let cancelled_ids = outcomes
571 .iter()
572 .filter_map(|outcome| match outcome {
573 crate::PendingTurnInputCancelOutcome::Cancelled(input) => {
574 Some(input.input_id.clone())
575 }
576 _ => None,
577 })
578 .collect::<Vec<_>>();
579 if !cancelled_ids.is_empty() {
580 self.record_queue_changed(SessionQueueEventKind::Cancelled, cancelled_ids);
581 }
582 })
583 }
584
585 pub async fn abandon_queued_work_claim(
592 &self,
593 claim: &crate::QueuedWorkClaim,
594 ) -> Result<(), crate::RuntimeError> {
595 let observation = self.observe();
596 let store = observation
597 .queue_store
598 .clone()
599 .ok_or_else(super::session_api::queued_turn_input_store_required)?;
600 store
601 .abandon_queued_work_claim(claim)
602 .await
603 .map_err(|err| {
604 crate::RuntimeError::new(
605 crate::RuntimeErrorCode::StoreCommitFailed,
606 err.to_string(),
607 )
608 })?;
609 self.record_queue_changed(
610 SessionQueueEventKind::Enqueued,
611 claim
612 .batches
613 .iter()
614 .map(|batch| batch.batch_id.clone())
615 .collect(),
616 );
617 Ok(())
618 }
619
620 pub async fn abandon_turn_input_claim(
624 &self,
625 claim: &crate::TurnInputClaim,
626 ) -> Result<(), crate::RuntimeError> {
627 let observation = self.observe();
628 let store = observation
629 .queue_store
630 .clone()
631 .ok_or_else(super::session_api::queued_turn_input_store_required)?;
632 store.abandon_turn_input_claim(claim).await.map_err(|err| {
633 crate::RuntimeError::new(crate::RuntimeErrorCode::StoreCommitFailed, err.to_string())
634 })?;
635 self.record_queue_changed(
636 SessionQueueEventKind::Enqueued,
637 claim
638 .inputs
639 .iter()
640 .map(|input| input.input_id.clone())
641 .collect(),
642 );
643 Ok(())
644 }
645
646 pub async fn cancel_queued_work_batch(
647 &self,
648 session_id: &str,
649 batch_id: &str,
650 ) -> Result<Option<crate::QueuedWorkBatch>, crate::RuntimeError> {
651 let observation = self.observe();
652 let store = observation
653 .queue_store
654 .clone()
655 .ok_or_else(super::session_api::queued_turn_input_store_required)?;
656 store
657 .cancel_queued_work_batch(session_id, batch_id)
658 .await
659 .map_err(|err| {
660 crate::RuntimeError::new(
661 crate::RuntimeErrorCode::StoreCommitFailed,
662 err.to_string(),
663 )
664 })
665 .inspect(|batch| {
666 if batch.is_some() {
667 self.record_queue_changed(
668 SessionQueueEventKind::Cancelled,
669 vec![batch_id.to_string()],
670 );
671 }
672 })
673 }
674
675 pub fn try_into_runtime(self) -> Result<LashRuntime, Self> {
676 match Arc::try_unwrap(self.runtime) {
677 Ok(mutex) => Ok(mutex.into_inner()),
678 Err(runtime) => Err(Self {
679 runtime,
680 observation: self.observation,
681 live_replay_store: self.live_replay_store,
682 }),
683 }
684 }
685}
686
687#[cfg(test)]
688mod tests {
689 use super::*;
690 use std::time::Instant;
691
692 struct PanicLiveReplayStore;
693
694 impl LiveReplayStore for PanicLiveReplayStore {
695 fn append(
696 &self,
697 _session_id: &str,
698 _revision: SessionRevision,
699 _payload: SessionObservationEventPayload,
700 ) -> Result<Arc<SessionObservationEvent>, LiveReplayStoreError> {
701 panic!("append should not be called by cursor rejection tests")
702 }
703
704 fn replay_after_cursor(
705 &self,
706 _cursor: &SessionCursor,
707 ) -> Result<LiveReplayResult, LiveReplayStoreError> {
708 panic!("replay_after_cursor should not be called for rejected cursors")
709 }
710
711 fn subscribe_after_cursor(
712 &self,
713 _cursor: &SessionCursor,
714 ) -> Result<LiveReplaySubscribeResult, LiveReplayStoreError> {
715 panic!("subscribe_after_cursor should not be called for rejected cursors")
716 }
717
718 fn current_cursor(&self, session_id: &str, revision: SessionRevision) -> SessionCursor {
719 SessionCursor::new(session_id, revision, 0)
720 }
721
722 fn trim_session(&self, _session_id: &str) -> Result<(), LiveReplayStoreError> {
723 Ok(())
724 }
725 }
726
727 #[tokio::test]
728 async fn runtime_rejects_bad_cursors_before_replay_store_gap_handling() {
729 let runtime = LashRuntime::builder()
730 .with_session_id("session-a")
731 .with_policy(crate::SessionPolicy {
732 model: crate::ModelSpec::from_token_limits(
733 "test-model",
734 Default::default(),
735 1024,
736 None,
737 )
738 .expect("model"),
739 ..Default::default()
740 })
741 .build()
742 .await
743 .expect("runtime");
744 let handle = RuntimeHandle::with_live_replay_store(runtime, Arc::new(PanicLiveReplayStore));
745 let wrong_session = SessionCursor::new("session-b", SessionRevision(0), 99);
746 let malformed = SessionCursor::from_raw_for_testing("bad");
747
748 assert!(matches!(
749 handle.resume_session_observation(&wrong_session),
750 Err(LiveReplayStoreError::Cursor(
751 SessionCursorError::WrongSession { .. }
752 ))
753 ));
754 assert!(matches!(
755 handle.subscribe_session_observation(&wrong_session),
756 Err(LiveReplayStoreError::Cursor(
757 SessionCursorError::WrongSession { .. }
758 ))
759 ));
760 assert!(matches!(
761 handle.resume_session_observation(&malformed),
762 Err(LiveReplayStoreError::Cursor(
763 SessionCursorError::Malformed { .. }
764 ))
765 ));
766 assert!(matches!(
767 handle.subscribe_session_observation(&malformed),
768 Err(LiveReplayStoreError::Cursor(
769 SessionCursorError::Malformed { .. }
770 ))
771 ));
772 }
773
774 #[tokio::test]
775 async fn publish_revision_matches_the_single_export_across_a_commit() {
776 let runtime = LashRuntime::builder()
777 .with_session_id("revision-equivalence")
778 .with_policy(crate::SessionPolicy {
779 model: crate::ModelSpec::from_token_limits(
780 "test-model",
781 Default::default(),
782 1024,
783 None,
784 )
785 .expect("model"),
786 ..Default::default()
787 })
788 .build()
789 .await
790 .expect("runtime");
791 let handle = RuntimeHandle::new(runtime);
792 let writer = handle.writer();
793 let mut runtime = writer.lock().await;
794 runtime.state.turn_index = 9;
795 runtime.state.head_revision = Some(17);
796
797 let exported = runtime.export_persisted_state();
798 let exported_revision = SessionRevision::from_state(&exported);
799 let accessor_revision = SessionRevision::from_runtime(&runtime);
800 assert_eq!(accessor_revision, exported_revision);
801
802 handle.publish_from(&runtime);
803 assert_eq!(handle.observe().session_revision(), exported_revision);
804 }
805
806 #[tokio::test]
807 async fn publish_keeps_frame_switch_immediately_before_commit() {
808 let runtime = LashRuntime::builder()
809 .with_session_id("publish-order")
810 .with_policy(crate::SessionPolicy {
811 model: crate::ModelSpec::from_token_limits(
812 "test-model",
813 Default::default(),
814 1024,
815 None,
816 )
817 .expect("model"),
818 ..Default::default()
819 })
820 .build()
821 .await
822 .expect("runtime");
823 let handle = RuntimeHandle::new(runtime);
824 let cursor = handle.observe().cursor().clone();
825 let writer = handle.writer();
826 let mut runtime = writer.lock().await;
827 runtime.state.current_agent_frame_id = "next-frame".to_string();
828
829 handle.publish_from(&runtime);
830 let SessionResume::Replayed { events } = handle
831 .resume_session_observation(&cursor)
832 .expect("replay publication")
833 else {
834 panic!("publication should remain replayable");
835 };
836 assert_eq!(events.len(), 2);
837 assert!(matches!(
838 events[0].payload,
839 SessionObservationEventPayload::AgentFrameSwitched { .. }
840 ));
841 assert!(matches!(
842 events[1].payload,
843 SessionObservationEventPayload::Committed { .. }
844 ));
845 }
846
847 #[tokio::test]
848 #[ignore = "manual lane-O publish_from timing measurement"]
849 async fn measure_publish_from_wall_clock() {
850 const COMMITS: usize = 5_000;
851 let runtime = LashRuntime::builder()
852 .with_session_id("publish-perf")
853 .with_policy(crate::SessionPolicy {
854 model: crate::ModelSpec::from_token_limits(
855 "test-model",
856 Default::default(),
857 1024,
858 None,
859 )
860 .expect("model"),
861 ..Default::default()
862 })
863 .build()
864 .await
865 .expect("runtime");
866 let handle = RuntimeHandle::with_live_replay_store(
867 runtime,
868 Arc::new(InMemoryLiveReplayStore::with_bounds(
869 COMMITS + 1,
870 std::time::Duration::from_secs(120),
871 )),
872 );
873 let writer = handle.writer();
874 let runtime = writer.lock().await;
875 for _ in 0..100 {
876 handle.publish_from(&runtime);
877 }
878 let started = Instant::now();
879 for _ in 0..COMMITS {
880 handle.publish_from(&runtime);
881 }
882 let elapsed = started.elapsed();
883 eprintln!(
884 "publish_from: commits={COMMITS} elapsed_ns={} ns_per_commit={:.3}",
885 elapsed.as_nanos(),
886 elapsed.as_nanos() as f64 / COMMITS as f64,
887 );
888 }
889}