fix(tui): order hook records and suppress done restores

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<TerminalPublicId, TerminalAgentRecord>,
+    pub(super) agent_hook_fences: HashMap<TerminalPublicId, super::HookFence>,
     pub(super) terminal_notifications: HashMap<TerminalPublicId, SurfaceNotification>,
     pub(super) notification_ledger: VecDeque<ResourceNotification>,
 }
@@ -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::<u64>().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));
+    }
 }

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::<Vec<_>>();
         agents.sort_by(|left, right| {

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,

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 })
 }
 

cmux-tui/crates/cmux-tui-core/src/workspace_registry/public_projection_store.rs

@@ -44,6 +44,14 @@ pub struct RegistryAgentProjection {
     pub source_session: Option<String>,
 }
 
+#[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<RegistryNotificationProjection>,
     pub agents: Vec<RegistryAgentProjection>,
+    pub(crate) agent_hook_states: Vec<RegistryAgentHookState>,
     pub terminal_defaults: Option<DefaultColors>,
     pub frontend_projections: Vec<FrontendProjection>,
 }
@@ -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<Vec<RegistryAgentProjection>> {
-        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<HashSet<TerminalPublicId>> {
@@ -338,6 +364,34 @@ impl WorkspaceRegistry {
         Ok(agents)
     }
 
+    fn durable_agent_hook_states(&self) -> anyhow::Result<Vec<RegistryAgentHookState>> {
+        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<Option<DefaultColors>> {
         let stored = self
             .connection