fix(multiplexer): restore session proxies on daemon restart

Fixes Docker container connectivity issue when daemon restarts by:
- Persisting port allocations from database to PortAllocator on startup
- Recreating session proxy listeners for running Docker sessions
- Enhanced reconciliation with orphaned proxy cleanup and auto-healing

Key changes:
- Added restore_allocations() to PortAllocator for state recovery
- Added restore_session_proxies() to ProxyManager with health checks
- Wired up restoration logic in daemon startup (server.rs)
- Enhanced reconcile() to clean up zombies and auto-heal missing proxies

This enables iterative development by allowing frequent daemon restarts
without breaking container network connectivity.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: shepherdjerred

packages/multiplexer/src/api/server.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
+use uuid::Uuid;
 
 use crate::backends::{DockerBackend, DockerProxyConfig};
 use crate::core::SessionManager;
@@ -97,6 +98,34 @@ pub async fn run_daemon_with_http(enable_proxy: bool, http_port: Option<u16>) ->
     };
     tracing::info!("Session manager initialized");
 
+    // Restore session proxies for active sessions (if proxy manager is enabled)
+    if let Some(ref pm) = proxy_manager {
+        tracing::info!("Restoring session proxies for active sessions...");
+
+        // Get all sessions from database
+        let sessions = manager.list_sessions().await;
+
+        // Extract port allocations for PortAllocator restoration
+        let port_allocations: Vec<(u16, Uuid)> = sessions
+            .iter()
+            .filter_map(|s| s.proxy_port.map(|port| (port, s.id)))
+            .collect();
+
+        // Restore port allocations in PortAllocator
+        if !port_allocations.is_empty() {
+            if let Err(e) = pm.port_allocator().restore_allocations(port_allocations).await {
+                tracing::error!("Failed to restore port allocations: {}", e);
+                tracing::warn!("New sessions may experience port conflicts");
+            }
+        }
+
+        // Restore session proxies (creates new proxy tasks on allocated ports)
+        if let Err(e) = pm.restore_session_proxies(&sessions).await {
+            tracing::error!("Failed to restore session proxies: {}", e);
+            tracing::warn!("Existing sessions may not have network connectivity");
+        }
+    }
+
     // Spawn both Unix socket and HTTP servers concurrently
     let unix_socket_future = run_unix_socket_server(Arc::clone(&manager));
 

packages/multiplexer/src/core/manager.rs

@@ -430,6 +430,63 @@ impl SessionManager {
 
                 if !exists {
                     report.missing_backends.push(session.id);
+
+                    // Clean up orphaned session proxy
+                    if session.backend == BackendType::Docker {
+                        if let Some(ref proxy_manager) = self.proxy_manager {
+                            tracing::info!(
+                                session_id = %session.id,
+                                "Destroying proxy for session with missing container"
+                            );
+                            let _ = proxy_manager.destroy_session_proxy(session.id).await;
+                        }
+                    }
+                } else {
+                    // Container exists but session is archived/failed - clean up zombie
+                    if matches!(session.status, SessionStatus::Archived | SessionStatus::Failed) {
+                        tracing::warn!(
+                            session_id = %session.id,
+                            status = ?session.status,
+                            "Found zombie container for non-active session, cleaning up"
+                        );
+
+                        match session.backend {
+                            BackendType::Docker => {
+                                let _ = self.docker.delete(backend_id).await;
+                                if let Some(ref proxy_manager) = self.proxy_manager {
+                                    let _ = proxy_manager.destroy_session_proxy(session.id).await;
+                                }
+                            }
+                            BackendType::Zellij => {
+                                let _ = self.zellij.delete(backend_id).await;
+                            }
+                        }
+                    }
+
+                    // Verify proxy exists for running Docker sessions
+                    if session.backend == BackendType::Docker
+                        && session.status == SessionStatus::Running
+                    {
+                        if let Some(ref proxy_manager) = self.proxy_manager {
+                            if let Some(port) = session.proxy_port {
+                                // Check if proxy is actually listening
+                                if tokio::net::TcpStream::connect(format!("127.0.0.1:{}", port))
+                                    .await
+                                    .is_err()
+                                {
+                                    tracing::warn!(
+                                        session_id = %session.id,
+                                        port = port,
+                                        "Session proxy not responding - attempting recreation"
+                                    );
+                                    // Attempt auto-recreation
+                                    let _ = proxy_manager
+                                        .restore_session_proxies(&[session.clone()])
+                                        .await;
+                                }
+                            }
+                        }
+                    }
                 }
             }
         }

packages/multiplexer/src/proxy/manager.rs

@@ -328,6 +328,138 @@ impl ProxyManager {
             anyhow::bail!("Session proxy not found for session {}", session_id)
         }
     }
+
+    /// Get reference to port allocator (for restoration)
+    ///
+    /// Exposed to allow daemon initialization code to restore port allocations
+    /// from database on startup.
+    pub fn port_allocator(&self) -> &Arc<PortAllocator> {
+        &self.port_allocator
+    }
+
+    /// Restore session proxies from database
+    ///
+    /// Called on daemon startup to recreate proxies for active sessions.
+    /// Only restores proxies for sessions with Running status.
+    ///
+    /// This enables containers to maintain network connectivity across daemon restarts
+    /// by recreating the proxy listeners on their allocated ports.
+    pub async fn restore_session_proxies(
+        &self,
+        sessions: &[crate::core::Session],
+    ) -> anyhow::Result<()> {
+        use crate::core::{BackendType, SessionStatus};
+
+        tracing::info!("Restoring session proxies from database...");
+
+        let mut restored = 0;
+        let mut skipped = 0;
+
+        for session in sessions {
+            // Only restore proxies for active Docker sessions with allocated ports
+            if session.backend != BackendType::Docker {
+                continue;
+            }
+
+            if session.status != SessionStatus::Running {
+                tracing::debug!(
+                    session_id = %session.id,
+                    status = ?session.status,
+                    "Skipping proxy restore for non-running session"
+                );
+                skipped += 1;
+                continue;
+            }
+
+            let Some(port) = session.proxy_port else {
+                tracing::debug!(
+                    session_id = %session.id,
+                    "Skipping proxy restore for session without allocated port"
+                );
+                skipped += 1;
+                continue;
+            };
+
+            // Create session proxy with the same port as before
+            let access_mode_lock = Arc::new(RwLock::new(session.access_mode));
+
+            let authority = self.ca.to_rcgen_authority()?;
+            let proxy = HttpAuthProxy::for_session(
+                port,
+                authority,
+                Arc::clone(&self.credentials),
+                Arc::clone(&self.audit_logger),
+                session.id,
+                Arc::clone(&access_mode_lock),
+            );
+
+            // Spawn proxy task
+            let session_id = session.id;
+            let task = tokio::spawn(async move {
+                if let Err(e) = proxy.run().await {
+                    tracing::error!(session_id = %session_id, "Session proxy error: {}", e);
+                }
+            });
+
+            // Wait for proxy to bind (health check)
+            let mut bound = false;
+            for attempt in 1..=10 {
+                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+                if tokio::net::TcpStream::connect(format!("127.0.0.1:{}", port))
+                    .await
+                    .is_ok()
+                {
+                    bound = true;
+                    break;
+                }
+                if attempt == 10 {
+                    tracing::warn!(
+                        port = port,
+                        session_id = %session.id,
+                        "Restored session proxy may not be ready (could not verify binding)"
+                    );
+                }
+            }
+
+            if !bound {
+                tracing::error!(
+                    session_id = %session.id,
+                    port = port,
+                    "Failed to restore session proxy - port may be in use"
+                );
+                task.abort();
+                skipped += 1;
+                continue;
+            }
+
+            // Store in session_proxies map
+            self.session_proxies.write().await.insert(
+                session.id,
+                SessionProxyHandle {
+                    port,
+                    access_mode: access_mode_lock,
+                    task,
+                },
+            );
+
+            tracing::info!(
+                session_id = %session.id,
+                port = port,
+                access_mode = ?session.access_mode,
+                "Restored session proxy"
+            );
+
+            restored += 1;
+        }
+
+        tracing::info!(
+            restored = restored,
+            skipped = skipped,
+            "Session proxy restoration complete"
+        );
+
+        Ok(())
+    }
 }
 
 impl Drop for ProxyManager {

packages/multiplexer/src/proxy/port_allocator.rs

@@ -59,6 +59,42 @@ impl PortAllocator {
     pub async fn get_session_id(&self, port: u16) -> Option<Uuid> {
         self.state.read().await.allocated.get(&port).copied()
     }
+
+    /// Restore port allocations from database
+    ///
+    /// Called on daemon startup to restore in-memory state from persistent storage.
+    /// Prevents port conflicts and maintains session-to-port mappings across restarts.
+    pub async fn restore_allocations(&self, allocations: Vec<(u16, Uuid)>) -> anyhow::Result<()> {
+        let mut state = self.state.write().await;
+
+        for (port, session_id) in allocations {
+            // Validate port is in our range
+            if port < Self::BASE_PORT || port >= Self::BASE_PORT + Self::MAX_SESSIONS {
+                tracing::warn!(
+                    port,
+                    session_id = %session_id,
+                    "Skipping invalid port allocation from database (out of range)"
+                );
+                continue;
+            }
+
+            state.allocated.insert(port, session_id);
+            tracing::debug!(port, session_id = %session_id, "Restored port allocation");
+        }
+
+        // Update next_port to avoid collisions with restored allocations
+        if let Some(&max_port) = state.allocated.keys().max() {
+            state.next_port = (max_port - Self::BASE_PORT + 1) % Self::MAX_SESSIONS;
+        }
+
+        tracing::info!(
+            count = state.allocated.len(),
+            "Restored {} port allocations from database",
+            state.allocated.len()
+        );
+
+        Ok(())
+    }
 }
 
 impl Default for PortAllocator {