diff --git a/cmux-tui/crates/cmux-tui-core/src/mux.rs b/cmux-tui/crates/cmux-tui-core/src/mux.rs index c98161cdf2b9..3fbaadf47c5e 100644 --- a/cmux-tui/crates/cmux-tui-core/src/mux.rs +++ b/cmux-tui/crates/cmux-tui-core/src/mux.rs @@ -5530,6 +5530,76 @@ impl Mux { } } + /// A positive terminal-host liveness death proof is evidence that the + /// exact runtime incarnation is gone. Record the interruption before exit + /// cleanup can collapse it into an ordinary detached runtime. + pub(crate) fn terminal_host_liveness_dead( + &self, + surface_id: SurfaceId, + identity: &TerminalHostIdentity, + ) -> bool { + if self.shutting_down.load(Ordering::Acquire) { + return false; + } + let Some((runtime_id, host_epoch, lease_generation)) = identity.runtime_attachment_tokens() + else { + return false; + }; + let mut registry = self.workspace_registry.lock().unwrap(); + let state = self.state.lock().unwrap(); + let identity_matches = state + .surfaces + .get(&surface_id) + .or_else(|| state.terminal_runtime_by_id(surface_id)) + .and_then(|surface| surface.terminal_host_identity()) + .is_some_and(|current| current == *identity); + drop(state); + if !identity_matches { + return false; + } + let Ok(Some(terminal)) = registry.terminal_record(&identity.terminal_id) else { + return false; + }; + if terminal.incarnation.as_deref() != Some(identity.incarnation.as_str()) + || matches!( + terminal.lifecycle, + TerminalLifecycle::Exited | TerminalLifecycle::Tombstoned + ) + { + return false; + } + let Ok(Some(public_id)) = registry.terminal_resource_id(&identity.terminal_id) else { + return false; + }; + let idempotency_key = format!( + "runtime-host-loss-{}-{runtime_id}-{host_epoch}-{lease_generation}-host-liveness-dead", + identity.terminal_id + ); + match registry.record_runtime_host_loss_proof( + "cmux-tui-runtime", + &idempotency_key, + &public_id, + runtime_id, + host_epoch, + lease_generation, + "host_liveness_dead", + ) { + Ok(commit) => { + if !commit.replayed { + self.publish_journal_event(); + } + true + } + Err(error) => { + self.emit(MuxEvent::Status(format!( + "could not persist terminal {} host loss proof: {error}", + identity.terminal_id + ))); + false + } + } + } + pub(crate) fn terminal_host_reconnected( self: &Arc, surface_id: SurfaceId, @@ -8736,6 +8806,11 @@ impl Mux { let Some(public_id) = registry.terminal_resource_id(&identity.terminal_id)? else { return Ok(()); }; + if state == "detached" + && registry.runtime_attachment_state(&public_id)?.as_deref() == Some("interrupted") + { + return Ok(()); + } let idempotency_key = format!("runtime-attachment-{state}-{}-{runtime_id}", identity.terminal_id); registry.record_runtime_attachment_update( @@ -16417,6 +16492,18 @@ mod tests { .collect() } + fn journal_event_payloads(mux: &Mux, kind: &str) -> Vec { + let registry = mux.workspace_registry.lock().unwrap(); + registry + .session_journal_after(0, 1024) + .unwrap() + .records + .into_iter() + .filter(|record| record.kind == kind) + .map(|record| record.payload) + .collect() + } + fn public_request( mux: &Arc, id: &str, @@ -20517,6 +20604,51 @@ mod tests { ); } + #[cfg(unix)] + #[test] + fn host_liveness_dead_journals_interruption_before_exit_detach() { + const TERMINAL: &str = "00000000000040008000000000000063"; + const INCARNATION: &str = "10000000000040008000000000000063"; + let mux = test_mux(); + let workspace = mux + .create_empty_workspace( + Some("runtime-host-loss-detector".into()), + Some("018f6e21-7b70-7e70-8000-000000001063".into()), + None, + ) + .unwrap(); + let surface_id = + mux.seed_running_terminal_for_test(TERMINAL, INCARNATION, &workspace.key).unwrap(); + let surface = mux.surface(surface_id).unwrap(); + let identity = + mux.resource_terminal_host_identity(&surface).expect("test terminal has host identity"); + let public_id = + mux.workspace_registry.lock().unwrap().terminal_resource_id(TERMINAL).unwrap().unwrap(); + + assert!(mux.terminal_host_liveness_dead(surface_id, &identity)); + assert!(mux.terminal_host_liveness_dead(surface_id, &identity)); + mux.surface_exited(surface_id); + + let loss_payloads = journal_event_payloads(&mux, "runtime.host_loss.proven"); + assert_eq!(loss_payloads.len(), 1); + assert_eq!(loss_payloads[0]["terminal_id"], public_id.as_str()); + assert_eq!(loss_payloads[0]["runtime_id"], INCARNATION); + assert_eq!(loss_payloads[0]["host_epoch"], INCARNATION); + assert_eq!( + loss_payloads[0]["lease_generation"], + runtime_attachment_lease_generation_for_test(INCARNATION) + ); + assert_eq!(loss_payloads[0]["proof"], "host_liveness_dead"); + let states = runtime_attachment_event_payloads(&mux) + .into_iter() + .map(|payload| payload["state"].as_str().unwrap().to_string()) + .collect::>(); + assert_eq!(states, vec!["attached"]); + let kinds = journal_event_kinds(&mux); + assert!(!kinds.iter().any(|kind| kind.starts_with("session.hibernate"))); + assert!(!kinds.iter().any(|kind| kind.starts_with("session.recover"))); + } + #[cfg(unix)] #[test] fn stale_host_exit_does_not_create_first_runtime_attachment_state() { diff --git a/cmux-tui/crates/cmux-tui-core/src/surface.rs b/cmux-tui/crates/cmux-tui-core/src/surface.rs index 4020ad2d41f1..10c1eb6f5aa2 100644 --- a/cmux-tui/crates/cmux-tui-core/src/surface.rs +++ b/cmux-tui/crates/cmux-tui-core/src/surface.rs @@ -3085,6 +3085,7 @@ impl Surface { ); pty.stream_progress.notify(); if let Some(mux) = mux.upgrade() { + mux.terminal_host_liveness_dead(surface.id, &identity); mux.surface_exited(surface.id); } return; diff --git a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/session_persistence_store.rs b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/session_persistence_store.rs index f9c485575efd..5ef73be24774 100644 --- a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/session_persistence_store.rs +++ b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/session_persistence_store.rs @@ -1234,3 +1234,20 @@ fn upsert_session_effect_workflow( )?; Ok(()) } + +impl WorkspaceRegistry { + pub(crate) fn runtime_attachment_state( + &self, + terminal_id: &TerminalPublicId, + ) -> anyhow::Result> { + self.connection + .query_row( + "SELECT state FROM journal_runtime_attachment_states + WHERE terminal_id = ?1", + [terminal_id.as_str()], + |row| row.get(0), + ) + .optional() + .map_err(Into::into) + } +}