Skip to content

Restart-safe observation identity and turn-scoped events (FIG-615) #135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SamGalanakis merged 3 commits into from
Jul 25, 2026
Merged

Restart-safe observation identity and turn-scoped events (FIG-615) #135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
81957d8
Make recoverable chat event identity restart-safe
SamGalanakis Jul 25, 2026
05fd60a
Scope session observations to their producing turn
SamGalanakis Jul 25, 2026
8fedd82
Scope workbench terminal cleanup to one turn
SamGalanakis Jul 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/lash-core/src/runtime/lifecycle.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,7 @@ impl LashRuntime {
process_sync_needed: Arc::new(AtomicBool::new(false)),
turn_phase_probe: None,
last_committed_lease_continuity: None,
last_committed_observation_turn: None,
graph_loaded_from_store: false,
residency: Residency::default(),
})
Expand Down
14 changes: 14 additions & 0 deletions crates/lash-core/src/runtime/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -905,6 +905,17 @@ pub trait TurnActivitySink: Send + Sync {
}

async fn emit(&self, activity: TurnActivity);

/// Emit activity with the identity of the physical turn that produced it.
///
/// Sinks that only consume turn-local activity can keep implementing
/// [`emit`](Self::emit). Observation sinks override this method to carry
/// turn identity on their enclosing event without adding it to
/// [`TurnActivity`].
async fn emit_for_turn(&self, turn_id: &str, activity: TurnActivity) {
let _ = turn_id;
self.emit(activity).await;
}
}

pub struct NoopTurnActivitySink;
Expand Down Expand Up @@ -1092,6 +1103,9 @@ pub struct LashRuntime {
/// next physical turn on this handle.
pub(in crate::runtime) last_committed_lease_continuity:
Option<session_execution_lease::SessionExecutionLeaseContinuity>,
/// Most recent physical turn committed by this runtime, paired with the
/// resulting session revision for observation-envelope attribution.
pub(in crate::runtime) last_committed_observation_turn: Option<(u64, String)>,
/// Set only after this handle itself has attempted a durable graph load.
pub(in crate::runtime) graph_loaded_from_store: bool,
/// Resident-graph policy chosen by the host. Controls whether
Expand Down
22 changes: 21 additions & 1 deletion crates/lash-core/src/runtime/observation.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,12 +290,16 @@ impl RuntimeHandle {
pub fn publish_from(&self, runtime: &LashRuntime) {
let revision = SessionRevision::from_runtime(runtime);
let previous = self.observation.load_full();
let turn_id = (previous.revision != revision)
.then(|| runtime.last_committed_turn_id_for_revision(revision))
.flatten();
let (state, read_view, usage_report) = export_observation_state(runtime);
if previous.persisted_state.current_agent_frame_id != state.current_agent_frame_id
&& !state.current_agent_frame_id.is_empty()
&& let Err(err) = self.live_replay_store.append(
runtime.session_id(),
revision,
None,
SessionObservationEventPayload::AgentFrameSwitched {
frame_id: state.current_agent_frame_id.clone(),
},
Expand All @@ -310,6 +314,7 @@ impl RuntimeHandle {
let cursor = match self.live_replay_store.append(
runtime.session_id(),
revision,
turn_id,
SessionObservationEventPayload::Committed {
read_view: read_view.clone(),
},
Expand Down Expand Up @@ -337,11 +342,12 @@ impl RuntimeHandle {
)));
}

pub fn record_turn_activity(&self, activity: crate::TurnActivity) {
pub fn record_turn_activity(&self, turn_id: Option<&str>, activity: crate::TurnActivity) {
let observation = self.observe();
if let Err(err) = self.live_replay_store.append(
observation.session_id(),
observation.session_revision(),
turn_id,
SessionObservationEventPayload::TurnActivity(activity),
) {
tracing::warn!(
Expand All @@ -357,6 +363,7 @@ impl RuntimeHandle {
if let Err(err) = self.live_replay_store.append(
observation.session_id(),
observation.session_revision(),
None,
SessionObservationEventPayload::QueueChanged { kind, batch_ids },
) {
tracing::warn!(
Expand All @@ -372,6 +379,7 @@ impl RuntimeHandle {
if let Err(err) = self.live_replay_store.append(
observation.session_id(),
observation.session_revision(),
None,
SessionObservationEventPayload::ProcessChanged { kind, process_ids },
) {
tracing::warn!(
Expand Down Expand Up @@ -671,6 +679,15 @@ impl RuntimeHandle {
}
}

impl LashRuntime {
fn last_committed_turn_id_for_revision(&self, revision: SessionRevision) -> Option<&str> {
self.last_committed_observation_turn
.as_ref()
.filter(|(committed_revision, _)| *committed_revision == revision.as_u64())
.map(|(_, turn_id)| turn_id.as_str())
}
}

#[cfg(test)]
mod tests {
use super::*;
Expand All @@ -683,6 +700,7 @@ mod tests {
&self,
_session_id: &str,
_revision: SessionRevision,
_turn_id: Option<&str>,
_payload: SessionObservationEventPayload,
) -> Result<Arc<SessionObservationEvent>, LiveReplayStoreError> {
panic!("append should not be called by cursor rejection tests")
Expand Down Expand Up @@ -825,10 +843,12 @@ mod tests {
events[0].payload,
SessionObservationEventPayload::AgentFrameSwitched { .. }
));
assert_eq!(events[0].turn_id, None);
assert!(matches!(
events[1].payload,
SessionObservationEventPayload::Committed { .. }
));
assert_eq!(events[1].turn_id, None);
}

#[tokio::test]
Expand Down
33 changes: 21 additions & 12 deletions crates/lash-core/src/runtime/observation/replay.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,8 @@ pub struct SessionObservation {
#[derive(Clone, Debug)]
pub struct SessionObservationEvent {
pub session_id: String,
pub replay_incarnation_id: String,
pub turn_id: Option<String>,
pub revision: SessionRevision,
pub cursor: SessionCursor,
pub payload: SessionObservationEventPayload,
Expand Down Expand Up @@ -358,6 +360,7 @@ pub trait LiveReplayStore: Send + Sync {
&self,
session_id: &str,
revision: SessionRevision,
turn_id: Option<&str>,
payload: SessionObservationEventPayload,
) -> Result<Arc<SessionObservationEvent>, LiveReplayStoreError>;

Expand Down Expand Up @@ -405,6 +408,7 @@ impl Default for InMemoryLiveReplayStoreConfig {

#[derive(Debug)]
pub struct InMemoryLiveReplayStore {
replay_incarnation_id: String,
config: InMemoryLiveReplayStoreConfig,
clock: Arc<dyn crate::Clock>,
sessions: StdMutex<HashMap<String, LiveReplaySessionBuffer>>,
Expand All @@ -417,6 +421,7 @@ impl InMemoryLiveReplayStore {

pub fn with_clock(config: InMemoryLiveReplayStoreConfig, clock: Arc<dyn crate::Clock>) -> Self {
Self {
replay_incarnation_id: uuid::Uuid::new_v4().to_string(),
config,
clock,
sessions: StdMutex::new(HashMap::new()),
Expand Down Expand Up @@ -529,6 +534,7 @@ impl LiveReplayStore for InMemoryLiveReplayStore {
&self,
session_id: &str,
revision: SessionRevision,
turn_id: Option<&str>,
payload: SessionObservationEventPayload,
) -> Result<Arc<SessionObservationEvent>, LiveReplayStoreError> {
let now = self.clock.now();
Expand All @@ -543,6 +549,8 @@ impl LiveReplayStore for InMemoryLiveReplayStore {
let cursor = SessionCursor::new(session_id, revision, buffer.tail_position);
let event = Arc::new(SessionObservationEvent {
session_id: session_id.to_string(),
replay_incarnation_id: self.replay_incarnation_id.clone(),
turn_id: turn_id.map(str::to_string),
revision,
cursor,
payload,
Expand Down Expand Up @@ -708,10 +716,10 @@ mod tests {
let store = InMemoryLiveReplayStore::default();
let start = store.current_cursor("s", SessionRevision(0));
store
.append("s", SessionRevision(0), activity("a"))
.append("s", SessionRevision(0), None, activity("a"))
.expect("append a");
store
.append("s", SessionRevision(0), activity("b"))
.append("s", SessionRevision(0), None, activity("b"))
.expect("append b");
let LiveReplayResult::Replayed(events) = store.replay_after_cursor(&start).expect("replay")
else {
Expand All @@ -732,10 +740,10 @@ mod tests {
let store = InMemoryLiveReplayStore::with_bounds(1, Duration::from_secs(120));
let start = store.current_cursor("s", SessionRevision(0));
store
.append("s", SessionRevision(0), activity("a"))
.append("s", SessionRevision(0), None, activity("a"))
.expect("append a");
store
.append("s", SessionRevision(0), activity("b"))
.append("s", SessionRevision(0), None, activity("b"))
.expect("append b");
assert!(matches!(
store.replay_after_cursor(&start).expect("gap"),
Expand All @@ -748,7 +756,7 @@ mod tests {
let store = InMemoryLiveReplayStore::with_bounds(16, Duration::from_millis(1));
let start = store.current_cursor("s", SessionRevision(0));
store
.append("s", SessionRevision(0), activity("a"))
.append("s", SessionRevision(0), None, activity("a"))
.expect("append a");
std::thread::sleep(Duration::from_millis(5));
assert!(matches!(
Expand All @@ -772,7 +780,7 @@ mod tests {
let store = InMemoryLiveReplayStore::default();
let start = store.current_cursor("s", SessionRevision(0));
store
.append("s", SessionRevision(0), activity("a"))
.append("s", SessionRevision(0), None, activity("a"))
.expect("append a");
let LiveReplaySubscribeResult::Subscribed(mut subscription) =
store.subscribe_after_cursor(&start).expect("subscribe")
Expand All @@ -782,7 +790,7 @@ mod tests {
let first = subscription.next_event().await.expect("replay");
assert_eq!(first.session_id, "s");
store
.append("s", SessionRevision(0), activity("b"))
.append("s", SessionRevision(0), None, activity("b"))
.expect("append b");
let second = subscription.next_event().await.expect("live");
match &second.payload {
Expand Down Expand Up @@ -815,6 +823,7 @@ mod tests {
.append(
"perf-session",
SessionRevision(7),
None,
activity(&format!("token-{ordinal}")),
)
.expect("append token event");
Expand Down Expand Up @@ -845,7 +854,7 @@ mod tests {
let store = InMemoryLiveReplayStore::default();
let start = store.current_cursor("s", SessionRevision(0));
store
.append("s", SessionRevision(0), activity("a"))
.append("s", SessionRevision(0), None, activity("a"))
.expect("append a");
{
let sessions = store.sessions.lock().expect("sessions");
Expand All @@ -862,7 +871,7 @@ mod tests {
}
drop(subscription);
store
.append("s", SessionRevision(0), activity("b"))
.append("s", SessionRevision(0), None, activity("b"))
.expect("append b");
let sessions = store.sessions.lock().expect("sessions");
assert!(sessions.get("s").expect("buffer").sender.is_none());
Expand All @@ -873,10 +882,10 @@ mod tests {
let store = InMemoryLiveReplayStore::with_bounds(1, Duration::from_secs(120));
let start = store.current_cursor("s", SessionRevision(0));
store
.append("s", SessionRevision(0), activity("a"))
.append("s", SessionRevision(0), None, activity("a"))
.expect("append a");
store
.append("s", SessionRevision(0), activity("b"))
.append("s", SessionRevision(0), None, activity("b"))
.expect("append b");
assert!(matches!(
store.subscribe_after_cursor(&start).expect("subscribe"),
Expand All @@ -889,7 +898,7 @@ mod tests {
let store = InMemoryLiveReplayStore::with_bounds(16, Duration::from_millis(1));
let start = store.current_cursor("s", SessionRevision(0));
store
.append("s", SessionRevision(0), activity("a"))
.append("s", SessionRevision(0), None, activity("a"))
.expect("append a");
std::thread::sleep(Duration::from_millis(5));
assert!(matches!(
Expand Down
Loading
Loading