From 858df09e267ad979abfb6a8e725aff97558ea506 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 1 Aug 2026 03:48:09 -0700 Subject: [PATCH 1/2] fix(dispatch): name target worktrees by project and allow follow-up turns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two problems showed up once dispatch was used for real work. A target checkout was created at `worktrees/`, so a dispatch from BitFun landed in a directory whose name said nothing about BitFun and carried a full job UUID. It now follows the same convention as a local managed worktree — `worktrees//-` — with the project name grouped under its repository's shared clone. The label is advisory input from the controller: the target sanitizes it, falls back to the remote URL's basename and then to a constant, and rejects anything that is not a single safe path component, so the path is never shaped by an untrusted string. The resolved path is recorded in the provision record and read back from there, which is also what lets retention find the checkout now that its name no longer contains the job id. A dispatch session also refused every message after the first, because a job was modelled as one exchange. It is a conversation: while a turn runs a message steers it through `append`, and once it has finished a message starts the next turn through the new `continue` verb. That rewinds only the job's run state, so the target session, its worktree, and its append-only event log all persist — the worker restores the session instead of creating it, which is what gives a follow-up the earlier turns as context, and the controller's cursor and transcript cache keep reading one growing transcript. Follow-ups carry a caller-generated turn id and are claimed under the job lock, so an ambiguous response resolves to the same turn rather than starting a second one. --- docs/architecture/detached-task-dispatch.md | 35 ++ src/apps/cli/src/dispatch/mod.rs | 52 ++- src/apps/cli/src/dispatch/protocol.rs | 34 ++ src/apps/cli/src/dispatch/store.rs | 391 ++++++++++++++---- src/apps/cli/src/dispatch/worker.rs | 72 ++-- src/apps/cli/src/dispatch/workspace.rs | 154 ++++++- src/apps/cli/src/peer_host/dispatch.rs | 1 + src/apps/desktop/src/api/dispatch_api.rs | 54 ++- src/apps/desktop/src/api/dispatch_host.rs | 1 + .../src/api/remote_workspace_policy.rs | 4 + src/apps/desktop/src/lib.rs | 1 + .../core/src/service/dispatch/baseline.rs | 24 ++ .../core/src/service/dispatch/controller.rs | 97 +++++ .../src/service/dispatch/device_controller.rs | 32 +- .../assembly/core/src/service/dispatch/mod.rs | 20 +- .../src/remote_ssh/dispatch_ssh.rs | 9 + src/web-ui/src/features/dispatch/README.md | 5 + .../src/features/dispatch/dispatchApi.ts | 19 + src/web-ui/src/features/dispatch/types.ts | 8 + .../flow-chat-manager/MessageModule.ts | 110 ++++- .../api/adapters/peer-device-adapter.ts | 1 + 21 files changed, 963 insertions(+), 161 deletions(-) diff --git a/docs/architecture/detached-task-dispatch.md b/docs/architecture/detached-task-dispatch.md index 96239bd0ba..b54e21bfa4 100644 --- a/docs/architecture/detached-task-dispatch.md +++ b/docs/architecture/detached-task-dispatch.md @@ -187,6 +187,7 @@ Public job verbs are: | `list` | List durable target jobs. | | `answer` | Resolve one persisted permission request for `remote` approval policy. | | `append` | Queue an idempotent steering message for the active turn. | +| `continue` | Queue the next turn for a job whose previous turn has finished. | Git delivery and synchronization use these internal data-plane verbs: @@ -242,6 +243,40 @@ available, probe reports the missing runner and submission remains disabled rather than falling back to local execution. Device dispatch never performs SSH-style installation through the Relay. +## Conversation model + +A dispatch session is a conversation, not a single exchange. One job owns one +target session, one worktree, and one append-only event log; each user message +is a turn inside it: + +- while a turn is running, a message is an `append` that steers it; +- once it has finished, a message is a `continue` that queues the next turn. + +`continue` rewinds only the job's run state to `queued` and clears the runtime +turn id, so a fresh detached worker picks it up. The worker restores the target +session rather than creating it, which is what gives the follow-up turn the +previous turns as context. Its `turnId` is caller-generated, so a retry after an +ambiguous response resolves to the same turn instead of starting a second one. + +Because the event log is per job rather than per turn, the controller's cursor, +transcript cache, and projection are unchanged by follow-ups: the observer keeps +reading one growing transcript. + +## Workspace naming + +A target checkout lives at: + +```text +~/.bitfun/dispatch/worktrees//- +``` + +`repoKey` groups every checkout of one source repository under its shared clone. +The leaf mirrors the local managed-worktree convention so a target directory is +recognizable rather than a bare job UUID. The project name is advisory input from +the controller: the target sanitizes it, falls back to the remote URL's basename +and then to a constant, and rejects anything that is not a single safe path +component — the path is never shaped by an untrusted string. + ## Event and observer contract The target event log is append-only within a retained window. A status response diff --git a/src/apps/cli/src/dispatch/mod.rs b/src/apps/cli/src/dispatch/mod.rs index 32805c73be..edbc69121d 100644 --- a/src/apps/cli/src/dispatch/mod.rs +++ b/src/apps/cli/src/dispatch/mod.rs @@ -15,12 +15,12 @@ use serde::de::DeserializeOwned; use protocol::{ DispatchAnswerRequest, DispatchAnswerResponse, DispatchAppendRequest, DispatchAppendResponse, - DispatchCancelRequest, DispatchCancelResponse, DispatchJobListEntry, DispatchJobState, - DispatchListRequest, DispatchProbeRequest, DispatchProbeResponse, DispatchStatusRequest, - DispatchStatusResponse, DispatchSubmitRequest, DispatchSubmitResponse, - DispatchWorkspaceBundleBeginRequest, DispatchWorkspaceBundleChunkRequest, - DispatchWorkspaceBundleCommitRequest, DispatchWorkspaceProbe, - DispatchWorkspaceProvisionRequest, DispatchWorkspaceSyncChunkRequest, + DispatchCancelRequest, DispatchCancelResponse, DispatchContinueRequest, + DispatchContinueResponse, DispatchJobListEntry, DispatchJobState, DispatchListRequest, + DispatchProbeRequest, DispatchProbeResponse, DispatchStatusRequest, DispatchStatusResponse, + DispatchSubmitRequest, DispatchSubmitResponse, DispatchWorkspaceBundleBeginRequest, + DispatchWorkspaceBundleChunkRequest, DispatchWorkspaceBundleCommitRequest, + DispatchWorkspaceProbe, DispatchWorkspaceProvisionRequest, DispatchWorkspaceSyncChunkRequest, DispatchWorkspaceSyncRequest, DISPATCH_PROTOCOL_VERSION, MAX_DISPATCH_TEXT_BYTES, }; use store::{CreateJobOutcome, DispatchStateRecord, DispatchStore}; @@ -61,6 +61,8 @@ pub(crate) async fn run_dispatch_verb( serde_json::to_value(answer(parse(input)?)?).context("encode permission answer") } "append" => serde_json::to_value(append(parse(input)?)?).context("encode appended message"), + "continue" => serde_json::to_value(continue_job(parse(input)?)?) + .context("encode follow-up turn response"), "workspace-provision" => serde_json::to_value(workspace::provision(parse::< DispatchWorkspaceProvisionRequest, >(input)?)?) @@ -209,6 +211,44 @@ async fn submit(mut request: DispatchSubmitRequest) -> Result Result { + if request.protocol_version != DISPATCH_PROTOCOL_VERSION { + bail!( + "unsupported dispatch protocolVersion {}; target requires {}", + request.protocol_version, + DISPATCH_PROTOCOL_VERSION + ); + } + if request.prompt.trim().is_empty() { + bail!("dispatch follow-up requires a prompt"); + } + if request.prompt.len() > MAX_DISPATCH_TEXT_BYTES { + bail!("dispatch follow-up prompt exceeds the 32 KiB safety limit"); + } + if !runner::is_supported() { + bail!("dispatch detached workers are supported only on Linux and macOS"); + } + let store = DispatchStore::open_default()?; + let job = store.load_job(&request.job_id)?; + // A worker may still be settling the previous turn's terminal state. + reconcile_worker_liveness(&store, &request.job_id)?; + let state = store.queue_follow_up_turn(&request)?; + ensure_worker_spawned(&store, &request.job_id, state.state)?; + Ok(DispatchContinueResponse { + accepted: true, + job_id: request.job_id, + session_id: job.request.session_id, + turn_id: request.turn_id, + state: state.state, + }) +} + fn ensure_worker_spawned( store: &DispatchStore, job_id: &str, diff --git a/src/apps/cli/src/dispatch/protocol.rs b/src/apps/cli/src/dispatch/protocol.rs index d7f1ea04e7..0ef4faaa1f 100644 --- a/src/apps/cli/src/dispatch/protocol.rs +++ b/src/apps/cli/src/dispatch/protocol.rs @@ -167,6 +167,11 @@ pub(crate) struct DispatchWorkspaceProvisionRequest { pub(crate) repo_key: String, #[serde(default)] pub(crate) remote_url: Option, + /// Readable name of the controller's project, used to build a worktree path + /// a human can recognize. Advisory: the target sanitizes it and falls back + /// to the remote's own basename, so a hostile value cannot shape the path. + #[serde(default)] + pub(crate) project_label: Option, /// Full 40-character commit id. A ref name would be ambiguous — it can move /// between the controller resolving it and the target fetching it. pub(crate) base_commit: String, @@ -316,6 +321,35 @@ pub(crate) struct DispatchWorkspaceSyncChunkResponse { pub(crate) eof: bool, } +/// Start the next turn in a dispatch session whose previous turn has finished. +/// +/// Distinct from `append`, which steers a turn that is still running. This is +/// what lets a dispatch session hold a conversation: the target session, its +/// worktree, and its event log all persist, and only the job's run state +/// rewinds so a new worker can pick it up. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct DispatchContinueRequest { + pub(crate) protocol_version: u32, + pub(crate) job_id: String, + /// Caller-generated identity for this turn, so a retried request cannot + /// start two. + pub(crate) turn_id: String, + pub(crate) prompt: String, + #[serde(default)] + pub(crate) display_content: Option, +} + +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct DispatchContinueResponse { + pub(crate) accepted: bool, + pub(crate) job_id: String, + pub(crate) session_id: String, + pub(crate) turn_id: String, + pub(crate) state: DispatchJobState, +} + #[derive(Clone, Debug, Deserialize, PartialEq, Eq)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub(crate) struct DispatchCancelRequest { diff --git a/src/apps/cli/src/dispatch/store.rs b/src/apps/cli/src/dispatch/store.rs index fa993ef0d2..0c64ac392a 100644 --- a/src/apps/cli/src/dispatch/store.rs +++ b/src/apps/cli/src/dispatch/store.rs @@ -7,8 +7,8 @@ use bitfun_agent_runtime::sdk::{PermissionReply, PermissionRequest}; use serde::{Deserialize, Serialize}; use super::protocol::{ - DispatchAppendRequest, DispatchEvent, DispatchJobListEntry, DispatchJobState, - DispatchSubmitRequest, DISPATCH_PROTOCOL_VERSION, + DispatchAppendRequest, DispatchContinueRequest, DispatchEvent, DispatchJobListEntry, + DispatchJobState, DispatchSubmitRequest, DISPATCH_PROTOCOL_VERSION, }; const JOB_RECORD_FILE: &str = "job.json"; @@ -25,6 +25,12 @@ const PERMISSION_ANSWERS_DIR: &str = "permissions/answers"; const RESOLVED_PERMISSIONS_DIR: &str = "permissions/resolved"; const PENDING_MESSAGES_DIR: &str = "messages/pending"; const CONSUMED_MESSAGES_DIR: &str = "messages/consumed"; +/// Follow-up turns queued against a job that has already finished one. +/// +/// Distinct from the append mailbox: an appended message steers the turn that +/// is already running, while a follow-up starts the next one. +const PENDING_TURNS_DIR: &str = "turns/pending"; +const CONSUMED_TURNS_DIR: &str = "turns/consumed"; const DEFAULT_MAX_EVENTS_BYTES: u64 = 64 * 1024 * 1024; // Keep a single projected event and a complete status page comfortably below // the server transport's 256 KiB WebSocket frame ceiling. @@ -45,6 +51,19 @@ const DISPATCH_REPOS_DIR: &str = "repos"; const DISPATCH_WORKTREES_DIR: &str = "worktrees"; pub(super) const REPO_CACHE_RECORD_FILE: &str = "repo.json"; const REPO_CACHE_RETENTION_DAYS: i64 = 30; +/// Written by the workspace layer; read here only to find a job's checkout. +const PROVISION_RECORD_FILE: &str = "provision.json"; + +/// The one field retention needs from a provision record. +/// +/// Deliberately not the workspace layer's full record: this only has to survive +/// that struct gaining fields, and reading fewer fields cannot fail on one. +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ProvisionedWorktreeRecord { + #[serde(default)] + workspace_path: Option, +} #[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] #[serde(rename_all = "camelCase")] @@ -134,6 +153,17 @@ struct StoredAppendMessage { created_at: String, } +/// One queued follow-up turn. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct StoredFollowUpTurn { + pub(crate) turn_id: String, + pub(crate) prompt: String, + #[serde(default)] + pub(crate) display_content: Option, + pub(crate) created_at: String, +} + #[derive(Clone, Debug, Deserialize)] #[serde(rename_all = "camelCase")] struct RepoCacheRetentionRecord { @@ -426,17 +456,6 @@ impl DispatchStore { Ok(state) } - pub(crate) fn record_turn_id(&self, job_id: &str, turn_id: &str) -> Result<()> { - let job_dir = self.existing_job_dir(job_id)?; - let _lock = JobLock::exclusive(&job_dir.join(".lock"))?; - let mut state = self.load_state_unlocked(&job_dir)?; - if !state.state.is_terminal() && state.turn_id.as_deref() != Some(turn_id) { - state.turn_id = Some(turn_id.to_string()); - atomic_write_json(&job_dir.join(STATE_FILE), &state)?; - } - Ok(()) - } - pub(crate) fn try_claim_worker_spawn(&self, job_id: &str) -> Result> { let job_dir = self.existing_job_dir(job_id)?; let Some(lease) = DispatchLease::try_acquire(&job_dir.join(SPAWN_LOCK_FILE))? else { @@ -676,6 +695,100 @@ impl DispatchStore { } } + /// Queue the next turn for a job whose previous turn has finished. + /// + /// This is what makes a dispatch session a conversation rather than a + /// one-shot: the target session, its worktree, and its event log all stay + /// put, and only the job's run state rewinds to `Queued`. + pub(crate) fn queue_follow_up_turn( + &self, + request: &DispatchContinueRequest, + ) -> Result { + validate_id("turnId", &request.turn_id)?; + let job_dir = self.existing_job_dir(&request.job_id)?; + let _lock = JobLock::exclusive(&job_dir.join(".lock"))?; + + let stored = StoredFollowUpTurn { + turn_id: request.turn_id.clone(), + prompt: request.prompt.clone(), + display_content: request.display_content.clone(), + created_at: chrono::Utc::now().to_rfc3339(), + }; + // A retried request must not start a second turn. Both mailboxes are + // checked because the worker may already have claimed this one. + let consumed_path = mailbox_path(&job_dir, CONSUMED_TURNS_DIR, &request.turn_id)?; + if let Some(existing) = read_optional_regular_json::(&consumed_path)? { + if existing.prompt != request.prompt { + bail!("dispatch turnId is already bound to different content"); + } + return self.load_state_unlocked(&job_dir); + } + let pending_path = mailbox_path(&job_dir, PENDING_TURNS_DIR, &request.turn_id)?; + if let Some(existing) = read_optional_regular_json::(&pending_path)? { + if existing.prompt != request.prompt { + bail!("dispatch turnId is already bound to different content"); + } + return self.load_state_unlocked(&job_dir); + } + + let mut state = self.load_state_unlocked(&job_dir)?; + if !state.state.is_terminal() { + bail!("this dispatch job is still running; steer it with an appended message instead"); + } + write_json_if_absent_or_equal(&pending_path, &stored)?; + + // Rewind only the run state. `started_at` is left alone so the job keeps + // reporting when its first turn began. + state.state = DispatchJobState::Queued; + state.turn_id = None; + state.finished_at = None; + state.last_error = None; + state.cancel_requested_at = None; + atomic_write_json(&job_dir.join(STATE_FILE), &state)?; + self.append_event_unlocked( + &job_dir, + &DispatchEvent::job_state(DispatchJobState::Queued, None), + )?; + Ok(state) + } + + /// Take the next queued turn and bind it to the runtime turn the worker is + /// about to submit. + /// + /// Consuming and recording the turn id happen under one lock so a crash can + /// never leave a turn that looks unclaimed but was already submitted. A + /// crash after this point settles the job as failed rather than replaying + /// the prompt, which is the same promise the first turn makes. + pub(crate) fn claim_follow_up_turn( + &self, + job_id: &str, + runtime_turn_id: &str, + ) -> Result> { + let job_dir = self.existing_job_dir(job_id)?; + let _lock = JobLock::exclusive(&job_dir.join(".lock"))?; + let mut pending = + read_json_directory::(&job_dir.join(PENDING_TURNS_DIR))?; + pending.sort_by(|left, right| { + left.created_at + .cmp(&right.created_at) + .then_with(|| left.turn_id.cmp(&right.turn_id)) + }); + let claimed = pending.into_iter().next(); + + let mut state = self.load_state_unlocked(&job_dir)?; + if !state.state.is_terminal() && state.turn_id.as_deref() != Some(runtime_turn_id) { + state.turn_id = Some(runtime_turn_id.to_string()); + atomic_write_json(&job_dir.join(STATE_FILE), &state)?; + } + + if let Some(turn) = claimed.as_ref() { + let consumed_path = mailbox_path(&job_dir, CONSUMED_TURNS_DIR, &turn.turn_id)?; + write_json_if_absent_or_equal(&consumed_path, turn)?; + remove_file_if_present(&mailbox_path(&job_dir, PENDING_TURNS_DIR, &turn.turn_id)?); + } + Ok(claimed) + } + pub(crate) fn enqueue_append_message(&self, request: DispatchAppendRequest) -> Result { validate_id("messageId", &request.message_id)?; let job_dir = self.existing_job_dir(&request.job_id)?; @@ -903,9 +1016,23 @@ impl DispatchStore { self.root.join(DISPATCH_WORKTREES_DIR) } - pub(crate) fn worktree_dir(&self, job_id: &str) -> Result { - validate_id("jobId", job_id)?; - Ok(self.worktrees_root().join(job_id)) + /// Checkout directory for one job, grouped under its repository's clone. + /// + /// `directory_name` is built by the workspace layer from a sanitized project + /// label; re-check it here so this path constructor is safe on its own and + /// cannot be walked out of the worktree root. + pub(crate) fn worktree_dir(&self, repo_key: &str, directory_name: &str) -> Result { + self.repo_dir(repo_key)?; + if directory_name.is_empty() + || directory_name.len() > 128 + || !directory_name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) + || directory_name.starts_with('.') + { + bail!("dispatch worktree directory name is not a safe path component"); + } + Ok(self.worktrees_root().join(repo_key).join(directory_name)) } fn maybe_collect_expired_terminal_jobs(&self) -> Result<()> { @@ -1098,6 +1225,12 @@ impl DispatchStore { else { continue; }; + // The checkout is named after the project, not the job, so the + // provision record is the only link back to it. Remove it before + // the record that points at it, or it becomes unreachable. + if !self.remove_recorded_worktree(&workspace_dir)? { + continue; + } let tombstone = workspaces_root.join(format!( ".gc-{}-{}", job_id, @@ -1117,89 +1250,71 @@ impl DispatchStore { self.remove_workspace_operation_locks(&job_id)?; removed += 1; } - removed += self.collect_orphaned_worktrees(&jobs_root)?; self.collect_expired_repo_clones(now)?; Ok(removed) } - /// Remove worktrees whose job record is gone. + /// Remove the checkout a departing job's provision record points at. /// /// The directory is only the checkout: every commit made in it was fetched - /// into the shared clone during sync, so removing it cannot lose work that - /// the controller pulled. Work the controller never pulled is discarded - /// along with the job it belonged to, which is the same retention promise - /// the event log makes. + /// into the shared clone during sync, so removing it cannot lose work the + /// controller pulled. Work the controller never pulled is discarded with the + /// job it belonged to, which is the same promise the event log makes. /// - /// Stale worktree administrative entries left inside the clone are pruned - /// by the next provision, which always runs `git worktree prune` first. - fn collect_orphaned_worktrees(&self, jobs_root: &Path) -> Result { - let worktrees_root = self.worktrees_root(); - let mut removed = 0; - for entry in fs::read_dir(&worktrees_root) - .with_context(|| format!("read dispatch worktrees {}", worktrees_root.display()))? - { - let entry = entry?; - let Some(job_id) = entry.file_name().to_str().map(ToOwned::to_owned) else { - continue; - }; - if validate_id("jobId", &job_id).is_err() || jobs_root.join(&job_id).exists() { - continue; - } - let worktree_dir = entry.path(); - let metadata = fs::symlink_metadata(&worktree_dir)?; - if metadata.file_type().is_symlink() || !metadata.is_dir() { - continue; - } - let old_enough = metadata - .modified() - .ok() - .and_then(|modified| modified.elapsed().ok()) - .is_some_and(|elapsed| { - elapsed.as_secs() >= (TERMINAL_JOB_RETENTION_DAYS as u64) * 24 * 60 * 60 - }); - if !old_enough { - continue; - } - let Some(operation_lock) = - JobLock::try_exclusive(&self.workspace_operation_lock_path(&job_id)?)? - else { - continue; - }; - let Some(git_operation_lock) = - JobLock::try_exclusive(&self.workspace_git_operation_lock_path(&job_id)?)? - else { - continue; - }; - let canonical_worktree = std::fs::canonicalize(&worktree_dir) - .unwrap_or_else(|_| worktree_dir.clone()) - .to_string_lossy() - .to_string(); - let Some(workspace_runtime_lock) = - WorkspaceLock::try_acquire(&self.workspace_lock_path(&canonical_worktree))? - else { - continue; - }; - let tombstone = worktrees_root.join(format!( - ".gc-{}-{}", - job_id, - uuid::Uuid::new_v4().as_simple() - )); - fs::rename(&worktree_dir, &tombstone).with_context(|| { - format!( - "quarantine orphaned dispatch worktree {}", - worktree_dir.display() - ) - })?; - fs::remove_dir_all(&tombstone).with_context(|| { - format!("remove orphaned dispatch worktree {}", tombstone.display()) - })?; - drop(workspace_runtime_lock); - drop(git_operation_lock); - drop(operation_lock); - self.remove_workspace_operation_locks(&job_id)?; - removed += 1; + /// Returns `false` when the worktree is still busy, so the caller leaves the + /// record in place and retries on the next sweep rather than orphaning it. + /// Stale worktree administrative entries inside the clone are pruned by the + /// next provision, which always runs `git worktree prune` first. + fn remove_recorded_worktree(&self, workspace_dir: &Path) -> Result { + let Ok(record) = + read_json::(&workspace_dir.join(PROVISION_RECORD_FILE)) + else { + // No record, or one written before checkouts were recorded: there is + // nothing this sweep can safely delete. + return Ok(true); + }; + let Some(path) = record.workspace_path else { + return Ok(true); + }; + let worktree_dir = PathBuf::from(&path); + // Only ever delete inside the managed worktree root, whatever the record + // claims. A record is target-owned, but this keeps one corrupt file from + // turning into an arbitrary recursive delete. + if !worktree_dir.starts_with(self.worktrees_root()) { + tracing::warn!("Skipping dispatch worktree outside the managed root: {path}"); + return Ok(true); } - Ok(removed) + let metadata = match fs::symlink_metadata(&worktree_dir) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(true), + Err(error) => return Err(error.into()), + }; + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Ok(true); + } + let canonical = std::fs::canonicalize(&worktree_dir) + .unwrap_or_else(|_| worktree_dir.clone()) + .to_string_lossy() + .to_string(); + let Some(workspace_runtime_lock) = + WorkspaceLock::try_acquire(&self.workspace_lock_path(&canonical))? + else { + return Ok(false); + }; + let tombstone = self + .worktrees_root() + .join(format!(".gc-{}", uuid::Uuid::new_v4().as_simple())); + fs::rename(&worktree_dir, &tombstone).with_context(|| { + format!( + "quarantine orphaned dispatch worktree {}", + worktree_dir.display() + ) + })?; + fs::remove_dir_all(&tombstone).with_context(|| { + format!("remove orphaned dispatch worktree {}", tombstone.display()) + })?; + drop(workspace_runtime_lock); + Ok(true) } fn collect_expired_repo_clones(&self, now: chrono::DateTime) -> Result<()> { @@ -2039,6 +2154,100 @@ mod tests { assert!(recovered.cursor > initial.cursor); } + fn continue_request(job_id: &str, turn_id: &str, prompt: &str) -> DispatchContinueRequest { + DispatchContinueRequest { + protocol_version: DISPATCH_PROTOCOL_VERSION, + job_id: job_id.to_string(), + turn_id: turn_id.to_string(), + prompt: prompt.to_string(), + display_content: None, + } + } + + #[test] + fn a_follow_up_turn_requeues_a_finished_job_and_is_claimed_once() { + let (_dir, store) = store(); + store + .create_job(request("job-1"), "job title".to_string()) + .expect("create job"); + store + .mark_state("job-1", DispatchJobState::Running, Some("turn-1"), None) + .expect("running"); + store + .mark_state("job-1", DispatchJobState::Succeeded, None, None) + .expect("succeeded"); + + let state = store + .queue_follow_up_turn(&continue_request("job-1", "turn-2", "and now this")) + .expect("queue follow-up"); + assert_eq!(state.state, DispatchJobState::Queued); + // The previous turn's identity must not survive, or the next worker + // would refuse to run believing a turn was already submitted. + assert!(state.turn_id.is_none()); + assert!(state.finished_at.is_none()); + + let claimed = store + .claim_follow_up_turn("job-1", "runtime-turn-2") + .expect("claim") + .expect("a queued turn"); + assert_eq!(claimed.prompt, "and now this"); + assert_eq!( + store.load_state("job-1").expect("state").turn_id.as_deref(), + Some("runtime-turn-2") + ); + // A second worker must find nothing left to run. + assert!(store + .claim_follow_up_turn("job-1", "runtime-turn-3") + .expect("second claim") + .is_none()); + } + + #[test] + fn a_retried_follow_up_request_never_starts_a_second_turn() { + let (_dir, store) = store(); + store + .create_job(request("job-1"), "job title".to_string()) + .expect("create job"); + store + .mark_state("job-1", DispatchJobState::Succeeded, None, None) + .expect("succeeded"); + + let queued = continue_request("job-1", "turn-2", "and now this"); + store.queue_follow_up_turn(&queued).expect("first"); + store.queue_follow_up_turn(&queued).expect("retry"); + store + .claim_follow_up_turn("job-1", "runtime-turn-2") + .expect("claim"); + // Even a retry that arrives after the worker claimed the turn is a + // no-op rather than a duplicate submission. + store.queue_follow_up_turn(&queued).expect("late retry"); + assert!(store + .claim_follow_up_turn("job-1", "runtime-turn-3") + .expect("claim again") + .is_none()); + + let conflicting = continue_request("job-1", "turn-2", "something else"); + assert!(store.queue_follow_up_turn(&conflicting).is_err()); + } + + #[test] + fn a_running_job_refuses_a_follow_up_turn() { + let (_dir, store) = store(); + store + .create_job(request("job-1"), "job title".to_string()) + .expect("create job"); + store + .mark_state("job-1", DispatchJobState::Running, Some("turn-1"), None) + .expect("running"); + + // Steering a live turn is what `append` is for; starting a second turn + // underneath a running one would race the worker. + let error = store + .queue_follow_up_turn(&continue_request("job-1", "turn-2", "next")) + .expect_err("a running job cannot take a follow-up"); + assert!(error.to_string().contains("still running")); + } + #[test] fn terminal_state_is_idempotent() { let (_dir, store) = store(); diff --git a/src/apps/cli/src/dispatch/worker.rs b/src/apps/cli/src/dispatch/worker.rs index bfab4f1969..96d1b0e6f1 100644 --- a/src/apps/cli/src/dispatch/worker.rs +++ b/src/apps/cli/src/dispatch/worker.rs @@ -4,9 +4,9 @@ use std::time::Duration; use anyhow::{anyhow, bail, Context, Result}; use bitfun_agent_runtime::sdk::{ - AgentDialogTurnRequest, AgentSessionCreateRequest, AgentTurnCancellationRequest, - AgentTurnSettlementRequest, PermissionReply, PermissionReplySource, PermissionRequest, - PermissionRequestEvent, + AgentDialogTurnRequest, AgentSessionCreateRequest, AgentSessionRestoreRequest, + AgentTurnCancellationRequest, AgentTurnSettlementRequest, PermissionReply, + PermissionReplySource, PermissionRequest, PermissionRequestEvent, }; use bitfun_events::{project_agentic_frontend_event, AgenticEvent}; use bitfun_runtime_ports::{AgentSubmissionSource, DialogSubmissionPolicy, SessionExecutionTarget}; @@ -114,35 +114,55 @@ async fn run_inner(store: &DispatchStore, job_id: &str) -> Result<()> { .map_err(|error| anyhow!(error.into_message()))?; let workspace_path = job.request.workspace_path.clone(); - agent_runtime - .create_session_with_id( - job.request.session_id.clone(), - AgentSessionCreateRequest { - session_name: job.title.clone(), - agent_type: job.request.agent_type.clone(), - workspace_path: Some(workspace_path.clone()), - project_workspace_path: Some(workspace_path.clone()), - execution_target: Some(SessionExecutionTarget::local(workspace_path.clone())), - workspace_id: None, - remote_connection_id: None, - remote_ssh_host: None, - model_id: job.request.model.clone(), - metadata: serde_json::Map::new(), - }, - ) + // A follow-up turn runs against the session the previous turn built, so its + // history is the agent's context. Restoring first is what makes a dispatch + // session a conversation; creating unconditionally would fail on the second + // turn because the persisted id already exists. + let restored = agent_runtime + .restore_session(AgentSessionRestoreRequest { + workspace_path: workspace_path.clone(), + session_id: job.request.session_id.clone(), + include_internal: false, + remote_connection_id: None, + remote_ssh_host: None, + }) .await - .map_err(|error| anyhow!(error.into_message())) - .context("create target-owned dispatch session")?; + .is_ok(); + if !restored { + agent_runtime + .create_session_with_id( + job.request.session_id.clone(), + AgentSessionCreateRequest { + session_name: job.title.clone(), + agent_type: job.request.agent_type.clone(), + workspace_path: Some(workspace_path.clone()), + project_workspace_path: Some(workspace_path.clone()), + execution_target: Some(SessionExecutionTarget::local(workspace_path.clone())), + workspace_id: None, + remote_connection_id: None, + remote_ssh_host: None, + model_id: job.request.model.clone(), + metadata: serde_json::Map::new(), + }, + ) + .await + .map_err(|error| anyhow!(error.into_message())) + .context("create target-owned dispatch session")?; + } let turn_id = uuid::Uuid::new_v4().to_string(); - // Persist the deterministic turn id before submission. A crash after the - // Runtime accepts the turn must never make a replacement worker submit the - // prompt a second time. - store.record_turn_id(job_id, &turn_id)?; + // Claim the queued follow-up and persist the turn id in one step. A crash + // after the Runtime accepts the turn must never make a replacement worker + // submit the prompt a second time. + let follow_up = store.claim_follow_up_turn(job_id, &turn_id)?; + let prompt = follow_up + .as_ref() + .map(|turn| turn.prompt.clone()) + .unwrap_or_else(|| job.request.prompt.clone()); agent_runtime .submit_dialog_turn(AgentDialogTurnRequest { session_id: job.request.session_id.clone(), - message: job.request.prompt.clone(), + message: prompt, original_message: None, turn_id: Some(turn_id.clone()), agent_type: job.request.agent_type.clone(), diff --git a/src/apps/cli/src/dispatch/workspace.rs b/src/apps/cli/src/dispatch/workspace.rs index 7630f2c58b..bad11e156a 100644 --- a/src/apps/cli/src/dispatch/workspace.rs +++ b/src/apps/cli/src/dispatch/workspace.rs @@ -41,6 +41,11 @@ const BUNDLE_RECORD_FILE: &str = "bundle.json"; const SYNC_OPERATION_FILE: &str = "sync-operation.json"; const INCOMING_BUNDLE_FILE: &str = "incoming.bundle"; const RESULT_BUNDLE_FILE: &str = "result.bundle"; +/// Short job-id suffix that keeps two dispatches of one project apart, matching +/// the local managed-worktree convention. +const WORKTREE_SUFFIX_CHARS: usize = 8; +/// Upper bound on the readable half of a worktree directory name. +const WORKTREE_LABEL_MAX_CHARS: usize = 48; const MAX_CHUNK_BYTES: usize = 256 * 1024; const MAX_CHUNK_BASE64_BYTES: usize = 384 * 1024; /// Ceiling for one delivered bundle. Generous for source history, small enough @@ -134,6 +139,8 @@ struct ProvisionRecord { base_commit: String, branch: String, created_at: String, + /// Resolved checkout directory. Recorded rather than recomputed so the path + /// stays stable even if the naming rules change under an existing job. #[serde(default)] workspace_path: Option, } @@ -323,7 +330,20 @@ fn provision_in_store( let _repo_lock = JobLock::exclusive(&store.repo_lock_path(&request.repo_key)?)?; - let worktree_path = store.worktree_dir(&request.job_id)?; + // Reuse the recorded path when there is one: a job keeps the directory it + // was first given, whatever the current naming rules would produce. + let existing_record: ProvisionRecord = read_json(&record_path)?; + let worktree_path = match existing_record.workspace_path.as_deref() { + Some(path) => PathBuf::from(path), + None => store.worktree_dir( + &request.repo_key, + &worktree_directory_name( + request.project_label.as_deref(), + request.remote_url.as_deref(), + &request.job_id, + ), + )?, + }; if let Some(existing) = existing_worktree(&worktree_path, &request.branch, &request.base_commit)? { @@ -971,7 +991,12 @@ fn sync_in_store( let provision: ProvisionRecord = read_json(&job_dir.join(PROVISION_RECORD_FILE)) .context("this job did not receive a Git workspace")?; let _repo_lock = JobLock::exclusive(&store.repo_lock_path(&provision.repo_key)?)?; - let worktree = store.worktree_dir(&request.job_id)?; + let worktree = PathBuf::from( + provision + .workspace_path + .as_deref() + .context("this job's Git workspace was never checked out")?, + ); if !is_real_directory(&worktree) { bail!("the dispatch worktree is missing"); } @@ -1327,6 +1352,64 @@ fn create_worktree( canonical_utf8(worktree_path) } +/// Leaf directory name for a job's checkout. +/// +/// Mirrors the local managed-worktree convention (`-`) so a +/// target directory is recognizable rather than a bare job UUID. The label is +/// advisory input from the controller, so it is sanitized here and falls back to +/// the remote URL's basename and finally to a constant — the path must never be +/// shaped by an untrusted string. +fn worktree_directory_name( + project_label: Option<&str>, + remote_url: Option<&str>, + job_id: &str, +) -> String { + let label = sanitize_label(project_label.unwrap_or_default()) + .or_else(|| sanitize_label(&remote_basename(remote_url.unwrap_or_default()))) + .unwrap_or_else(|| "workspace".to_string()); + let suffix = job_id + .chars() + .filter(|character| character.is_ascii_alphanumeric()) + .take(WORKTREE_SUFFIX_CHARS) + .collect::(); + if suffix.is_empty() { + label + } else { + format!("{label}-{suffix}") + } +} + +fn sanitize_label(value: &str) -> Option { + let cleaned = value + .chars() + .map(|character| { + if character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.') { + character + } else { + '-' + } + }) + .collect::(); + let trimmed = cleaned.trim_matches(|character| character == '-' || character == '.'); + let bounded = trimmed + .chars() + .take(WORKTREE_LABEL_MAX_CHARS) + .collect::(); + let bounded = bounded.trim_end_matches(['-', '.']).to_string(); + (!bounded.is_empty()).then_some(bounded) +} + +/// `git@host:acme/app.git` and `https://host/acme/app.git` both yield `app`. +fn remote_basename(remote_url: &str) -> String { + remote_url + .trim_end_matches('/') + .rsplit(['/', ':']) + .next() + .unwrap_or_default() + .trim_end_matches(".git") + .to_string() +} + fn commit_exists(repo: &Path, commit: &str) -> Result { git_succeeds(repo, &["cat-file", "-e", &format!("{commit}^{{commit}}")]) } @@ -1598,6 +1681,7 @@ mod tests { protocol_version: DISPATCH_PROTOCOL_VERSION, job_id: "job-1".to_string(), repo_key: "abcdef0123456789".to_string(), + project_label: Some("BitFun".to_string()), remote_url: None, base_commit: "0".repeat(40), branch: "bitfun/dispatch/job-1".to_string(), @@ -1624,6 +1708,7 @@ mod tests { job_id: "job-1".to_string(), repo_key: "abcdef0123456789".to_string(), remote_url: None, + project_label: Some("BitFun".to_string()), base_commit: base_commit.clone(), branch: "main".to_string(), }; @@ -1677,7 +1762,7 @@ mod tests { let base_commit = init_source_repository(&source); provision_from_bundle(&store, &source, &base_commit); - let worktree = store.worktree_dir("job-1").expect("worktree"); + let worktree = provisioned_worktree(&store, "job-1"); fs::write(worktree.join("agent.txt"), b"valuable work").expect("edit"); git(&worktree, &["add", "-A"]).expect("stage"); git( @@ -1719,6 +1804,7 @@ mod tests { protocol_version: DISPATCH_PROTOCOL_VERSION, job_id: "job-1".to_string(), repo_key: "abcdef0123456789".to_string(), + project_label: Some("BitFun".to_string()), remote_url: None, base_commit, branch: "main".to_string(), @@ -1768,7 +1854,7 @@ mod tests { let base_commit = init_source_repository(&source); provision_from_bundle(&store, &source, &base_commit); - let worktree = store.worktree_dir("job-1").expect("worktree"); + let worktree = provisioned_worktree(&store, "job-1"); git(&worktree, &["config", "user.email", "dispatch@example.com"]).expect("email"); git(&worktree, &["config", "user.name", "Dispatch Test"]).expect("name"); fs::write(worktree.join("file.txt"), b"changed by the agent").expect("edit"); @@ -1816,7 +1902,7 @@ mod tests { let source = temp.path().join("source"); let base_commit = init_source_repository(&source); provision_from_bundle(&store, &source, &base_commit); - let worktree = store.worktree_dir("job-1").expect("worktree"); + let worktree = provisioned_worktree(&store, "job-1"); fs::write(worktree.join("first.txt"), b"first checkpoint").expect("first edit"); let first = sync_in_store( @@ -2135,7 +2221,7 @@ mod tests { let source = temp.path().join("source"); let base_commit = init_source_repository(&source); provision_from_bundle(&store, &source, &base_commit); - let worktree = store.worktree_dir("job-1").expect("worktree"); + let worktree = provisioned_worktree(&store, "job-1"); git(&worktree, &["switch", "--quiet", "-c", "agent/other"]).expect("switch branch"); let error = sync_in_store( @@ -2167,6 +2253,7 @@ mod tests { protocol_version: DISPATCH_PROTOCOL_VERSION, job_id: "job-1".to_string(), repo_key: "abcdef0123456789".to_string(), + project_label: Some("BitFun".to_string()), remote_url: None, base_commit: "0".repeat(40), branch: "bitfun/dispatch/job-1".to_string(), @@ -2193,6 +2280,18 @@ mod tests { ); } + /// Resolve a job's checkout the way production does: from its record. + fn provisioned_worktree(store: &DispatchStore, job_id: &str) -> PathBuf { + let record: ProvisionRecord = read_json( + &store + .workspace_upload_dir(job_id) + .expect("job dir") + .join(PROVISION_RECORD_FILE), + ) + .expect("provision record"); + PathBuf::from(record.workspace_path.expect("checked-out workspace")) + } + fn provision_from_bundle(store: &DispatchStore, source: &Path, base_commit: &str) { let bundle = source.parent().expect("parent").join("base.bundle"); bundle_everything(source, &bundle); @@ -2201,6 +2300,7 @@ mod tests { job_id: "job-1".to_string(), repo_key: "abcdef0123456789".to_string(), remote_url: None, + project_label: Some("BitFun".to_string()), base_commit: base_commit.to_string(), branch: "main".to_string(), }; @@ -2227,6 +2327,48 @@ mod tests { provision_in_store(store, request).expect("second provision"); } + #[test] + fn worktree_directories_are_named_after_the_project_not_the_job() { + assert_eq!( + worktree_directory_name(Some("BitFun"), None, "dispatch-3d82ff46-bbf9-44c3"), + "BitFun-dispatch" + ); + // No label: the remote's own basename is the next most recognizable name. + assert_eq!( + worktree_directory_name(None, Some("git@example.com:acme/app.git"), "abcdef123456"), + "app-abcdef12" + ); + assert_eq!( + worktree_directory_name(None, Some("https://example.com/acme/app/"), "abcdef123456"), + "app-abcdef12" + ); + // Neither available: a constant, never an empty or job-shaped path. + assert_eq!( + worktree_directory_name(None, None, "abcdef123456"), + "workspace-abcdef12" + ); + } + + #[test] + fn a_hostile_project_label_cannot_shape_the_worktree_path() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = DispatchStore::open(temp.path().join("dispatch")).expect("store"); + + for label in ["../../etc", "..", "/absolute", ".hidden", "", " "] { + let name = worktree_directory_name(Some(label), None, "job1"); + assert!(!name.contains('/'), "{label} produced a path separator"); + assert!(!name.contains(".."), "{label} produced a traversal"); + assert!(!name.starts_with('.'), "{label} produced a hidden entry"); + let path = store.worktree_dir("abcdef0123456789", &name).expect("path"); + assert!(path.starts_with(store.worktrees_root())); + } + + // The store refuses anything the workspace layer did not sanitize. + assert!(store.worktree_dir("abcdef0123456789", "../escape").is_err()); + assert!(store.worktree_dir("abcdef0123456789", ".git").is_err()); + assert!(store.worktree_dir("abcdef0123456789", "").is_err()); + } + #[test] fn hostile_provisioning_inputs_are_rejected_before_git_runs() { assert!(validate_repo_key("../escape").is_err()); diff --git a/src/apps/cli/src/peer_host/dispatch.rs b/src/apps/cli/src/peer_host/dispatch.rs index 329555b154..a3bb692c1f 100644 --- a/src/apps/cli/src/peer_host/dispatch.rs +++ b/src/apps/cli/src/peer_host/dispatch.rs @@ -128,6 +128,7 @@ fn dispatch_target_verb(command: &str) -> Option<&'static str> { "dispatch_target_list" => Some("list"), "dispatch_target_answer" => Some("answer"), "dispatch_target_append" => Some("append"), + "dispatch_target_continue" => Some("continue"), "dispatch_target_workspace_provision" => Some("workspace-provision"), "dispatch_target_workspace_bundle_begin" => Some("workspace-bundle-begin"), "dispatch_target_workspace_bundle_chunk" => Some("workspace-bundle-chunk"), diff --git a/src/apps/desktop/src/api/dispatch_api.rs b/src/apps/desktop/src/api/dispatch_api.rs index b466173b13..68a714cd16 100644 --- a/src/apps/desktop/src/api/dispatch_api.rs +++ b/src/apps/desktop/src/api/dispatch_api.rs @@ -14,16 +14,17 @@ use bitfun_core::infrastructure::PathManager; use bitfun_core::service::dispatch::{ answer_device_dispatch, answer_dispatch, append_device_dispatch, append_dispatch, cancel_device_dispatch, cancel_dispatch, cancel_dispatch_cli_install, - get_device_dispatch_status, get_dispatch_status, list_device_dispatch_jobs, list_dispatch_jobs, - list_dispatch_targets, poll_dispatch_cli_install, probe_device_dispatch_target, - probe_dispatch_target, start_dispatch_cli_install, start_dispatch_cli_source_build, - submit_device_dispatch, submit_dispatch, sync_device_dispatch_result, - sync_dispatch_model_config, sync_dispatch_result, DeviceDispatchRpc, DispatchAnswerRequest, - DispatchAppendRequest, DispatchConnectionRequest, DispatchInstallPollRequest, - DispatchInstallStartRequest, DispatchJobRequest, DispatchListJobsRequest, - DispatchListTargetsRequest, DispatchProbeTargetRequest, DispatchSaveTranscriptRequest, - DispatchStatusRequest, DispatchSubmitRequest, DispatchSyncResultRequest, DispatchTarget, - DispatchTargetOption, DispatchTargetRequest, DispatchTranscriptRequest, OutboundDispatchStore, + continue_device_dispatch_job, continue_dispatch_job, get_device_dispatch_status, + get_dispatch_status, list_device_dispatch_jobs, list_dispatch_jobs, list_dispatch_targets, + poll_dispatch_cli_install, probe_device_dispatch_target, probe_dispatch_target, + start_dispatch_cli_install, start_dispatch_cli_source_build, submit_device_dispatch, + submit_dispatch, sync_device_dispatch_result, sync_dispatch_model_config, sync_dispatch_result, + DeviceDispatchRpc, DispatchAnswerRequest, DispatchAppendRequest, DispatchConnectionRequest, + DispatchContinueRequest, DispatchInstallPollRequest, DispatchInstallStartRequest, + DispatchJobRequest, DispatchListJobsRequest, DispatchListTargetsRequest, + DispatchProbeTargetRequest, DispatchSaveTranscriptRequest, DispatchStatusRequest, + DispatchSubmitRequest, DispatchSyncResultRequest, DispatchTarget, DispatchTargetOption, + DispatchTargetRequest, DispatchTranscriptRequest, OutboundDispatchStore, }; use bitfun_core::service::remote_ssh::dispatch_ssh::{ DispatchInstallPoll, DispatchInstallStart, DispatchSshProbe, @@ -392,6 +393,39 @@ pub async fn dispatch_sync_result( .map_err(|error| error.to_string()) } +/// Start the next turn of a dispatch session. +/// +/// A dispatch session is a conversation, not a single exchange: this reuses the +/// target's session, worktree, and event log, so the projection keeps growing +/// as one transcript. +#[tauri::command] +pub async fn dispatch_continue( + state: State<'_, AppState>, + path_manager: State<'_, Arc>, + request: DispatchContinueRequest, +) -> Result { + let store = OutboundDispatchStore::new(path_manager.as_ref()); + if matches!( + store + .get(&request.job_id) + .await + .map_err(|error| error.to_string())? + .map(|record| record.target), + Some(DispatchTarget::Device { .. }) + ) { + return continue_device_dispatch_job(&AccountDeviceDispatchRpc, &store, request) + .await + .map_err(|error| error.to_string()); + } + let manager = state + .get_ssh_manager_async() + .await + .map_err(|error| error.to_string())?; + continue_dispatch_job(&manager, &store, request) + .await + .map_err(|error| error.to_string()) +} + #[tauri::command] pub async fn dispatch_cancel( state: State<'_, AppState>, diff --git a/src/apps/desktop/src/api/dispatch_host.rs b/src/apps/desktop/src/api/dispatch_host.rs index 1c7d654502..9c66eed426 100644 --- a/src/apps/desktop/src/api/dispatch_host.rs +++ b/src/apps/desktop/src/api/dispatch_host.rs @@ -39,6 +39,7 @@ fn target_cli_verb(command: &str) -> Option<&'static str> { "dispatch_target_list" => Some("list"), "dispatch_target_answer" => Some("answer"), "dispatch_target_append" => Some("append"), + "dispatch_target_continue" => Some("continue"), "dispatch_target_workspace_provision" => Some("__workspace_provision"), "dispatch_target_workspace_bundle_begin" => Some("__workspace_bundle_begin"), "dispatch_target_workspace_bundle_chunk" => Some("__workspace_bundle_chunk"), diff --git a/src/apps/desktop/src/api/remote_workspace_policy.rs b/src/apps/desktop/src/api/remote_workspace_policy.rs index bf4a34f805..452d70e836 100644 --- a/src/apps/desktop/src/api/remote_workspace_policy.rs +++ b/src/apps/desktop/src/api/remote_workspace_policy.rs @@ -379,6 +379,10 @@ pub const REMOTE_WORKSPACE_COMMAND_POLICIES: &[(&str, RemoteWorkspacePolicy)] = ), ("dispatch_answer", RemoteWorkspacePolicy::WorkspaceAgnostic), ("dispatch_append", RemoteWorkspacePolicy::WorkspaceAgnostic), + ( + "dispatch_continue", + RemoteWorkspacePolicy::WorkspaceAgnostic, + ), ( "dispatch_load_transcript", RemoteWorkspacePolicy::WorkspaceAgnostic, diff --git a/src/apps/desktop/src/lib.rs b/src/apps/desktop/src/lib.rs index 6dff7f5834..44cec9957d 100644 --- a/src/apps/desktop/src/lib.rs +++ b/src/apps/desktop/src/lib.rs @@ -1759,6 +1759,7 @@ pub async fn run() { api::dispatch_api::dispatch_list_jobs, api::dispatch_api::dispatch_answer, api::dispatch_api::dispatch_append, + api::dispatch_api::dispatch_continue, api::dispatch_api::dispatch_load_transcript, api::dispatch_api::dispatch_save_transcript, // Relay self-deploy API diff --git a/src/crates/assembly/core/src/service/dispatch/baseline.rs b/src/crates/assembly/core/src/service/dispatch/baseline.rs index 0df0e0e144..fb2e4c2d3b 100644 --- a/src/crates/assembly/core/src/service/dispatch/baseline.rs +++ b/src/crates/assembly/core/src/service/dispatch/baseline.rs @@ -29,6 +29,8 @@ const UNCOMMITTED_COMMIT_MESSAGE: &str = "BitFun dispatch: uncommitted baseline #[derive(Debug, Clone)] pub(super) struct PreparedBaseline { + /// Readable project name the target uses to name its checkout. + pub(super) project_label: String, pub(super) delivery: DispatchWorkspaceDelivery, /// Absolute path of the managed worktree on this controller. pub(super) worktree_path: String, @@ -116,6 +118,7 @@ pub(super) async fn prepare_baseline( let repo_key = repo_key(remote_url.as_deref(), &repository.common_git_dir); Ok(PreparedBaseline { + project_label: project_label(&project_workspace_path), delivery: DispatchWorkspaceDelivery { source_workspace_path: project.to_string(), project_workspace_path: project_workspace_path.clone(), @@ -519,6 +522,18 @@ fn sanitized_stem(value: &str) -> String { .collect() } +/// Readable name for the project a dispatch came from. +/// +/// The target sanitizes this again before it becomes a path component, so this +/// only has to be recognizable — a directory basename is exactly that, and it +/// matches what the local managed-worktree naming already shows the user. +fn project_label(project_workspace_path: &str) -> String { + Path::new(project_workspace_path) + .file_name() + .map(|name| name.to_string_lossy().to_string()) + .unwrap_or_default() +} + /// Stable directory name for the target's shared clone. /// /// Keyed on the remote URL when there is one, so unrelated controllers cloning @@ -585,6 +600,14 @@ mod tests { assert_eq!(dispatch_branch_name("///", "job1"), "dispatch/job1"); } + #[test] + fn project_labels_come_from_the_workspace_basename() { + assert_eq!(project_label("/Users/me/code/BitFun"), "BitFun"); + assert_eq!(project_label("/Users/me/code/BitFun/"), "BitFun"); + // Nothing recognizable to use; the target falls back on its own side. + assert_eq!(project_label("/"), ""); + } + #[test] fn repo_keys_are_hex_stable_and_separate_remote_from_local_identity() { let remote = repo_key(Some("git@example.com:acme/app.git"), Path::new("/a/.git")); @@ -620,6 +643,7 @@ mod tests { let store = OutboundDispatchStore::new_in_root_for_tests(temp.path().join("dispatch-outbound")); let baseline = PreparedBaseline { + project_label: "BitFun".to_string(), delivery: DispatchWorkspaceDelivery { source_workspace_path: repository.to_string_lossy().to_string(), project_workspace_path: repository.to_string_lossy().to_string(), diff --git a/src/crates/assembly/core/src/service/dispatch/controller.rs b/src/crates/assembly/core/src/service/dispatch/controller.rs index 9ec243a604..75f8544614 100644 --- a/src/crates/assembly/core/src/service/dispatch/controller.rs +++ b/src/crates/assembly/core/src/service/dispatch/controller.rs @@ -121,6 +121,22 @@ pub struct DispatchAnswerRequest { pub feedback: Option, } +/// Start the next turn of an existing dispatch session. +/// +/// Separate from `append`, which steers a turn that is still running: this one +/// is for a job whose previous turn has finished, and it is what makes a +/// dispatch session hold a conversation instead of a single exchange. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DispatchContinueRequest { + pub job_id: String, + /// Caller-generated identity so a retry cannot start two turns. + pub turn_id: String, + pub prompt: String, + #[serde(default)] + pub display_content: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct DispatchAppendRequest { @@ -658,6 +674,7 @@ async fn provision_ssh_workspace( "protocolVersion": DISPATCH_PROTOCOL_VERSION, "jobId": job_id, "repoKey": baseline.repo_key, + "projectLabel": baseline.project_label, "remoteUrl": baseline.delivery.remote_url, "baseCommit": baseline.delivery.base_commit, "branch": baseline.delivery.branch, @@ -920,6 +937,69 @@ pub async fn append( dispatch_ssh::append(manager, connection_id, &serde_json::to_value(request)?).await } +/// Send the next turn of a dispatch session to its SSH target. +pub async fn continue_job( + manager: &SSHConnectionManager, + store: &OutboundDispatchStore, + request: DispatchContinueRequest, +) -> anyhow::Result { + validate_continue_request(&request)?; + let record = store + .get(&request.job_id) + .await? + .ok_or_else(|| anyhow::anyhow!("Outbound dispatch job was not found"))?; + let DispatchTarget::Ssh { connection_id, .. } = &record.target else { + anyhow::bail!("SSH dispatch follow-up requires an SSH target"); + }; + let response = + dispatch_ssh::continue_job(manager, connection_id, &continue_payload(&request)).await?; + record_follow_up_state(store, &record, &response).await; + Ok(response) +} + +/// The wire payload both transports send for a follow-up turn. +pub(super) fn continue_payload(request: &DispatchContinueRequest) -> Value { + let mut payload = json!({ + "protocolVersion": DISPATCH_PROTOCOL_VERSION, + "jobId": request.job_id, + "turnId": request.turn_id, + "prompt": request.prompt, + }); + if let Some(display) = request + .display_content + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + { + payload["displayContent"] = Value::String(display.to_string()); + } + payload +} + +/// Move the observer record back out of its terminal state. +/// +/// Best effort: the next status poll reconciles it from the target anyway, but +/// updating here keeps the composer from briefly re-offering "send" as if the +/// follow-up had not been accepted. +pub(super) async fn record_follow_up_state( + store: &OutboundDispatchStore, + record: &OutboundDispatchRecord, + response: &Value, +) { + let Some(state) = response.get("state").and_then(Value::as_str) else { + return; + }; + if let Err(error) = store + .update_progress(&record.job_id, record.last_cursor, state) + .await + { + log::warn!( + "Failed to record dispatch follow-up state: job_id={} error={error}", + record.job_id + ); + } +} + pub async fn list_jobs( manager: &SSHConnectionManager, store: &OutboundDispatchStore, @@ -1000,6 +1080,23 @@ pub(super) fn validate_append_request(request: &DispatchAppendRequest) -> anyhow Ok(()) } +pub(super) fn validate_continue_request(request: &DispatchContinueRequest) -> anyhow::Result<()> { + if request.turn_id.trim().is_empty() || request.turn_id.len() > 128 { + anyhow::bail!("Dispatch turnId must contain 1-128 bytes"); + } + if request.prompt.trim().is_empty() { + anyhow::bail!("Dispatch follow-up prompt cannot be empty"); + } + let total_bytes = request + .prompt + .len() + .saturating_add(request.display_content.as_ref().map_or(0, String::len)); + if total_bytes > MAX_DISPATCH_TEXT_BYTES { + anyhow::bail!("Dispatch follow-up exceeds the 32 KiB request limit"); + } + Ok(()) +} + pub(super) fn validate_answer_request(request: &DispatchAnswerRequest) -> anyhow::Result<()> { if request.request_id.trim().is_empty() || request.request_id.len() > 512 { anyhow::bail!("Dispatch permission requestId is invalid"); diff --git a/src/crates/assembly/core/src/service/dispatch/device_controller.rs b/src/crates/assembly/core/src/service/dispatch/device_controller.rs index 1f19b59229..68e602d958 100644 --- a/src/crates/assembly/core/src/service/dispatch/device_controller.rs +++ b/src/crates/assembly/core/src/service/dispatch/device_controller.rs @@ -14,10 +14,11 @@ use super::baseline::{ build_base_bundle, prepare_baseline, release_prepared_baseline, PreparedBaseline, }; use super::controller::{ - bind_outbound_record, finish_sync, provisioned_path, release_unbound_preparation_baseline, - result_bundle_path, same_target_identity, target_have_tips, validate_answer_request, - validate_append_request, validate_submission_preflight, validate_submit_ack, - validate_submit_request, DispatchAnswerRequest, DispatchAppendRequest, DispatchJobRequest, + bind_outbound_record, continue_payload, finish_sync, provisioned_path, record_follow_up_state, + release_unbound_preparation_baseline, result_bundle_path, same_target_identity, + target_have_tips, validate_answer_request, validate_append_request, validate_continue_request, + validate_submission_preflight, validate_submit_ack, validate_submit_request, + DispatchAnswerRequest, DispatchAppendRequest, DispatchContinueRequest, DispatchJobRequest, DispatchListJobsRequest, DispatchProbeTargetRequest, DispatchStatusRequest, DispatchSubmitRequest, DispatchSyncResultRequest, DISPATCH_PROTOCOL_VERSION, }; @@ -441,6 +442,28 @@ pub async fn append_device( .await } +/// Send the next turn of a dispatch session to an account device. +pub async fn continue_device_job( + rpc: &dyn DeviceDispatchRpc, + store: &OutboundDispatchStore, + request: DispatchContinueRequest, +) -> anyhow::Result { + validate_continue_request(&request)?; + let record = load_device_record(store, &request.job_id).await?; + let DispatchTarget::Device { device_id, .. } = &record.target else { + unreachable!("load_device_record validates target kind") + }; + let response = rpc + .invoke( + device_id, + "dispatch_target_continue", + continue_payload(&request), + ) + .await?; + record_follow_up_state(store, &record, &response).await; + Ok(response) +} + pub async fn list_device_jobs( rpc: &dyn DeviceDispatchRpc, store: &OutboundDispatchStore, @@ -483,6 +506,7 @@ async fn provision_device_workspace( "protocolVersion": DISPATCH_PROTOCOL_VERSION, "jobId": job_id, "repoKey": baseline.repo_key, + "projectLabel": baseline.project_label, "remoteUrl": baseline.delivery.remote_url, "baseCommit": baseline.delivery.base_commit, "branch": baseline.delivery.branch, diff --git a/src/crates/assembly/core/src/service/dispatch/mod.rs b/src/crates/assembly/core/src/service/dispatch/mod.rs index a4acb602aa..a21cd4d121 100644 --- a/src/crates/assembly/core/src/service/dispatch/mod.rs +++ b/src/crates/assembly/core/src/service/dispatch/mod.rs @@ -23,7 +23,7 @@ use crate::infrastructure::PathManager; #[cfg(feature = "ssh-remote")] pub use controller::{ answer as answer_dispatch, append as append_dispatch, cancel as cancel_dispatch, - install_cli_cancel as cancel_dispatch_cli_install, + continue_job as continue_dispatch_job, install_cli_cancel as cancel_dispatch_cli_install, install_cli_poll as poll_dispatch_cli_install, install_cli_source_start as start_dispatch_cli_source_build, install_cli_start as start_dispatch_cli_install, list_jobs as list_dispatch_jobs, @@ -31,18 +31,18 @@ pub use controller::{ status as get_dispatch_status, submit as submit_dispatch, sync_model_config as sync_dispatch_model_config, sync_result as sync_dispatch_result, DispatchAnswerRequest, DispatchAppendRequest, DispatchConnectionRequest, - DispatchInstallPollRequest, DispatchInstallStartRequest, DispatchJobRequest, - DispatchListJobsRequest, DispatchListTargetsRequest, DispatchPermissionReplyKind, - DispatchProbeTargetRequest, DispatchStatusRequest, DispatchSubmitRequest, - DispatchSyncResultRequest, DispatchTargetOption, + DispatchContinueRequest, DispatchInstallPollRequest, DispatchInstallStartRequest, + DispatchJobRequest, DispatchListJobsRequest, DispatchListTargetsRequest, + DispatchPermissionReplyKind, DispatchProbeTargetRequest, DispatchStatusRequest, + DispatchSubmitRequest, DispatchSyncResultRequest, DispatchTargetOption, }; #[cfg(feature = "ssh-remote")] pub use device_controller::{ answer_device as answer_device_dispatch, append_device as append_device_dispatch, - cancel_device as cancel_device_dispatch, list_device_jobs as list_device_dispatch_jobs, - probe_device as probe_device_dispatch_target, status_device as get_device_dispatch_status, - submit_device as submit_device_dispatch, sync_device_result as sync_device_dispatch_result, - DeviceDispatchRpc, + cancel_device as cancel_device_dispatch, continue_device_job as continue_device_dispatch_job, + list_device_jobs as list_device_dispatch_jobs, probe_device as probe_device_dispatch_target, + status_device as get_device_dispatch_status, submit_device as submit_device_dispatch, + sync_device_result as sync_device_dispatch_result, DeviceDispatchRpc, }; pub use target::{DispatchTarget, DispatchTargetRequest, DispatchWorkspaceDelivery}; @@ -461,7 +461,7 @@ impl OutboundDispatchStore { match fs::remove_file(&path).await { Ok(()) => Ok(true), Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false), - Err(error) => return Err(error.into()), + Err(error) => Err(error.into()), } } diff --git a/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs b/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs index fe042837ec..ac392143b6 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs @@ -1385,6 +1385,15 @@ pub async fn append( invoke_json(manager, connection_id, "append", request).await } +/// Start the next turn of a dispatch session that has already finished one. +pub async fn continue_job( + manager: &SSHConnectionManager, + connection_id: &str, + request: &Value, +) -> Result { + invoke_json(manager, connection_id, "continue", request).await +} + /// Commit the target's worktree and fetch the Git bundle it produced. /// /// Downloads only. The controller decides separately whether to fast-forward diff --git a/src/web-ui/src/features/dispatch/README.md b/src/web-ui/src/features/dispatch/README.md index 3a4893764d..68712a93e8 100644 --- a/src/web-ui/src/features/dispatch/README.md +++ b/src/web-ui/src/features/dispatch/README.md @@ -7,6 +7,11 @@ dispatch. 1. A dispatch target is selected while creating a session and is immutable after the first turn. +1a. A dispatch session accepts follow-up messages. While a turn runs, a message + is an `append` that steers it; once it has finished, a message is a + `dispatch_continue` that queues the next turn against the same target + session, worktree, and event log. The follow-up carries a caller-generated + `turnId` so an ambiguous response cannot start two turns. 2. `local` uses the current session, worktree, persistence, and dialog-turn paths unchanged. 3. The dispatch picker follows the same Git-workspace visibility condition as diff --git a/src/web-ui/src/features/dispatch/dispatchApi.ts b/src/web-ui/src/features/dispatch/dispatchApi.ts index 3ad6bcca16..02f93fee0d 100644 --- a/src/web-ui/src/features/dispatch/dispatchApi.ts +++ b/src/web-ui/src/features/dispatch/dispatchApi.ts @@ -5,6 +5,7 @@ import type { DispatchCliRelease, DispatchInstallPoll, DispatchInstallStart, + DispatchContinueResponse, DispatchJobListEntry, DispatchSshProbe, DispatchStatusResponse, @@ -95,6 +96,24 @@ export const dispatchApi = { }); }, + /** + * Start the next turn of a dispatch session. + * + * Distinct from `append`, which steers a turn that is still running: this is + * for a job whose previous turn has finished. The target keeps its session, + * worktree, and event log, so the projection stays one continuous transcript. + */ + async continueJob( + jobId: string, + turnId: string, + prompt: string, + displayContent?: string, + ): Promise { + return api.invoke('dispatch_continue', { + request: { jobId, turnId, prompt, displayContent }, + }); + }, + async status(jobId: string, cursor: number): Promise { return api.invoke('dispatch_status', { request: { jobId, cursor }, diff --git a/src/web-ui/src/features/dispatch/types.ts b/src/web-ui/src/features/dispatch/types.ts index c585a07956..8b475ee6b3 100644 --- a/src/web-ui/src/features/dispatch/types.ts +++ b/src/web-ui/src/features/dispatch/types.ts @@ -205,6 +205,14 @@ export interface DispatchStatusResponse { lastError?: string; } +export interface DispatchContinueResponse { + accepted: boolean; + jobId: string; + sessionId: string; + turnId: string; + state: Exclude; +} + export interface DispatchCancelResponse { cancelled: boolean; } diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts index 9d6c0d8ce5..29120109e8 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts @@ -28,7 +28,7 @@ import { sessionWorktreeMaterializationPlan } from '../../utils/sessionWorktree' import { dispatchApi } from '@/features/dispatch/dispatchApi'; import { dispatchJobStore } from '@/features/dispatch/dispatchJobStore'; import { requestDispatchJobRefresh } from '@/features/dispatch/DispatchJobObserver'; -import { isNonLocalDispatchTarget } from '@/features/dispatch/types'; +import { isDispatchJobTerminal, isNonLocalDispatchTarget } from '@/features/dispatch/types'; import { markOptimisticDispatchTurnMetadata } from '@/features/dispatch/optimisticDispatchTurn'; import { isSessionInUseError } from '@/infrastructure/api/errors/TauriCommandError'; import { i18nService } from '@/infrastructure/i18n'; @@ -85,6 +85,16 @@ interface PendingDispatchAppendRetry { // instead of injecting the steering text twice. const pendingDispatchAppendRetries = new Map(); +interface PendingDispatchContinueRetry { + content: string; + displayContent?: string; + turnId: string; +} + +// Same reasoning as the append retries above: an ambiguous transport failure +// must be retryable without starting the follow-up turn twice on the target. +const pendingDispatchContinueRetries = new Map(); + function acpClientIdFromMode(mode: string | undefined): string | null { const value = mode?.trim(); if (!value?.startsWith('acp:')) return null; @@ -216,6 +226,83 @@ export async function sendMessage( } const sendAttempt = beginSessionSend(sessionId); + /** + * Start the next turn of a finished dispatch job. + * + * The optimistic turn mirrors the first-message path so the user sees their + * message immediately; the target's own `DialogTurnStarted` adopts it once + * the follow-up worker starts. + */ + const continueDispatchJob = async (jobId: string): Promise => { + const followUpSession = context.flowChatStore.getState().sessions.get(sessionId) ?? session; + const followUpAgentType = (agentType?.trim() || followUpSession.mode || 'agentic').trim(); + const current = pendingDispatchContinueRetries.get(sessionId); + const retry = + current?.content === message && current.displayContent === displayMessage + ? current + : { + content: message, + displayContent: displayMessage, + // Reused across retries so a lost response cannot start two turns. + turnId: + globalThis.crypto?.randomUUID?.() + ?? `dispatch-turn-${Date.now()}-${Math.random().toString(36).slice(2)}`, + }; + pendingDispatchContinueRetries.set(sessionId, retry); + + const optimisticTurnId = `dispatch_pending_${jobId}`; + context.flowChatStore.addDialogTurn(sessionId, { + id: optimisticTurnId, + sessionId, + agentType: followUpAgentType, + userMessage: { + id: `user_dispatch_${retry.turnId}`, + content: displayMessage || message, + timestamp: Date.now(), + metadata: markOptimisticDispatchTurnMetadata( + options?.userMessageMetadata, + jobId, + ), + }, + modelRounds: [], + status: 'pending', + startTime: Date.now(), + }); + globalEventBus.emit( + FLOWCHAT_PIN_TURN_TO_TOP_EVENT, + { + sessionId, + turnId: optimisticTurnId, + behavior: 'auto', + source: 'send-message', + pinMode: 'sticky-latest', + } satisfies FlowChatPinTurnToTopRequest, + 'MessageModule', + ); + + try { + const response = await dispatchApi.continueJob( + jobId, + retry.turnId, + message, + displayMessage, + ); + if (!response.accepted) { + throw new Error('Dispatch target did not accept the follow-up turn'); + } + // The target owns the job state; the refresh below reads it back rather + // than this side guessing what the follow-up did to it. + } catch (error) { + context.flowChatStore.deleteDialogTurn(sessionId, optimisticTurnId); + throw error; + } finally { + if (pendingDispatchContinueRetries.get(sessionId)?.turnId === retry.turnId) { + pendingDispatchContinueRetries.delete(sessionId); + } + } + requestDispatchJobRefresh(jobId); + }; + const appendToDispatchJob = async (jobId: string): Promise => { const current = pendingDispatchAppendRetries.get(sessionId); const retry = @@ -361,18 +448,25 @@ export async function sendMessage( if ((options?.imageContexts?.length ?? 0) > 0) { throw new Error('Image attachments are not supported for detached dispatch yet'); } - if ( - readySession.config.dispatchJobState === 'queued' - || readySession.config.dispatchJobState === 'running' - ) { + const dispatchState = readySession.config.dispatchJobState; + if (dispatchState === 'queued' || dispatchState === 'running') { + // A turn is already in flight; this message steers it rather than + // starting another one underneath it. await appendToDispatchJob(jobId); return; } + if (dispatchState && isDispatchJobTerminal(dispatchState)) { + // The previous turn finished. A dispatch session is a conversation, so + // this starts the next turn against the same target session, worktree, + // and event log instead of refusing the message. + await continueDispatchJob(jobId); + return; + } if ( - readySession.config.dispatchJobState !== 'submitting' && - readySession.config.dispatchJobState !== 'submission_unknown' + dispatchState !== 'submitting' && + dispatchState !== 'submission_unknown' ) { - throw new Error('This detached dispatch job has already been submitted'); + throw new Error('This dispatch session is not ready to accept a message'); } if (isFirstMessage) { handleTitleGeneration(context, sessionId, message); diff --git a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts b/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts index 00a52e36f7..88206c1a75 100644 --- a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts +++ b/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.ts @@ -72,6 +72,7 @@ const LOCAL_ONLY_COMMANDS = new Set([ 'dispatch_list_jobs', 'dispatch_answer', 'dispatch_append', + 'dispatch_continue', 'dispatch_sync_result', 'dispatch_load_transcript', 'dispatch_save_transcript', From 1e60558ffc9aabcb19523b2ef3e2e2df728e9f10 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Sat, 1 Aug 2026 04:01:17 -0700 Subject: [PATCH 2/2] fix(dispatch): report the restore failure behind a failed follow-up turn A follow-up turn only reaches session creation when its restore failed, and creating then fails on the already-persisted id. Carrying the restore error into that context makes the report name the real cause instead of the "already exists" symptom it produces. --- src/apps/cli/src/dispatch/worker.rs | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/src/apps/cli/src/dispatch/worker.rs b/src/apps/cli/src/dispatch/worker.rs index 96d1b0e6f1..1e69de9339 100644 --- a/src/apps/cli/src/dispatch/worker.rs +++ b/src/apps/cli/src/dispatch/worker.rs @@ -118,7 +118,7 @@ async fn run_inner(store: &DispatchStore, job_id: &str) -> Result<()> { // history is the agent's context. Restoring first is what makes a dispatch // session a conversation; creating unconditionally would fail on the second // turn because the persisted id already exists. - let restored = agent_runtime + let restore_error = agent_runtime .restore_session(AgentSessionRestoreRequest { workspace_path: workspace_path.clone(), session_id: job.request.session_id.clone(), @@ -127,8 +127,12 @@ async fn run_inner(store: &DispatchStore, job_id: &str) -> Result<()> { remote_ssh_host: None, }) .await - .is_ok(); - if !restored { + .err(); + if let Some(error) = restore_error.as_ref() { + // Expected on the first turn — there is nothing to restore yet. + tracing::debug!("Dispatch session restore did not apply: {error}"); + } + if restore_error.is_some() { agent_runtime .create_session_with_id( job.request.session_id.clone(), @@ -147,7 +151,16 @@ async fn run_inner(store: &DispatchStore, job_id: &str) -> Result<()> { ) .await .map_err(|error| anyhow!(error.into_message())) - .context("create target-owned dispatch session")?; + // A follow-up turn reaches this only when its restore failed, and + // creating then fails on the existing persisted id. Carry the + // restore error so the report names the real cause instead of the + // "already exists" symptom. + .with_context(|| match restore_error { + Some(restore) => { + format!("create target-owned dispatch session after restore failed: {restore}") + } + None => "create target-owned dispatch session".to_string(), + })?; } let turn_id = uuid::Uuid::new_v4().to_string();