pulpod 0.0.42

Pulpo daemon — manages agent sessions via tmux/Docker
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

#[cfg(not(coverage))]
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    use clap::Parser;
    use tracing::info;

    let cli = pulpod::Cli::parse();

    match &cli.command {
        Some(pulpod::CliCommand::Mcp) => {
            // MCP mode: no tracing to stdout (would corrupt STDIO protocol)
            let server = pulpod::build_mcp_server(&cli).await?;
            pulpod::mcp::run_stdio(server).await?;
        }
        None => {
            // HTTP daemon mode
            let config = pulpod::config::load(&cli.config)?;
            let data_dir = std::path::PathBuf::from(config.data_dir());
            let _log_guard = pulpod::init_tracing(Some(&data_dir), config.node.log_retain_days)?;
            let (app, addr, shutdown_handle) = pulpod::build_app(&cli).await?;
            let listener = tokio::net::TcpListener::bind(&addr).await?;

            // After the shutdown signal fires, give in-flight streaming
            // connections (SSE, WebSocket) 3 seconds to close before forcing exit.
            let (shutdown_done_tx, shutdown_done_rx) = tokio::sync::oneshot::channel::<()>();
            let server = axum::serve(listener, app).with_graceful_shutdown(async move {
                shutdown_signal(shutdown_handle).await;
                let _ = shutdown_done_tx.send(());
            });

            tokio::select! {
                result = server => result?,
                () = async {
                    let _ = shutdown_done_rx.await;
                    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
                } => {
                    info!("Streaming connections still open — forcing shutdown");
                }
            }
            info!("pulpod shut down cleanly");
        }
    }

    Ok(())
}

/// Wait for SIGTERM or Ctrl+C, then signal all background loops to stop.
#[cfg(not(coverage))]
async fn shutdown_signal(handle: pulpod::ShutdownHandle) {
    use tracing::info;

    let ctrl_c = async {
        tokio::signal::ctrl_c()
            .await
            .expect("failed to install Ctrl+C handler");
    };

    #[cfg(unix)]
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("failed to install SIGTERM handler")
            .recv()
            .await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        () = ctrl_c => {},
        () = terminate => {},
    }

    info!("Shutdown signal received, stopping background tasks...");
    handle.shutdown();
}

// Dummy main for coverage builds — the real logic is tested via lib.rs
#[cfg(coverage)]
fn main() {}