fix(dispatch): close result-pull gaps left by the first pass

src/apps/cli/src/dispatch/mod.rs

@@ -19,7 +19,7 @@ use protocol::{
     DispatchListRequest, DispatchProbeRequest, DispatchProbeResponse, DispatchStatusRequest,
     DispatchStatusResponse, DispatchSubmitRequest, DispatchSubmitResponse,
     DispatchWorkspaceBeginRequest, DispatchWorkspaceChunkRequest, DispatchWorkspaceCommitRequest,
-    DispatchWorkspaceResultRequest,
+    DispatchWorkspaceResultChunkRequest, DispatchWorkspaceResultRequest,
     DispatchWorkspaceProbe, DISPATCH_PROTOCOL_VERSION, MAX_DISPATCH_TEXT_BYTES,
 };
 use store::{CreateJobOutcome, DispatchStateRecord, DispatchStore};
@@ -76,6 +76,10 @@ pub(crate) async fn run_dispatch_verb(
             DispatchWorkspaceResultRequest,
         >(input)?)?)
         .context("encode workspace result response"),
+        "workspace-result-chunk" => serde_json::to_value(workspace::result_chunk(parse::<
+            DispatchWorkspaceResultChunkRequest,
+        >(input)?)?)
+        .context("encode workspace result chunk response"),
         _ => bail!("unsupported dispatch verb: {verb}"),
     }
 }

src/apps/cli/src/dispatch/protocol.rs

@@ -206,6 +206,28 @@ pub(crate) struct DispatchWorkspaceResultResponse {
     pub(crate) summary: WorkspaceResultSummary,
 }
 
+/// Read a slice of an already-built result bundle.
+///
+/// Exists for transports with no file channel of their own: SSH pulls the
+/// bundle over SFTP, but an account device can only carry JSON, so it streams
+/// the same bytes back in chunks — the mirror of the upload path.
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(crate) struct DispatchWorkspaceResultChunkRequest {
+    pub(crate) job_id: String,
+    pub(crate) offset: u64,
+    pub(crate) length: u64,
+}
+
+#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct DispatchWorkspaceResultChunkResponse {
+    pub(crate) offset: u64,
+    pub(crate) data_base64: String,
+    /// True once this chunk reaches the end of the bundle.
+    pub(crate) eof: bool,
+}
+
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub(crate) struct DispatchCancelRequest {

src/apps/cli/src/dispatch/workspace.rs

@@ -14,7 +14,8 @@ use serde::{Deserialize, Serialize};
 use super::protocol::{
     DispatchWorkspaceBeginRequest, DispatchWorkspaceBeginResponse, DispatchWorkspaceChunkRequest,
     DispatchWorkspaceChunkResponse, DispatchWorkspaceCommitRequest,
-    DispatchWorkspaceCommitResponse, DispatchWorkspaceResultRequest,
+    DispatchWorkspaceCommitResponse, DispatchWorkspaceResultChunkRequest,
+    DispatchWorkspaceResultChunkResponse, DispatchWorkspaceResultRequest,
     DispatchWorkspaceResultResponse, DISPATCH_PROTOCOL_VERSION,
 };
 use super::store::{
@@ -341,6 +342,39 @@ pub(crate) fn result(
     })
 }
 
+/// Stream back a slice of the bundle `result` already produced.
+///
+/// Read-only and bounded: it never rebuilds the bundle, so the digest the
+/// controller verified stays the digest it receives.
+pub(crate) fn result_chunk(
+    request: DispatchWorkspaceResultChunkRequest,
+) -> Result<DispatchWorkspaceResultChunkResponse> {
+    if request.length == 0 || request.length > MAX_CHUNK_BYTES as u64 {
+        bail!("workspace result chunk length must be between 1 and {MAX_CHUNK_BYTES} bytes");
+    }
+    let store = DispatchStore::open_default()?;
+    let upload_dir = store.workspace_upload_dir(&request.job_id)?;
+    let bundle_path = upload_dir.join(RESULT_BUNDLE_FILE);
+    let mut file = fs::File::open(&bundle_path)
+        .context("build the dispatch result bundle before reading it")?;
+    let size = file.metadata()?.len();
+    if request.offset > size {
+        bail!("workspace result chunk offset is past the end of the bundle");
+    }
+    file.seek(SeekFrom::Start(request.offset))?;
+    let remaining = size - request.offset;
+    let take = request.length.min(remaining) as usize;
+    let mut buffer = vec![0_u8; take];
+    file.read_exact(&mut buffer)
+        .context("read dispatch result bundle")?;
+    let next_offset = request.offset + take as u64;
+    Ok(DispatchWorkspaceResultChunkResponse {
+        offset: next_offset,
+        data_base64: base64::engine::general_purpose::STANDARD.encode(&buffer),
+        eof: next_offset >= size,
+    })
+}
+
 /// Detached target-side materialization. The short `workspace-commit` RPC
 /// starts this process and subsequent commit calls poll the durable record, so
 /// extraction is not bounded by an SSH or Relay request timeout.

src/apps/cli/src/main.rs

@@ -606,6 +606,8 @@ pub(crate) enum DispatchAction {
     WorkspaceCommit,
     #[command(name = "__workspace_result", hide = true)]
     WorkspaceResult,
+    #[command(name = "__workspace_result_chunk", hide = true)]
+    WorkspaceResultChunk,
     #[command(name = "__workspace_materialize", hide = true)]
     WorkspaceMaterialize {
         #[arg(long)]

src/apps/cli/src/peer_host/dispatch.rs

@@ -131,6 +131,8 @@ fn dispatch_target_verb(command: &str) -> Option<&'static str> {
         "dispatch_target_workspace_begin" => Some("workspace-begin"),
         "dispatch_target_workspace_chunk" => Some("workspace-chunk"),
         "dispatch_target_workspace_commit" => Some("workspace-commit"),
+        "dispatch_target_workspace_result" => Some("workspace-result"),
+        "dispatch_target_workspace_result_chunk" => Some("workspace-result-chunk"),
         _ => None,
     }
 }

src/apps/cli/src/root_handlers.rs

@@ -60,6 +60,7 @@ pub(crate) async fn handle_dispatch_action(action: DispatchAction) -> Result<()>
         DispatchAction::WorkspaceChunk => "workspace-chunk",
         DispatchAction::WorkspaceCommit => "workspace-commit",
         DispatchAction::WorkspaceResult => "workspace-result",
+        DispatchAction::WorkspaceResultChunk => "workspace-result-chunk",
     };
     let result = async {
         use std::io::{IsTerminal, Read};

src/apps/desktop/src/api/dispatch_api.rs

@@ -14,7 +14,8 @@ use bitfun_core::service::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, pull_dispatch_result, start_dispatch_cli_install,
+    probe_dispatch_target, pull_device_dispatch_result, pull_dispatch_result,
+    start_dispatch_cli_install,
     start_dispatch_cli_source_build, submit_device_dispatch, submit_dispatch,
     sync_dispatch_model_config,
     DeviceDispatchRpc, DispatchAnswerRequest, DispatchAppendRequest, DispatchConnectionRequest,
@@ -277,6 +278,20 @@ pub async fn dispatch_pull_result(
     request: DispatchJobRequest,
 ) -> Result<Value, String> {
     let store = OutboundDispatchStore::new(path_manager.as_ref());
+    // Both transports stage the bundle and its summary identically, so the
+    // apply step below is transport-blind.
+    if matches!(
+        store
+            .get(&request.job_id)
+            .await
+            .map_err(|error| error.to_string())?
+            .map(|record| record.target),
+        Some(DispatchTarget::Device { .. })
+    ) {
+        return pull_device_dispatch_result(&AccountDeviceDispatchRpc, &store, request)
+            .await
+            .map_err(|error| error.to_string());
+    }
     let manager = state
         .get_ssh_manager_async()
         .await

src/apps/desktop/src/api/dispatch_host.rs

@@ -42,6 +42,8 @@ fn target_cli_verb(command: &str) -> Option<&'static str> {
         "dispatch_target_workspace_begin" => Some("__workspace_begin"),
         "dispatch_target_workspace_chunk" => Some("__workspace_chunk"),
         "dispatch_target_workspace_commit" => Some("__workspace_commit"),
+        "dispatch_target_workspace_result" => Some("__workspace_result"),
+        "dispatch_target_workspace_result_chunk" => Some("__workspace_result_chunk"),
         _ => None,
     }
 }

src/crates/assembly/core/src/service/dispatch/controller.rs

@@ -499,28 +499,49 @@ pub async fn pull_result(
         .await?
         .ok_or_else(|| anyhow::anyhow!("Outbound dispatch job was not found"))?;
     let DispatchTarget::Ssh { connection_id, .. } = &record.target else {
-        anyhow::bail!("Pulling dispatch results requires an SSH target");
+        anyhow::bail!("SSH dispatch result pull requires an SSH target");
     };
     let destination = result_bundle_path(store, &request.job_id);
     let response =
         dispatch_ssh::pull_result(manager, connection_id, &request.job_id, &destination).await?;
-    // Persist the summary next to the bundle so applying reads both from disk.
-    // The digests that decide whether a local file may be overwritten must come
-    // from the verified pull, not from whatever the caller hands back later.
+    record_result_summary(store, &request.job_id, &response)?;
+    Ok(response)
+}
+
+/// Persist the summary next to the bundle so applying reads both from disk.
+///
+/// The digests that decide whether a local file may be overwritten must come
+/// from the verified pull, not from whatever the caller hands back later.
+pub(super) fn record_result_summary(
+    store: &OutboundDispatchStore,
+    job_id: &str,
+    response: &Value,
+) -> anyhow::Result<()> {
     if let Some(summary) = response.get("summary") {
-        let summary_path = result_summary_path(store, &request.job_id);
-        std::fs::write(&summary_path, serde_json::to_vec(summary)?)
+        // Owner-only like the bundle beside it: this records which paths of the
+        // user's workspace changed.
+        let summary_path = result_summary_path(store, job_id);
+        dispatch_ssh::write_private_file(&summary_path, &serde_json::to_vec(summary)?)
             .with_context(|| format!("record result summary {}", summary_path.display()))?;
     }
-    Ok(response)
+    Ok(())
 }
 
-fn result_bundle_path(store: &OutboundDispatchStore, job_id: &str) -> std::path::PathBuf {
-    store.root().join(".results").join(format!("{job_id}.tar.gz"))
+pub(super) fn result_bundle_path(
+    store: &OutboundDispatchStore,
+    job_id: &str,
+) -> std::path::PathBuf {
+    store
+        .root()
+        .join(super::OUTBOUND_RESULTS_DIR)
+        .join(format!("{job_id}.tar.gz"))
 }
 
 fn result_summary_path(store: &OutboundDispatchStore, job_id: &str) -> std::path::PathBuf {
-    store.root().join(".results").join(format!("{job_id}.json"))
+    store
+        .root()
+        .join(super::OUTBOUND_RESULTS_DIR)
+        .join(format!("{job_id}.json"))
 }
 
 /// Apply a pulled result bundle to a local workspace.