agentmux_srv/bootstrap/
plumbing.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Split out of the single-file bootstrap module unchanged (SPEC_LARGE_FILE_MODULE_ANALYSIS_2026_09_30.md §4.3).
5
6use super::*;
7
8/// Output of [`spawn_reducer_plumbing`].
9pub struct ReducerPlumbing {
10    pub srv_state: Arc<tokio::sync::Mutex<state::State>>,
11    pub srv_events_tx: tokio::sync::broadcast::Sender<agentmux_common::ipc::Event>,
12    #[cfg(target_os = "windows")]
13    pub srv_event_log: Arc<event_log::EventLog>,
14}
15
16/// Phase E.2 / E.2c.2 — srv reducer plumbing, hoisted out of the
17/// (conditional) pipe-IPC bind block so HTTP/WS RPC handlers in
18/// dispatch_service can route through the reducer. State, event
19/// bus, event log, and persist subscriber all live unconditionally;
20/// the pipe IPC server is still conditional on
21/// `AGENTMUX_SRV_PIPE_PATH` being set (absent in `task dev` mode).
22///
23/// Bootstraps reducer state from SQLite, spawns the disk writer (forensic
24/// log of every reducer event), the persist subscriber (idempotent SQLite
25/// write-back), the MuxObjUpdate bridge, and the subagent-watcher block-delete
26/// cascade backstop.
27pub async fn spawn_reducer_plumbing(
28    mstore: &Arc<Store>,
29    event_bus: &Arc<EventBus>,
30    subagent_watcher: &Arc<backend::subagent_watcher::SubagentWatcher>,
31) -> ReducerPlumbing {
32    let mstore_for_persist = Arc::clone(mstore);
33    let srv_state = std::sync::Arc::new(tokio::sync::Mutex::new(state::State::default()));
34    let (srv_events_tx, _) =
35        tokio::sync::broadcast::channel::<agentmux_common::ipc::Event>(1024);
36    let srv_event_log = std::sync::Arc::new(event_log::EventLog::new(Some(
37        base::get_mux_data_dir().join("srv-events.log"),
38    )));
39
40    // Bootstrap reducer state from SQLite. Always runs (even in
41    // `task dev` where there's no pipe IPC server) so RPC handlers
42    // dispatching through the reducer see populated state.
43    persist::bootstrap_state_from_mstore(&srv_state, &mstore_for_persist).await;
44
45    // Spawn the disk writer (forensic log of every reducer event)
46    // and the persist subscriber (idempotent SQLite write-back).
47    let disk_writer_rx = srv_events_tx.subscribe();
48    let log_for_writer = std::sync::Arc::clone(&srv_event_log);
49    tokio::spawn(event_log::run_disk_writer(log_for_writer, disk_writer_rx));
50    let subscriber_rx = srv_events_tx.subscribe();
51    persist_subscriber::spawn_persist_subscriber(
52        subscriber_rx,
53        std::sync::Arc::clone(&mstore_for_persist),
54        std::sync::Arc::clone(&srv_state),
55    );
56
57    // Phase 1 of the MuxObjUpdate bridge: subscribe to srv_events_tx and
58    // translate workspace mutations into `waveobj:update` WS broadcasts.
59    // Fixes the workspace-rename reactivity gap where UpdateWorkspace
60    // returned `success_empty()` and the response loop had nothing to
61    // broadcast — see docs/specs/SPEC_OBJ_UPDATE_BRIDGE_2026-05-14.md.
62    //
63    // Watchdog: capture the JoinHandle and observe it from a sibling task
64    // so a panic in the bridge's loop scaffolding (vs. an inner
65    // dispatch_event panic, which is already caught per-event) is logged
66    // loudly. Without this, a silent bridge death would manifest as
67    // "renaming a workspace stopped propagating" with no log evidence.
68    // (Per ReAgent P2 follow-up on PR #852.)
69    let bridge_rx = srv_events_tx.subscribe();
70    let bridge_handle = server::mux_obj_bridge::spawn_mux_obj_bridge(
71        bridge_rx,
72        std::sync::Arc::clone(&mstore_for_persist),
73        std::sync::Arc::clone(event_bus),
74    );
75    tokio::spawn(async move {
76        match bridge_handle.await {
77            Ok(()) => tracing::info!(
78                target: "wave-obj-bridge",
79                "bridge task exited normally (events channel closed at srv shutdown)"
80            ),
81            Err(e) if e.is_panic() => tracing::error!(
82                target: "wave-obj-bridge",
83                "bridge task PANICKED at top level — frontend MOS will stop receiving updates until srv restart. Panic: {}",
84                e
85            ),
86            Err(e) => tracing::error!(
87                target: "wave-obj-bridge",
88                "bridge task terminated unexpectedly (non-panic JoinError): {}",
89                e
90            ),
91        }
92    });
93
94    // Block-delete cascade backstop for the subagent watcher: prunes a
95    // closed block's subagents/dispatches on Event::BlockDeleted/TabDeleted/
96    // WorkspaceDeleted, independent of whether the frontend's normal
97    // /agentmux/reactive/unregister teardown path fires for this close (see
98    // SubagentWatcher::prune_block's doc comment). Without this, closing an
99    // agent pane left a ghost row in the Swarm pane until srv restart.
100    let block_prune_rx = srv_events_tx.subscribe();
101    backend::subagent_watcher::spawn_block_prune_subscriber(subagent_watcher.clone(), block_prune_rx);
102
103    ReducerPlumbing {
104        srv_state,
105        srv_events_tx,
106        #[cfg(target_os = "windows")]
107        srv_event_log,
108    }
109}
110
111/// Assembles the final `AppState` from every bundle produced above, plus the
112/// handful of state pieces (auth session manager, install sessions, container
113/// manager, shell sessions, cron scheduler) that only ever existed inline in
114/// the `AppState` struct literal.
115pub fn build_app_state(
116    config: &config::Config,
117    version: String,
118    stores: Stores,
119    bg: BackgroundSubsystems,
120    net: &NetworkBundle,
121    reducer: &ReducerPlumbing,
122) -> AppState {
123    // Clone before move into AppState for cron_scheduler construction.
124    let shared_store_for_cron = stores.shared_store.clone();
125    // Live settings for the agent CPU-priority policy (process_tracker).
126    net.process_tracker.set_config(bg.config_watcher.clone());
127    let broker = bg.broker;
128
129    // Built before `container_manager` below so its constructor block can
130    // wire the two together (`set_dev_proxy_registry`) — `stop`/`remove`
131    // then clear an agent's routes when its container goes away. See
132    // docs/specs/SPEC_NATIVE_CONTAINER_DEV_PROXY_2026_09_19.md.
133    let dev_proxy = crate::backend::dev_proxy::DevProxyRegistry::new();
134
135    // The viewer (backend::viewer): its certificate lives in this instance's
136    // data directory, and its port is advertised on the fleet feed and the
137    // UDP replies while a viewer listener is up.
138    let viewer = crate::backend::viewer::ViewerService::new(Some(base::get_mux_data_dir()), &broker);
139    viewer.install_global();
140
141    let fleet_feed = Arc::new(
142        crate::backend::fleet_feed::FleetFeed::new(
143            net.hostname.clone(),
144            crate::backend::reactive::registry::local_channel_id(),
145            version.clone(),
146        )
147        .with_install_id(crate::backend::fleet_source::install_id())
148        .with_viewer_port(viewer.advertised_port_handle()),
149    );
150
151    AppState {
152        auth_key: config.auth_key.clone(),
153        lan_key: config.lan_key.clone(),
154        boot_id: Arc::from(uuid::Uuid::new_v4().to_string()),
155        version,
156        hostname: net.hostname.clone(),
157        app_path: config.app_path.clone(),
158        mstore: stores.mstore,
159        shared_store: stores.shared_store,
160        id_store: stores.id_store,
161        identity_store: stores.identity_store,
162        filestore: stores.filestore,
163        global_transcript_store: stores.global_transcript_store,
164        event_bus: bg.event_bus,
165        broker: broker.clone(),
166        reactive_handler: bg.reactive_handler,
167        poller: bg.poller,
168        config_watcher: bg.config_watcher,
169        messagebus: bg.messagebus,
170        subagent_watcher: bg.subagent_watcher,
171        history_service: bg.history_service,
172        lan_discovery: net.lan_discovery.clone(),
173        lan_listeners: net.lan_listeners.clone(),
174        fleet_feed,
175        viewer,
176        lsp_supervisor: net.lsp_supervisor.clone(),
177        local_web_url: net.local_web_url.clone(),
178        // Bounded request timeout: cross-instance reactive-inject forwards
179        // (Tier 2/2b/3) chain through this client, and an unbounded client
180        // combined with a forwarding cycle (two channels each holding a
181        // stale-but-PID-alive shared-registry entry pointing at the other)
182        // could otherwise hang a request indefinitely. The hop-count guard
183        // in server/reactive.rs bounds the CYCLE; this bounds each
184        // individual HOP (reagent P1 on PR #2350).
185        http_client: reqwest::Client::builder()
186            .timeout(std::time::Duration::from_secs(10))
187            .build()
188            .unwrap_or_else(|e| {
189                tracing::warn!(error = %e, "failed to build http_client with timeout, using default");
190                reqwest::Client::new()
191            }),
192        host_ipc: std::sync::Arc::new(tokio::sync::Mutex::new(None)),
193        host_reg_secret: config.host_reg_secret.clone(),
194        process_tracker: net.process_tracker.clone(),
195        process_broker: net.process_broker.clone(),
196        dock_snapshots: std::sync::Arc::new(crate::backend::dock_snapshot::DockSnapshotCache::new()),
197        pending_background_pids: std::sync::Arc::new(crate::backend::pending_background_pids::PendingBackgroundPids::new()),
198        narrated_events: Arc::new(crate::backend::narrated_events::NarratedEvents::new()),
199        // Phase E.2c.2 — reducer state + event bus exposed to HTTP/WS
200        // dispatch handlers. Workspace handlers route through the
201        // reducer and publish events to `srv_events_tx`; the persist
202        // subscriber writes back to SQLite asynchronously.
203        srv_state: std::sync::Arc::clone(&reducer.srv_state),
204        srv_events_tx: reducer.srv_events_tx.clone(),
205        // Phase E.5.5 — saga-id allocator. Seeded from
206        // `SagaLog::max_saga_id()` so restarts don't collide with
207        // prior runs' IDs. First new saga after restart gets
208        // `seed + 1`; on a fresh DB seed=0, first saga gets id 1.
209        saga_id_alloc: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(stores.saga_id_seed)),
210        saga_log: stores.saga_log,
211        auth_session_manager: {
212            let mgr = std::sync::Arc::new(crate::identity::auth_session::AuthSessionManager::new());
213            // Ends timed-out logins and drops finished ones, polled or not.
214            mgr.spawn_sweeper();
215            mgr
216        },
217        install_sessions: crate::server::install_handlers::InstallSessionRegistry::new(),
218        container_manager: {
219            // Self-healing: unlike a plain `Option<ContainerManager>` fixed at
220            // boot, `ContainerRuntimeHandle` retries the connect on demand, so
221            // a daemon that starts after this point is picked up by later
222            // calls without an app restart. See
223            // docs/retro/RETRO_DOCKER_DETECTION_DIVERGENCE_2026_07_04.md.
224            let handle = std::sync::Arc::new(
225                crate::backend::container::ContainerRuntimeHandle::connect_at_startup(),
226            );
227            handle.set_dev_proxy_registry(dev_proxy.clone());
228            let handle_check = handle.clone();
229            tokio::spawn(async move {
230                if handle_check.is_available().await {
231                    tracing::info!("Docker daemon available — container agent panes enabled");
232                } else {
233                    tracing::warn!(
234                        "Docker daemon not reachable at startup; container agent panes will \
235                         become available automatically once Docker is running"
236                    );
237                }
238            });
239            handle
240        },
241        dev_proxy,
242        shell_sessions: crate::backend::shell_node::ShellSessionRegistry::new(),
243        cron_scheduler: crate::backend::cron::CronScheduler::new(
244            shared_store_for_cron,
245            reqwest::Client::new(),
246            net.local_web_url.clone(),
247            config.auth_key.clone(),
248            Arc::clone(&broker),
249        ),
250        editor_file_watcher: bg.editor_file_watcher,
251        media_file_watcher: bg.media_file_watcher,
252        files_watcher: bg.files_watcher,
253        fs_watch_pool: bg.fs_watch_pool,
254    }
255}
256
257/// Phase E.1b — srv pipe IPC server. Bound when launcher passes
258/// `AGENTMUX_SRV_PIPE_PATH`; absent in `task dev` mode (no
259/// launcher in the loop).
260///
261/// Bind happens BEFORE the AGENTMUXSRV-ESTART line so the
262/// launcher knows the pipe is ready when host starts. Non-fatal
263/// if the bind fails — srv keeps running with HTTP/WS only.
264#[cfg(target_os = "windows")]
265pub fn bind_srv_pipe_ipc(
266    version: &str,
267    srv_state: std::sync::Arc<tokio::sync::Mutex<state::State>>,
268    srv_events_tx: tokio::sync::broadcast::Sender<agentmux_common::ipc::Event>,
269    srv_event_log: std::sync::Arc<event_log::EventLog>,
270) {
271    if let Ok(srv_pipe_path) = std::env::var("AGENTMUX_SRV_PIPE_PATH") {
272        if !srv_pipe_path.is_empty() {
273            match crate::srv_ipc::server::bind_first_pipe_instance(&srv_pipe_path) {
274                Ok(first_pipe) => {
275                    // Phase E.2c.2 — pipe IPC server reuses the
276                    // hoisted srv_state / events_tx / event_log so
277                    // pipe-originated commands and HTTP/WS-originated
278                    // commands mutate the same canonical state.
279                    let srv_ctx = crate::srv_ipc::ServerCtx {
280                        srv_pid: std::process::id(),
281                        srv_version: version.to_string(),
282                        state: srv_state,
283                        events_tx: srv_events_tx,
284                        event_log: srv_event_log,
285                    };
286                    let _srv_ipc_handle = crate::srv_ipc::run_srv_ipc_server(
287                        srv_pipe_path.clone(),
288                        first_pipe,
289                        srv_ctx,
290                    );
291                    tracing::info!(
292                        target: "srv-ipc",
293                        "[srv-ipc] bound + spawned on {}",
294                        srv_pipe_path
295                    );
296                }
297                Err(e) => {
298                    tracing::error!(
299                        target: "srv-ipc",
300                        "[srv-ipc] bind failed on {}: {} — srv runs without pipe IPC",
301                        srv_pipe_path,
302                        e
303                    );
304                }
305            }
306        }
307    }
308}
agentmux_srv/bootstrap/plumbing.rs

agentmux_srv/bootstrap/
plumbing.rs
