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..1e69de9339 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,68 @@ 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 restore_error = 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")?; + .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(), + 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())) + // 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(); - // 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',