From f6389959548f7ade356cfaee11c610ce07daedae Mon Sep 17 00:00:00 2001 From: jamescui <54262737+Tinghecui@users.noreply.github.com> Date: Fri, 6 Nov 2026 19:26:49 +0000 Subject: [PATCH] refactor(startup): share the daemon boot between the desktop app and codeg-server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The desktop app's Tauri `setup` hook and `bin/codeg_server.rs` each carried their own copy of the daemon's startup — the bundled skill installs, the chat-dir GC, the delegation broker and its tool configs, the persisted settings, the delegation listener, the chat channels, the LifecycleSubscriber, the pet state mapper, the idle and office-watch sweeps, the automation and work-task engines, config sync, the Codex model list refresher — kept in step by "mirrors lib.rs setup" comments. Move those steps into `src-tauri/src/boot.rs`, which both hosts call in the same order. `boot` does not depend on Tauri: events go out through `EventEmitter`, and the native pieces (browser tabs, computer use) are handed in by the host. What only one host has stays where it was: the tray, windows, deep links, close behavior, pet panel, embedded web server and binary-cache startup sweep on the desktop; the listener, token, proxy-JWT, computer-use opt-in and upload staging purge in the server. No HTTP/WS API change. Co-Authored-By: Claude Opus 5.5 --- src-tauri/src/bin/codeg_server.rs | 415 +++------------------ src-tauri/src/boot.rs | 470 ++++++++++++++++++++++++ src-tauri/src/lib.rs | 582 ++++++------------------------ 3 files changed, 630 insertions(+), 837 deletions(-) create mode 100644 src-tauri/src/boot.rs diff --git a/src-tauri/src/bin/codeg_server.rs b/src-tauri/src/bin/codeg_server.rs index 7179953157f..710838d3b8f 100644 --- a/src-tauri/src/bin/codeg_server.rs +++ b/src-tauri/src/bin/codeg_server.rs @@ -316,28 +316,6 @@ async fn async_main( // config at build time, so this must run before the first one is constructed. codeg_lib::init_proxy_from_db(&db.conn).await; - // Reclaim orphaned chat scratch dirs (pre-send drafts that never bound to a - // conversation, plus dirs left behind by deleted chat conversations). - // Background, non-blocking; failures are logged but non-fatal. - { - let gc_conn = db.conn.clone(); - let gc_data_dir = data_dir.clone(); - tokio::spawn(async move { - match codeg_lib::commands::conversations::gc_orphan_chat_dirs_core( - &gc_conn, - &gc_data_dir, - ) - .await - { - Ok(n) if n > 0 => { - tracing::info!("[SERVER] chat-dir GC: reclaimed {n} orphan scratch dir(s)") - } - Ok(_) => {} - Err(err) => tracing::error!("[SERVER] chat-dir GC failed: {err}"), - } - }); - } - // Create shared broadcaster + internal ACP event bus. let broadcaster = Arc::new(WebEventBroadcaster::new()); let event_bus_metrics = Arc::new(codeg_lib::acp::EventBusMetrics::default()); @@ -358,21 +336,8 @@ async fn async_main( codeg_lib::cli_run::wire::RuntimeKind::Server, ); let cli_access = codeg_lib::cli_run::CliAccess::new(cli_runs.clone(), data_dir.clone()); - let ( - delegation_broker, - delegation_tokens, - delegation_socket_path, - feedback_config, - question_config, - session_info_config, - chat_authoring_config, - browser_tools_config, - computer_tools_config, - ) = codeg_lib::app_state::build_delegation_stack( - &connection_manager, - db.conn.clone(), - data_dir.clone(), - ); + let delegation = + codeg_lib::boot::DelegationStack::build(&connection_manager, &db.conn, &data_dir); let state = Arc::new(AppState { db, connection_manager, @@ -386,85 +351,43 @@ async fn async_main( workspace_transfer: Arc::new( codeg_lib::workspace_transfer::WorkspaceTransferManager::new_from_env(), ), - pet_state: pet_state_handle.clone(), - delegation_broker: delegation_broker.clone(), - delegation_tokens: delegation_tokens.clone(), - delegation_socket_path: delegation_socket_path.clone(), - feedback_config: feedback_config.clone(), - question_config: question_config.clone(), - session_info_config: session_info_config.clone(), - chat_authoring_config: chat_authoring_config.clone(), - browser_tools_config: browser_tools_config.clone(), - computer_tools_config: computer_tools_config.clone(), + pet_state: pet_state_handle, + delegation_broker: delegation.broker.clone(), + delegation_tokens: delegation.tokens.clone(), + delegation_socket_path: delegation.socket_path.clone(), + feedback_config: delegation.feedback.clone(), + question_config: delegation.question.clone(), + session_info_config: delegation.session_info.clone(), + chat_authoring_config: delegation.chat_authoring.clone(), + browser_tools_config: delegation.browser_tools.clone(), + computer_tools_config: delegation.computer_tools.clone(), computer_service: std::sync::OnceLock::new(), system_op_lock: codeg_lib::app_state::default_system_op_lock(), update_state: codeg_lib::app_state::default_update_state(), cli_runs, cli_access, }); - state - .connection_manager - .install_chat_channel(state.chat_channel_manager.clone_ref()); - // Logging phase 3: wire the emitter so the Logs viewer's live tail // (`logs://appended`) reaches WS clients. if let Some(hub) = codeg_lib::logging::hub::log_hub() { hub.set_emitter(state.emitter.clone()); } - // Apply persisted delegation settings (depth, enabled) before - // the listener starts accepting so even the first companion request - // sees the operator's configured behavior. Cancellation is handled - // out-of-band via MCP `notifications/cancelled` — no broker-side - // timeout to apply here. - codeg_lib::commands::delegation::apply_persisted_config(&state.db.conn, &delegation_broker) - .await; - // Same for the live-feedback enable flag, so the first companion launch - // sees the operator's configured behavior. - codeg_lib::commands::feedback::apply_persisted_feedback_config( - &state.db.conn, - &feedback_config, - ) - .await; - // Same for the ask-user-question enable flag. - codeg_lib::commands::question::apply_persisted_question_config( - &state.db.conn, - &question_config, - ) - .await; - // Same for the get-session-info enable flag. - codeg_lib::commands::session_info::apply_persisted_session_info_config( - &state.db.conn, - &session_info_config, - ) - .await; - // Same for the chat-authoring flags, so the first companion launch knows - // whether to advertise `create_automation` / `create_work_task`. - codeg_lib::commands::chat_authoring::apply_persisted_chat_authoring_config( - &state.db.conn, - &chat_authoring_config, - ) - .await; - // And the browser-tools switch, so the popover reports it truthfully. - // Server mode never advertises the group (there are no native tabs here), - // but the flag is one setting shared by both runtimes. - codeg_lib::commands::browser_tools::apply_persisted_browser_tools_config( - &state.db.conn, - &state.browser_tools_config, - ) - .await; - // And the computer-use switches: the popover reports them, and — where - // the person running this server lets it share the screen it runs on — - // the computer service below starts from them. - codeg_lib::commands::computer_tools::apply_persisted_computer_tools_config( + // The daemon, started the same way the desktop app starts it (`boot`, + // steps in its order); what sits between the steps here is what only the + // server has. + codeg_lib::boot::spawn_startup_housekeeping(&state.db.conn, &state.data_dir, &state.emitter); + + // Persisted terminal and tool settings into their live runtimes before + // the delegation listener accepts and before the server takes a request, + // so even the first companion launch sees the operator's configured + // behavior. + codeg_lib::boot::apply_persisted_settings( &state.db.conn, - &state.computer_tools_config, + &state.connection_manager, + &delegation, ) .await; - // The generative-UI skill's central copy, and the move from 0.35.0's - // `json-render`: before the server takes a request, so the move can never - // undo a change made on the settings page. - codeg_lib::commands::generative_ui::install_generative_ui_skill(&state.db.conn).await; // Computer use: only where whoever runs this server says so, by // CODEG_COMPUTER_USE — the server's web clients then share this // machine's windows with agents, and Stop them, from the panel. Nothing @@ -501,263 +424,43 @@ async fn async_main( install.ps1 to move it to %LOCALAPPDATA%\\codeg-server." ); } - // Before accepting connections: keep ACP model terminal fallbacks aligned - // with the same default-shell preference the built-in terminal uses, and - // seed the command-color opt-in that every launch env is built from. - let terminal_shell_config = state.connection_manager.terminal_shell_config(); - codeg_lib::commands::system_settings::apply_persisted_terminal_settings( - &state.db.conn, - &terminal_shell_config, - ) - .await; - - // Spawn the delegation listener so companion processes can round-trip - // through the broker. Path is PID-scoped, so the listener owns it for - // the lifetime of the process. - { - let listener = codeg_lib::acp::delegation::listener::DelegationListener::new( - delegation_broker, - delegation_tokens, - Arc::new(codeg_lib::acp::manager::ConnectionManagerParentLookup { - manager: Arc::new(state.connection_manager.clone_ref()), - }), - Arc::new(codeg_lib::acp::manager::ConnectionManagerFeedbackLookup { - manager: Arc::new(state.connection_manager.clone_ref()), - }), - Arc::new(codeg_lib::acp::manager::ConnectionManagerQuestionLookup { - manager: Arc::new(state.connection_manager.clone_ref()), - }), - Arc::new(codeg_lib::commands::session_info::DbSessionInfoLookup::new( - Arc::new(codeg_lib::db::AppDatabase { - conn: state.db.conn.clone(), - }), - )), - Arc::new(codeg_lib::work_task::EngineWorkTaskTools), - Arc::new(codeg_lib::commands::chat_authoring::DbChatAuthoring::new( - Arc::new(codeg_lib::db::AppDatabase { - conn: state.db.conn.clone(), - }), - state.emitter.clone(), - chat_authoring_config.clone(), - )), - // No native webviews in this process: what a web user sees in a - // "browser tab" is an iframe their own browser renders, which - // nothing here can reach. - Arc::new(codeg_lib::acp::browser_tools::NoBrowserTabs), - // The screen this server runs on, where it is let share it - // (`CODEG_COMPUTER_USE`); none otherwise. - match state.computer_service.get() { - Some(service) => Arc::new(codeg_lib::commands::computer::McpComputerTools::new( - service.clone(), - )) - as Arc, - None => Arc::new(codeg_lib::acp::computer_tools::NoComputerDesktop), - }, - Arc::new(codeg_lib::acp::result_tool::ConnectionManagerResultLookup { - manager: Arc::new(state.connection_manager.clone_ref()), - }), - ); - // Bind through the service handle rather than a bare `listener.run` - // spawn: it keeps the bind error and the accept-loop handle around, so - // the workspace status indicator can report why the broker socket is - // down and rebind it without restarting the server. - let service = codeg_lib::acp::delegation::service::DelegationService::new( - listener, - delegation_socket_path.clone(), - ); - codeg_lib::acp::delegation::service::install(service.clone()); - tokio::spawn(async move { - if let Err(e) = service.start().await { - tracing::error!("[delegation] listener failed to start: {e}"); - } - }); - } - - // Install bundled expert skills into the central store - // (`~/.codeg/skills/`). Runs in the background; failures are logged - // but non-fatal. - tokio::spawn(async move { - let report = codeg_lib::commands::experts::ensure_central_experts_installed().await; - if !report.errors.is_empty() { - tracing::error!( - "[Experts] install finished with {} error(s): {:?}", - report.errors.len(), - report.errors - ); - } else { - tracing::info!( - "[Experts] install ok: installed={} updated={} pending_review={}", - report.installed_count, - report.updated_count, - report.pending_user_review.len() - ); - } - }); - - // Install bundled scientific-research skills into the same central store. - // Runs in the background; failures are logged but non-fatal. - tokio::spawn(async move { - let report = codeg_lib::commands::science::ensure_central_science_installed().await; - if !report.errors.is_empty() { - tracing::error!( - "[Science] install finished with {} error(s): {:?}", - report.errors.len(), - report.errors - ); - } else { - tracing::info!( - "[Science] install ok: installed={} updated={} pending_review={}", - report.installed_count, - report.updated_count, - report.pending_user_review.len() - ); - } - }); - - // Start chat channel background tasks (event subscriber, command dispatcher, scheduler, auto-connect) - state - .chat_channel_manager - .start_background( - state.event_broadcaster.clone(), - state.acp_event_bus.clone(), - state.db.conn.clone(), - state.data_dir.clone(), - state.connection_manager.clone_ref(), - state.emitter.clone(), - ) - .await; - - // Spawn the LifecycleSubscriber for cross-connection DB writes. The - // broker is supplied so TurnComplete on a delegation child resolves the - // parent's pending `delegate_to_agent` tool_use_id and emits - // `DelegationCompleted`. - tokio::spawn(codeg_lib::lifecycle_subscriber_task( - state.db.conn.clone(), - state.connection_manager.clone_ref(), - state.acp_event_bus.clone(), - Some(state.delegation_broker.clone()), - )); - - // Spawn the desktop pet state mapper so server-mode browsers viewing - // /pet receive `pet://state` and `pet://oneshot` over the WebSocket - // bridge, just like the Tauri webview does in desktop mode. ACP events - // come through the typed bus; folder/app side-channels stay on the - // JSON broadcaster. - tokio::spawn(codeg_lib::pet_state_mapper::pet_state_subscriber_task( - state.acp_event_bus.clone(), - state.event_broadcaster.clone(), - state.emitter.clone(), - pet_state_handle, - )); - - // Spawn the idle sweep so connections abandoned without an explicit - // disconnect (e.g. browser tab closed, panic survivors) are reaped. - // Override the 60-second default via `CODEG_ACP_IDLE_TIMEOUT_SECS` - // (set to `0` to disable). - if let Some(idle_timeout) = codeg_lib::idle_timeout_from_env() { - tokio::spawn(codeg_lib::idle_sweep_task( - state.connection_manager.clone_ref(), - idle_timeout, - std::time::Duration::from_secs(codeg_lib::SWEEP_INTERVAL_SECS), - )); - } - - // Reclaim scratch directories lost track of mid-session. Deliberately NOT - // gated on `idle_timeout_from_env` like the sweep above: setting - // `CODEG_ACP_IDLE_TIMEOUT_SECS=0` disables idle disconnects, not disk - // reclamation. - tokio::spawn(codeg_lib::scratch_sweep_task()); - - // Office watch preview servers: reap dead children + ref0 stragglers. - if let Some(idle_timeout) = codeg_lib::office_watch::idle_timeout_from_env() { - tokio::spawn(codeg_lib::office_watch::office_watch_idle_sweep_task( - idle_timeout, - std::time::Duration::from_secs(codeg_lib::office_watch::SWEEP_INTERVAL_SECS), - )); - } - - // Automation engine (mirrors lib.rs setup): manual + scheduled fires, - // event-bus completion, reconcile, boot recovery. One per process. - if let Some(engine) = codeg_lib::automation::build_engine( - codeg_lib::db::AppDatabase { - conn: state.db.conn.clone(), + let daemon = codeg_lib::boot::Daemon { + conn: &state.db.conn, + data_dir: &state.data_dir, + connection_manager: &state.connection_manager, + chat_channels: &state.chat_channel_manager, + broadcaster: &state.event_broadcaster, + bus: &state.acp_event_bus, + emitter: &state.emitter, + pet_state: &state.pet_state, + delegation: &delegation, + }; + codeg_lib::boot::start_delegation_service( + &daemon, + // No native webviews in this process: what a web user sees in a + // "browser tab" is an iframe their own browser renders, which nothing + // here can reach. + Arc::new(codeg_lib::acp::browser_tools::NoBrowserTabs), + // The screen this server runs on, where it is let share it + // (`CODEG_COMPUTER_USE`); none otherwise. + match state.computer_service.get() { + Some(service) => Arc::new(codeg_lib::commands::computer::McpComputerTools::new( + service.clone(), + )) + as Arc, + None => Arc::new(codeg_lib::acp::computer_tools::NoComputerDesktop), }, - state.connection_manager.clone_ref(), - state.emitter.clone(), - state.acp_event_bus.clone(), - state.data_dir.clone(), - ) { - tokio::spawn(codeg_lib::automation::run_automation_engine(engine)); - } - - // Headless `codeg-cli` runs (mirrors lib.rs setup): the policy backstop - // here; the loopback listener and its discovery file once the server has - // its own socket (below), so a server that fails to start publishes none. + ); + // Awaited: every enabled channel has had its auto-connect attempt before + // the server binds. + codeg_lib::boot::start_chat_channels(&daemon).await; + let daemon_services = codeg_lib::boot::spawn_daemon_services(&daemon); + + // Headless `codeg-cli` runs: the policy backstop here; the loopback + // listener and its discovery file once the server has its own socket + // (below), so a server that fails to start publishes none. state.cli_runs.spawn_background(); - // Work-task engine (mirrors lib.rs setup): manual pipeline, event-bus - // settlement, merging git-truth recovery. One per process. - if let Some(engine) = codeg_lib::work_task::build_task_engine( - codeg_lib::db::AppDatabase { - conn: state.db.conn.clone(), - }, - state.connection_manager.clone_ref(), - state.emitter.clone(), - state.acp_event_bus.clone(), - state.data_dir.clone(), - ) { - tokio::spawn(codeg_lib::work_task::run_task_engine(engine)); - // Repository-panel auto-processing (mirrors lib.rs setup): only where - // the task engine runs, since it mints tasks unattended. - if let Some(processor) = codeg_lib::forge::auto::build( - codeg_lib::db::AppDatabase { - conn: state.db.conn.clone(), - }, - state.emitter.clone(), - ) { - tokio::spawn(codeg_lib::forge::auto::run(processor)); - } - } - - // Config-sync uploader (mirrors lib.rs setup): sleeps a minute, then - // compares the configuration's hash every interval and uploads only when - // it changed. Does nothing at all until a WebDAV endpoint is configured. - { - let db_for_sync = state.db.conn.clone(); - let emitter = std::sync::Arc::new(state.emitter.clone()); - tokio::spawn(async move { - codeg_lib::commands::config_sync::auto_sync::run_auto_sync_loop( - db_for_sync, - emitter, - codeg_lib::commands::config_sync::APP_VERSION.to_string(), - ) - .await; - }); - } - - // The Codex model list refresher (mirrors lib.rs setup). - if let Some(scheduler) = codeg_lib::acp::codex_catalog_refresh::build(state.emitter.clone()) { - tokio::spawn(codeg_lib::acp::codex_catalog_refresh::run(scheduler)); - } - - // Label worktree folders registered before aliases were seeded at creation - // with the branch they have checked out (mirrors lib.rs setup). Background; - // changed folders are broadcast, so a browser that already fetched its - // folder list still picks them up. - { - let db = codeg_lib::db::AppDatabase { - conn: state.db.conn.clone(), - }; - let emitter = state.emitter.clone(); - tokio::spawn(async move { - let n = - codeg_lib::commands::folders::backfill_worktree_folder_aliases(&emitter, &db).await; - if n > 0 { - tracing::info!("[folders] labeled {n} worktree folder(s) by branch"); - } - }); - } - // Sweep abandoned upload staging files from any prior run before // serving the first request. The quota log/validate ran earlier in // `main` so strict-mode misconfigurations abort before we touch @@ -867,7 +570,7 @@ async fn async_main( state.cli_access.stop().await; // Graceful shutdown: release any live office watch preview servers // (kill_on_drop is the backstop, but this frees their ports promptly). - codeg_lib::office_watch::stop_all_office_watches(); + daemon_services.stop(); ExitCode::SUCCESS } diff --git a/src-tauri/src/boot.rs b/src-tauri/src/boot.rs new file mode 100644 index 00000000000..6f955f012d2 --- /dev/null +++ b/src-tauri/src/boot.rs @@ -0,0 +1,470 @@ +//! The daemon's startup, shared by both runtimes that host one: the desktop +//! app (`lib.rs`, the Tauri `setup` hook) and `codeg-server` +//! (`bin/codeg_server.rs`). +//! +//! Before this module each of the two carried its own copy of the same steps, +//! kept in step by "mirrors lib.rs setup" comments. Everything a daemon does +//! whatever hosts it lives here now; what stays with the hosts is what only +//! one of them has: +//! +//! * the desktop app: the tray, windows, deep links, the updater, the close +//! behavior, the pet panel, native browser tabs, computer use on the screen +//! it runs on, the optional embedded web server, the startup sweep of the +//! ACP binary cache; +//! * `codeg-server`: binding the HTTP listener, the access token, proxy-JWT, +//! computer use only when asked for, the upload staging purge. +//! +//! Both also keep building their own `codeg-cli` run service, which each +//! constructs at a different point (managed Tauri state vs. a field of the +//! server's `AppState`). +//! +//! The steps are in the order a host calls them: +//! +//! 1. [`spawn_startup_housekeeping`]: background cleanups and installs. +//! 2. [`DelegationStack::build`]: the broker and the per-tool runtime configs. +//! 3. [`apply_persisted_settings`]: stored settings into those live configs +//! (and the terminal's), before anything can start an agent. +//! 4. The host's computer-use service, if any (it reads the configs from 3). +//! 5. [`start_delegation_service`]: the broker's socket for companions. +//! 6. [`start_chat_channels`]: chat-channel background tasks and auto-connect. +//! 7. [`spawn_daemon_services`]: the long-lived background services; the +//! returned [`DaemonServices`] stops what has to be stopped at exit. +//! +//! Nothing here touches Tauri: events go out through [`EventEmitter`], which +//! the desktop app builds over its `AppHandle` and the server over its web +//! broadcaster. Every function that spawns must be called inside a tokio +//! runtime context (the server's `#[tokio::main]` runtime, or the desktop's +//! `tauri::async_runtime::block_on`). + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use sea_orm::DatabaseConnection; + +use crate::acp::browser_tools::{BrowserToolAccess, BrowserToolsRuntimeConfig}; +use crate::acp::chat_authoring::ChatAuthoringRuntimeConfig; +use crate::acp::computer_tools::{ComputerToolAccess, ComputerToolsRuntimeConfig}; +use crate::acp::delegation::broker::DelegationBroker; +use crate::acp::delegation::listener::TokenRegistry; +use crate::acp::feedback::FeedbackRuntimeConfig; +use crate::acp::manager::ConnectionManager; +use crate::acp::question::QuestionRuntimeConfig; +use crate::acp::session_info::SessionInfoRuntimeConfig; +use crate::acp::InternalEventBus; +use crate::chat_channel::manager::ChatChannelManager; +use crate::db::AppDatabase; +use crate::pet_state_mapper::PetStateHandle; +use crate::web::event_bridge::{EventEmitter, WebEventBroadcaster}; + +/// The pieces of a running daemon both hosts own, borrowed from wherever the +/// host keeps them (Tauri managed state, or the server's `AppState`). +pub struct Daemon<'a> { + pub conn: &'a DatabaseConnection, + pub data_dir: &'a Path, + pub connection_manager: &'a ConnectionManager, + pub chat_channels: &'a ChatChannelManager, + pub broadcaster: &'a Arc, + pub bus: &'a Arc, + pub emitter: &'a EventEmitter, + pub pet_state: &'a PetStateHandle, + pub delegation: &'a DelegationStack, +} + +impl Daemon<'_> { + fn db(&self) -> AppDatabase { + AppDatabase { + conn: self.conn.clone(), + } + } +} + +/// Step 1. Background cleanups and installs that need nothing else running: +/// the periodic scratch-directory sweep, the bundled expert and science +/// skills, orphaned chat scratch directories, and branch labels for worktree +/// folders registered before those were seeded. All detached; failures are +/// logged and never fatal. +pub fn spawn_startup_housekeeping( + conn: &DatabaseConnection, + data_dir: &Path, + emitter: &EventEmitter, +) { + // Reclaim scratch directories this process loses track of mid-session. + // Its own timer on purpose: the ACP idle sweep is not spawned at all when + // `CODEG_ACP_IDLE_TIMEOUT_SECS=0`, and turning off idle disconnects must + // not also turn off disk reclamation on a machine leaking gigabytes per + // launch. + tokio::spawn(crate::scratch_sweep_task()); + + // Bundled expert skills into the central store (`~/.codeg/skills/`). + tokio::spawn(async move { + let report = crate::commands::experts::ensure_central_experts_installed().await; + if !report.errors.is_empty() { + tracing::error!( + "[Experts] install finished with {} error(s): {:?}", + report.errors.len(), + report.errors + ); + } else { + tracing::info!( + "[Experts] install ok: installed={} updated={} pending_review={}", + report.installed_count, + report.updated_count, + report.pending_user_review.len() + ); + } + }); + + // Bundled scientific-research skills into the same central store. + tokio::spawn(async move { + let report = crate::commands::science::ensure_central_science_installed().await; + if !report.errors.is_empty() { + tracing::error!( + "[Science] install finished with {} error(s): {:?}", + report.errors.len(), + report.errors + ); + } else { + tracing::info!( + "[Science] install ok: installed={} updated={} pending_review={}", + report.installed_count, + report.updated_count, + report.pending_user_review.len() + ); + } + }); + + // Orphaned chat scratch dirs: pre-send drafts that never bound to a + // conversation, plus dirs left behind by deleted chat conversations. + // Anything still in use is left for next startup. + { + let conn = conn.clone(); + let data_dir = data_dir.to_path_buf(); + tokio::spawn(async move { + match crate::commands::conversations::gc_orphan_chat_dirs_core(&conn, &data_dir).await { + Ok(n) if n > 0 => tracing::info!( + "[conversations] chat-dir GC: reclaimed {n} orphan scratch dir(s)" + ), + Ok(_) => {} + Err(err) => tracing::error!("[conversations] chat-dir GC failed: {err}"), + } + }); + } + + // Worktree folders registered before aliases were seeded at creation get + // the branch they have checked out, so the sidebar names them by branch + // rather than by their (long, derived) directory. Changed folders are + // broadcast, so a client that already fetched its folder list still picks + // them up. + { + let db = AppDatabase { conn: conn.clone() }; + let emitter = emitter.clone(); + tokio::spawn(async move { + let n = crate::commands::folders::backfill_worktree_folder_aliases(&emitter, &db).await; + if n > 0 { + tracing::info!("[folders] labeled {n} worktree folder(s) by branch"); + } + }); + } +} + +/// Step 2. The delegation broker, its token registry and socket path, and the +/// live runtime config of every tool group a companion can be offered. The +/// broker's listener is started later ([`start_delegation_service`]), once +/// the host holds what it borrows. +#[derive(Clone)] +pub struct DelegationStack { + pub broker: Arc, + pub tokens: Arc, + pub socket_path: PathBuf, + pub feedback: FeedbackRuntimeConfig, + pub question: QuestionRuntimeConfig, + pub session_info: SessionInfoRuntimeConfig, + pub chat_authoring: ChatAuthoringRuntimeConfig, + pub browser_tools: BrowserToolsRuntimeConfig, + pub computer_tools: ComputerToolsRuntimeConfig, +} + +impl DelegationStack { + pub fn build( + connection_manager: &ConnectionManager, + conn: &DatabaseConnection, + data_dir: &Path, + ) -> Self { + let ( + broker, + tokens, + socket_path, + feedback, + question, + session_info, + chat_authoring, + browser_tools, + computer_tools, + ) = crate::app_state::build_delegation_stack( + connection_manager, + conn.clone(), + data_dir.to_path_buf(), + ); + Self { + broker, + tokens, + socket_path, + feedback, + question, + session_info, + chat_authoring, + browser_tools, + computer_tools, + } + } +} + +/// Step 3. Stored settings into their live runtimes, before anything can +/// start an agent: the default shell and command-color flag every launch env +/// is built from, then the delegation depth/enable flags and each tool +/// group's switches (so even the first companion sees the configured +/// behavior), then the generative-UI skill's central copy (before any command +/// runs, so its one-time move from 0.35.0's `json-render` can never undo a +/// change made on the settings page). +pub async fn apply_persisted_settings( + conn: &DatabaseConnection, + connection_manager: &ConnectionManager, + delegation: &DelegationStack, +) { + crate::commands::system_settings::apply_persisted_terminal_settings( + conn, + &connection_manager.terminal_shell_config(), + ) + .await; + crate::commands::delegation::apply_persisted_config(conn, &delegation.broker).await; + crate::commands::feedback::apply_persisted_feedback_config(conn, &delegation.feedback).await; + crate::commands::question::apply_persisted_question_config(conn, &delegation.question).await; + crate::commands::session_info::apply_persisted_session_info_config( + conn, + &delegation.session_info, + ) + .await; + crate::commands::chat_authoring::apply_persisted_chat_authoring_config( + conn, + &delegation.chat_authoring, + ) + .await; + crate::commands::browser_tools::apply_persisted_browser_tools_config( + conn, + &delegation.browser_tools, + ) + .await; + crate::commands::computer_tools::apply_persisted_computer_tools_config( + conn, + &delegation.computer_tools, + ) + .await; + crate::commands::generative_ui::install_generative_ui_skill(conn).await; +} + +/// Step 5. Bind the delegation broker's socket so companion processes can +/// round-trip through it. `browser` and `computer` are the host's: native +/// browser tabs and the screen exist only where it has them. +/// +/// Bound through the service handle rather than a bare `listener.run` spawn: +/// it keeps the bind error and the accept-loop handle around, so the +/// workspace status indicator can report why the broker socket is down and +/// rebind it without a restart. +pub fn start_delegation_service( + daemon: &Daemon<'_>, + browser: Arc, + computer: Arc, +) { + use crate::acp::manager::{ + ConnectionManagerFeedbackLookup, ConnectionManagerParentLookup, + ConnectionManagerQuestionLookup, + }; + + let manager = || Arc::new(daemon.connection_manager.clone_ref()); + let listener = crate::acp::delegation::listener::DelegationListener::new( + daemon.delegation.broker.clone(), + daemon.delegation.tokens.clone(), + Arc::new(ConnectionManagerParentLookup { manager: manager() }), + Arc::new(ConnectionManagerFeedbackLookup { manager: manager() }), + Arc::new(ConnectionManagerQuestionLookup { manager: manager() }), + Arc::new(crate::commands::session_info::DbSessionInfoLookup::new( + Arc::new(daemon.db()), + )), + Arc::new(crate::work_task::EngineWorkTaskTools), + Arc::new(crate::commands::chat_authoring::DbChatAuthoring::new( + Arc::new(daemon.db()), + daemon.emitter.clone(), + daemon.delegation.chat_authoring.clone(), + )), + browser, + computer, + Arc::new(crate::acp::result_tool::ConnectionManagerResultLookup { manager: manager() }), + ); + let service = crate::acp::delegation::service::DelegationService::new( + listener, + daemon.delegation.socket_path.clone(), + ); + crate::acp::delegation::service::install(service.clone()); + tokio::spawn(async move { + if let Err(e) = service.start().await { + tracing::error!("[delegation] listener failed to start: {e}"); + } + }); +} + +/// Step 6. Chat channels: hand the manager to the connection manager, then +/// start the event subscribers, command dispatcher, daily-report scheduler, +/// and auto-connect every enabled channel. The hand-over comes first: a +/// `/new` that landed before it would write its live ACP title against an +/// `install_chat_channel` that hasn't happened yet, and skip the topic rename +/// for good (the later reconciliation passes only sync titles their own +/// conditional UPDATE wrote, and this one already converged). +/// +/// Resolves once auto-connect has been attempted for every channel. The +/// desktop app spawns it (no window waits on a network round-trip); the +/// server awaits it before it binds. +pub fn start_chat_channels( + daemon: &Daemon<'_>, +) -> impl std::future::Future + Send + 'static { + daemon + .connection_manager + .install_chat_channel(daemon.chat_channels.clone_ref()); + let channels = daemon.chat_channels.clone_ref(); + let broadcaster = daemon.broadcaster.clone(); + let bus = daemon.bus.clone(); + let conn = daemon.conn.clone(); + let data_dir = daemon.data_dir.to_path_buf(); + let manager = daemon.connection_manager.clone_ref(); + let emitter = daemon.emitter.clone(); + async move { + channels + .start_background(broadcaster, bus, conn, data_dir, manager, emitter) + .await; + } +} + +/// Step 7. The long-lived background services: +/// +/// * the LifecycleSubscriber (cross-connection DB writes off the emit hot +/// path; resolves a delegation child's pending `delegate_to_agent` with the +/// broker, which is why it comes after [`DelegationStack::build`]); +/// * the pet state mapper (`pet://state` / `pet://oneshot`, to a webview or a +/// `/pet` browser alike); +/// * the ACP idle sweep and the office-watch preview sweep; +/// * the automation engine; the work-task engine and, only beside it, the +/// repository panel's auto-processing (it mints tasks unattended). One of +/// each per process; +/// * the config-sync uploader; +/// * the Codex model list refresher. +pub fn spawn_daemon_services(daemon: &Daemon<'_>) -> DaemonServices { + // `subscribe()` runs synchronously inside the task constructor, before + // the future is returned, so subscribe-before-spawn holds. + tokio::spawn(crate::acp::lifecycle_subscriber_task( + daemon.conn.clone(), + daemon.connection_manager.clone_ref(), + daemon.bus.clone(), + Some(daemon.delegation.broker.clone()), + )); + + // ACP events come through the typed bus; folder/app side-channels stay on + // the JSON broadcaster. + tokio::spawn(crate::pet_state_mapper::pet_state_subscriber_task( + daemon.bus.clone(), + daemon.broadcaster.clone(), + daemon.emitter.clone(), + daemon.pet_state.clone(), + )); + + // Connections abandoned without an explicit disconnect (window or tab + // closed without teardown, panic survivors) are reaped. Override the + // 60-second default via `CODEG_ACP_IDLE_TIMEOUT_SECS` (`0` disables). + if let Some(idle_timeout) = crate::acp::idle_timeout_from_env() { + tokio::spawn(crate::acp::idle_sweep_task( + daemon.connection_manager.clone_ref(), + idle_timeout, + std::time::Duration::from_secs(crate::acp::SWEEP_INTERVAL_SECS), + )); + } + + // Office watch preview servers: reap dead children + ref0 stragglers + // (live previews are never swept). Override via + // `CODEG_OFFICE_WATCH_IDLE_TIMEOUT_SECS` (`0` disables). + if let Some(idle_timeout) = crate::office_watch::idle_timeout_from_env() { + tokio::spawn(crate::office_watch::office_watch_idle_sweep_task( + idle_timeout, + std::time::Duration::from_secs(crate::office_watch::SWEEP_INTERVAL_SECS), + )); + } + + // Automation engine: drives manual + scheduled fires, settles runs off + // the event bus, reconciles, and recovers on boot. + if let Some(engine) = crate::automation::build_engine( + daemon.db(), + daemon.connection_manager.clone_ref(), + daemon.emitter.clone(), + daemon.bus.clone(), + daemon.data_dir.to_path_buf(), + ) { + tokio::spawn(crate::automation::run_automation_engine(engine)); + } + + // Work-task engine: drives the todo→…→done pipeline, settles runs off the + // event bus, recovers merges from git truth on boot. The repository + // panel's auto-processing mints tasks unattended, so it runs only where + // the task engine does — the one process holding this data directory's + // task lock. + if let Some(engine) = crate::work_task::build_task_engine( + daemon.db(), + daemon.connection_manager.clone_ref(), + daemon.emitter.clone(), + daemon.bus.clone(), + daemon.data_dir.to_path_buf(), + ) { + tokio::spawn(crate::work_task::run_task_engine(engine)); + if let Some(processor) = crate::forge::auto::build(daemon.db(), daemon.emitter.clone()) { + tokio::spawn(crate::forge::auto::run(processor)); + } + } + + // Config-sync uploader: sleeps a minute, then compares the + // configuration's hash every interval and uploads only when it changed; + // reads its settings every tick, so toggling sync takes effect without a + // restart. Does nothing at all until a WebDAV endpoint is configured. + { + let conn = daemon.conn.clone(); + let emitter = Arc::new(daemon.emitter.clone()); + tokio::spawn(async move { + crate::commands::config_sync::auto_sync::run_auto_sync_loop( + conn, + emitter, + crate::commands::config_sync::APP_VERSION.to_string(), + ) + .await; + }); + } + + // Keeps the Codex model list codeg published current (a provider list + // past its age, a codex upgraded outside codeg) and clears what a release + // left behind. + if let Some(scheduler) = crate::acp::codex_catalog_refresh::build(daemon.emitter.clone()) { + tokio::spawn(crate::acp::codex_catalog_refresh::run(scheduler)); + } + + DaemonServices { _private: () } +} + +/// Handle on what [`spawn_daemon_services`] started that outlives a dropped +/// task: child processes, which no runtime shutdown reaches. +#[must_use = "call `stop` at exit, or the office-watch preview servers outlive the daemon"] +#[derive(Clone)] +pub struct DaemonServices { + _private: (), +} + +impl DaemonServices { + /// Release the office-watch preview servers (`kill_on_drop` is the + /// backstop, but this frees their ports promptly). Idempotent. + pub fn stop(&self) { + crate::office_watch::stop_all_office_watches(); + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index e5900e5aef0..e6664c993e3 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -28,6 +28,7 @@ pub mod app_state; pub mod appimage_env; pub mod automation; pub mod backgrounds; +pub mod boot; /// Built-in browser. Only its wire types and its grant rules compile in server /// mode — see `browser/mod.rs` for why those two, and only those two. pub mod browser; @@ -173,7 +174,10 @@ mod tauri_app { if let Some(tm) = app.try_state::() { tm.kill_all(); } - crate::office_watch::stop_all_office_watches(); + // Office watch previews (`boot::DaemonServices`). + if let Some(services) = app.try_state::() { + services.stop(); + } if let Some(cm) = app.try_state::() { tauri::async_runtime::block_on(cm.disconnect_all()); } @@ -925,231 +929,59 @@ mod tauri_app { }); }); - // Reclaim scratch directories this process loses track of - // mid-session. Its own timer on purpose: the ACP idle sweep is - // not spawned at all when `CODEG_ACP_IDLE_TIMEOUT_SECS=0`, and - // turning off idle disconnects must not also turn off disk - // reclamation on a machine leaking gigabytes per launch. - tauri::async_runtime::spawn(crate::scratch_sweep_task()); - - // Install bundled expert skills into the central store - // (`~/.codeg/skills/`). Runs in the background and does - // not block startup; failures are logged but non-fatal. - tauri::async_runtime::spawn(async move { - let report = crate::commands::experts::ensure_central_experts_installed().await; - if !report.errors.is_empty() { - tracing::error!( - "[Experts] install finished with {} error(s): {:?}", - report.errors.len(), - report.errors - ); - } else { - tracing::info!( - "[Experts] install ok: installed={} updated={} pending_review={}", - report.installed_count, - report.updated_count, - report.pending_user_review.len() - ); - } + // The daemon, started the same way `codeg-server` starts it + // (`boot`, steps in its order); what sits between the steps + // here is what only the desktop app has. Every `boot` call + // that spawns runs inside `block_on`, which gives it Tauri's + // tokio runtime. + let emitter = web::event_bridge::EventEmitter::Tauri(app.handle().clone()); + tauri::async_runtime::block_on(async { + crate::boot::spawn_startup_housekeeping(&db.conn, &effective_data_dir, &emitter) }); - // Install bundled scientific-research skills into the same - // central store (`~/.codeg/skills/`). Background, non-blocking; - // failures are logged but non-fatal. - tauri::async_runtime::spawn(async move { - let report = crate::commands::science::ensure_central_science_installed().await; - if !report.errors.is_empty() { - tracing::error!( - "[Science] install finished with {} error(s): {:?}", - report.errors.len(), - report.errors - ); - } else { - tracing::info!( - "[Science] install ok: installed={} updated={} pending_review={}", - report.installed_count, - report.updated_count, - report.pending_user_review.len() - ); - } - }); - - // Reclaim orphaned chat scratch dirs (pre-send drafts that never - // bound to a conversation, plus dirs left behind by deleted chat - // conversations). Background, non-blocking; failures are logged - // but non-fatal — anything still in use is left for next startup. - { - let gc_conn = app.state::().conn.clone(); - let gc_data_dir = effective_data_dir.clone(); - tauri::async_runtime::spawn(async move { - match crate::commands::conversations::gc_orphan_chat_dirs_core( - &gc_conn, - &gc_data_dir, - ) - .await - { - Ok(n) if n > 0 => tracing::info!( - "[conversations] chat-dir GC: reclaimed {n} orphan scratch dir(s)" - ), - Ok(_) => {} - Err(err) => { - tracing::error!("[conversations] chat-dir GC failed: {err}") - } - } - }); - } + // Delegation broker + per-tool configs, managed alongside the + // other per-resource states so commands (Tauri + web) resolve + // them by type. Built before the LifecycleSubscriber + // (`spawn_daemon_services`), which takes the broker. + let delegation = crate::boot::DelegationStack::build( + &app.state::(), + &db.conn, + &effective_data_dir, + ); + app.manage(delegation.broker.clone()); + app.manage(delegation.tokens.clone()); + app.manage(delegation.feedback.clone()); + app.manage(delegation.question.clone()); + app.manage(delegation.session_info.clone()); + app.manage(delegation.chat_authoring.clone()); + app.manage(delegation.browser_tools.clone()); + // The desktop app always serves computer use (its service + // starts below): said now, before any session could be + // started and ask whether to offer the tools. + delegation.computer_tools.mark_served(); + app.manage(delegation.computer_tools.clone()); + app.manage(crate::commands::delegation::DelegationSocketPath( + delegation.socket_path.clone(), + )); - // Push the persisted terminal settings into their live + // Persisted terminal and tool settings into their live // runtimes BEFORE any background task that can spawn an agent - // (the chat-channel dispatcher below is one). For the shell - // handle a late seed would only ever be a narrow race, since - // it is read at terminal-create time; the command-color flag - // is read while BUILDING a launch's env, so a late seed there - // would silently hand the first agent of the run the wrong - // one. "Seeded before anything can connect" covers both, and - // matches server startup, which seeds before it binds. - { - let db_for_shell = app.state::().conn.clone(); - let shell_config = app.state::().terminal_shell_config(); - tauri::async_runtime::block_on(async move { - crate::commands::system_settings::apply_persisted_terminal_settings( - &db_for_shell, - &shell_config, - ) - .await; - }); - } + // (the chat-channel dispatcher below is one), matching server + // startup, which seeds before it binds. + tauri::async_runtime::block_on(crate::boot::apply_persisted_settings( + &db.conn, + &app.state::(), + &delegation, + )); // Seed the close-behavior atomic. `CloseRequested` is a // synchronous callback that reads the cache, not the database, // so an unseeded cache would serve "ask" to a user who pinned // a choice months ago. Blocking here keeps that impossible // even for a close in the first moments after launch. - { - let db_for_close = app.state::().conn.clone(); - tauri::async_runtime::block_on(async move { - crate::commands::system_settings::apply_persisted_close_behavior( - &db_for_close, - ) - .await; - }); - } - - // Start the config-sync uploader. Background and detached: - // it sleeps a minute before its first hash compare, reads its - // settings every tick (so toggling sync in the UI takes effect - // without a restart), and does nothing at all until the user - // configures a WebDAV endpoint. - { - let db_for_sync = app.state::().conn.clone(); - let emitter = std::sync::Arc::new(web::event_bridge::EventEmitter::Tauri( - app.handle().clone(), - )); - tauri::async_runtime::spawn(async move { - crate::commands::config_sync::auto_sync::run_auto_sync_loop( - db_for_sync, - emitter, - env!("CARGO_PKG_VERSION").to_string(), - ) - .await; - }); - } - - // Keeps the Codex model list codeg published current (a provider - // list past its age, a codex upgraded outside codeg) and clears - // what a release left behind. One per process; mirrored in - // `bin/codeg_server.rs`. - if let Some(scheduler) = crate::acp::codex_catalog_refresh::build( - web::event_bridge::EventEmitter::Tauri(app.handle().clone()), - ) { - tauri::async_runtime::spawn(crate::acp::codex_catalog_refresh::run(scheduler)); - } - - // Label worktree folders registered before aliases were seeded at - // creation with the branch they have checked out, so the sidebar - // names them by branch rather than by their (long, derived) - // directory. Background, non-blocking; changed folders are - // broadcast, so a client that already fetched its folder list - // still picks them up. - { - let db = db::AppDatabase { - conn: app.state::().conn.clone(), - }; - let emitter = web::event_bridge::EventEmitter::Tauri(app.handle().clone()); - tauri::async_runtime::spawn(async move { - let n = - crate::commands::folders::backfill_worktree_folder_aliases(&emitter, &db) - .await; - if n > 0 { - tracing::info!("[folders] labeled {n} worktree folder(s) by branch"); - } - }); - } - - // Hand the chat-channel manager to the connection manager - // BEFORE the chat background tasks below start accepting - // messages: a `/new` that lands first would write its live ACP - // title against an `install_chat_channel` that hasn't happened - // yet, and skip the topic rename for good (the later - // reconciliation passes only sync titles their own conditional - // UPDATE wrote, and this one already converged). - { - let cm = app.state::(); - let ccm = app.state::(); - cm.install_chat_channel(ccm.clone_ref()); - } - - // Start chat channel background tasks - { - let ccm = app.state::(); - let broadcaster = - app.state::>(); - let db_conn = app.state::().conn.clone(); - let data_dir = effective_data_dir.clone(); - let ccm_ref = ccm.clone_ref(); - let br = broadcaster.inner().clone(); - let bus = app - .state::>() - .inner() - .clone(); - let cm = app.state::().clone_ref(); - let emitter = web::event_bridge::EventEmitter::Tauri(app.handle().clone()); - tauri::async_runtime::spawn(async move { - ccm_ref - .start_background(br, bus, db_conn, data_dir, cm, emitter) - .await; - }); - } - - // Spawn the desktop pet state mapper: subscribes to ACP events - // (typed envelopes via the in-process bus) AND folder/app - // side-channel notifications (JSON via the broadcaster), and - // emits `pet://state` whenever the aggregated pet state - // changes. The renderer in the floating pet window listens - // for these events to drive its sprite animation row. - { - let bus = app - .state::>() - .inner() - .clone(); - let broadcaster = app - .state::>() - .inner() - .clone(); - let emitter = web::event_bridge::EventEmitter::Tauri(app.handle().clone()); - let pet_state_handle = app - .state::() - .inner() - .clone(); - tauri::async_runtime::spawn( - crate::pet_state_mapper::pet_state_subscriber_task( - bus, - broadcaster, - emitter, - pet_state_handle, - ), - ); - } + tauri::async_runtime::block_on( + crate::commands::system_settings::apply_persisted_close_behavior(&db.conn), + ); // Spawn the pet panel active-session aggregator: rebuilds the // `PetSessionsPayload` (running/waiting/error counts + per- @@ -1181,204 +1013,61 @@ mod tauri_app { ); } - // Delegation broker + UDS listener. Built from the managed - // ConnectionManager + DB so spawn / depth-lookup work against - // live state. Managed alongside the existing per-resource - // states so commands (Tauri + web) can resolve them by type. - // MUST run before the LifecycleSubscriber spawn below so the - // broker handle is available to it. - let broker_for_lifecycle = { - let cm_state = app.state::(); - let db_conn = app.state::().conn.clone(); - let ( - broker, - tokens, - socket_path, - feedback_config, - question_config, - session_info_config, - chat_authoring_config, - browser_tools_config, - computer_tools_config, - ) = crate::app_state::build_delegation_stack( - &cm_state, - db_conn.clone(), - effective_data_dir.clone(), - ); - app.manage(broker.clone()); - app.manage(tokens.clone()); - app.manage(feedback_config.clone()); - app.manage(question_config.clone()); - app.manage(session_info_config.clone()); - app.manage(chat_authoring_config.clone()); - app.manage(browser_tools_config.clone()); - // The desktop app always serves computer use (its service - // starts below): said now, before any session could be - // started and ask whether to offer the tools. - computer_tools_config.mark_served(); - app.manage(computer_tools_config.clone()); - app.manage(crate::commands::delegation::DelegationSocketPath( - socket_path.clone(), - )); - - // Push persisted settings into the broker + feedback + question - // + session-info config before listener accept. - let broker_for_init = broker.clone(); - let db_for_init = db_conn.clone(); - let feedback_for_init = feedback_config.clone(); - let question_for_init = question_config.clone(); - let session_info_for_init = session_info_config.clone(); - let chat_authoring_for_init = chat_authoring_config.clone(); - let browser_tools_for_init = browser_tools_config.clone(); - let computer_tools_for_init = computer_tools_config.clone(); - tauri::async_runtime::block_on(async move { - delegation_commands::apply_persisted_config( - &db_for_init, - &broker_for_init, - ) - .await; - crate::commands::feedback::apply_persisted_feedback_config( - &db_for_init, - &feedback_for_init, - ) - .await; - crate::commands::question::apply_persisted_question_config( - &db_for_init, - &question_for_init, - ) - .await; - crate::commands::session_info::apply_persisted_session_info_config( - &db_for_init, - &session_info_for_init, - ) - .await; - crate::commands::chat_authoring::apply_persisted_chat_authoring_config( - &db_for_init, - &chat_authoring_for_init, - ) - .await; - crate::commands::browser_tools::apply_persisted_browser_tools_config( - &db_for_init, - &browser_tools_for_init, - ) - .await; - crate::commands::computer_tools::apply_persisted_computer_tools_config( - &db_for_init, - &computer_tools_for_init, - ) - .await; - // The generative-UI skill's central copy, and the move - // from 0.35.0's `json-render`: here, before any command - // runs, so the move can never undo a change made on - // the settings page. - crate::commands::generative_ui::install_generative_ui_skill( - &db_for_init, - ) - .await; - }); - - // Computer use: the helper, the window table, and the two - // background duties (end every grant when the group is - // switched off; end grants that go unused). Started after - // the persisted settings are applied, so the watcher's - // first view of the switch is the stored one. - let computer_service = crate::commands::computer::ComputerService::start( - crate::commands::computer::ComputerHost::Desktop(app.handle().clone()), - computer_tools_config.clone(), - ); - app.manage(computer_service.clone()); - - let listener_broker = broker.clone(); - let listener = crate::acp::delegation::listener::DelegationListener::new( - listener_broker, - tokens, - std::sync::Arc::new( - crate::acp::manager::ConnectionManagerParentLookup { - manager: std::sync::Arc::new(cm_state.clone_ref()), - }, - ), - std::sync::Arc::new( - crate::acp::manager::ConnectionManagerFeedbackLookup { - manager: std::sync::Arc::new(cm_state.clone_ref()), - }, - ), - std::sync::Arc::new( - crate::acp::manager::ConnectionManagerQuestionLookup { - manager: std::sync::Arc::new(cm_state.clone_ref()), - }, - ), - std::sync::Arc::new( - crate::commands::session_info::DbSessionInfoLookup::new( - std::sync::Arc::new(db::AppDatabase { - conn: db_conn.clone(), - }), - ), - ), - std::sync::Arc::new(crate::work_task::EngineWorkTaskTools), - std::sync::Arc::new( - crate::commands::chat_authoring::DbChatAuthoring::new( - std::sync::Arc::new(db::AppDatabase { - conn: db_conn.clone(), - }), - crate::web::event_bridge::EventEmitter::Tauri( - app.handle().clone(), - ), - chat_authoring_config.clone(), - ), - ), - std::sync::Arc::new( - crate::commands::browser::McpBrowserTools::new( + // Computer use: the helper, the window table, and the two + // background duties (end every grant when the group is + // switched off; end grants that go unused). Started after + // the persisted settings are applied, so the watcher's + // first view of the switch is the stored one. + let computer_service = crate::commands::computer::ComputerService::start( + crate::commands::computer::ComputerHost::Desktop(app.handle().clone()), + delegation.computer_tools.clone(), + ); + app.manage(computer_service.clone()); + + let pet_state = app + .state::() + .inner() + .clone(); + let broadcaster = app + .state::>() + .inner() + .clone(); + let bus = app + .state::>() + .inner() + .clone(); + let daemon_services = { + let connection_manager = app.state::(); + let chat_channels = app.state::(); + let daemon = crate::boot::Daemon { + conn: &db.conn, + data_dir: &effective_data_dir, + connection_manager: &connection_manager, + chat_channels: &chat_channels, + broadcaster: &broadcaster, + bus: &bus, + emitter: &emitter, + pet_state: &pet_state, + delegation: &delegation, + }; + tauri::async_runtime::block_on(async { + crate::boot::start_delegation_service( + &daemon, + std::sync::Arc::new(crate::commands::browser::McpBrowserTools::new( app.handle().clone(), - browser_tools_config.clone(), - ), - ), - std::sync::Arc::new(crate::commands::computer::McpComputerTools::new( - computer_service, - )), - std::sync::Arc::new( - crate::acp::result_tool::ConnectionManagerResultLookup { - manager: std::sync::Arc::new(cm_state.clone_ref()), - }, - ), - ); - // Bind through the service handle rather than a bare - // `listener.run` spawn: it keeps the bind error and the - // accept-loop handle around, which is what lets the - // workspace status indicator report why the broker socket - // is down and rebind it without an app restart. - let service = crate::acp::delegation::service::DelegationService::new( - listener, - socket_path, - ); - crate::acp::delegation::service::install(service.clone()); - tauri::async_runtime::spawn(async move { - if let Err(e) = service.start().await { - tracing::error!("[delegation] listener failed to start: {e}"); - } - }); - broker + delegation.browser_tools.clone(), + )), + std::sync::Arc::new(crate::commands::computer::McpComputerTools::new( + computer_service, + )), + ); + // Spawned rather than awaited: no window waits on + // auto-connecting the chat channels. + tauri::async_runtime::spawn(crate::boot::start_chat_channels(&daemon)); + crate::boot::spawn_daemon_services(&daemon) + }) }; - - // Spawn the LifecycleSubscriber: persists cross-connection DB state - // (currently `external_id` on conversation rows when SessionStarted fires) - // off the emit hot path. `subscribe()` runs synchronously inside - // `lifecycle_subscriber_task` before the future is returned, so the - // subscribe-before-spawn invariant holds. The setup callback runs - // outside any tokio runtime, so we use `tauri::async_runtime::spawn`. - { - let db_conn = app.state::().conn.clone(); - let cm = app.state::().clone_ref(); - let bus = app - .state::>() - .inner() - .clone(); - tauri::async_runtime::spawn(crate::acp::lifecycle_subscriber_task( - db_conn, - cm, - bus, - Some(broker_for_lifecycle), - )); - } + app.manage(daemon_services); // Headless `codeg-cli` runs: the service and its loopback // access. Managed before the web service may start below, @@ -1429,75 +1118,6 @@ mod tauri_app { Err(err) => tracing::error!("[WEB] failed to load auto-start config: {err}"), } - // Spawn the idle sweep so connections abandoned without an - // explicit disconnect (e.g. window/tab closed without - // teardown, panic survivors) are reaped. Override the - // 60-second default via `CODEG_ACP_IDLE_TIMEOUT_SECS` - // (set to `0` to disable). - if let Some(idle_timeout) = crate::acp::idle_timeout_from_env() { - let cm = app.state::().clone_ref(); - tauri::async_runtime::spawn(crate::acp::idle_sweep_task( - cm, - idle_timeout, - std::time::Duration::from_secs(crate::acp::SWEEP_INTERVAL_SECS), - )); - } - - // Office watch preview servers: reap dead children + ref0 - // stragglers (live previews are never swept). Override via - // `CODEG_OFFICE_WATCH_IDLE_TIMEOUT_SECS` (`0` disables). - if let Some(idle_timeout) = crate::office_watch::idle_timeout_from_env() { - tauri::async_runtime::spawn(crate::office_watch::office_watch_idle_sweep_task( - idle_timeout, - std::time::Duration::from_secs(crate::office_watch::SWEEP_INTERVAL_SECS), - )); - } - - // Automation engine: drives manual + scheduled fires, settles - // runs off the event bus, reconciles, and recovers on boot. One - // per process; mirrored in `bin/codeg_server.rs`. - if let Some(engine) = crate::automation::build_engine( - crate::db::AppDatabase { - conn: app.state::().conn.clone(), - }, - app.state::().clone_ref(), - crate::web::event_bridge::EventEmitter::Tauri(app.handle().clone()), - app.state::>() - .inner() - .clone(), - effective_data_dir.clone(), - ) { - tauri::async_runtime::spawn(crate::automation::run_automation_engine(engine)); - } - - // Work-task engine: drives the todo→…→done pipeline, settles - // runs off the event bus, recovers merges from git truth on - // boot. One per process; mirrored in `bin/codeg_server.rs`. - if let Some(engine) = crate::work_task::build_task_engine( - crate::db::AppDatabase { - conn: app.state::().conn.clone(), - }, - app.state::().clone_ref(), - crate::web::event_bridge::EventEmitter::Tauri(app.handle().clone()), - app.state::>() - .inner() - .clone(), - effective_data_dir.clone(), - ) { - tauri::async_runtime::spawn(crate::work_task::run_task_engine(engine)); - // The repository panel's auto-processing mints tasks - // unattended, so it runs only where the task engine does — - // the one process holding this data directory's task lock. - if let Some(processor) = crate::forge::auto::build( - crate::db::AppDatabase { - conn: app.state::().conn.clone(), - }, - crate::web::event_bridge::EventEmitter::Tauri(app.handle().clone()), - ) { - tauri::async_runtime::spawn(crate::forge::auto::run(processor)); - } - } - // OS `codeg://` URLs. Register the listener after the DB is // live so a warm-start click can look the conversation up. // Cold-start URLs are also read here and baked into the main