diff --git a/cmux-tui/Cargo.lock b/cmux-tui/Cargo.lock index 859ffecf1abd..725bf09eb3d8 100644 --- a/cmux-tui/Cargo.lock +++ b/cmux-tui/Cargo.lock @@ -760,8 +760,8 @@ dependencies = [ "cmux-tui-core", "cmux-tui-machine-agent-protocol", "cmux-tui-machine-protocol", - "crossterm", "crossbeam-channel", + "crossterm", "fs4", "getrandom 0.3.4", "ghostty-vt", diff --git a/cmux-tui/crates/cmux-tui-core/src/mux.rs b/cmux-tui/crates/cmux-tui-core/src/mux.rs index dd9fbc0f698d..8c4c4dce60f7 100644 --- a/cmux-tui/crates/cmux-tui-core/src/mux.rs +++ b/cmux-tui/crates/cmux-tui-core/src/mux.rs @@ -1086,6 +1086,24 @@ impl AgentSource { } } +/// The agent-record state a committed hook journal event implies, or `None` +/// for events that carry no lifecycle transition (child agents, unclassified +/// state changes) so they never churn the record. +fn agent_state_for_hook_kind(kind: &str) -> Option { + Some(match kind { + // A freshly started session sits at its prompt; a completed turn + // returns to it. + "agent.session.started" | "agent.turn.completed" => AgentState::Idle, + "agent.turn.started" => AgentState::Working, + "agent.approval.requested" + | "agent.question.requested" + | "agent.plan_review.requested" + | "agent.error.reported" => AgentState::Blocked, + "agent.session.ended" => AgentState::Done, + _ => return None, + }) +} + #[derive(Debug, Clone)] pub struct AgentRecord { pub surface: SurfaceId, @@ -1096,6 +1114,70 @@ pub struct AgentRecord { pub updated_at_ms: u64, } +#[derive(Debug, Clone, PartialEq, Eq)] +struct HookFence { + session_id: String, + sequence: u64, + ended: bool, +} + +/// Session-less adapters get a local generation token. The journal sequence +/// is durable and strictly increasing, so a new legacy lifecycle cannot reuse +/// the previous fence identity after restart. +pub(super) fn legacy_hook_session_id(terminal_id: &TerminalPublicId, sequence: u64) -> String { + format!("legacy:{terminal_id}:{sequence}") +} + +const AGENT_HOOK_RETRY_ERROR: &str = "agent hook projection retry deferred"; + +#[derive(Debug)] +struct AgentHookTerminalUnavailable; + +impl fmt::Display for AgentHookTerminalUnavailable { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("terminal is not available for agent hook projection") + } +} + +impl std::error::Error for AgentHookTerminalUnavailable {} + +#[derive(Debug)] +struct AgentHookTerminalGone; + +impl fmt::Display for AgentHookTerminalGone { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("terminal no longer exists for agent hook projection") + } +} + +impl std::error::Error for AgentHookTerminalGone {} + +fn agent_hook_terminal_gone(error: &anyhow::Error) -> bool { + error.downcast_ref::().is_some() +} + +fn agent_hook_retry_class(error: &anyhow::Error) -> crate::workspace_registry::AgentHookRetryClass { + if error.downcast_ref::().is_some() + || error.chain().any(|cause| { + matches!( + cause.downcast_ref::(), + Some(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseBusy + | rusqlite::ErrorCode::DatabaseLocked, + .. + }, + _ + )) + ) + }) + { + crate::workspace_registry::AgentHookRetryClass::Transient + } else { + crate::workspace_registry::AgentHookRetryClass::Permanent + } +} + #[derive(Debug, Clone)] struct TerminalAgentRecord { state: AgentState, @@ -2007,6 +2089,7 @@ pub struct Mux { durable_terminal_defaults: AtomicBool, sidebar_plugin: Mutex, agent_records: Mutex>, + agent_hook_fences: Mutex>, /// Nonterminal notifications remain placement-local. Terminal unread /// state is keyed separately by stable content identity so every view of /// one terminal shares the same attention marker. @@ -2275,6 +2358,7 @@ impl Mux { has_terminal_defaults, next_notification_id, agent_records, + agent_hook_fences, terminal_notifications, notification_ledger, } = restore_public_projections(&state, registry.public_projections()?)?; @@ -2384,6 +2468,7 @@ impl Mux { durable_terminal_defaults: AtomicBool::new(has_terminal_defaults), sidebar_plugin: Mutex::new(SidebarPluginRuntime::default()), agent_records: Mutex::new(agent_records), + agent_hook_fences: Mutex::new(agent_hook_fences), placement_notifications: Mutex::new(HashMap::new()), terminal_notifications: Mutex::new(terminal_notifications), notification_ledger: Mutex::new(notification_ledger), @@ -2446,10 +2531,108 @@ impl Mux { } std::thread::sleep(Duration::from_millis(25)); } + mux.retry_pending_agent_hooks()?; crate::journal_hooks::start(&mux)?; Ok(mux) } + fn retry_pending_agent_hooks(&self) -> anyhow::Result<()> { + let mut cursor = None; + // Keep startup work bounded. Remaining rows stay durable for a later + // availability signal or restart. + for _ in 0..crate::workspace_registry::AGENT_HOOK_MAX_RETRY_PAGES_PER_WAKE { + let (pending, next_cursor) = self + .workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections_page(cursor.clone())?; + let Some(next_cursor) = next_cursor else { break }; + cursor = Some(next_cursor); + self.retry_pending_agent_hooks_rows(pending)?; + } + Ok(()) + } + + fn retry_pending_agent_hooks_for_terminal( + &self, + terminal_id: &TerminalPublicId, + ) -> anyhow::Result<()> { + // Drain in fixed-size pages, with a hard per-signal cap. Rows beyond + // the cap remain durable for the next terminal availability signal. + for _ in 0..crate::workspace_registry::AGENT_HOOK_MAX_RETRY_PAGES_PER_WAKE { + let pending = self + .workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections_for_terminal(terminal_id)?; + if pending.is_empty() { + break; + } + let applied = self.retry_pending_agent_hooks_rows(pending)?; + if applied == 0 { + break; + } + } + Ok(()) + } + + fn retry_pending_agent_hooks_rows( + &self, + pending: Vec<(String, String, String, u64, crate::JournalIngress)>, + ) -> anyhow::Result { + let mut applied = 0; + for (producer_id, origin, key, sequence, ingress) in pending { + match self.apply_agent_hook_record(&ingress, sequence) { + Ok(()) => { + if self + .workspace_registry + .lock() + .unwrap() + .clear_agent_hook_pending(&producer_id, &origin, &key) + .is_ok() + { + applied += 1; + } else { + self.report_internal_diagnostic("agent hook retry cleanup deferred"); + } + } + Err(error) if agent_hook_terminal_gone(&error) => { + if self + .workspace_registry + .lock() + .unwrap() + .clear_agent_hook_pending(&producer_id, &origin, &key) + .is_ok() + { + applied += 1; + } else { + self.report_internal_diagnostic("agent hook retry cleanup deferred"); + } + } + Err(error) => { + if self + .workspace_registry + .lock() + .unwrap() + .enqueue_agent_hook_pending( + &producer_id, + &origin, + &key, + sequence, + &ingress, + AGENT_HOOK_RETRY_ERROR, + agent_hook_retry_class(&error), + ) + .is_err() + { + self.report_internal_diagnostic("agent hook retry bookkeeping deferred"); + } + } + } + } + Ok(applied) + } + /// Rehydrate workspace rows that belong to an interrupted correlated /// creation without exposing them through the public snapshot. Terminal /// host adoption then restores the live content, and creation settlement @@ -3032,6 +3215,12 @@ impl Mux { }; drop(state); self.emit_terminal_registry_changed(®istry, revision); + drop(registry); + // Adoption makes the terminal's resource surface available. Retry + // only hooks scoped to this terminal, not the entire pending table. + if let Ok(terminal_id) = TerminalPublicId::parse(terminal_id) { + let _ = self.retry_pending_agent_hooks_for_terminal(&terminal_id); + } Ok(()) } @@ -3656,6 +3845,8 @@ impl Mux { self.surface(surface).context("created surface disappeared") } + /// Report an agent state for a selected terminal resource and reconcile + /// any durable hook projections waiting for that terminal. #[allow(clippy::too_many_arguments)] fn commit_resource_mutation_plan( &self, @@ -4380,6 +4571,8 @@ impl Mux { ) } + /// Commit an agent report and its resource projection under the sequence + /// fence used to preserve hook ordering across retries and restarts. #[allow(clippy::too_many_arguments)] pub(crate) fn prepare_resource_effect( &self, @@ -5207,24 +5400,202 @@ impl Mux { idempotency_key: &str, ) -> anyhow::Result { let validated = self.journal_kernel.validate_ingress(ingress)?; - if self.journal_ingress.enabled() { - return self.journal_ingress.send_producer( + let commit = if self.journal_ingress.enabled() { + self.journal_ingress.send_producer( ingress.clone(), validated, origin.into(), idempotency_key.into(), + )? + } else { + let commit = self.workspace_registry.lock().unwrap().append_journal_ingress( + ingress, + &validated, + origin, + idempotency_key, + )?; + if !commit.replayed { + self.publish_journal_event(); + } + commit + }; + // Replayed journal commits still need projection reconciliation. A + // process can crash after the durable journal commit and before the + // in-memory/resource projection update. The sequence guard makes this + // a no-op for already-applied events while allowing restart repair. + if let Err(error) = self.apply_agent_hook_record(ingress, commit.sequence) { + if agent_hook_terminal_gone(&error) { + if let Err(_bookkeeping_error) = self + .workspace_registry + .lock() + .unwrap() + .clear_agent_hook_pending(&ingress.producer_id, origin, idempotency_key) + { + self.report_internal_diagnostic("terminal-gone agent hook cleanup deferred"); + } + } else { + if let Err(_bookkeeping_error) = + self.workspace_registry.lock().unwrap().enqueue_agent_hook_pending( + &ingress.producer_id, + origin, + idempotency_key, + commit.sequence, + ingress, + AGENT_HOOK_RETRY_ERROR, + agent_hook_retry_class(&error), + ) + { + self.report_internal_diagnostic( + "durable agent hook receipt remains staged after retry bookkeeping failure", + ); + } + } + } else if let Err(_bookkeeping_error) = self + .workspace_registry + .lock() + .unwrap() + .clear_agent_hook_pending(&ingress.producer_id, origin, idempotency_key) + { + self.report_internal_diagnostic( + "agent hook projection applied; retry bookkeeping cleanup deferred", ); } - let commit = self.workspace_registry.lock().unwrap().append_journal_ingress( - ingress, - &validated, - origin, - idempotency_key, + Ok(commit) + } + + /// Committed agent hook events double as the live agent-status feed: + /// a fresh `agent.*` journal event updates its terminal's agent record, + /// so agents views show working/blocked/idle/done without a separate + /// reporting channel. The journal commit remains durable even when the + /// projection fails. The ingress path catches the projection error, stages + /// a durable pending row, and still returns the journal receipt so callers + /// can retain exactly-once semantics while the terminal becomes available. + fn apply_agent_hook_record( + &self, + ingress: &crate::JournalIngress, + sequence: u64, + ) -> anyhow::Result<()> { + if ingress.producer_id != crate::agent_hooks::AGENT_HOOK_PRODUCER_ID { + return Ok(()); + } + let Some(state) = agent_state_for_hook_kind(&ingress.kind) else { return Ok(()) }; + let Some(terminal_subject) = + ingress.subjects.iter().find(|subject| subject.kind == "terminal") + else { + return Ok(()); + }; + let terminal_id = TerminalPublicId::parse(&terminal_subject.id) + .with_context(|| format!("invalid terminal subject {:?}", terminal_subject.id))?; + let Some(surface) = self.resource_surface_for_terminal(&terminal_id) else { + // A durable terminal that is still live may be between journal + // commit and host materialization. A missing or tombstoned + // terminal cannot become available, so do not retain its receipt + // in the retry queue forever. + let terminal_is_retryable = self + .workspace_registry + .lock() + .unwrap() + .agent_hook_terminal_retryable(&terminal_id)?; + if !terminal_is_retryable { + return Err(anyhow::Error::new(AgentHookTerminalGone)); + } + return Err(anyhow::Error::new(AgentHookTerminalUnavailable).context(format!( + "terminal {terminal_id} is not available for agent hook projection" + ))); + }; + // Serialize the sequence check, projection commit, and sequence + // update as one operation. The projection path takes registry, state, + // then agent-record locks, so teardown acquires sequence before those + // locks as well. + let mut fences = self.agent_hook_fences.lock().unwrap(); + let explicit_session_id = ingress + .payload + .get("normalized") + .and_then(|value| value.get("agent_session_id")) + .and_then(Value::as_str) + .filter(|session_id| !session_id.is_empty()) + .map(str::to_owned); + let is_session_start = ingress.kind == "agent.session.started"; + let previous_fence = fences.get(&terminal_id).cloned(); + // Once an adapter has supplied a native session identity, a later + // session-less event is ambiguous. Do not assign delayed events from + // that generation to the currently active session. + if explicit_session_id.is_none() + && previous_fence.as_ref().is_some_and(|fence| !fence.session_id.starts_with("legacy:")) + { + return Ok(()); + } + // A non-start event without an adapter identity belongs to the live + // legacy generation. A start event must carry a new identity, or it + // receives a fresh local generation token below. Reusing the active + // fence for a start would let a delayed, session-less start mutate a + // newer lifecycle. + let agent_session_id = explicit_session_id + .clone() + .or_else(|| { + (!is_session_start) + .then(|| previous_fence.as_ref().filter(|fence| !fence.ended)) + .flatten() + .map(|fence| fence.session_id.clone()) + }) + .unwrap_or_else(|| legacy_hook_session_id(&terminal_id, sequence)); + if let Some(fence) = previous_fence.as_ref() { + if fence.session_id == agent_session_id { + if fence.ended || sequence <= fence.sequence { + return Ok(()); + } + } else if !is_session_start || !fence.ended || sequence <= fence.sequence { + // A mismatched event cannot cross an active lifecycle fence. + // A start may replace an ended fence only as a new lifecycle: + // an explicit adapter id must differ from the ended id, while + // a legacy adapter receives the sequence-scoped token above. + return Ok(()); + } + } + if previous_fence.as_ref().is_some_and(|fence| sequence <= fence.sequence) { + return Ok(()); + } + // The record's session field is a human-facing label; native agent + // session ids are opaque, so views fall back to their own context. + let marker = if state == AgentState::Done { + format!("cmux-hook-ended:{sequence}") + } else { + format!("cmux-hook-sequence:{sequence}") + }; + let hook_state = crate::workspace_registry::AgentHookProjectionState { + agent_session_id: agent_session_id.clone(), + applied_sequence: sequence, + ended: state == AgentState::Done, + }; + self.report_agent_with_sequence_lock( + surface, + state, + AgentSource::Hook, + Some(marker), + true, + Some(hook_state), + Some(sequence), )?; - if !commit.replayed { - self.publish_journal_event(); + fences.insert( + terminal_id.clone(), + HookFence { session_id: agent_session_id, sequence, ended: state == AgentState::Done }, + ); + // Projection ordering is complete. Do not carry the fence guard into + // cleanup or any retry/reentrant path. + drop(fences); + // An ended session leaves the roster entirely: the done state was + // committed and broadcast above (so remote caches converge), and the + // live record is dropped so agents views stop listing the terminal + // and a fresh agent there starts clean. Hooks of one terminal are + // sequential (they follow one agent process's lifecycle), so nothing + // races this removal. + if state == AgentState::Done { + let mut records = self.agent_records.lock().unwrap(); + if records.get(&terminal_id).is_some_and(|record| record.state == AgentState::Done) { + records.remove(&terminal_id); + } } - Ok(commit) + Ok(()) } pub(crate) fn journal_hook_states( @@ -5311,22 +5682,11 @@ impl Mux { Ok(()) } - /// Logs a skipped terminal-host reconnect checkpoint at most once - /// until a later reconnect checkpoint succeeds. A checkpoint is a - /// journal-replay optimization: skipping one only moves the next replay - /// boundary back, so repeated skips are daemon-log noise, not per-toast - /// news. - pub(crate) fn report_skipped_reconnect_checkpoint( - &self, - terminal_id: impl fmt::Display, - error: &anyhow::Error, - ) { - let message = format!( - "skipped terminal {terminal_id} reconnect checkpoint (replay starts from the previous boundary): {error:#}" - ); - if self.reconnect_checkpoint_skip_reported.swap(true, Ordering::AcqRel) { - return; - } + /// Sends a diagnostic to the frontend-owned sink without writing to a + /// frontend terminal. One message is retained when startup races sink + /// installation. + fn report_internal_diagnostic(&self, message: impl Into) { + let message = message.into(); if let Some(reporter) = self.diagnostic_reporter.get().cloned() { reporter(&message); return; @@ -5345,6 +5705,25 @@ impl Mux { } } + /// Logs a skipped terminal-host reconnect checkpoint at most once + /// until a later reconnect checkpoint succeeds. A checkpoint is a + /// journal-replay optimization: skipping one only moves the next replay + /// boundary back, so repeated skips are daemon-log noise, not per-toast + /// news. + pub(crate) fn report_skipped_reconnect_checkpoint( + &self, + terminal_id: impl fmt::Display, + error: &anyhow::Error, + ) { + let message = format!( + "skipped terminal {terminal_id} reconnect checkpoint (replay starts from the previous boundary): {error:#}" + ); + if self.reconnect_checkpoint_skip_reported.swap(true, Ordering::AcqRel) { + return; + } + self.report_internal_diagnostic(message); + } + /// Installs the frontend-owned sink for diagnostics emitted by the mux. /// /// A mux has one owner for its lifetime, so accepting the first reporter @@ -8566,6 +8945,19 @@ impl Mux { state: AgentState, source: AgentSource, session: Option, + ) -> anyhow::Result { + self.report_agent_with_sequence_lock(surface, state, source, session, false, None, None) + } + + fn report_agent_with_sequence_lock( + &self, + surface: SurfaceId, + state: AgentState, + source: AgentSource, + session: Option, + sequence_lock_held: bool, + hook_state: Option, + journal_sequence: Option, ) -> anyhow::Result { let mutation = WorkspaceMutation::new( format!("raw-agent-{}", crate::workspace_registry::new_uuid_v4()), @@ -8586,8 +8978,17 @@ impl Mux { None, &mutation, &fingerprint, + sequence_lock_held, + hook_state.as_ref(), + journal_sequence, )?; - record.context("fresh raw agent report unexpectedly replayed") + let record = record.context("fresh raw agent report unexpectedly replayed")?; + if source != AgentSource::Hook { + // A successful report means this terminal is available. Retry only + // durable hooks for that terminal, never the entire pending table. + let _ = self.retry_pending_agent_hooks_for_terminal(&record.terminal_id); + } + Ok(record) } #[allow(clippy::too_many_arguments)] @@ -8609,7 +9010,7 @@ impl Mux { "source":source.as_str(), "source_session":source_session, }); - self.commit_agent_report( + let result = self.commit_agent_report( AgentReportTarget::Resource { selectors: &selectors, terminal_id }, agent_state, source, @@ -8617,8 +9018,14 @@ impl Mux { expected_revision, mutation, &fingerprint, - ) - .map(|(commit, _)| commit) + false, + None, + None, + ); + if result.is_ok() && source != AgentSource::Hook { + let _ = self.retry_pending_agent_hooks_for_terminal(terminal_id); + } + result.map(|(commit, _)| commit) } #[allow(clippy::too_many_arguments)] @@ -8631,11 +9038,24 @@ impl Mux { expected_revision: Option, mutation: &WorkspaceMutation, fingerprint: &Value, + sequence_lock_held: bool, + hook_state: Option<&crate::workspace_registry::AgentHookProjectionState>, + journal_sequence: Option, ) -> anyhow::Result<(ResourcePatchCommit, Option)> { + // Hook replay already owns this guard to serialize sequence checks and + // projection commits. Other report sources acquire it before the + // registry/state locks, so all paths use one lock order. + let mut sequence_guard = + (!sequence_lock_held).then(|| self.agent_hook_fences.lock().unwrap()); let mut registry = self.workspace_registry.lock().unwrap(); if let Some(replay) = registry.replay_resource_patch(mutation, "agent.report", fingerprint)? { + if let Some(sequence) = journal_sequence { + // The projection transaction committed before this replay was + // observed. Repair the durable apply watermark separately. + registry.advance_agent_hook_apply_cursor(sequence)?; + } return Ok((replay, None)); } let mut state = self.state.lock().unwrap(); @@ -8671,14 +9091,87 @@ impl Mux { (surface, (*terminal_id).clone()) } }; + let mut direct_hook_state = None; + if source != AgentSource::Hook { + let ended_fence = sequence_guard + .as_ref() + .and_then(|guard| guard.get(&terminal_id)) + .filter(|fence| fence.ended); + let fresh_session = source_session.as_deref().filter(|session| { + !session.starts_with("cmux-hook-sequence:") + && !session.starts_with("cmux-hook-ended:") + }); + if ended_fence.is_some_and(|fence| { + fresh_session.is_none_or(|session| session == fence.session_id) + }) { + anyhow::bail!("agent_session_ended"); + } + } else if let Some(fence) = sequence_guard + .as_ref() + .and_then(|guard| guard.get(&terminal_id)) + .filter(|fence| fence.ended) + { + let supplied_session = source_session.as_deref().filter(|session| { + !session.is_empty() + && !session.starts_with("cmux-hook-sequence:") + && !session.starts_with("cmux-hook-ended:") + }); + let journal_hook_session = hook_state.is_some_and(|state| { + !state.ended + && state.applied_sequence > fence.sequence + && state.agent_session_id != fence.session_id + && source_session + .as_deref() + .is_some_and(|session| session.starts_with("cmux-hook-sequence:")) + }); + let direct_hook_session = hook_state.is_none() + && supplied_session.is_some_and(|session| session != fence.session_id); + if direct_hook_session { + direct_hook_state = Some(crate::workspace_registry::AgentHookProjectionState { + agent_session_id: supplied_session + .expect("direct hook session was checked above") + .to_owned(), + applied_sequence: fence.sequence, + ended: false, + }); + } + if !journal_hook_session && !direct_hook_session { + anyhow::bail!("agent_session_ended"); + } + } + let effective_hook_state = direct_hook_state.as_ref().or(hook_state); + let persisted_source_session = if source == AgentSource::Hook { + source_session.clone().filter(|value| { + !value.starts_with("cmux-hook-sequence:") && !value.starts_with("cmux-hook-ended:") + }) + } else { + if source_session.as_deref().is_some_and(|value| { + value.starts_with("cmux-hook-sequence:") || value.starts_with("cmux-hook-ended:") + }) { + anyhow::bail!("reserved hook marker is invalid for non-hook agent source"); + } + // Preserve a valid fresh socket identity. The internal hook + // marker is only a compatibility fallback when no identity was + // supplied, while durable hook state carries the fence itself. + source_session.clone().or_else(|| { + sequence_guard + .as_ref() + .and_then(|guard| guard.get(&terminal_id)) + .map(|fence| format!("cmux-hook-sequence:{}", fence.sequence)) + }) + }; + let source_session = source_session.filter(|value| { + !value.starts_with("cmux-hook-sequence:") && !value.starts_with("cmux-hook-ended:") + }); let now = now_ms(); let mut records = self.agent_records.lock().unwrap(); + let socket_report_ignored = records.get(&terminal_id).is_some_and(|existing| { + existing.source == AgentSource::Hook + && source == AgentSource::Socket + && !effective_hook_state.is_some_and(|state| state.ended) + }); let record = match records.get(&terminal_id) { - Some(existing) - if existing.source == AgentSource::Hook && source == AgentSource::Socket => - { - existing.clone() - } + Some(existing) if socket_report_ignored => existing.clone(), _ => TerminalAgentRecord { state: agent_state, source, @@ -8686,6 +9179,11 @@ impl Mux { updated_at_ms: now, }, }; + // A socket report that is intentionally ignored by the hook-owned + // record must persist that effective record, not the discarded socket + // identity. Otherwise durable and in-memory projections diverge. + let persisted_source_session = + if socket_report_ignored { record.session.clone() } else { persisted_source_session }; let digest = Sha256::digest(format!("cmux.protocol/2/agent/{terminal_id}").as_bytes()); let payload = digest[..16].iter().map(|byte| format!("{byte:02x}")).collect::(); let agent_id = @@ -8698,23 +9196,61 @@ impl Mux { "state":record.state.as_str(), "source":record.source.as_str(), "updated_at_ms":record.updated_at_ms.to_string(), - "source_session":record.session, + "source_session":persisted_source_session.clone().or(record.session.clone()), }); - let deltas = serde_json::json!([{ - "kind":"upsert", - "sequence":0, - "resource":"agent", - "id":agent_id, - "value":value, - }]); - let commit = registry.commit_agent_projection( - mutation, - fingerprint, - expected_revision, - &terminal_id, - &value, - &deltas, - )?; + let mut public_value = value.clone(); + public_value["source_session"] = serde_json::json!(record.session.clone()); + let deltas = if effective_hook_state.is_some_and(|state| state.ended) { + serde_json::json!([{ + "kind":"delete", + "sequence":0, + "resource":"agent", + "id":agent_id, + }]) + } else { + serde_json::json!([{ + "kind":"upsert", + "sequence":0, + "resource":"agent", + "id":agent_id, + "value":public_value, + }]) + }; + let commit = match journal_sequence { + Some(sequence) => registry.commit_agent_projection_with_hook_state_and_sequence( + mutation, + fingerprint, + expected_revision, + &terminal_id, + &value, + &deltas, + effective_hook_state, + sequence, + )?, + None => registry.commit_agent_projection_with_hook_state( + mutation, + fingerprint, + expected_revision, + &terminal_id, + &value, + &deltas, + effective_hook_state, + )?, + }; + if !commit.replayed { + if let (Some(direct_state), Some(sequence_guard)) = + (direct_hook_state.as_ref(), sequence_guard.as_mut()) + { + sequence_guard.insert( + terminal_id.clone(), + HookFence { + session_id: direct_state.agent_session_id.clone(), + sequence: direct_state.applied_sequence, + ended: false, + }, + ); + } + } state.resource_revision = commit.revision; if !commit.replayed { records.insert(terminal_id.clone(), record.clone()); @@ -8760,7 +9296,16 @@ impl Mux { } fn purge_terminal_side_tables(&self, terminal_id: &TerminalPublicId) { - self.agent_records.lock().unwrap().remove(terminal_id); + let pending_cleanup = { + let mut registry = self.workspace_registry.lock().unwrap(); + registry.purge_agent_hook_pending_for_terminal(terminal_id) + }; + if pending_cleanup.is_err() { + self.report_internal_diagnostic("terminal agent hook cleanup deferred"); + } + // The registry guard is dropped before acquiring the fence guard. + self.agent_hook_fences.lock().unwrap().remove(terminal_id); + self.agent_records.lock().unwrap().remove(terminal_id); self.terminal_notifications.lock().unwrap().remove(terminal_id); } @@ -21557,6 +22102,1068 @@ mod tests { assert_eq!(batches[1].changes[0]["value"]["state"], "blocked"); } + #[test] + fn committed_agent_hook_events_drive_the_terminal_agent_record() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let surface_id = surface.id; + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface_id).unwrap() { + ContentPublicId::Terminal(terminal_id) => terminal_id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + + let append = |event: &str, key: &str| { + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({"session_id":"native-1"}), + ) + .unwrap(); + mux.append_journal_ingress(&ingress, "test", key).unwrap(); + }; + + append("SessionStart", "hook-1"); + let records = mux.list_agents(Some(surface_id), None); + assert_eq!(records.len(), 1); + assert_eq!(records[0].state, AgentState::Idle); + assert_eq!(records[0].source, AgentSource::Hook); + + append("UserPromptSubmit", "hook-2"); + assert_eq!(mux.list_agents(Some(surface_id), None)[0].state, AgentState::Working); + + append("PermissionRequest", "hook-3"); + assert_eq!(mux.list_agents(Some(surface_id), None)[0].state, AgentState::Blocked); + + append("Stop", "hook-4"); + assert_eq!(mux.list_agents(Some(surface_id), None)[0].state, AgentState::Idle); + + // Child-agent events carry no top-level lifecycle transition. + append("SubagentStart", "hook-5"); + assert_eq!(mux.list_agents(Some(surface_id), None)[0].state, AgentState::Idle); + + // A socket report cannot downgrade a hook-owned record. + mux.report_agent(surface_id, AgentState::Working, AgentSource::Socket, None).unwrap(); + assert_eq!(mux.list_agents(Some(surface_id), None)[0].state, AgentState::Idle); + + // An exited agent leaves the roster; a fresh one starts clean. + append("SessionEnd", "hook-6"); + assert!(mux.list_agents(Some(surface_id), None).is_empty()); + append("SessionStart", "hook-7"); + let records = mux.list_agents(Some(surface_id), None); + assert_eq!(records.len(), 1); + assert_eq!(records[0].state, AgentState::Idle); + } + + #[test] + fn stale_same_terminal_hook_sequence_cannot_overwrite_newer_state() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let newer = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "PermissionRequest", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + let older = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.apply_agent_hook_record(&newer, 2).unwrap(); + mux.apply_agent_hook_record(&older, 1).unwrap(); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].state, AgentState::Blocked); + } + + #[test] + fn hook_projection_does_not_relock_its_held_sequence_guard() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let hook = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + + // apply_agent_hook_record intentionally holds the sequence guard while + // it commits. The Hook report path must use its supplied marker and + // must not try to acquire that guard again. + mux.apply_agent_hook_record(&hook, 1).unwrap(); + + assert_eq!(mux.list_agents(Some(surface.id), None)[0].state, AgentState::Working); + assert_eq!( + mux.agent_hook_fences.lock().unwrap().get(&terminal_id).map(|fence| fence.sequence), + Some(1) + ); + } + + #[test] + fn replayed_agent_hook_events_do_not_rewrite_the_record() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let surface_id = surface.id; + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface_id).unwrap() { + ContentPublicId::Terminal(terminal_id) => terminal_id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let started = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({"session_id":"native-1"}), + ) + .unwrap(); + mux.append_journal_ingress(&started, "test", "hook-replay").unwrap(); + let working_at = mux.list_agents(Some(surface_id), None)[0].updated_at_ms; + + let replay = mux.append_journal_ingress(&started, "test", "hook-replay").unwrap(); + assert!(replay.replayed); + let record = &mux.list_agents(Some(surface_id), None)[0]; + assert_eq!(record.state, AgentState::Working); + assert_eq!(record.updated_at_ms, working_at); + } + + #[test] + fn failed_agent_hook_projection_does_not_consume_sequence() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + let validated = mux.journal_kernel.validate_ingress(&ingress).unwrap(); + let receipt = mux + .workspace_registry + .lock() + .unwrap() + .append_journal_ingress(&ingress, &validated, "test", "hook-pending-retry") + .unwrap(); + assert!(receipt.sequence > 0); + assert_eq!( + mux.workspace_registry.lock().unwrap().pending_agent_hook_projections().unwrap().len(), + 1 + ); + let validated = mux.journal_kernel.validate_ingress(&ingress).unwrap(); + mux.workspace_registry + .lock() + .unwrap() + .append_journal_ingress(&ingress, &validated, "other-origin", "hook-pending-retry") + .unwrap(); + assert_eq!( + mux.workspace_registry.lock().unwrap().pending_agent_hook_projections().unwrap().len(), + 2 + ); + assert!(mux.list_agents(Some(surface.id), None).is_empty()); + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(true).unwrap(); + mux.retry_pending_agent_hooks().unwrap(); + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(false).unwrap(); + let replay = mux.append_journal_ingress(&ingress, "test", "hook-pending-retry").unwrap(); + assert!(replay.replayed); + assert_eq!( + mux.workspace_registry.lock().unwrap().pending_agent_hook_projections().unwrap().len(), + 1 + ); + mux.append_journal_ingress(&ingress, "other-origin", "hook-pending-retry").unwrap(); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections() + .unwrap() + .is_empty() + ); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].state, AgentState::Working); + } + + #[test] + fn journal_commit_preseeds_hook_retry_before_projection() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + let validated = mux.journal_kernel.validate_ingress(&ingress).unwrap(); + let commit = mux + .workspace_registry + .lock() + .unwrap() + .append_journal_ingress(&ingress, &validated, "test", "crash-window") + .unwrap(); + assert!(!commit.replayed); + assert_eq!( + mux.workspace_registry.lock().unwrap().pending_agent_hook_projections().unwrap().len(), + 1 + ); + + let replay = mux.append_journal_ingress(&ingress, "test", "crash-window").unwrap(); + assert!(replay.replayed); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections() + .unwrap() + .is_empty() + ); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].state, AgentState::Working); + } + + #[test] + fn available_agent_report_wakes_pending_hook_projection_retry() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(true).unwrap(); + mux.append_journal_ingress(&ingress, "test", "hook-wake").unwrap(); + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(false).unwrap(); + + mux.report_agent(surface.id, AgentState::Working, AgentSource::Socket, None).unwrap(); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections() + .unwrap() + .is_empty() + ); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].source, AgentSource::Hook); + } + + #[test] + fn unavailable_terminal_hook_is_retained_for_projection_retry() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + // Keep the durable terminal in its Running lifecycle while removing + // the in-memory catalog entry, which models a recoverable adoption + // gap rather than a terminal that can never return. + assert!(mux.remove_terminal_catalog_for_test(&terminal_id).is_some()); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + + let receipt = mux.append_journal_ingress(&ingress, "test", "missing-terminal").unwrap(); + assert!(receipt.sequence > 0); + assert_eq!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections_for_terminal(&terminal_id) + .unwrap() + .len(), + 1 + ); + } + + #[test] + fn unavailable_terminal_hook_retries_without_consuming_attempt_budget() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + assert!(mux.remove_terminal_catalog_for_test(&terminal_id).is_some()); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + + for attempt in 0..3 { + mux.append_journal_ingress(&ingress, "test", "transient-budget").unwrap(); + let retry_state = mux + .workspace_registry + .lock() + .unwrap() + .agent_hook_pending_retry_state_for_test( + crate::agent_hooks::AGENT_HOOK_PRODUCER_ID, + "test", + "transient-budget", + ) + .unwrap() + .expect("unavailable terminal hook remains pending"); + assert_eq!(retry_state.0, 0, "transient retry consumed attempt {attempt}"); + } + } + + #[test] + fn agent_hook_retry_class_uses_typed_transient_errors() { + let unavailable = anyhow::Error::new(AgentHookTerminalUnavailable) + .context("terminal term_secret is not available for agent hook projection"); + assert_eq!( + agent_hook_retry_class(&unavailable), + crate::workspace_registry::AgentHookRetryClass::Transient + ); + assert_eq!( + agent_hook_retry_class(&anyhow::anyhow!("invalid terminal subject")), + crate::workspace_registry::AgentHookRetryClass::Permanent + ); + } + + #[test] + fn agent_report_retries_only_pending_hooks_for_its_terminal() { + let mux = test_mux(); + let first = mux.new_workspace(None, None).unwrap(); + let second = mux.new_workspace(None, None).unwrap(); + let terminal_id = + |surface: &Surface| surface.terminal_public_id().cloned().expect("workspace terminal"); + let first_terminal = terminal_id(&first); + let second_terminal = terminal_id(&second); + let hook = |terminal_id: &TerminalPublicId, key: &str| { + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + let validated = mux.journal_kernel.validate_ingress(&ingress).unwrap(); + mux.workspace_registry + .lock() + .unwrap() + .append_journal_ingress(&ingress, &validated, "test", key) + .unwrap(); + }; + + hook(&first_terminal, "first-pending"); + hook(&second_terminal, "second-pending"); + + mux.report_agent(first.id, AgentState::Working, AgentSource::Socket, None).unwrap(); + let pending = + mux.workspace_registry.lock().unwrap().pending_agent_hook_projections().unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].0, crate::agent_hooks::AGENT_HOOK_PRODUCER_ID); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections_for_terminal(&first_terminal) + .unwrap() + .is_empty() + ); + assert_eq!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections_for_terminal(&second_terminal) + .unwrap() + .len(), + 1 + ); + assert_eq!(mux.list_agents(Some(first.id), None)[0].source, AgentSource::Hook); + assert!(mux.list_agents(Some(second.id), None).is_empty()); + + mux.report_agent(second.id, AgentState::Working, AgentSource::Socket, None).unwrap(); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections() + .unwrap() + .is_empty() + ); + assert_eq!(mux.list_agents(Some(second.id), None)[0].source, AgentSource::Hook); + } + + #[test] + fn terminal_report_drains_more_than_one_pending_retry_page() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + let validated = mux.journal_kernel.validate_ingress(&ingress).unwrap(); + for index in 0..65 { + mux.workspace_registry + .lock() + .unwrap() + .append_journal_ingress( + &ingress, + &validated, + "test", + &format!("pending-page-{index}"), + ) + .unwrap(); + } + assert_eq!( + mux.workspace_registry.lock().unwrap().pending_agent_hook_projections().unwrap().len(), + 65 + ); + + mux.report_agent(surface.id, AgentState::Working, AgentSource::Socket, None).unwrap(); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections() + .unwrap() + .is_empty() + ); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].source, AgentSource::Hook); + } + + #[test] + fn permanently_failed_hook_projection_is_quarantined() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(true).unwrap(); + let receipt = mux.append_journal_ingress(&ingress, "test", "dead-letter").unwrap(); + assert!(receipt.sequence > 0); + for _ in 0..crate::workspace_registry::AGENT_HOOK_MAX_RETRY_PAGES_PER_WAKE { + mux.retry_pending_agent_hooks_for_terminal(&terminal_id).unwrap(); + } + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(false).unwrap(); + + let retry_state = mux + .workspace_registry + .lock() + .unwrap() + .agent_hook_pending_retry_state_for_test( + crate::agent_hooks::AGENT_HOOK_PRODUCER_ID, + "test", + "dead-letter", + ) + .unwrap() + .expect("failed hook remains durable"); + assert_eq!(retry_state.0, crate::workspace_registry::AGENT_HOOK_MAX_ATTEMPTS); + assert_eq!(retry_state.1, "agent hook retry limit reached"); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections_for_terminal(&terminal_id) + .unwrap() + .is_empty() + ); + assert_eq!( + mux.workspace_registry.lock().unwrap().pending_agent_hook_projections().unwrap().len(), + 1 + ); + } + + #[test] + fn sessionless_legacy_hook_identity_survives_restart() { + let root = std::env::temp_dir() + .join(format!("cmux-agent-legacy-marker-{}", crate::workspace_registry::new_uuid_v4())); + let mux = Mux::open_persistent("legacy-marker", SurfaceOptions::default(), &root).unwrap(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let ingress = |event: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap() + }; + + mux.apply_agent_hook_record(&ingress("SessionStart"), 1).unwrap(); + let first_session_id = + mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id.clone(); + assert_eq!(first_session_id, format!("legacy:{terminal_id}:1")); + mux.shutdown(); + drop(mux); + + let mux = Mux::open_persistent("legacy-marker", SurfaceOptions::default(), &root).unwrap(); + assert_eq!( + mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id, + first_session_id + ); + let reopened_surface = mux.resource_surface_for_terminal(&terminal_id).unwrap(); + let next = ingress("UserPromptSubmit"); + assert_eq!(next.kind, "agent.turn.started"); + mux.apply_agent_hook_record(&next, 2).unwrap(); + assert_eq!(mux.list_agents(Some(reopened_surface), None)[0].state, AgentState::Working); + mux.shutdown(); + drop(mux); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn late_socket_report_cannot_resurrect_hook_ended_agent() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let initial_revision = mux.with_state(|state| state.resource_revision); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let ended = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "SessionEnd", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.apply_agent_hook_record(&ended, 9).unwrap(); + let ended_events = mux.resource_events_after(initial_revision).unwrap(); + assert_eq!(ended_events.batches.len(), 1); + let ended_change = &ended_events.batches[0].changes[0]; + assert_eq!(ended_change["kind"], "delete"); + assert_eq!(ended_change["resource"], "agent"); + assert!(ended_change.get("value").is_none()); + let delayed = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.apply_agent_hook_record(&delayed, 10).unwrap(); + assert!(mux.list_agents(Some(surface.id), None).is_empty()); + let snapshot = crate::resource_api::public_session_snapshot(&mux).unwrap(); + assert!(snapshot["agents"].as_array().unwrap().is_empty()); + assert!( + mux.report_agent(surface.id, AgentState::Working, AgentSource::Socket, None).is_err() + ); + assert!( + mux.report_agent(surface.id, AgentState::Working, AgentSource::Hook, None).is_err() + ); + assert!(mux.list_agents(Some(surface.id), None).is_empty()); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .public_agent_projections(None, None) + .unwrap() + .is_empty() + ); + } + + #[test] + fn new_non_hook_session_can_start_after_hook_session_end() { + let root = std::env::temp_dir() + .join(format!("cmux-agent-hook-restart-{}", crate::workspace_registry::new_uuid_v4())); + let mux = + Mux::open_persistent("agent-hook-restart", SurfaceOptions::default(), &root).unwrap(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let hook = |event: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({"session_id":"old-hook"}), + ) + .unwrap() + }; + mux.apply_agent_hook_record(&hook("SessionEnd"), 1).unwrap(); + + mux.report_agent( + surface.id, + AgentState::Working, + AgentSource::Socket, + Some("new-socket-session".into()), + ) + .unwrap(); + let record = &mux.list_agents(Some(surface.id), None)[0]; + assert_eq!(record.source, AgentSource::Socket); + assert_eq!(record.session.as_deref(), Some("new-socket-session")); + + let public = crate::resource_api::public_session_snapshot(&mux).unwrap(); + assert_eq!(public["agents"][0]["source_session"], "new-socket-session"); + + // A late event from the ended hook session remains fenced. + mux.apply_agent_hook_record(&hook("UserPromptSubmit"), 2).unwrap(); + let record = &mux.list_agents(Some(surface.id), None)[0]; + assert_eq!(record.source, AgentSource::Socket); + assert_eq!(record.session.as_deref(), Some("new-socket-session")); + + mux.shutdown(); + drop(mux); + let reopened = + Mux::open_persistent("agent-hook-restart", SurfaceOptions::default(), &root).unwrap(); + let reopened_record = &reopened.list_agents(Some(surface.id), None)[0]; + assert_eq!(reopened_record.source, AgentSource::Socket); + assert_eq!(reopened_record.session.as_deref(), Some("new-socket-session")); + assert_eq!( + crate::resource_api::public_session_snapshot(&reopened).unwrap()["agents"][0]["source_session"], + "new-socket-session" + ); + reopened.shutdown(); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn direct_hook_report_with_new_session_can_follow_hook_end() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let hook = |event: &str, session_id: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({"session_id": session_id}), + ) + .unwrap() + }; + mux.apply_agent_hook_record(&hook("SessionEnd", "old"), 1).unwrap(); + assert!( + mux.report_agent( + surface.id, + AgentState::Working, + AgentSource::Hook, + Some("new".into()), + ) + .is_ok() + ); + let record = &mux.list_agents(Some(surface.id), None)[0]; + assert_eq!(record.source, AgentSource::Hook); + assert_eq!(record.session.as_deref(), Some("new")); + mux.apply_agent_hook_record(&hook("UserPromptSubmit", "new"), 2).unwrap(); + assert_eq!(mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id, "new"); + assert_eq!(mux.agent_hook_fences.lock().unwrap()[&terminal_id].sequence, 2); + assert!( + mux.report_agent( + surface.id, + AgentState::Working, + AgentSource::Hook, + Some("old".into()), + ) + .is_err() + ); + } + + #[test] + fn old_hook_session_end_cannot_fence_new_session() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().unwrap(); + let ingress = |event: &str, session_id: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({"session_id": session_id}), + ) + .unwrap() + }; + mux.apply_agent_hook_record(&ingress("SessionEnd", "old"), 1).unwrap(); + mux.apply_agent_hook_record(&ingress("SessionStart", "new"), 2).unwrap(); + mux.apply_agent_hook_record(&ingress("UserPromptSubmit", "old"), 3).unwrap(); + let records = mux.list_agents(Some(surface.id), None); + assert_eq!(records.len(), 1); + assert_eq!(records[0].state, AgentState::Idle); + assert!( + !mux.agent_hook_fences + .lock() + .unwrap() + .get(&terminal_id) + .is_some_and(|fence| fence.ended) + ); + } + + #[test] + fn delayed_old_hook_session_start_cannot_replace_active_session() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().unwrap(); + let ingress = |event: &str, session_id: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({"session_id": session_id}), + ) + .unwrap() + }; + mux.apply_agent_hook_record(&ingress("SessionStart", "old"), 1).unwrap(); + mux.apply_agent_hook_record(&ingress("SessionEnd", "old"), 2).unwrap(); + // The old session can arrive after its end marker. Matching the ended + // identity is still a stale event, not permission to reopen it. + mux.apply_agent_hook_record(&ingress("SessionStart", "old"), 3).unwrap(); + assert!(mux.list_agents(Some(surface.id), None).is_empty()); + assert!(mux.agent_hook_fences.lock().unwrap()[&terminal_id].ended); + mux.apply_agent_hook_record(&ingress("SessionStart", "new"), 4).unwrap(); + mux.apply_agent_hook_record(&ingress("SessionStart", "old"), 5).unwrap(); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].state, AgentState::Idle); + assert_eq!(mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id, "new"); + } + + #[test] + fn late_sessionless_hook_event_cannot_attach_to_new_session() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().unwrap(); + let explicit = |event: &str, session_id: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({"session_id": session_id}), + ) + .unwrap() + }; + let sessionless = |event: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap() + }; + + mux.apply_agent_hook_record(&explicit("SessionStart", "old"), 1).unwrap(); + mux.apply_agent_hook_record(&explicit("SessionEnd", "old"), 2).unwrap(); + mux.apply_agent_hook_record(&explicit("SessionStart", "new"), 3).unwrap(); + mux.apply_agent_hook_record(&sessionless("UserPromptSubmit"), 4).unwrap(); + + assert_eq!(mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id, "new"); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].state, AgentState::Idle); + } + + #[test] + fn sessionless_hook_lifecycles_get_distinct_fence_identity() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().unwrap(); + let ingress = |event: &str| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap() + }; + mux.apply_agent_hook_record(&ingress("SessionStart"), 1).unwrap(); + let first = mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id.clone(); + // A session-less start cannot silently reuse the active generation. + mux.apply_agent_hook_record(&ingress("SessionStart"), 2).unwrap(); + assert_eq!(mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id, first); + mux.apply_agent_hook_record(&ingress("SessionEnd"), 2).unwrap(); + mux.apply_agent_hook_record(&ingress("SessionStart"), 3).unwrap(); + let second = mux.agent_hook_fences.lock().unwrap()[&terminal_id].session_id.clone(); + assert_ne!(first, second); + mux.apply_agent_hook_record(&ingress("UserPromptSubmit"), 4).unwrap(); + assert_eq!(mux.list_agents(Some(surface.id), None)[0].state, AgentState::Working); + } + + #[test] + fn socket_report_preserves_hook_sequence_marker() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let initial_revision = mux.with_state(|state| state.resource_revision); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let hook = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.apply_agent_hook_record(&hook, 4).unwrap(); + mux.report_agent( + surface.id, + AgentState::Working, + AgentSource::Socket, + Some("raw-session".into()), + ) + .unwrap(); + let projections = mux.workspace_registry.lock().unwrap().public_projections().unwrap(); + let projection = + projections.agents.into_iter().find(|agent| agent.terminal_id == terminal_id).unwrap(); + assert_eq!(projection.source_session, None); + assert_eq!( + mux.agent_hook_fences.lock().unwrap().get(&terminal_id).map(|fence| fence.sequence), + Some(4) + ); + let public_agents = + mux.workspace_registry.lock().unwrap().public_agent_projections(None, None).unwrap(); + assert_eq!(public_agents.len(), 1); + assert_eq!(public_agents[0].source_session, None); + let snapshot = crate::resource_api::public_session_snapshot(&mux).unwrap(); + let snapshot_agent = snapshot["agents"].as_array().unwrap().first().unwrap(); + assert!(snapshot_agent["source_session"].is_null()); + let public_events = mux.resource_events_after(initial_revision).unwrap(); + assert!( + public_events + .batches + .iter() + .all(|batch| { batch.changes[0]["value"]["source_session"].is_null() }) + ); + } + + #[test] + fn non_hook_reports_reject_reserved_hook_markers() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let error = mux + .report_agent( + surface.id, + AgentState::Working, + AgentSource::Socket, + Some("cmux-hook-sequence:9".into()), + ) + .unwrap_err(); + assert!(error.to_string().contains("reserved hook marker")); + assert!(mux.list_agents(Some(surface.id), None).is_empty()); + } + + #[test] + fn persistent_hook_watermark_restores_from_journal() { + let root = std::env::temp_dir() + .join(format!("cmux-agent-watermark-{}", crate::workspace_registry::new_uuid_v4())); + let mux = + Mux::open_persistent("agent-watermark", SurfaceOptions::default(), &root).unwrap(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = mux.with_state(|state| { + match state.resource_indexes.content_ids.get(&surface.id).unwrap() { + ContentPublicId::Terminal(id) => id.clone(), + ContentPublicId::Browser(_) => panic!("workspace opened a browser"), + } + }); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "SessionEnd", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + let commit = mux.append_journal_ingress(&ingress, "test", "watermark-end").unwrap(); + assert!(commit.sequence > 0); + assert_eq!( + mux.workspace_registry.lock().unwrap().agent_hook_apply_cursor().unwrap(), + commit.sequence + ); + let projections = mux.workspace_registry.lock().unwrap().public_projections().unwrap(); + let restored = + mux.with_state(|state| restore_public_projections(state, projections)).unwrap(); + assert_eq!(restored.agent_hook_fences[&terminal_id].sequence, commit.sequence); + assert!(restored.agent_hook_fences[&terminal_id].ended); + drop(mux); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn failed_hook_projection_does_not_consume_sequence_and_replay_succeeds() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(true).unwrap(); + let error = mux.apply_agent_hook_record(&ingress, 7).unwrap_err(); + assert!(error.to_string().contains("forced resource patch failure")); + assert_eq!(mux.workspace_registry.lock().unwrap().agent_hook_apply_cursor().unwrap(), 0); + assert!(mux.agent_hook_fences.lock().unwrap().get(&terminal_id).is_none()); + + mux.workspace_registry.lock().unwrap().set_resource_patch_failure(false).unwrap(); + mux.apply_agent_hook_record(&ingress, 7).unwrap(); + assert_eq!(mux.workspace_registry.lock().unwrap().agent_hook_apply_cursor().unwrap(), 7); + assert_eq!(mux.agent_hook_fences.lock().unwrap()[&terminal_id].sequence, 7); + assert_eq!(mux.list_agents(Some(surface.id), None).len(), 1); + + // A replay of the committed sequence is an idempotent no-op. + mux.apply_agent_hook_record(&ingress, 7).unwrap(); + assert_eq!(mux.workspace_registry.lock().unwrap().agent_hook_apply_cursor().unwrap(), 7); + assert_eq!(mux.list_agents(Some(surface.id), None).len(), 1); + } + + #[test] + fn reopened_ended_hook_fence_rejects_late_events() { + let root = std::env::temp_dir() + .join(format!("cmux-agent-ended-reopen-{}", crate::workspace_registry::new_uuid_v4())); + let mux = + Mux::open_persistent("agent-ended-reopen", SurfaceOptions::default(), &root).unwrap(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let ingress = |event: &str, session_id: Option<&str>| { + crate::agent_hooks::agent_hook_journal_ingress( + "claude", + event, + Some(&terminal_id.to_string()), + session_id + .map(|session_id| serde_json::json!({"session_id": session_id})) + .unwrap_or_else(|| serde_json::json!({})), + ) + .unwrap() + }; + mux.apply_agent_hook_record(&ingress("SessionStart", Some("ended-session")), 1).unwrap(); + mux.apply_agent_hook_record(&ingress("SessionEnd", Some("ended-session")), 2).unwrap(); + assert!(mux.list_agents(Some(surface.id), None).is_empty()); + mux.shutdown(); + drop(mux); + + let reopened = + Mux::open_persistent("agent-ended-reopen", SurfaceOptions::default(), &root).unwrap(); + let reopened_surface = reopened.resource_surface_for_terminal(&terminal_id).unwrap(); + assert!(reopened.list_agents(Some(reopened_surface), None).is_empty()); + assert!(reopened.agent_hook_fences.lock().unwrap()[&terminal_id].ended); + + // Events from the ended identity, a different identity, and a + // session-less adapter must all remain behind the durable fence. + reopened + .apply_agent_hook_record(&ingress("UserPromptSubmit", Some("ended-session")), 3) + .unwrap(); + reopened + .apply_agent_hook_record(&ingress("UserPromptSubmit", Some("different-session")), 4) + .unwrap(); + reopened.apply_agent_hook_record(&ingress("UserPromptSubmit", None), 5).unwrap(); + assert!(reopened.list_agents(Some(reopened_surface), None).is_empty()); + assert_eq!(reopened.agent_hook_fences.lock().unwrap()[&terminal_id].sequence, 2); + reopened.shutdown(); + drop(reopened); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn agent_hook_events_without_a_live_terminal_still_append() { + let mux = test_mux(); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + None, + serde_json::json!({}), + ) + .unwrap(); + mux.append_journal_ingress(&ingress, "test", "hook-no-terminal").unwrap(); + assert!(mux.list_agents(None, None).is_empty()); + } + + #[test] + fn unknown_terminal_hook_receipt_is_not_retained_for_retry() { + let mux = test_mux(); + let terminal_id = TerminalPublicId::parse("term_000000000000000000000000000000ff").unwrap(); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.append_journal_ingress(&ingress, "test", "unknown-terminal").unwrap(); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections() + .unwrap() + .is_empty() + ); + } + + #[test] + fn terminal_cleanup_purges_pending_hook_receipts() { + let mux = test_mux(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let ingress = crate::agent_hooks::agent_hook_journal_ingress( + "claude", + "UserPromptSubmit", + Some(&terminal_id.to_string()), + serde_json::json!({}), + ) + .unwrap(); + mux.workspace_registry + .lock() + .unwrap() + .enqueue_agent_hook_pending( + &ingress.producer_id, + "test", + "pending-before-close", + 1, + &ingress, + AGENT_HOOK_RETRY_ERROR, + crate::workspace_registry::AgentHookRetryClass::Transient, + ) + .unwrap(); + mux.purge_terminal_side_tables(&terminal_id); + assert!( + mux.workspace_registry + .lock() + .unwrap() + .pending_agent_hook_projections() + .unwrap() + .is_empty() + ); + } + #[test] fn failed_raw_agent_report_rolls_back_projection_memory_revision_and_event() { let mux = test_mux(); diff --git a/cmux-tui/crates/cmux-tui-core/src/mux/public_projections.rs b/cmux-tui/crates/cmux-tui-core/src/mux/public_projections.rs index 965968c0c1b8..8b222d314f26 100644 --- a/cmux-tui/crates/cmux-tui-core/src/mux/public_projections.rs +++ b/cmux-tui/crates/cmux-tui-core/src/mux/public_projections.rs @@ -9,6 +9,7 @@ pub(super) struct RestoredPublicProjections { pub(super) has_terminal_defaults: bool, pub(super) next_notification_id: u64, pub(super) agent_records: HashMap, + pub(super) agent_hook_fences: HashMap, pub(super) terminal_notifications: HashMap, pub(super) notification_ledger: VecDeque, } @@ -59,13 +60,56 @@ pub(super) fn restore_public_projections( .saturating_add(1); let mut agent_records = HashMap::with_capacity(projections.agents.len()); + let mut agent_hook_fences = HashMap::new(); + for hook_state in projections.agent_hook_states { + agent_hook_fences.insert( + hook_state.terminal_id, + super::HookFence { + session_id: hook_state.agent_session_id, + sequence: hook_state.applied_sequence, + ended: hook_state.ended, + }, + ); + } for agent in projections.agents { + let state = agent_state(&agent.state)?; + let internal_marker = agent.source_session.as_deref().is_some_and(|value| { + value.starts_with("cmux-hook-sequence:") || value.starts_with("cmux-hook-ended:") + }); + if let Some(source_session) = agent.source_session.as_deref() { + let marker = source_session.strip_prefix("cmux-hook-sequence:"); + let ended = source_session.strip_prefix("cmux-hook-ended:"); + if let Some(value) = marker.or(ended).and_then(|value| value.parse::().ok()) { + // Marker-only projections predate durable hook state. Use the + // marker sequence as their legacy generation token so a + // session-less event continues the same lifecycle after a + // restart without reusing a terminal-wide identity. + agent_hook_fences.entry(agent.terminal_id.clone()).or_insert(super::HookFence { + session_id: super::legacy_hook_session_id(&agent.terminal_id, value), + sequence: value, + ended: ended.is_some(), + }); + } + } + if state == AgentState::Done && agent.source == "hook" { + // Older projections predate the internal ended marker. Keep + // their terminal fenced after restart so a late socket report + // cannot resurrect the completed session. Sequence zero is the + // one-release compatibility generation for records without a + // marker. + agent_hook_fences.entry(agent.terminal_id.clone()).or_insert(super::HookFence { + session_id: super::legacy_hook_session_id(&agent.terminal_id, 0), + sequence: 0, + ended: true, + }); + continue; + } let previous = agent_records.insert( agent.terminal_id.clone(), TerminalAgentRecord { - state: agent_state(&agent.state)?, + state, source: agent_source(&agent.source)?, - session: agent.source_session, + session: (!internal_marker).then_some(agent.source_session).flatten(), updated_at_ms: agent.updated_at_ms, }, ); @@ -81,6 +125,7 @@ pub(super) fn restore_public_projections( has_terminal_defaults, next_notification_id, agent_records, + agent_hook_fences, terminal_notifications, notification_ledger, }) @@ -123,6 +168,10 @@ mod tests { use crate::terminal_host_runtime::TerminalHostIdentity; use crate::workspace_registry::{RegistryAgentProjection, RegistryNotificationProjection}; + fn terminal_id(value: u8) -> TerminalPublicId { + TerminalPublicId::parse(format!("term_{value:032x}")).unwrap() + } + fn empty_state() -> State { State { workspaces: Vec::new(), @@ -177,6 +226,7 @@ mod tests { updated_at_ms: 1, source_session: None, }], + agent_hook_states: Vec::new(), terminal_defaults: None, frontend_projections: Vec::new(), }; @@ -206,6 +256,7 @@ mod tests { unread: true, }], agents: Vec::new(), + agent_hook_states: Vec::new(), terminal_defaults: None, frontend_projections: Vec::new(), }; @@ -229,6 +280,7 @@ mod tests { unread: true, }], agents: Vec::new(), + agent_hook_states: Vec::new(), terminal_defaults: None, frontend_projections: Vec::new(), }; @@ -238,4 +290,73 @@ mod tests { assert_eq!(restored.notification_ledger[0].terminal_id, Some(terminal)); assert!(restored.terminal_notifications.is_empty()); } + + #[test] + fn done_agent_records_are_not_restored_into_live_roster() { + let terminal = terminal_id(9); + let projections = RegistryPublicProjections { + notifications: Vec::new(), + agents: vec![RegistryAgentProjection { + id: AgentPublicId::parse("agent_00000000000000000000000000000009").unwrap(), + terminal_id: terminal.clone(), + state: "done".into(), + source: "hook".into(), + updated_at_ms: 1, + source_session: None, + }], + agent_hook_states: Vec::new(), + terminal_defaults: None, + frontend_projections: Vec::new(), + }; + let restored = restore_public_projections(&empty_state(), projections).unwrap(); + assert!(restored.agent_records.is_empty()); + assert!(restored.agent_hook_fences[&terminal].ended); + } + + #[test] + fn hook_marker_restores_watermark_without_exposing_session() { + let terminal = terminal_id(10); + let projections = RegistryPublicProjections { + notifications: Vec::new(), + agents: vec![RegistryAgentProjection { + id: AgentPublicId::parse("agent_00000000000000000000000000000010").unwrap(), + terminal_id: terminal.clone(), + state: "working".into(), + source: "hook".into(), + updated_at_ms: 1, + source_session: Some("cmux-hook-sequence:12".into()), + }], + agent_hook_states: Vec::new(), + terminal_defaults: None, + frontend_projections: Vec::new(), + }; + let restored = restore_public_projections(&empty_state(), projections).unwrap(); + assert_eq!(restored.agent_hook_fences[&terminal].sequence, 12); + assert_eq!(restored.agent_records[&terminal].session, None); + } + + #[test] + fn socket_done_agent_restores_into_live_record_map() { + let terminal = terminal_id(11); + let projections = RegistryPublicProjections { + notifications: Vec::new(), + agents: vec![RegistryAgentProjection { + id: AgentPublicId::parse("agent_00000000000000000000000000000011").unwrap(), + terminal_id: terminal.clone(), + state: "done".into(), + source: "socket".into(), + updated_at_ms: 3, + source_session: Some("socket-session".into()), + }], + agent_hook_states: Vec::new(), + terminal_defaults: None, + frontend_projections: Vec::new(), + }; + let restored = restore_public_projections(&empty_state(), projections).unwrap(); + let record = &restored.agent_records[&terminal]; + assert_eq!(record.state, AgentState::Done); + assert_eq!(record.source, AgentSource::Socket); + assert_eq!(record.session.as_deref(), Some("socket-session")); + assert!(!restored.agent_hook_fences.contains_key(&terminal)); + } } diff --git a/cmux-tui/crates/cmux-tui-core/src/resource_api.rs b/cmux-tui/crates/cmux-tui-core/src/resource_api.rs index 100144e9db7d..0cf4ad9dc4eb 100644 --- a/cmux-tui/crates/cmux-tui-core/src/resource_api.rs +++ b/cmux-tui/crates/cmux-tui-core/src/resource_api.rs @@ -724,6 +724,22 @@ pub(crate) fn public_session_snapshot_with_journal_head( let mut agents = public_projections .agents .into_iter() + .filter(|agent| { + !(agent.source == "hook" && agent.state == "done") + && !agent + .source_session + .as_deref() + .is_some_and(|value| value.starts_with("cmux-hook-ended:")) + }) + .map(|mut agent| { + if agent.source_session.as_deref().is_some_and(|value| { + value.starts_with("cmux-hook-sequence:") + || value.starts_with("cmux-hook-ended:") + }) { + agent.source_session = None; + } + agent + }) .map(|agent| agent.into_public_snapshot(&topology.session_id)) .collect::>(); agents.sort_by(|left, right| { diff --git a/cmux-tui/crates/cmux-tui-core/src/workspace_registry.rs b/cmux-tui/crates/cmux-tui-core/src/workspace_registry.rs index 7f6b7caed7b7..a166ecf313a1 100644 --- a/cmux-tui/crates/cmux-tui-core/src/workspace_registry.rs +++ b/cmux-tui/crates/cmux-tui-core/src/workspace_registry.rs @@ -61,6 +61,10 @@ pub use public_projection_store::RegistryPublicProjections; #[cfg(test)] pub use public_projection_store::{RegistryAgentProjection, RegistryNotificationProjection}; pub(crate) use resource_store::validate_registry_screen_projection; +pub(crate) use resource_store::{ + AGENT_HOOK_MAX_ATTEMPTS, AGENT_HOOK_MAX_RETRY_PAGES_PER_WAKE, AgentHookProjectionState, + AgentHookRetryClass, +}; #[allow(unused_imports)] pub use resource_store::{ RegistryBrowser, RegistryBrowserLaunch, RegistryBrowserReconnect, RegistryBrowserSource, diff --git a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/journal_extensions.rs b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/journal_extensions.rs index aaa8245e32df..c70c1082c0d0 100644 --- a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/journal_extensions.rs +++ b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/journal_extensions.rs @@ -1428,6 +1428,16 @@ fn append_journal_ingress_transaction( if let Some(commit) = ingress_receipt(tx, &ingress.producer_id, origin, idempotency_key, fingerprint.as_slice())? { + if ingress.producer_id == crate::AGENT_HOOK_PRODUCER_ID { + WorkspaceRegistry::stage_agent_hook_pending( + tx, + &ingress.producer_id, + origin, + idempotency_key, + commit.sequence, + ingress, + )?; + } return Ok(commit); } let installed = tx @@ -1528,6 +1538,16 @@ fn append_journal_ingress_transaction( canonical_json(&result)?, ], )?; + if built_in_agent { + WorkspaceRegistry::stage_agent_hook_pending( + tx, + &ingress.producer_id, + origin, + idempotency_key, + sequence, + ingress, + )?; + } Ok(JournalAppendCommit { sequence, event_id, replayed: false }) } diff --git a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/public_projection_store.rs b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/public_projection_store.rs index 6774a65694b6..2555de3cccfb 100644 --- a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/public_projection_store.rs +++ b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/public_projection_store.rs @@ -44,6 +44,14 @@ pub struct RegistryAgentProjection { pub source_session: Option, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RegistryAgentHookState { + pub terminal_id: TerminalPublicId, + pub agent_session_id: String, + pub applied_sequence: u64, + pub ended: bool, +} + impl RegistryAgentProjection { pub(crate) fn into_public_snapshot(self, session_id: &SessionPublicId) -> Value { json!({ @@ -63,6 +71,7 @@ pub struct RegistryPublicProjections { /// Oldest first, matching the in-memory notification ledger. pub notifications: Vec, pub agents: Vec, + pub(crate) agent_hook_states: Vec, pub terminal_defaults: Option, pub frontend_projections: Vec, } @@ -183,11 +192,13 @@ impl WorkspaceRegistry { let live_terminals = self.live_terminal_public_ids()?; let notifications = self.durable_notifications(&live_terminals)?; let agents = self.durable_agents(None, None)?; + let agent_hook_states = self.durable_agent_hook_states()?; let terminal_defaults = self.durable_terminal_defaults()?; let frontend_projections = self.public_frontend_projections()?; Ok(RegistryPublicProjections { notifications, agents, + agent_hook_states, terminal_defaults, frontend_projections, }) @@ -198,7 +209,22 @@ impl WorkspaceRegistry { terminal: Option<&TerminalPublicId>, state: Option<&str>, ) -> anyhow::Result> { - self.durable_agents(terminal, state) + let mut agents = self.durable_agents(terminal, state)?; + agents.retain(|agent| { + !(agent.source == "hook" && agent.state == "done") + && !agent + .source_session + .as_deref() + .is_some_and(|value| value.starts_with("cmux-hook-ended:")) + }); + for agent in &mut agents { + if agent.source_session.as_deref().is_some_and(|value| { + value.starts_with("cmux-hook-sequence:") || value.starts_with("cmux-hook-ended:") + }) { + agent.source_session = None; + } + } + Ok(agents) } fn live_terminal_public_ids(&self) -> anyhow::Result> { @@ -338,6 +364,34 @@ impl WorkspaceRegistry { Ok(agents) } + fn durable_agent_hook_states(&self) -> anyhow::Result> { + let mut statement = self.connection.prepare( + "SELECT terminal_id, agent_session_id, applied_sequence, ended + FROM resource_agent_hook_state + ORDER BY terminal_id ASC", + )?; + statement + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, bool>(3)?, + )) + })? + .map(|row| { + let (terminal_id, agent_session_id, applied_sequence, ended) = row?; + Ok(RegistryAgentHookState { + terminal_id: TerminalPublicId::parse(terminal_id)?, + agent_session_id, + applied_sequence: u64::try_from(applied_sequence) + .context("agent hook sequence is negative")?, + ended, + }) + }) + .collect() + } + fn durable_terminal_defaults(&self) -> anyhow::Result> { let stored = self .connection diff --git a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/resource_store.rs b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/resource_store.rs index a1a5549c40f2..7340c8088333 100644 --- a/cmux-tui/crates/cmux-tui-core/src/workspace_registry/resource_store.rs +++ b/cmux-tui/crates/cmux-tui-core/src/workspace_registry/resource_store.rs @@ -7,6 +7,18 @@ use super::*; pub(super) const RESOURCE_MUTATION_REPLAY_CAPACITY: usize = 4096; pub(super) const RESOURCE_MUTATION_PRUNE_INTERVAL: u64 = 128; const RESOURCE_EVENT_PAGE_SIZE: usize = 1024; +pub(super) const AGENT_HOOK_RETRY_PAGE_SIZE: i64 = 64; +// Rows that reach this cap stay durable as dead-letter records. Selectors +// exclude them, so a permanent projection failure cannot spin forever. +pub(crate) const AGENT_HOOK_MAX_ATTEMPTS: i64 = 8; +pub(crate) const AGENT_HOOK_MAX_RETRY_PAGES_PER_WAKE: usize = 16; +pub(crate) const AGENT_HOOK_DEAD_LETTER_CAP: i64 = 1024; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AgentHookRetryClass { + Transient, + Permanent, +} pub(super) fn create_resource_schema(transaction: &Transaction<'_>) -> anyhow::Result<()> { transaction.execute_batch( @@ -150,6 +162,29 @@ pub(super) fn create_resource_schema(transaction: &Transaction<'_>) -> anyhow::R ) ) ); + CREATE TABLE IF NOT EXISTS resource_agent_hook_state ( + terminal_id TEXT PRIMARY KEY NOT NULL + REFERENCES resource_terminals(public_id) ON DELETE CASCADE, + agent_session_id TEXT NOT NULL, + applied_sequence INTEGER NOT NULL CHECK(applied_sequence >= 0), + ended INTEGER NOT NULL CHECK(ended IN (0, 1)), + committed_revision INTEGER NOT NULL CHECK(committed_revision >= 0) + ); + CREATE TABLE IF NOT EXISTS resource_agent_hook_apply_cursor ( + id INTEGER PRIMARY KEY CHECK(id = 1), + sequence INTEGER NOT NULL CHECK(sequence >= 0) + ); + CREATE TABLE IF NOT EXISTS resource_agent_hook_pending ( + producer_id TEXT NOT NULL, + origin TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + terminal_id TEXT, + event_sequence INTEGER NOT NULL CHECK(event_sequence >= 0), + ingress_json TEXT NOT NULL CHECK(json_valid(ingress_json)), + error TEXT NOT NULL, + attempt INTEGER NOT NULL CHECK(attempt >= 0), + PRIMARY KEY(producer_id, origin, idempotency_key) + ); DROP TRIGGER IF EXISTS resource_agent_projection_terminal_tombstone; CREATE INDEX IF NOT EXISTS resource_mutations_by_operation_revision ON resource_mutations(operation, committed_revision DESC); @@ -162,6 +197,67 @@ pub(super) fn create_resource_schema(transaction: &Transaction<'_>) -> anyhow::R terminal_id DESC );", )?; + let has_scoped_pending = transaction + .prepare("PRAGMA table_info(resource_agent_hook_pending)")? + .query_map([], |row| row.get::<_, String>(1))? + .collect::, _>>()? + .iter() + .any(|column| column == "producer_id"); + if !has_scoped_pending { + transaction.execute_batch( + "ALTER TABLE resource_agent_hook_pending RENAME TO resource_agent_hook_pending_legacy; + CREATE TABLE resource_agent_hook_pending ( + producer_id TEXT NOT NULL, + origin TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + terminal_id TEXT, + event_sequence INTEGER NOT NULL CHECK(event_sequence >= 0), + ingress_json TEXT NOT NULL CHECK(json_valid(ingress_json)), + error TEXT NOT NULL, + attempt INTEGER NOT NULL CHECK(attempt >= 0), + PRIMARY KEY(producer_id, origin, idempotency_key) + ); + INSERT INTO resource_agent_hook_pending( + producer_id, origin, idempotency_key, terminal_id, event_sequence, ingress_json, error, attempt + ) SELECT COALESCE(NULLIF(json_extract(ingress_json, '$.producer_id'), ''), 'cmux_agent'), + 'agent-hook', idempotency_key, + (SELECT json_extract(value, '$.id') + FROM json_each(resource_agent_hook_pending_legacy.ingress_json, '$.subjects') + WHERE json_extract(value, '$.kind') = 'terminal' LIMIT 1), + event_sequence, ingress_json, error, attempt + FROM resource_agent_hook_pending_legacy; + DROP TABLE resource_agent_hook_pending_legacy;", + )?; + } + transaction.execute( + "INSERT OR IGNORE INTO resource_agent_hook_apply_cursor(id, sequence) VALUES(1, 0)", + [], + )?; + let has_pending_terminal_id = transaction + .prepare("PRAGMA table_info(resource_agent_hook_pending)")? + .query_map([], |row| row.get::<_, String>(1))? + .collect::, _>>()? + .iter() + .any(|column| column == "terminal_id"); + if !has_pending_terminal_id { + transaction + .execute("ALTER TABLE resource_agent_hook_pending ADD COLUMN terminal_id TEXT", [])?; + transaction.execute( + "UPDATE resource_agent_hook_pending + SET terminal_id = ( + SELECT json_extract(value, '$.id') + FROM json_each(resource_agent_hook_pending.ingress_json, '$.subjects') + WHERE json_extract(value, '$.kind') = 'terminal' LIMIT 1 + ) + WHERE terminal_id IS NULL", + [], + )?; + } + transaction.execute_batch( + "DROP INDEX IF EXISTS resource_agent_hook_pending_by_terminal; + CREATE INDEX IF NOT EXISTS resource_agent_hook_pending_by_terminal + ON resource_agent_hook_pending(terminal_id, event_sequence, idempotency_key);", + )?; Ok(()) } @@ -276,6 +372,13 @@ pub(super) fn resource_tabs_needs_multiview_normalization( Ok(!saw_browser_view_index) } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct AgentHookProjectionState { + pub agent_session_id: String, + pub applied_sequence: u64, + pub ended: bool, +} + pub(super) fn migrate_resource_agent_projections( transaction: &Transaction<'_>, ) -> anyhow::Result<()> { @@ -407,7 +510,364 @@ pub(super) fn migrate_resource_browser_metadata( Ok(()) } +fn advance_agent_hook_apply_cursor_transaction( + transaction: &Transaction<'_>, + sequence: u64, +) -> anyhow::Result<()> { + let sequence = i64::try_from(sequence).context("agent hook sequence exceeds SQLite range")?; + let changed = transaction.execute( + "UPDATE resource_agent_hook_apply_cursor + SET sequence = CASE WHEN sequence < ?1 THEN ?1 ELSE sequence END + WHERE id = 1", + [sequence], + )?; + anyhow::ensure!(changed == 1, "agent hook apply cursor row is missing"); + Ok(()) +} + impl WorkspaceRegistry { + /// Return the highest journal sequence committed with a hook projection. + /// This recovery watermark is not an admission cursor. It advances only + /// after the projection transaction commits. + pub fn agent_hook_apply_cursor(&self) -> anyhow::Result { + self.connection + .query_row( + "SELECT sequence FROM resource_agent_hook_apply_cursor WHERE id = 1", + [], + |row| row.get::<_, i64>(0), + ) + .map(|value| u64::try_from(value).context("agent hook apply cursor is negative"))? + } + + pub fn advance_agent_hook_apply_cursor(&mut self, sequence: u64) -> anyhow::Result<()> { + let tx = self.connection.transaction()?; + advance_agent_hook_apply_cursor_transaction(&tx, sequence)?; + tx.commit()?; + Ok(()) + } + pub(super) fn stage_agent_hook_pending( + transaction: &Transaction<'_>, + producer_id: &str, + origin: &str, + idempotency_key: &str, + sequence: u64, + ingress: &crate::JournalIngress, + ) -> anyhow::Result<()> { + let ingress_json = serde_json::to_string(ingress)?; + let terminal_id = ingress + .subjects + .iter() + .find(|subject| subject.kind == "terminal") + .map(|subject| subject.id.as_str()); + transaction.execute( + "INSERT INTO resource_agent_hook_pending( + producer_id, origin, idempotency_key, terminal_id, event_sequence, ingress_json, error, attempt + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, '', 0) + ON CONFLICT(producer_id, origin, idempotency_key) DO UPDATE SET + terminal_id = excluded.terminal_id, + event_sequence = excluded.event_sequence, + ingress_json = excluded.ingress_json", + params![producer_id, origin, idempotency_key, terminal_id, i64::try_from(sequence)?, ingress_json], + )?; + Ok(()) + } + + pub fn enqueue_agent_hook_pending( + &mut self, + producer_id: &str, + origin: &str, + idempotency_key: &str, + sequence: u64, + ingress: &crate::JournalIngress, + error: &str, + retry_class: AgentHookRetryClass, + ) -> anyhow::Result<()> { + const MAX_ERROR_CHARS: usize = 1_024; + let ingress_json = serde_json::to_string(ingress)?; + let terminal_id = ingress + .subjects + .iter() + .find(|subject| subject.kind == "terminal") + .map(|subject| subject.id.as_str()); + let bounded_error = error.chars().take(MAX_ERROR_CHARS).collect::(); + // Projection failures caused by temporary availability or storage + // conditions keep the attempt budget unchanged. Other failures consume + // the bounded budget and become quarantined at the cap. + let transient = matches!(retry_class, AgentHookRetryClass::Transient); + self.connection.execute( + "INSERT INTO resource_agent_hook_pending( + producer_id, origin, idempotency_key, terminal_id, event_sequence, ingress_json, error, attempt + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, 1) + ON CONFLICT(producer_id, origin, idempotency_key) DO UPDATE SET + terminal_id = excluded.terminal_id, + event_sequence = excluded.event_sequence, + ingress_json = excluded.ingress_json, + error = CASE + WHEN ?8 = 1 THEN excluded.error + WHEN resource_agent_hook_pending.attempt + 1 >= ?9 + THEN 'agent hook retry limit reached' + ELSE excluded.error + END, + attempt = CASE + WHEN ?8 = 1 THEN resource_agent_hook_pending.attempt + WHEN resource_agent_hook_pending.attempt < ?9 + THEN resource_agent_hook_pending.attempt + 1 + ELSE resource_agent_hook_pending.attempt + END", + params![ + producer_id, + origin, + idempotency_key, + terminal_id, + i64::try_from(sequence)?, + ingress_json, + bounded_error, + transient as i64, + AGENT_HOOK_MAX_ATTEMPTS, + ], + )?; + // Keep quarantined failures bounded. Live retry rows remain untouched; + // only the oldest dead letters beyond the retention cap are evicted. + self.connection.execute( + "DELETE FROM resource_agent_hook_pending + WHERE attempt >= ?1 + AND rowid NOT IN ( + SELECT rowid + FROM resource_agent_hook_pending + WHERE attempt >= ?1 + ORDER BY rowid DESC + LIMIT ?2 + )", + params![AGENT_HOOK_MAX_ATTEMPTS, AGENT_HOOK_DEAD_LETTER_CAP], + )?; + Ok(()) + } + + pub(crate) fn purge_agent_hook_pending_for_terminal( + &mut self, + terminal_id: &crate::resource::TerminalPublicId, + ) -> anyhow::Result<()> { + self.connection.execute( + "DELETE FROM resource_agent_hook_pending WHERE terminal_id = ?1", + [terminal_id.as_str()], + )?; + Ok(()) + } + + pub fn clear_agent_hook_pending( + &mut self, + producer_id: &str, + origin: &str, + idempotency_key: &str, + ) -> anyhow::Result<()> { + self.connection.execute( + "DELETE FROM resource_agent_hook_pending + WHERE producer_id = ?1 AND origin = ?2 AND idempotency_key = ?3", + params![producer_id, origin, idempotency_key], + )?; + Ok(()) + } + + fn record_agent_hook_pending_failure( + &self, + producer_id: &str, + origin: &str, + idempotency_key: &str, + ) -> anyhow::Result<()> { + self.connection.execute( + "UPDATE resource_agent_hook_pending + SET error = CASE + WHEN attempt + 1 >= ?4 THEN 'agent hook retry limit reached' + ELSE 'invalid pending agent hook payload' + END, + attempt = CASE + WHEN attempt < ?4 THEN attempt + 1 + ELSE attempt + END + WHERE producer_id = ?1 AND origin = ?2 AND idempotency_key = ?3", + params![producer_id, origin, idempotency_key, AGENT_HOOK_MAX_ATTEMPTS], + )?; + Ok(()) + } + + pub fn pending_agent_hook_projections( + &self, + ) -> anyhow::Result> { + let mut statement = self.connection.prepare( + "SELECT producer_id, origin, idempotency_key, event_sequence, ingress_json + FROM resource_agent_hook_pending ORDER BY event_sequence ASC, idempotency_key ASC", + )?; + statement + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, i64>(3)?, + row.get::<_, String>(4)?, + )) + })? + .map(|row| { + let (producer_id, origin, key, sequence, ingress_json) = row?; + Ok(( + producer_id, + origin, + key, + u64::try_from(sequence).context("pending hook sequence is negative")?, + serde_json::from_str(&ingress_json)?, + )) + }) + .collect() + } + + pub fn pending_agent_hook_projections_for_terminal( + &self, + terminal_id: &crate::resource::TerminalPublicId, + ) -> anyhow::Result> { + let mut statement = self.connection.prepare( + "SELECT producer_id, origin, idempotency_key, event_sequence, ingress_json + FROM resource_agent_hook_pending + WHERE terminal_id = ?1 AND attempt < ?2 + ORDER BY event_sequence ASC, idempotency_key ASC + LIMIT ?3", + )?; + let rows = statement + .query_map( + params![terminal_id.as_str(), AGENT_HOOK_MAX_ATTEMPTS, AGENT_HOOK_RETRY_PAGE_SIZE], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, i64>(3)?, + row.get::<_, String>(4)?, + )) + }, + )? + .collect::, _>>()?; + drop(statement); + let mut pending = Vec::with_capacity(rows.len()); + for (producer_id, origin, key, sequence, ingress_json) in rows { + let ingress = match serde_json::from_str(&ingress_json) { + Ok(ingress) => ingress, + Err(_) => { + self.record_agent_hook_pending_failure(&producer_id, &origin, &key)?; + continue; + } + }; + pending.push(( + producer_id, + origin, + key, + u64::try_from(sequence).context("pending hook sequence is negative")?, + ingress, + )); + } + Ok(pending) + } + + pub fn pending_agent_hook_projections_page( + &self, + after: Option<(u64, String, i64)>, + ) -> anyhow::Result<( + Vec<(String, String, String, u64, crate::JournalIngress)>, + Option<(u64, String, i64)>, + )> { + let (after_sequence, after_key, after_rowid) = after.unwrap_or((0, String::new(), 0)); + let mut statement = self.connection.prepare( + "SELECT rowid, producer_id, origin, idempotency_key, event_sequence, ingress_json + FROM resource_agent_hook_pending + WHERE attempt < ?1 + AND (event_sequence > ?2 + OR (event_sequence = ?2 AND idempotency_key > ?3) + OR (event_sequence = ?2 AND idempotency_key = ?3 AND rowid > ?4)) + ORDER BY event_sequence ASC, idempotency_key ASC, rowid ASC + LIMIT ?5", + )?; + let rows = statement + .query_map( + params![ + AGENT_HOOK_MAX_ATTEMPTS, + i64::try_from(after_sequence)?, + after_key, + after_rowid, + AGENT_HOOK_RETRY_PAGE_SIZE + ], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, i64>(4)?, + row.get::<_, String>(5)?, + )) + }, + )? + .collect::, _>>()?; + drop(statement); + let mut pending = Vec::with_capacity(rows.len()); + let mut next_cursor = None; + for (rowid, producer_id, origin, key, sequence, ingress_json) in rows { + let sequence = u64::try_from(sequence).context("pending hook sequence is negative")?; + next_cursor = Some((sequence, key.clone(), rowid)); + let ingress = match serde_json::from_str(&ingress_json) { + Ok(ingress) => ingress, + Err(_) => { + self.record_agent_hook_pending_failure(&producer_id, &origin, &key)?; + continue; + } + }; + pending.push((producer_id, origin, key, sequence, ingress)); + } + Ok((pending, next_cursor)) + } + + pub fn commit_agent_projection_with_hook_state( + &mut self, + mutation: &WorkspaceMutation, + fingerprint: &Value, + expected_revision: Option, + terminal_id: &TerminalPublicId, + result: &Value, + deltas: &Value, + hook_state: Option<&AgentHookProjectionState>, + ) -> anyhow::Result { + self.commit_agent_projection_inner( + mutation, + fingerprint, + expected_revision, + terminal_id, + result, + deltas, + hook_state, + None, + ) + } + + pub fn commit_agent_projection_with_hook_state_and_sequence( + &mut self, + mutation: &WorkspaceMutation, + fingerprint: &Value, + expected_revision: Option, + terminal_id: &TerminalPublicId, + result: &Value, + deltas: &Value, + hook_state: Option<&AgentHookProjectionState>, + journal_sequence: u64, + ) -> anyhow::Result { + self.commit_agent_projection_inner( + mutation, + fingerprint, + expected_revision, + terminal_id, + result, + deltas, + hook_state, + Some(journal_sequence), + ) + } + pub fn replay_resource_patch( &self, mutation: &WorkspaceMutation, @@ -429,6 +889,30 @@ impl WorkspaceRegistry { terminal_id: &TerminalPublicId, result: &Value, deltas: &Value, + ) -> anyhow::Result { + self.commit_agent_projection_inner( + mutation, + fingerprint, + expected_revision, + terminal_id, + result, + deltas, + None, + None, + ) + } + + #[allow(clippy::too_many_arguments)] + fn commit_agent_projection_inner( + &mut self, + mutation: &WorkspaceMutation, + fingerprint: &Value, + expected_revision: Option, + terminal_id: &TerminalPublicId, + result: &Value, + deltas: &Value, + hook_state: Option<&AgentHookProjectionState>, + journal_sequence: Option, ) -> anyhow::Result { const OPERATION: &str = "agent.report"; validate_identifier("mutation id", &mutation.id)?; @@ -441,6 +925,10 @@ impl WorkspaceRegistry { let result_json = canonical_json(result)?; let tx = self.connection.transaction()?; if let Some(replayed) = resource_patch_replay(&tx, mutation, OPERATION, &fingerprint)? { + if let Some(sequence) = journal_sequence { + advance_agent_hook_apply_cursor_transaction(&tx, sequence)?; + tx.commit()?; + } return Ok(replayed); } let terminal_is_live = tx @@ -475,6 +963,27 @@ impl WorkspaceRegistry { committed_revision = excluded.committed_revision", params![terminal_id.as_str(), result_json, sqlite_revision], )?; + if let Some(hook_state) = hook_state { + let applied_sequence = i64::try_from(hook_state.applied_sequence) + .context("agent hook sequence exceeds SQLite range")?; + tx.execute( + "INSERT INTO resource_agent_hook_state( + terminal_id, agent_session_id, applied_sequence, ended, committed_revision + ) VALUES(?1, ?2, ?3, ?4, ?5) + ON CONFLICT(terminal_id) DO UPDATE SET + agent_session_id = excluded.agent_session_id, + applied_sequence = excluded.applied_sequence, + ended = excluded.ended, + committed_revision = excluded.committed_revision", + params![ + terminal_id.as_str(), + hook_state.agent_session_id, + applied_sequence, + hook_state.ended, + sqlite_revision, + ], + )?; + } tx.execute( "UPDATE meta SET value = ?1 WHERE key = 'resource_revision'", [revision.to_string()], @@ -504,6 +1013,9 @@ impl WorkspaceRegistry { deltas, )?; prune_resource_mutations(&tx)?; + if let Some(sequence) = journal_sequence { + advance_agent_hook_apply_cursor_transaction(&tx, sequence)?; + } tx.commit()?; Ok(ResourcePatchCommit { revision, result: result.clone(), replayed: false }) } @@ -576,6 +1088,25 @@ impl WorkspaceRegistry { .map_err(Into::into) } + /// A missing in-memory surface can be a startup race only while the + /// durable terminal is launching, adopting, or running. Exited and + /// tombstoned terminals cannot recover a hook projection. + pub fn agent_hook_terminal_retryable( + &self, + public_id: &TerminalPublicId, + ) -> anyhow::Result { + let Some(host_id) = self.live_terminal_host_id(public_id)? else { + return Ok(false); + }; + let Some(terminal) = self.terminal_record(&host_id)? else { + return Ok(false); + }; + Ok(matches!( + terminal.lifecycle, + TerminalLifecycle::Launching | TerminalLifecycle::Adopting | TerminalLifecycle::Running + )) + } + pub fn resource_topology_snapshot(&self) -> anyhow::Result { let revision = current_resource_revision(&self.connection)?; let active_workspace = meta_value(&self.connection, "active_workspace_id")? @@ -1057,6 +1588,37 @@ impl WorkspaceRegistry { )?; u64::try_from(count).context("resource agent projection count is negative") } + + #[cfg(test)] + pub(crate) fn delete_agent_hook_state_for_test( + &mut self, + terminal_id: &crate::resource::TerminalPublicId, + ) -> anyhow::Result<()> { + self.connection.execute( + "DELETE FROM resource_agent_hook_state WHERE terminal_id = ?1", + [terminal_id.as_str()], + )?; + Ok(()) + } + + #[cfg(test)] + pub(crate) fn agent_hook_pending_retry_state_for_test( + &self, + producer_id: &str, + origin: &str, + idempotency_key: &str, + ) -> anyhow::Result> { + self.connection + .query_row( + "SELECT attempt, error + FROM resource_agent_hook_pending + WHERE producer_id = ?1 AND origin = ?2 AND idempotency_key = ?3", + params![producer_id, origin, idempotency_key], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional() + .map_err(Into::into) + } } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]