cmux-tui: journal reducer framework; derive the agent roster from the session journal

cmux-tui/crates/cmux-tui-cdp/src/client.rs

@@ -10,7 +10,6 @@ use std::sync::mpsc::{
 use std::sync::{Arc, Condvar, Mutex, Weak};
 use std::time::{Duration, Instant};
 
-use anyhow::Context;
 use base64::Engine;
 use serde_json::{Value, json};
 use tungstenite::client::IntoClientRequest;

cmux-tui/crates/cmux-tui-core/src/agent_hooks.rs

@@ -9,7 +9,7 @@ use crate::{
 
 pub const AGENT_HOOK_PRODUCER_ID: &str = "cmux_agent";
 pub const AGENT_HOOK_MANIFEST_VERSION: u32 = 1;
-const AGENT_HOOK_FORMAT: &str = "cmux.agent-hook.v1";
+pub(crate) const AGENT_HOOK_FORMAT: &str = "cmux.agent-hook.v1";
 const MAX_AGENT_SOURCE_BYTES: usize = 64;
 const MAX_NATIVE_EVENT_BYTES: usize = 128;
 const NORMALIZED_TEXT_BYTES: usize = 8 * 1024;
@@ -220,6 +220,11 @@ fn validate_agent_source(source: &str) -> anyhow::Result<()> {
             }),
         "agent source must contain 1 to {MAX_AGENT_SOURCE_BYTES} lowercase ASCII letters, digits, hyphens, or underscores"
     );
+    anyhow::ensure!(
+        source != crate::journal_reducers::SOCKET_REPORT_ADAPTER
+            && source != crate::journal_reducers::DETECTED_REPORT_ADAPTER,
+        "agent source is reserved for internal report adapters"
+    );
     Ok(())
 }
 
@@ -880,6 +885,13 @@ mod tests {
         assert_eq!(grok.kind, "agent.session.ended");
     }
 
+    #[test]
+    fn socket_adapter_name_is_reserved_for_internal_reports() {
+        let error =
+            agent_hook_journal_ingress("socket", "PermissionRequest", None, json!({})).unwrap_err();
+        assert!(error.to_string().contains("reserved for internal report adapters"));
+    }
+
     #[test]
     fn provider_envelopes_normalize_structural_fields_and_redact_message_data() {
         let native = json!({

cmux-tui/crates/cmux-tui-core/src/event_bus.rs

@@ -466,13 +466,15 @@ mod tests {
                 state: format!("one-{index}").into(),
                 source: "hook".into(),
                 session: None,
+                agent: None,
                 updated_at_ms: index,
             });
             broadcaster.emit(MuxEvent::AgentChanged {
                 surface: 2,
                 state: format!("two-{index}").into(),
                 source: "socket".into(),
                 session: Some("agent-session".into()),
+                agent: None,
                 updated_at_ms: index,
             });
         }
@@ -493,6 +495,7 @@ mod tests {
                 state,
                 source,
                 session: Some(session),
+                agent: None,
                 updated_at_ms: 9_999,
             } if state.as_ref() == "two-9999"
                 && source.as_ref() == "socket"
@@ -558,6 +561,7 @@ mod tests {
             state: "working".into(),
             source: "hook".into(),
             session: None,
+            agent: None,
             updated_at_ms: 1,
         });
         broadcaster.emit(MuxEvent::SurfaceExited(4));

cmux-tui/crates/cmux-tui-core/src/journal_ingress.rs

@@ -21,7 +21,7 @@ const JOURNAL_DURABLE_BATCH_BYTES: usize = 8 * 1024 * 1024;
 pub(crate) const TERMINAL_OUTPUT_INGRESS_BYTES: usize = 64 * 1024;
 const TERMINAL_OUTPUT_BATCH_BYTES: usize = 256 * 1024;
 const JOURNAL_TERMINAL_FAILURE_RETRY_ATTEMPTS: usize = 6;
-const JOURNAL_DURABLE_WAIT: Duration = Duration::from_secs(2);
+pub(crate) const JOURNAL_DURABLE_WAIT: Duration = Duration::from_secs(2);
 const JOURNAL_COMMIT_RESULT_WAIT: Duration = Duration::from_secs(1);
 const JOURNAL_WRITER_SHUTDOWN_WAIT: Duration = Duration::from_secs(1);
 const JOURNAL_SQLITE_RETRY_SLICE: Duration = Duration::from_millis(100);
@@ -1010,8 +1010,14 @@ fn run(mux: Weak<Mux>, receivers: JournalIngressReceivers) {
                             return;
                         }
                         if retryable_sqlite_error(&error) {
-                            let epoch = mux.journal_event_epoch();
-                            mux.wait_for_journal_event(epoch, delay.min(remaining));
+                            let journal = mux.shared_journal_handle();
+                            let epoch = journal.epoch();
+                            drop(mux);
+                            // Keep the retry wake independent from the owning
+                            // mux so a blocked writer cannot keep shutdown
+                            // alive. The journal kernel receives the same
+                            // commit and shutdown notifications.
+                            journal.wait(epoch, delay.min(remaining));
                             delay = (delay * 2).min(Duration::from_secs(1));
                             continue;
                         }
@@ -1028,8 +1034,10 @@ fn run(mux: Weak<Mux>, receivers: JournalIngressReceivers) {
                             if receivers.state.pause_nonretryable_failure_for_test() {
                                 continue;
                             }
-                            let epoch = mux.journal_event_epoch();
-                            mux.wait_for_journal_event(epoch, delay);
+                            let journal = mux.shared_journal_handle();
+                            let epoch = journal.epoch();
+                            drop(mux);
+                            journal.wait(epoch, delay);
                             delay = (delay * 2).min(Duration::from_secs(1));
                             continue;
                         } else if batch[0].completion.is_none() {