diff --git a/migrations/20260305000002_worker_directory.sql b/migrations/20260305000002_worker_directory.sql new file mode 100644 index 000000000..f06b23621 --- /dev/null +++ b/migrations/20260305000002_worker_directory.sql @@ -0,0 +1,3 @@ +-- Persist the working directory for opencode workers so that idle workers +-- can be resumed into the correct directory after a restart. +ALTER TABLE worker_runs ADD COLUMN directory TEXT; diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 3dda516d1..113e691f1 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -2250,6 +2250,7 @@ impl Channel { worker_type, &self.deps.agent_id, *interactive, + None, ); } ProcessEvent::WorkerStatus { diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index d6df99e11..667186199 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -466,6 +466,12 @@ pub async fn spawn_opencode_worker_from_state( directory: &str, interactive: bool, ) -> std::result::Result { + if !interactive { + return Err(AgentError::Other(anyhow::anyhow!( + "OpenCode workers must be interactive" + ))); + } + check_worker_limit(state).await?; ensure_dispatch_readiness(state, "opencode_worker"); let task = task.into(); @@ -487,6 +493,17 @@ pub async fn spawn_opencode_worker_from_state( let server_pool = rc.opencode_server_pool.load().clone(); + // Prevent multiple opencode workers on the same directory. + server_pool + .claim_directory(&directory) + .await + .map_err(AgentError::Other)?; + + // Clone for the release call in the async worker task. + let release_pool = server_pool.clone(); + let release_directory = directory.clone(); + let persist_directory = directory.clone(); + let oc_secrets_store = state.deps.runtime_config.secrets.load().as_ref().clone(); let worker = if interactive { @@ -504,10 +521,11 @@ pub async fn spawn_opencode_worker_from_state( .write() .await .insert(worker_id, input_tx); - match &oc_secrets_store { + let worker = match &oc_secrets_store { Some(store) => worker.with_secrets_store(store.clone()), None => worker, - } + }; + worker.with_sqlite_pool(state.deps.sqlite_pool.clone()) } else { let worker = crate::opencode::OpenCodeWorker::new( Some(state.channel_id.clone()), @@ -517,10 +535,11 @@ pub async fn spawn_opencode_worker_from_state( server_pool, state.deps.event_tx.clone(), ); - match &oc_secrets_store { + let worker = match &oc_secrets_store { Some(store) => worker.with_secrets_store(store.clone()), None => worker, - } + }; + worker.with_sqlite_pool(state.deps.sqlite_pool.clone()) }; let worker_id = worker.id; @@ -540,7 +559,12 @@ pub async fn spawn_opencode_worker_from_state( Some(state.channel_id.clone()), oc_secrets_store, async move { - let result = worker.run().await.map_err(SpacebotError::from)?; + let result = worker.run().await.map_err(SpacebotError::from); + + // Release the directory claim regardless of success or failure. + release_pool.release_directory(&release_directory).await; + + let result = result?; // Persist the transcript built from SSE events so the worker detail // view can show the full conversation (text + tool calls + results). @@ -592,6 +616,12 @@ pub async fn spawn_opencode_worker_from_state( }) .ok(); + // Persist the directory so idle workers can be resumed into the correct + // directory after a restart. + state + .process_run_logger + .log_worker_directory(worker_id, &persist_directory); + tracing::info!(worker_id = %worker_id, task = %task, interactive, "OpenCode worker spawned"); Ok(worker_id) @@ -704,6 +734,236 @@ where }) } +/// Resume an idle interactive worker into a channel's state after restart. +/// +/// Loads the prior transcript, creates a resumed worker (builtin or opencode), +/// registers it into the channel's worker_inputs/worker_handles/status_block, +/// and spawns the follow-up loop. Returns `Ok(worker_id)` on success, or +/// an error string if the worker couldn't be resumed. +pub async fn resume_idle_worker_into_state( + state: &ChannelState, + idle_worker: &crate::conversation::history::IdleWorkerRow, +) -> std::result::Result { + let worker_id: WorkerId = idle_worker + .id + .parse::() + .map_err(|error| format!("invalid worker ID '{}': {error}", idle_worker.id))?; + + match idle_worker.worker_type.as_str() { + "opencode" => { + let session_id = idle_worker + .opencode_session_id + .as_deref() + .ok_or("opencode worker has no session_id, cannot resume")?; + + let rc = &state.deps.runtime_config; + let opencode_config = rc.opencode.load(); + if !opencode_config.enabled { + return Err("OpenCode workers are not enabled".into()); + } + + let directory = idle_worker + .directory + .as_deref() + .map(std::path::PathBuf::from) + .unwrap_or_else(|| rc.workspace_dir.clone()); + let server_pool = rc.opencode_server_pool.load().clone(); + + let result = crate::opencode::OpenCodeWorker::resume_interactive( + worker_id, + Some(state.channel_id.clone()), + state.deps.agent_id.clone(), + &idle_worker.task, + directory, + server_pool, + state.deps.event_tx.clone(), + session_id.to_string(), + idle_worker.transcript.clone(), + ) + .await; + + let (mut worker, input_tx) = result.ok_or_else(|| { + "failed to reconnect to OpenCode session (server dead or session expired)" + .to_string() + })?; + + // Apply builder chain (same as spawn_opencode_worker_from_state). + let oc_secrets_store = state.deps.runtime_config.secrets.load().as_ref().clone(); + if let Some(store) = &oc_secrets_store { + worker = worker.with_secrets_store(store.clone()); + } + worker = worker.with_sqlite_pool(state.deps.sqlite_pool.clone()); + + state + .worker_inputs + .write() + .await + .insert(worker_id, input_tx); + + let worker_span = tracing::info_span!( + "worker.resume", + worker_id = %worker_id, + channel_id = %state.channel_id, + task = %idle_worker.task, + worker_type = "opencode", + ); + let sqlite_pool = state.deps.sqlite_pool.clone(); + let handle = spawn_worker_task( + worker_id, + state.deps.event_tx.clone(), + state.deps.agent_id.clone(), + Some(state.channel_id.clone()), + oc_secrets_store, + async move { + let result = worker.run().await.map_err(SpacebotError::from)?; + // Persist final transcript. + if !result.transcript.is_empty() { + let blob = crate::conversation::worker_transcript::serialize_steps( + &result.transcript, + ); + let tool_calls = result.tool_calls; + let wid = worker_id.to_string(); + let pool = sqlite_pool.clone(); + tokio::spawn(async move { + if let Err(error) = sqlx::query( + "UPDATE worker_runs SET transcript = ?, tool_calls = ? WHERE id = ?", + ) + .bind(&blob) + .bind(tool_calls) + .bind(&wid) + .execute(&pool) + .await + { + tracing::warn!(%error, worker_id = wid, "failed to persist OpenCode transcript"); + } + }); + } + Ok::(result.result_text) + } + .instrument(worker_span), + ); + + state.worker_handles.write().await.insert(worker_id, handle); + + let opencode_task = format!("[opencode] {}", idle_worker.task); + { + let mut status = state.status_block.write().await; + status.add_worker(worker_id, &opencode_task, false, true); + } + + state + .deps + .event_tx + .send(ProcessEvent::WorkerStarted { + agent_id: state.deps.agent_id.clone(), + worker_id, + channel_id: Some(state.channel_id.clone()), + task: opencode_task, + worker_type: "opencode".into(), + interactive: true, + }) + .ok(); + + tracing::info!(worker_id = %worker_id, task = %idle_worker.task, "OpenCode worker resumed"); + Ok(worker_id) + } + _ => { + // Builtin worker resume: deserialize transcript blob back into + // Rig message history so the LLM can continue the conversation. + let prior_history = if let Some(blob) = &idle_worker.transcript { + let steps = crate::conversation::worker_transcript::deserialize_transcript(blob) + .map_err(|error| format!("failed to deserialize transcript: {error}"))?; + crate::conversation::worker_transcript::transcript_to_history(&steps) + } else { + return Err("no transcript blob to restore history from".into()); + }; + + let rc = &state.deps.runtime_config; + let prompt_engine = rc.prompts.load(); + let sandbox_enabled = state.deps.sandbox.mode_enabled(); + let sandbox_containment_active = state.deps.sandbox.containment_active(); + let sandbox_read_allowlist = state.deps.sandbox.prompt_read_allowlist(); + let sandbox_write_allowlist = state.deps.sandbox.prompt_write_allowlist(); + let secrets_guard = rc.secrets.load(); + let tool_secret_names = match (*secrets_guard).as_ref() { + Some(store) => store.tool_secret_names(), + None => Vec::new(), + }; + let system_prompt = prompt_engine + .render_worker_prompt( + &rc.instance_dir.display().to_string(), + &rc.workspace_dir.display().to_string(), + sandbox_enabled, + sandbox_containment_active, + sandbox_read_allowlist, + sandbox_write_allowlist, + &tool_secret_names, + ) + .map_err(|error| format!("failed to render worker prompt: {error}"))?; + let browser_config = (**rc.browser_config.load()).clone(); + let brave_search_key = (**rc.brave_search_key.load()).clone(); + + let (worker, input_tx) = Worker::resume_interactive( + worker_id, + Some(state.channel_id.clone()), + &idle_worker.task, + &system_prompt, + state.deps.clone(), + browser_config, + state.screenshot_dir.clone(), + brave_search_key, + state.logs_dir.clone(), + prior_history, + ); + + state + .worker_inputs + .write() + .await + .insert(worker_id, input_tx); + + let worker_span = tracing::info_span!( + "worker.resume", + worker_id = %worker_id, + channel_id = %state.channel_id, + task = %idle_worker.task, + ); + let secrets_store = state.deps.runtime_config.secrets.load().as_ref().clone(); + let handle = spawn_worker_task( + worker_id, + state.deps.event_tx.clone(), + state.deps.agent_id.clone(), + Some(state.channel_id.clone()), + secrets_store, + worker.run().instrument(worker_span), + ); + + state.worker_handles.write().await.insert(worker_id, handle); + + { + let mut status = state.status_block.write().await; + status.add_worker(worker_id, &idle_worker.task, false, true); + } + + state + .deps + .event_tx + .send(ProcessEvent::WorkerStarted { + agent_id: state.deps.agent_id.clone(), + worker_id, + channel_id: Some(state.channel_id.clone()), + task: idle_worker.task.clone(), + worker_type: "builtin".into(), + interactive: true, + }) + .ok(); + + tracing::info!(worker_id = %worker_id, task = %idle_worker.task, "builtin worker resumed"); + Ok(worker_id) + } + } +} + /// Expand a leading `~` or `~/` in a path to the user's home directory. /// /// LLMs consistently produce tilde-prefixed paths because that's what appears diff --git a/src/agent/channel_history.rs b/src/agent/channel_history.rs index 72bfaaccd..27c455449 100644 --- a/src/agent/channel_history.rs +++ b/src/agent/channel_history.rs @@ -416,8 +416,11 @@ pub(crate) fn event_is_for_channel(event: &ProcessEvent, channel_id: &ChannelId) channel_id: event_channel, .. } => event_channel.as_ref() == Some(channel_id), - ProcessEvent::OpenCodeSessionCreated { .. } - | ProcessEvent::OpenCodePartUpdated { .. } + ProcessEvent::OpenCodeSessionCreated { + channel_id: event_channel, + .. + } => event_channel.as_ref() == Some(channel_id), + ProcessEvent::OpenCodePartUpdated { .. } | ProcessEvent::StatusUpdate { .. } | ProcessEvent::TaskUpdated { .. } => false, } diff --git a/src/agent/cortex.rs b/src/agent/cortex.rs index 8cb091745..62c593ef3 100644 --- a/src/agent/cortex.rs +++ b/src/agent/cortex.rs @@ -2491,6 +2491,7 @@ async fn pickup_one_ready_task(deps: &AgentDeps, logger: &CortexLogger) -> anyho "task", &deps.agent_id, false, + None, ); let task_store = deps.task_store.clone(); @@ -3613,6 +3614,7 @@ mod tests { ProcessEvent::OpenCodeSessionCreated { agent_id: Arc::from("agent"), worker_id, + channel_id: Some(channel_id.clone()), session_id: "session-1".to_string(), port: 19898, }, diff --git a/src/agent/worker.rs b/src/agent/worker.rs index 5e1d67c9a..b0f40706d 100644 --- a/src/agent/worker.rs +++ b/src/agent/worker.rs @@ -62,6 +62,8 @@ pub struct Worker { /// Status updates. pub status_tx: watch::Sender, pub status_rx: watch::Receiver, + /// Prior conversation history for resumed workers (set by `resume_interactive`). + pub prior_history: Option>, } impl Worker { @@ -103,6 +105,7 @@ impl Worker { logs_dir, status_tx, status_rx, + prior_history: None, } } @@ -159,6 +162,54 @@ impl Worker { (worker, input_tx) } + /// Resume an interactive worker that was idle at shutdown. + /// + /// Instead of running the initial task, skips directly to the follow-up + /// loop with the prior conversation history restored from the transcript + /// blob. The worker keeps its original ID so the DB row stays linked. + #[allow(clippy::too_many_arguments)] + pub fn resume_interactive( + existing_id: WorkerId, + channel_id: Option, + task: impl Into, + system_prompt: impl Into, + deps: AgentDeps, + browser_config: BrowserConfig, + screenshot_dir: PathBuf, + brave_search_key: Option, + logs_dir: PathBuf, + prior_history: Vec, + ) -> (Self, mpsc::Sender) { + let (input_tx, input_rx) = mpsc::channel(32); + let mut worker = Self::build( + channel_id, + task, + system_prompt, + deps, + browser_config, + screenshot_dir, + brave_search_key, + logs_dir, + Some(input_rx), + ); + // Reuse the original worker ID so DB row stays linked. + worker.id = existing_id; + // Rebuild the hook so it publishes events under the correct worker ID + // (Self::build creates it with a fresh random ID). + let process_id = ProcessId::Worker(existing_id); + worker.hook = SpacebotHook::new( + worker.deps.agent_id.clone(), + process_id, + ProcessType::Worker, + worker.channel_id.clone(), + worker.deps.event_tx.clone(), + ); + worker.state = WorkerState::WaitingForInput; + // Stash the prior history so `run_follow_up_loop()` can pick it up. + worker.prior_history = Some(prior_history); + (worker, input_tx) + } + /// Check if the worker can transition to a new state. pub fn can_transition_to(&self, target: WorkerState) -> bool { use WorkerState::*; @@ -229,97 +280,117 @@ impl Worker { .tool_server_handle(worker_tool_server) .build(); - // Fresh history for the worker (no channel context) - let mut history = Vec::new(); + // If this is a resumed worker, load the prior history into `history` + // (not `compacted_history`) so the LLM sees it as conversation context + // on the next follow-up call. + let resuming = self.prior_history.is_some(); + let mut history = self.prior_history.take().unwrap_or_default(); let mut compacted_history = Vec::new(); + if resuming { + tracing::info!( + worker_id = %self.id, + prior_messages = history.len(), + "resuming interactive worker with prior history" + ); + self.hook.send_status("resumed — waiting for input"); + self.hook.send_worker_idle(); + } + // Run the initial task in segments with compaction checkpoints + // (skipped entirely for resumed workers). let mut prompt = self.task.clone(); let mut segments_run = 0; let mut overflow_retries = 0; - let result = loop { - segments_run += 1; + let result = if resuming { + // For resumed workers, synthesize a "result" from the task + // since the original initial result was already relayed. + String::new() + } else { + loop { + segments_run += 1; - match self - .hook - .prompt_with_tool_nudge_retry(&agent, &mut history, &prompt) - .await - { - Ok(response) => { - break response; - } - Err(rig::completion::PromptError::MaxTurnsError { .. }) => { - overflow_retries = 0; + match self + .hook + .prompt_with_tool_nudge_retry(&agent, &mut history, &prompt) + .await + { + Ok(response) => { + break response; + } + Err(rig::completion::PromptError::MaxTurnsError { .. }) => { + overflow_retries = 0; - if segments_run >= MAX_SEGMENTS { - tracing::warn!( + if segments_run >= MAX_SEGMENTS { + tracing::warn!( + worker_id = %self.id, + segments = segments_run, + "worker hit max segments, returning partial result" + ); + self.hook.send_status("done (max segments)"); + break crate::agent::extract_last_assistant_text(&history) + .unwrap_or_else(|| { + "Worker reached maximum segments without a final response." + .to_string() + }); + } + + self.maybe_compact_history(&mut compacted_history, &mut history) + .await; + prompt = + "Continue where you left off. Do not repeat completed work.".into(); + self.hook + .send_status(format!("working (segment {segments_run})")); + + tracing::debug!( worker_id = %self.id, - segments = segments_run, - "worker hit max segments, returning partial result" - ); - self.hook.send_status("done (max segments)"); - break crate::agent::extract_last_assistant_text(&history).unwrap_or_else( - || { - "Worker reached maximum segments without a final response." - .to_string() - }, + segment = segments_run, + history_len = history.len(), + "continuing to next segment" ); } - - self.maybe_compact_history(&mut compacted_history, &mut history) - .await; - prompt = "Continue where you left off. Do not repeat completed work.".into(); - self.hook - .send_status(format!("working (segment {segments_run})")); - - tracing::debug!( - worker_id = %self.id, - segment = segments_run, - history_len = history.len(), - "continuing to next segment" - ); - } - Err(rig::completion::PromptError::PromptCancelled { reason, .. }) => { - self.state = WorkerState::Failed; - self.hook.send_status("cancelled"); - self.write_failure_log(&history, &format!("cancelled: {reason}")); - self.persist_transcript(&compacted_history, &history); - tracing::info!(worker_id = %self.id, %reason, "worker cancelled"); - return Err(crate::error::AgentError::Cancelled { reason }.into()); - } - Err(error) if is_context_overflow_error(&error.to_string()) => { - overflow_retries += 1; - if overflow_retries > MAX_OVERFLOW_RETRIES { + Err(rig::completion::PromptError::PromptCancelled { reason, .. }) => { self.state = WorkerState::Failed; - self.hook.send_status("failed"); - self.write_failure_log(&history, &format!("context overflow after {MAX_OVERFLOW_RETRIES} compaction attempts: {error}")); - self.persist_transcript(&compacted_history, &history); - tracing::error!(worker_id = %self.id, %error, "worker context overflow unrecoverable"); - return Err(crate::error::AgentError::Other(error.into()).into()); + self.hook.send_status("cancelled"); + self.write_failure_log(&history, &format!("cancelled: {reason}")); + self.persist_transcript(&compacted_history, &history).await; + tracing::info!(worker_id = %self.id, %reason, "worker cancelled"); + return Err(crate::error::AgentError::Cancelled { reason }.into()); } + Err(error) if is_context_overflow_error(&error.to_string()) => { + overflow_retries += 1; + if overflow_retries > MAX_OVERFLOW_RETRIES { + self.state = WorkerState::Failed; + self.hook.send_status("failed"); + self.write_failure_log(&history, &format!("context overflow after {MAX_OVERFLOW_RETRIES} compaction attempts: {error}")); + self.persist_transcript(&compacted_history, &history).await; + tracing::error!(worker_id = %self.id, %error, "worker context overflow unrecoverable"); + return Err(crate::error::AgentError::Other(error.into()).into()); + } - tracing::warn!( - worker_id = %self.id, - attempt = overflow_retries, - %error, - "context overflow, compacting and retrying" - ); - self.hook.send_status("compacting (overflow recovery)"); - self.force_compact_history(&mut compacted_history, &mut history) - .await; - prompt = "Continue where you left off. Do not repeat completed work. \ + tracing::warn!( + worker_id = %self.id, + attempt = overflow_retries, + %error, + "context overflow, compacting and retrying" + ); + self.hook.send_status("compacting (overflow recovery)"); + self.force_compact_history(&mut compacted_history, &mut history) + .await; + prompt = "Continue where you left off. Do not repeat completed work. \ Your previous attempt exceeded the context limit, so older history \ has been compacted." - .into(); - } - Err(error) => { - self.state = WorkerState::Failed; - self.hook.send_status("failed"); - self.write_failure_log(&history, &error.to_string()); - self.persist_transcript(&compacted_history, &history); - tracing::error!(worker_id = %self.id, %error, "worker LLM call failed"); - return Err(crate::error::AgentError::Other(error.into()).into()); + .into(); + } + Err(error) => { + self.state = WorkerState::Failed; + self.hook.send_status("failed"); + self.write_failure_log(&history, &error.to_string()); + self.persist_transcript(&compacted_history, &history).await; + tracing::error!(worker_id = %self.id, %error, "worker LLM call failed"); + return Err(crate::error::AgentError::Other(error.into()).into()); + } } } }; @@ -327,9 +398,14 @@ impl Worker { // For interactive workers, enter a follow-up loop let mut follow_up_failure: Option = None; if let Some(mut input_rx) = self.input_rx.take() { - self.state = WorkerState::WaitingForInput; - self.hook.send_status("waiting for input"); - self.hook.send_worker_idle(); + if !resuming { + // Fresh worker: persist transcript and signal idle for the first time. + // Resumed workers already did this in the preamble above. + self.state = WorkerState::WaitingForInput; + self.persist_transcript(&compacted_history, &history).await; + self.hook.send_status("waiting for input"); + self.hook.send_worker_idle(); + } while let Some(follow_up) = input_rx.recv().await { self.state = WorkerState::Running; @@ -417,13 +493,14 @@ impl Worker { } self.state = WorkerState::WaitingForInput; + self.persist_transcript(&compacted_history, &history).await; self.hook.send_status("waiting for input"); self.hook.send_worker_idle(); } } if let Some(failure_reason) = follow_up_failure { - self.persist_transcript(&compacted_history, &history); + self.persist_transcript(&compacted_history, &history).await; tracing::error!(worker_id = %self.id, reason = %failure_reason, "worker failed"); return Err(crate::error::AgentError::Other(anyhow::anyhow!(failure_reason)).into()); } @@ -437,8 +514,8 @@ impl Worker { self.write_success_log(&history); } - // Persist transcript blob (fire-and-forget) - self.persist_transcript(&compacted_history, &history); + // Persist transcript blob + self.persist_transcript(&compacted_history, &history).await; tracing::info!(worker_id = %self.id, "worker completed"); Ok(result) @@ -528,8 +605,11 @@ impl Worker { ); } - /// Persist the compressed transcript blob to worker_runs. Fire-and-forget. - fn persist_transcript( + /// Persist the compressed transcript blob to worker_runs. + /// + /// Awaited directly so that at idle boundaries "idle implies persisted" + /// and concurrent snapshots cannot land out of order. + async fn persist_transcript( &self, compacted_history: &[rig::message::Message], history: &[rig::message::Message], @@ -538,7 +618,6 @@ impl Worker { full_history.extend(history.iter().cloned()); let transcript_blob = crate::conversation::worker_transcript::serialize_transcript(&full_history); - let pool = self.deps.sqlite_pool.clone(); let worker_id = self.id.to_string(); // Count tool calls from the Rig history (each ToolCall in an Assistant message) @@ -555,18 +634,16 @@ impl Worker { }) .sum(); - tokio::spawn(async move { - if let Err(error) = - sqlx::query("UPDATE worker_runs SET transcript = ?, tool_calls = ? WHERE id = ?") - .bind(&transcript_blob) - .bind(tool_calls) - .bind(&worker_id) - .execute(&pool) - .await - { - tracing::warn!(%error, worker_id, "failed to persist worker transcript"); - } - }); + if let Err(error) = + sqlx::query("UPDATE worker_runs SET transcript = ?, tool_calls = ? WHERE id = ?") + .bind(&transcript_blob) + .bind(tool_calls) + .bind(&worker_id) + .execute(&self.deps.sqlite_pool) + .await + { + tracing::warn!(%error, worker_id, "failed to persist worker transcript"); + } } /// Check if worker is in a terminal state. diff --git a/src/conversation/history.rs b/src/conversation/history.rs index 1bbe098ad..f29728a39 100644 --- a/src/conversation/history.rs +++ b/src/conversation/history.rs @@ -333,6 +333,7 @@ impl ProcessRunLogger { } /// Record a worker starting. Fire-and-forget. + #[allow(clippy::too_many_arguments)] pub fn log_worker_started( &self, channel_id: Option<&ChannelId>, @@ -341,6 +342,7 @@ impl ProcessRunLogger { worker_type: &str, agent_id: &crate::AgentId, interactive: bool, + directory: Option<&std::path::Path>, ) { let pool = self.pool.clone(); let id = worker_id.to_string(); @@ -348,11 +350,12 @@ impl ProcessRunLogger { let task = task.to_string(); let worker_type = worker_type.to_string(); let agent_id = agent_id.to_string(); + let directory = directory.map(|d| d.to_string_lossy().to_string()); tokio::spawn(async move { if let Err(error) = sqlx::query( - "INSERT OR IGNORE INTO worker_runs (id, channel_id, task, worker_type, agent_id, interactive) \ - VALUES (?, ?, ?, ?, ?, ?)", + "INSERT OR IGNORE INTO worker_runs (id, channel_id, task, worker_type, agent_id, interactive, directory) \ + VALUES (?, ?, ?, ?, ?, ?, ?)", ) .bind(&id) .bind(&channel_id) @@ -360,6 +363,7 @@ impl ProcessRunLogger { .bind(&worker_type) .bind(&agent_id) .bind(interactive) + .bind(&directory) .execute(&pool) .await { @@ -368,6 +372,26 @@ impl ProcessRunLogger { }); } + /// Persist the working directory for a worker. Fire-and-forget. + /// + /// Called from `spawn_opencode_worker_from_state` after the worker row is + /// created, so the directory survives for idle-worker resume. + pub fn log_worker_directory(&self, worker_id: WorkerId, directory: &std::path::Path) { + let pool = self.pool.clone(); + let id = worker_id.to_string(); + let dir = directory.to_string_lossy().to_string(); + tokio::spawn(async move { + if let Err(error) = sqlx::query("UPDATE worker_runs SET directory = ? WHERE id = ?") + .bind(&dir) + .bind(&id) + .execute(&pool) + .await + { + tracing::warn!(%error, worker_id = %id, "failed to persist worker directory"); + } + }); + } + /// Update a worker's status. Fire-and-forget. /// Most status text updates are transient — they're available via the /// in-memory StatusBlock for live workers and don't need to be persisted. @@ -464,10 +488,13 @@ impl ProcessRunLogger { }); } - /// Mark all orphaned running/idle workers as failed for an agent. + /// Mark orphaned **running** workers as failed for an agent. /// - /// Called at startup to reconcile rows that were left in `running` or `idle` + /// Called at startup to reconcile rows that were left in `running` status /// when the process exited before a `WorkerComplete` event was persisted. + /// + /// Idle interactive workers are intentionally left alone — they will be + /// resumed by `get_idle_interactive_workers()` + the reconnection logic. pub async fn reconcile_running_workers_for_agent( &self, agent_id: &str, @@ -481,7 +508,7 @@ impl ProcessRunLogger { WHEN result IS NULL OR result = '' THEN ? \ ELSE result \ END \ - WHERE status IN ('running', 'idle') AND (agent_id = ? OR agent_id IS NULL)", + WHERE status = 'running' AND (agent_id = ? OR agent_id IS NULL)", ) .bind(failure_message) .bind(agent_id) @@ -492,6 +519,74 @@ impl ProcessRunLogger { Ok(result.rows_affected()) } + /// Load all idle interactive workers for an agent. + /// + /// Called at startup to find workers that were waiting for follow-up input + /// when the process exited. These can potentially be reconnected to their + /// sessions and resumed rather than marked as failed. + pub async fn get_idle_interactive_workers( + &self, + agent_id: &str, + ) -> crate::error::Result> { + let rows = sqlx::query_as::<_, IdleWorkerRow>( + "SELECT id, task, channel_id, worker_type, transcript, \ + COALESCE(tool_calls, 0) AS tool_calls, \ + opencode_session_id, opencode_port, directory \ + FROM worker_runs \ + WHERE status = 'idle' AND interactive = TRUE \ + AND (agent_id = ? OR agent_id IS NULL)", + ) + .bind(agent_id) + .fetch_all(&self.pool) + .await + .map_err(|error| anyhow::anyhow!(error))?; + + Ok(rows) + } + + /// Mark an idle worker as failed (used when reconnection fails at startup). + pub async fn fail_idle_worker( + &self, + worker_id: &str, + reason: &str, + ) -> crate::error::Result<()> { + sqlx::query( + "UPDATE worker_runs \ + SET status = 'failed', \ + completed_at = COALESCE(completed_at, CURRENT_TIMESTAMP), \ + result = CASE \ + WHEN result IS NULL OR result = '' THEN ? \ + ELSE result \ + END \ + WHERE id = ? AND status = 'idle'", + ) + .bind(reason) + .bind(worker_id) + .execute(&self.pool) + .await + .map_err(|error| anyhow::anyhow!(error))?; + Ok(()) + } + + /// Retire an idle worker whose session can no longer be resumed. + /// + /// Marks the row as `done` (not `failed`) because the worker completed its + /// work successfully — only the follow-up session expired. The existing + /// result and transcript are preserved. + pub async fn retire_idle_worker(&self, worker_id: &str) -> crate::error::Result<()> { + sqlx::query( + "UPDATE worker_runs \ + SET status = 'done', \ + completed_at = COALESCE(completed_at, CURRENT_TIMESTAMP) \ + WHERE id = ? AND status = 'idle'", + ) + .bind(worker_id) + .execute(&self.pool) + .await + .map_err(|error| anyhow::anyhow!(error))?; + Ok(()) + } + /// Mark a detached running worker as cancelled. /// /// Used by API cancellation when the in-memory channel state no longer has @@ -796,6 +891,20 @@ pub struct WorkerRunRow { pub interactive: bool, } +/// A worker that was idle at shutdown, loaded for reconnection at startup. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct IdleWorkerRow { + pub id: String, + pub task: String, + pub channel_id: Option, + pub worker_type: String, + pub transcript: Option>, + pub tool_calls: i64, + pub opencode_session_id: Option, + pub opencode_port: Option, + pub directory: Option, +} + /// A worker run row with full detail including the transcript blob. #[derive(Debug, Clone)] pub struct WorkerDetailRow { diff --git a/src/conversation/worker_transcript.rs b/src/conversation/worker_transcript.rs index 7dd858d8f..09d6aa9f9 100644 --- a/src/conversation/worker_transcript.rs +++ b/src/conversation/worker_transcript.rs @@ -20,6 +20,12 @@ const MAX_TOOL_ARGS_BYTES: usize = 2_000; pub enum TranscriptStep { /// Agent reasoning and/or tool calls. Action { content: Vec }, + /// User-originated text (task prompt, follow-up input). + /// + /// Distinct from `Action` so that `transcript_to_history()` can reconstruct + /// the correct `Message::User` role instead of treating everything as + /// `Message::Assistant`. + UserText { text: String }, /// Tool execution result. ToolResult { call_id: String, @@ -106,12 +112,16 @@ pub fn convert_opencode_messages(messages: &[serde_json::Value]) -> (Vec { @@ -297,6 +307,82 @@ pub fn convert_opencode_parts( steps } +/// Convert a persisted `Vec` back into a Rig `Vec` history. +/// +/// Used when resuming an idle interactive worker after restart: the transcript +/// blob is the only surviving record of the worker's conversation, so we +/// reconstruct the Rig message history from it. +/// +/// The mapping is lossy (reasoning, images, etc. are not round-tripped), but +/// it preserves all text and tool call/result pairs which is sufficient for +/// the LLM to pick up the conversation. +pub fn transcript_to_history(steps: &[TranscriptStep]) -> Vec { + use rig::message::{ + AssistantContent, Message, Text, ToolCall, ToolFunction, ToolResult, ToolResultContent, + UserContent, + }; + use rig::one_or_many::OneOrMany; + + let mut messages: Vec = Vec::new(); + + for step in steps { + match step { + TranscriptStep::Action { content } => { + let mut parts: Vec = Vec::new(); + for item in content { + match item { + ActionContent::Text { text } => { + parts.push(AssistantContent::Text(Text { text: text.clone() })); + } + ActionContent::ToolCall { id, name, args } => { + let arguments = serde_json::from_str(args) + .unwrap_or_else(|_| serde_json::Value::String(args.clone())); + parts.push(AssistantContent::ToolCall(ToolCall { + id: id.clone(), + call_id: None, + function: ToolFunction { + name: name.clone(), + arguments, + }, + signature: None, + additional_params: None, + })); + } + } + } + if !parts.is_empty() { + // OneOrMany::many returns Err only for empty vecs; we checked above. + let content = OneOrMany::many(parts).expect("parts is non-empty"); + messages.push(Message::Assistant { id: None, content }); + } + } + TranscriptStep::UserText { text } => { + if !text.is_empty() { + messages.push(Message::User { + content: OneOrMany::one(UserContent::Text(Text { text: text.clone() })), + }); + } + } + TranscriptStep::ToolResult { + call_id, + name: _, + text, + } => { + let result = ToolResult { + id: call_id.clone(), + call_id: Some(call_id.clone()), + content: OneOrMany::one(ToolResultContent::Text(Text { text: text.clone() })), + }; + messages.push(Message::User { + content: OneOrMany::one(UserContent::ToolResult(result)), + }); + } + } + } + + messages +} + /// Convert Rig `Vec` to `Vec`. fn convert_history(history: &[rig::message::Message]) -> Vec { let mut steps = Vec::new(); @@ -367,10 +453,8 @@ fn convert_history(history: &[rig::message::Message]) -> Vec { rig::message::UserContent::Text(text) => { // Skip compaction markers and system-injected messages if !text.text.is_empty() && !text.text.starts_with("[System:") { - steps.push(TranscriptStep::Action { - content: vec![ActionContent::Text { - text: text.text.clone(), - }], + steps.push(TranscriptStep::UserText { + text: text.text.clone(), }); } } diff --git a/src/lib.rs b/src/lib.rs index 7e3f03c1a..3f20573f8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -258,6 +258,7 @@ pub enum ProcessEvent { OpenCodeSessionCreated { agent_id: AgentId, worker_id: WorkerId, + channel_id: Option, session_id: String, port: u16, }, diff --git a/src/main.rs b/src/main.rs index b7890d9a2..ecd26328c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1554,6 +1554,293 @@ async fn run( // Active conversation channels: conversation_id -> ActiveChannel let mut active_channels: HashMap = HashMap::new(); + // Resume idle interactive workers that survived the restart. + // For each idle worker, pre-create the channel if needed and spawn + // the resumed worker into its state so follow-ups route correctly. + if agents_initialized { + for (agent_id, agent) in agents.iter() { + let run_logger = spacebot::conversation::ProcessRunLogger::new(agent.db.sqlite.clone()); + let idle_workers = match run_logger + .get_idle_interactive_workers(&agent.config.id) + .await + { + Ok(workers) => workers, + Err(error) => { + tracing::warn!(agent_id = %agent_id, %error, "failed to query idle workers"); + continue; + } + }; + if idle_workers.is_empty() { + continue; + } + tracing::info!( + agent_id = %agent_id, + idle_count = idle_workers.len(), + "found idle interactive workers to resume" + ); + + // Group idle workers by channel_id + let mut by_channel: HashMap< + String, + Vec<&spacebot::conversation::history::IdleWorkerRow>, + > = HashMap::new(); + for worker in &idle_workers { + if let Some(channel_id) = &worker.channel_id { + by_channel + .entry(channel_id.clone()) + .or_default() + .push(worker); + } else { + // Workers without a channel_id can't be resumed (no follow-up + // routing). Leave them as idle — the transcript is preserved + // for inspection in the UI. + tracing::warn!( + worker_id = %worker.id, + "idle worker has no channel_id, cannot resume (leaving as idle)" + ); + } + } + + for (conversation_id, workers) in by_channel { + // Ensure the channel exists. If it's already in active_channels + // (unlikely at startup), use its state. Otherwise, pre-create it. + if !active_channels.contains_key(&conversation_id) { + // First pass: retire any workers whose sessions can't be + // reconnected. Only create the channel if at least one + // worker has a chance of resuming. + let mut resumable: Vec<&spacebot::conversation::history::IdleWorkerRow> = + Vec::new(); + for idle_worker in &workers { + if idle_worker.worker_type == "opencode" + && idle_worker.opencode_session_id.is_none() + { + // OpenCode workers without session metadata can never + // resume — the server died with kill_on_drop. + if let Err(error) = run_logger.retire_idle_worker(&idle_worker.id).await + { + tracing::warn!( + worker_id = %idle_worker.id, + %error, + "failed to retire idle worker" + ); + } + tracing::info!( + worker_id = %idle_worker.id, + channel_id = %conversation_id, + "retired idle opencode worker (no session metadata)" + ); + } else { + resumable.push(idle_worker); + } + } + if resumable.is_empty() { + continue; + } + + let (response_tx, mut response_rx) = + mpsc::channel::(32); + let event_rx = agent.deps.event_tx.subscribe(); + let channel_id: spacebot::ChannelId = Arc::from(conversation_id.as_str()); + + let (channel, channel_tx) = spacebot::agent::channel::Channel::new( + channel_id, + agent.deps.clone(), + response_tx, + event_rx, + agent.config.screenshot_dir(), + agent.config.logs_dir(), + ); + agent + .deps + .process_control_registry + .register_channel(channel.id.clone(), channel.control_handle().downgrade()) + .await; + api_state + .register_channel_status( + conversation_id.clone(), + channel.state.status_block.clone(), + ) + .await; + api_state + .register_channel_state(conversation_id.clone(), channel.state.clone()) + .await; + + // Resume workers into the channel state before spawning the event loop. + let mut any_resumed = false; + for idle_worker in &resumable { + match spacebot::agent::channel_dispatch::resume_idle_worker_into_state( + &channel.state, + idle_worker, + ) + .await + { + Ok(worker_id) => { + any_resumed = true; + tracing::info!( + worker_id = %worker_id, + channel_id = %conversation_id, + "resumed idle worker" + ); + } + Err(reason) => { + // Resume failed at runtime (e.g. OpenCode disabled, + // transcript corrupt). Retire the worker. + if let Err(error) = + run_logger.retire_idle_worker(&idle_worker.id).await + { + tracing::warn!( + worker_id = %idle_worker.id, + %error, + "failed to retire idle worker" + ); + } + tracing::info!( + worker_id = %idle_worker.id, + channel_id = %conversation_id, + %reason, + "retired idle worker (session expired)" + ); + } + } + } + + // Spawn the channel event loop. + let cleanup_channel_id = conversation_id.clone(); + let process_control_registry = agent.deps.process_control_registry.clone(); + let api_state_for_cleanup = api_state.clone(); + tokio::spawn(async move { + if let Err(error) = channel.run().await { + tracing::error!(%error, "channel event loop failed"); + } + let scoped_channel_id: spacebot::ChannelId = + Arc::from(cleanup_channel_id.as_str()); + process_control_registry + .unregister_channel(&scoped_channel_id) + .await; + api_state_for_cleanup + .unregister_channel_status(&cleanup_channel_id) + .await; + api_state_for_cleanup + .unregister_channel_state(&cleanup_channel_id) + .await; + }); + + // Outbound response routing for this pre-created channel. + // Since there's no inbound message yet, we create a placeholder. + let latest_message = + Arc::new(tokio::sync::RwLock::new(spacebot::InboundMessage { + id: uuid::Uuid::new_v4().to_string(), + source: "internal".to_string(), + adapter: None, + conversation_id: conversation_id.clone(), + content: spacebot::MessageContent::Text(String::new()), + sender_id: String::new(), + formatted_author: None, + metadata: std::collections::HashMap::new(), + agent_id: Some(agent_id.clone()), + timestamp: chrono::Utc::now(), + })); + let outbound_message = latest_message.clone(); + let messaging_for_outbound = messaging_manager.clone(); + let api_event_tx = api_state.event_tx.clone(); + let sse_agent_id = agent_id.to_string(); + let sse_channel_id = conversation_id.clone(); + let outbound_handle = tokio::spawn(async move { + while let Some(response) = response_rx.recv().await { + match &response { + spacebot::OutboundResponse::Text(text) => { + api_event_tx + .send(spacebot::api::ApiEvent::OutboundMessage { + agent_id: sse_agent_id.clone(), + channel_id: sse_channel_id.clone(), + text: text.clone(), + }) + .ok(); + } + spacebot::OutboundResponse::RichMessage { text, .. } => { + api_event_tx + .send(spacebot::api::ApiEvent::OutboundMessage { + agent_id: sse_agent_id.clone(), + channel_id: sse_channel_id.clone(), + text: text.clone(), + }) + .ok(); + } + spacebot::OutboundResponse::ThreadReply { text, .. } => { + api_event_tx + .send(spacebot::api::ApiEvent::OutboundMessage { + agent_id: sse_agent_id.clone(), + channel_id: sse_channel_id.clone(), + text: text.clone(), + }) + .ok(); + } + spacebot::OutboundResponse::Status( + spacebot::StatusUpdate::Thinking, + ) => { + api_event_tx + .send(spacebot::api::ApiEvent::TypingState { + agent_id: sse_agent_id.clone(), + channel_id: sse_channel_id.clone(), + is_typing: true, + }) + .ok(); + } + spacebot::OutboundResponse::Status( + spacebot::StatusUpdate::StopTyping, + ) => { + api_event_tx + .send(spacebot::api::ApiEvent::TypingState { + agent_id: sse_agent_id.clone(), + channel_id: sse_channel_id.clone(), + is_typing: false, + }) + .ok(); + } + _ => {} + } + let current_message = outbound_message.read().await.clone(); + match response { + spacebot::OutboundResponse::Status(status) => { + if let Err(error) = messaging_for_outbound + .send_status(¤t_message, status) + .await + { + tracing::warn!(%error, "failed to send status update"); + } + } + response => { + if let Err(error) = messaging_for_outbound + .respond(¤t_message, response) + .await + { + tracing::error!(%error, "failed to send outbound response"); + } + } + } + } + }); + + active_channels.insert( + conversation_id.clone(), + ActiveChannel { + message_tx: channel_tx, + latest_message, + _outbound_handle: outbound_handle, + }, + ); + + tracing::info!( + conversation_id = %conversation_id, + agent_id = %agent_id, + any_resumed, + "pre-created channel for idle worker resumption" + ); + } + } + } + } + // Main event loop: route inbound messages to agent channels loop { // Poll the inbound stream if it exists, otherwise yield a never-resolving future diff --git a/src/opencode/server.rs b/src/opencode/server.rs index c047c418c..0aad4b014 100644 --- a/src/opencode/server.rs +++ b/src/opencode/server.rs @@ -12,8 +12,8 @@ use crate::opencode::types::*; use anyhow::{Context as _, bail}; use reqwest::Client; -use std::collections::HashMap; use std::collections::hash_map::DefaultHasher; +use std::collections::{HashMap, HashSet}; use std::hash::{Hash, Hasher}; use std::path::{Path, PathBuf}; use std::process::Stdio; @@ -541,6 +541,9 @@ impl Drop for OpenCodeServer { /// No file persistence needed -- just health-check the expected port. pub struct OpenCodeServerPool { servers: Mutex>>>, + /// Directories that currently have an active OpenCode worker running. + /// Prevents spawning a second worker targeting the same directory. + active_directories: Mutex>, opencode_path: String, permissions: OpenCodePermissions, max_servers: usize, @@ -555,6 +558,7 @@ impl OpenCodeServerPool { ) -> Self { Self { servers: Mutex::new(HashMap::new()), + active_directories: Mutex::new(HashSet::new()), opencode_path: opencode_path.into(), permissions, max_servers, @@ -637,6 +641,57 @@ impl OpenCodeServerPool { pub async fn server_count(&self) -> usize { self.servers.lock().await.len() } + + /// Reserve a directory for an active OpenCode worker. + /// + /// Returns an error if another worker is already running in this directory. + /// The caller must call [`release_directory`] when the worker finishes. + pub async fn claim_directory(&self, directory: &Path) -> anyhow::Result<()> { + let canonical = directory + .canonicalize() + .with_context(|| format!("directory '{}' does not exist", directory.display()))?; + + let mut active = self.active_directories.lock().await; + if !active.insert(canonical.clone()) { + bail!( + "An OpenCode worker is already active in directory '{}'", + canonical.display() + ); + } + tracing::debug!( + directory = %canonical.display(), + "claimed directory for OpenCode worker" + ); + Ok(()) + } + + /// Release a directory previously claimed by [`claim_directory`]. + pub async fn release_directory(&self, directory: &Path) { + let canonical = match directory.canonicalize() { + Ok(path) => path, + Err(error) => { + tracing::warn!( + directory = %directory.display(), + %error, + "could not canonicalize directory during release, using raw path" + ); + directory.to_path_buf() + } + }; + + let mut active = self.active_directories.lock().await; + if !active.remove(&canonical) { + tracing::warn!( + directory = %canonical.display(), + "released directory that was not claimed" + ); + } else { + tracing::debug!( + directory = %canonical.display(), + "released directory for OpenCode worker" + ); + } + } } /// Derive a deterministic port from a directory path. diff --git a/src/opencode/worker.rs b/src/opencode/worker.rs index 2906edba2..a8b159bd5 100644 --- a/src/opencode/worker.rs +++ b/src/opencode/worker.rs @@ -16,6 +16,13 @@ use std::sync::Arc; use tokio::sync::{Mutex, broadcast, mpsc}; use uuid::Uuid; +/// State for resuming an idle OpenCode session after restart. +pub struct ResumeSession { + pub session_id: String, + pub accumulated_parts: Vec, + pub tool_calls: i64, +} + /// An OpenCode-backed worker that drives a coding session via subprocess. pub struct OpenCodeWorker { pub id: WorkerId, @@ -33,6 +40,10 @@ pub struct OpenCodeWorker { pub model: Option, /// Secrets store for exact-match scrubbing of tool secret values in SSE output. pub secrets_store: Option>, + /// SQLite pool for incremental transcript persistence (set by channel_dispatch). + pub sqlite_pool: Option, + /// Pre-populated session state for resumed workers (set by `resume_interactive`). + pub resuming_session: Option, } /// Accumulated state from SSE event processing. @@ -96,6 +107,8 @@ impl OpenCodeWorker { system_prompt: None, model: None, secrets_store: None, + sqlite_pool: None, + resuming_session: None, } } @@ -132,6 +145,101 @@ impl OpenCodeWorker { self } + /// Set the SQLite pool for incremental transcript persistence. + pub fn with_sqlite_pool(mut self, pool: sqlx::SqlitePool) -> Self { + self.sqlite_pool = Some(pool); + self + } + + /// Create a resumed interactive OpenCode worker for an idle session. + /// + /// Instead of creating a new session, reconnects to `session_id` on the + /// existing OpenCode server. The prior transcript (from the DB blob) is + /// loaded into `accumulated_parts` so subsequent `persist_transcript_snapshot` + /// calls produce a complete history. + /// + /// Returns `None` if reconnection fails (server dead, session gone). + #[allow(clippy::too_many_arguments)] + pub async fn resume_interactive( + existing_id: WorkerId, + channel_id: Option, + agent_id: AgentId, + task: impl Into, + directory: PathBuf, + server_pool: Arc, + event_tx: broadcast::Sender, + session_id: String, + _prior_transcript_blob: Option>, + ) -> Option<(Self, mpsc::Sender)> { + // Try to reconnect to the OpenCode server for this directory. + let server = match server_pool.get_or_create(&directory).await { + Ok(server) => server, + Err(error) => { + tracing::warn!( + worker_id = %existing_id, + %error, + directory = %directory.display(), + "failed to reconnect to OpenCode server for idle worker" + ); + return None; + } + }; + + // Verify the session still exists by fetching its messages. + let messages = { + let guard = server.lock().await; + guard.get_messages(&session_id).await + }; + if let Err(error) = &messages { + tracing::warn!( + worker_id = %existing_id, + %error, + session_id = %session_id, + "OpenCode session no longer exists, cannot resume" + ); + return None; + } + + // Reconstruct accumulated_parts from the session messages (preferred) + // or from the persisted transcript blob (fallback). + let accumulated_parts = if let Ok(messages) = &messages { + // Re-parse the parts from the session messages API. + // This gives us the authoritative state. + let mut parts = Vec::new(); + for message in messages { + if let Some(msg_parts) = message.get("parts").and_then(|p| p.as_array()) { + for part_value in msg_parts { + if let Ok(part) = serde_json::from_value::(part_value.clone()) + { + parts.push(part); + } + } + } + } + parts + } else { + Vec::new() + }; + + // Count tool calls from accumulated parts + let tool_calls = accumulated_parts + .iter() + .filter(|p| matches!(p, OpenCodePart::Tool { .. })) + .count() as i64; + + let (input_tx, input_rx) = mpsc::channel(32); + let mut worker = Self::new(channel_id, agent_id, task, directory, server_pool, event_tx); + worker.id = existing_id; + worker.input_rx = Some(input_rx); + worker.resuming_session = Some(ResumeSession { + session_id, + accumulated_parts, + tool_calls, + }); + + Some((worker, input_tx)) + } + /// Scrub tool secret values from text, replacing each with `[REDACTED:]`. /// Returns the scrubbed text. If no secrets store is set, returns the input unchanged. fn scrub_text(&self, text: &str) -> String { @@ -144,108 +252,160 @@ impl OpenCodeWorker { /// Run the worker: spawn/reuse an OpenCode server, create a session, /// send the task, monitor via SSE, and return the result. pub async fn run(mut self) -> anyhow::Result { - self.send_status("starting OpenCode server"); - - // Get or create server for this directory - let server = self - .server_pool - .get_or_create(&self.directory) - .await - .with_context(|| { - format!( - "failed to get OpenCode server for '{}'", - self.directory.display() - ) - })?; - - self.send_status("creating session"); - - // Create a session - let session = { - let guard = server.lock().await; - guard - .create_session(Some(format!("spacebot-worker-{}", self.id))) - .await? - }; - let session_id = session.id.clone(); + let resuming = self.resuming_session.is_some(); - // Record metadata so the web UI can embed the OpenCode interface - let opencode_port = { - let guard = server.lock().await; - guard.port() - }; - self.event_tx - .send(ProcessEvent::OpenCodeSessionCreated { - agent_id: self.agent_id.clone(), - worker_id: self.id, - session_id: session_id.clone(), - port: opencode_port, - }) - .ok(); + // --- Session setup: either resume an existing session or create a new one --- + let (server, session_id, mut event_state, result_text) = + if let Some(resume) = self.resuming_session.take() { + // Resumed worker: reconnect to the existing server + session. + self.send_status("reconnecting to OpenCode session"); - tracing::info!( - worker_id = %self.id, - session_id = %session_id, - port = opencode_port, - directory = %self.directory.display(), - "OpenCode session created" - ); + let server = self + .server_pool + .get_or_create(&self.directory) + .await + .with_context(|| { + format!( + "failed to reconnect to OpenCode server for '{}'", + self.directory.display() + ) + })?; + + let opencode_port = { + let guard = server.lock().await; + guard.port() + }; - // Subscribe to SSE events before sending the prompt - let event_response = { - let guard = server.lock().await; - guard.subscribe_events().await? - }; + // Re-emit session metadata so the frontend can show the embed. + self.event_tx + .send(ProcessEvent::OpenCodeSessionCreated { + agent_id: self.agent_id.clone(), + worker_id: self.id, + channel_id: self.channel_id.clone(), + session_id: resume.session_id.clone(), + port: opencode_port, + }) + .ok(); - // Build the prompt request - let model_param = self.model.as_ref().and_then(|m| parse_model_param(m)); - let prompt_request = SendPromptRequest { - parts: vec![PartInput::Text { - text: self.task.clone(), - synthetic: None, - }], - system: self.system_prompt.clone(), - model: model_param, - agent: None, - }; + tracing::info!( + worker_id = %self.id, + session_id = %resume.session_id, + port = opencode_port, + prior_parts = resume.accumulated_parts.len(), + "resumed OpenCode worker, reconnected to session" + ); - // Send prompt async so we can process SSE events while it runs - self.send_status("sending task to OpenCode"); - { - let guard = server.lock().await; - guard - .send_prompt_async(&session_id, &prompt_request) - .await?; - } + let mut event_state = EventState::new(); + event_state.accumulated_parts = resume.accumulated_parts; + event_state.tool_calls = resume.tool_calls; + event_state.has_received_event = true; + event_state.has_assistant_message = true; - // Process SSE events until session goes idle or errors. - // EventState tracks status and last_text for the initial result delivery. - // The full transcript is fetched from the OpenCode API on completion. - let mut event_state = EventState::new(); - self.process_events(event_response, &session_id, &server, &mut event_state) - .await?; + (server, resume.session_id, event_state, String::new()) + } else { + // Fresh worker: create a new server + session. + self.send_status("starting OpenCode server"); - // last_text is our best signal for the initial result (used for - // WorkerInitialResult and the fallback if API fetch fails). - let result_text = event_state.last_text.clone(); + let server = self + .server_pool + .get_or_create(&self.directory) + .await + .with_context(|| { + format!( + "failed to get OpenCode server for '{}'", + self.directory.display() + ) + })?; + + self.send_status("creating session"); + + let session = { + let guard = server.lock().await; + guard + .create_session(Some(format!("spacebot-worker-{}", self.id))) + .await? + }; + let session_id = session.id.clone(); + + let opencode_port = { + let guard = server.lock().await; + guard.port() + }; + self.event_tx + .send(ProcessEvent::OpenCodeSessionCreated { + agent_id: self.agent_id.clone(), + worker_id: self.id, + channel_id: self.channel_id.clone(), + session_id: session_id.clone(), + port: opencode_port, + }) + .ok(); + + tracing::info!( + worker_id = %self.id, + session_id = %session_id, + port = opencode_port, + directory = %self.directory.display(), + "OpenCode session created" + ); + + // Subscribe to SSE events before sending the prompt + let event_response = { + let guard = server.lock().await; + guard.subscribe_events().await? + }; + + let model_param = self.model.as_ref().and_then(|m| parse_model_param(m)); + let prompt_request = SendPromptRequest { + parts: vec![PartInput::Text { + text: self.task.clone(), + synthetic: None, + }], + system: self.system_prompt.clone(), + model: model_param, + agent: None, + }; + + self.send_status("sending task to OpenCode"); + { + let guard = server.lock().await; + guard + .send_prompt_async(&session_id, &prompt_request) + .await?; + } + + let mut event_state = EventState::new(); + self.process_events(event_response, &session_id, &server, &mut event_state) + .await?; + + let result_text = event_state.last_text.clone(); + (server, session_id, event_state, result_text) + }; // Interactive follow-up loop if let Some(mut input_rx) = self.input_rx.take() { - // Emit the initial result immediately so the channel can retrigger - // and tell the user what the worker found. Without this, the channel - // would block until the entire follow-up loop exits — which only - // happens when the channel drops the input sender, creating a deadlock. - let scrubbed_result = self.scrub_text(&result_text); - let scrubbed_result = crate::secrets::scrub::scrub_leaks(&scrubbed_result); - let _ = self.event_tx.send(ProcessEvent::WorkerInitialResult { - agent_id: self.agent_id.clone(), - worker_id: self.id, - channel_id: self.channel_id.clone(), - result: scrubbed_result, - }); - - self.send_status("waiting for follow-up"); - self.send_idle(); + if resuming { + // Resumed worker: go straight to idle without emitting initial result + // (it was already relayed before the restart). Persist the recovered + // transcript so a second crash doesn't lose it. + self.persist_transcript_snapshot(&event_state).await; + self.send_status("resumed — waiting for follow-up"); + self.send_idle(); + } else { + // Fresh worker: emit the initial result so the channel can retrigger. + let scrubbed_result = self.scrub_text(&result_text); + let scrubbed_result = crate::secrets::scrub::scrub_leaks(&scrubbed_result); + let _ = self.event_tx.send(ProcessEvent::WorkerInitialResult { + agent_id: self.agent_id.clone(), + worker_id: self.id, + channel_id: self.channel_id.clone(), + result: scrubbed_result, + }); + + self.persist_transcript_snapshot(&event_state).await; + self.send_status("waiting for follow-up"); + self.send_idle(); + } while let Some(follow_up) = input_rx.recv().await { self.send_status("processing follow-up"); @@ -291,6 +451,7 @@ impl OpenCodeWorker { result: scrubbed, }); } + self.persist_transcript_snapshot(&event_state).await; self.send_status("waiting for follow-up"); self.send_idle(); } @@ -722,6 +883,42 @@ impl OpenCodeWorker { channel_id: self.channel_id.clone(), }); } + + /// Persist a snapshot of the transcript built from accumulated SSE parts. + /// + /// Called each time the worker goes idle so that if spacebot restarts + /// while the worker is waiting for follow-up, the transcript survives. + /// Awaited directly so "idle implies persisted" — no out-of-order writes. + async fn persist_transcript_snapshot(&self, event_state: &EventState) { + let Some(pool) = &self.sqlite_pool else { + return; + }; + if event_state.accumulated_parts.is_empty() { + return; + } + + let steps = crate::conversation::worker_transcript::convert_opencode_parts( + &event_state.accumulated_parts, + ); + if steps.is_empty() { + return; + } + + let blob = crate::conversation::worker_transcript::serialize_steps(&steps); + let tool_calls = event_state.tool_calls; + let worker_id = self.id.to_string(); + + if let Err(error) = + sqlx::query("UPDATE worker_runs SET transcript = ?, tool_calls = ? WHERE id = ?") + .bind(&blob) + .bind(tool_calls) + .bind(&worker_id) + .execute(pool) + .await + { + tracing::warn!(%error, worker_id, "failed to persist transcript snapshot"); + } + } } /// Extract a human-readable description from a tool's input JSON. diff --git a/src/tools/spawn_worker.rs b/src/tools/spawn_worker.rs index 814fc0bba..4f269fd58 100644 --- a/src/tools/spawn_worker.rs +++ b/src/tools/spawn_worker.rs @@ -104,7 +104,7 @@ impl Tool for SpawnWorkerTool { "interactive": { "type": "boolean", "default": false, - "description": "If true, the worker stays alive and accepts follow-up messages via route_to_worker. If false (default), the worker runs once and returns." + "description": "If true, the worker stays alive and accepts follow-up messages via route_to_worker. If false (default), the worker runs once and returns. OpenCode workers are always interactive regardless of this flag." }, "suggested_skills": { "type": "array", @@ -152,7 +152,8 @@ impl Tool for SpawnWorkerTool { SpawnWorkerError("directory is required for opencode workers".into()) })?; - spawn_opencode_worker_from_state(&self.state, &args.task, directory, args.interactive) + // OpenCode workers are always interactive — ignore args.interactive. + spawn_opencode_worker_from_state(&self.state, &args.task, directory, true) .await .map_err(|e| SpawnWorkerError(format!("{e}")))? } else { @@ -171,7 +172,9 @@ impl Tool for SpawnWorkerTool { }; let worker_type_label = if is_opencode { "OpenCode" } else { "builtin" }; - let message = if args.interactive { + // OpenCode workers are always interactive regardless of args.interactive. + let effectively_interactive = args.interactive || is_opencode; + let message = if effectively_interactive { format!( "Interactive {worker_type_label} worker {worker_id} spawned for: {}. Route follow-ups with route_to_worker.", args.task @@ -198,7 +201,7 @@ impl Tool for SpawnWorkerTool { Ok(SpawnWorkerOutput { worker_id, spawned: true, - interactive: args.interactive, + interactive: effectively_interactive, message: format!("{message}{readiness_note}"), }) } diff --git a/src/tools/worker_inspect.rs b/src/tools/worker_inspect.rs index 039b6e4d8..edac9db8e 100644 --- a/src/tools/worker_inspect.rs +++ b/src/tools/worker_inspect.rs @@ -132,6 +132,9 @@ impl Tool for WorkerInspectTool { } } } + worker_transcript::TranscriptStep::UserText { text } => { + summary.push_str(&format!("**User:** {text}\n\n")); + } worker_transcript::TranscriptStep::ToolResult { name, text, .. } => {