fix(dispatch): harden session delivery and snapshot reuse

docs/architecture/detached-task-dispatch.md

@@ -36,14 +36,28 @@ storage is not used for workspace contents.
 
 `workspacePath` in a submit request identifies a directory on the target. It
 does not imply that similarly named directories on two machines are related.
-Dispatch therefore supports two explicit delivery modes.
+Dispatch therefore supports three explicit delivery modes.
 
 ### Existing target directory
 
 `existing` uses a directory that already exists on the target. Probe returns its
 canonical path and Git facts before submit. BitFun never clones, fetches,
 checks out, stashes, or rewrites that directory as part of dispatch.
 
+### One-shot source snapshot
+
+`snapshot-source` captures the controller workspace while honoring repository
+ignore rules. It includes tracked and non-ignored source files, including
+hidden source such as `.github/`, while excluding ignored dependency caches,
+build output, and local secrets. It uses the same verified, one-shot upload,
+materialization, result, and conflict rules as an exact snapshot. The filtered
+input set is carried in the existing exact-snapshot wire envelope, so compatible
+targets do not need a second materialization protocol.
+
+This is the default snapshot choice for ordinary source workspaces. Users who
+need ignored runtime inputs must choose the exact mode explicitly and confirm
+its wider data boundary.
+
 ### One-shot exact snapshot
 
 `snapshot-exact` captures the controller workspace at submit time and materializes it
@@ -90,11 +104,29 @@ the package fail, but coordinated edits across multiple files can still span
 the traversal interval. Callers that require an application-consistent source
 must quiesce the source or select a filesystem snapshot as the source path.
 
+The controller retains the latest verified archive for each canonical source
+path and capture mode. Before packaging a later job, it recomputes a lightweight
+fingerprint from the selected paths and their filesystem identity, size,
+executable state, and write/change timestamps. An unchanged fingerprint
+hard-links the cached immutable archive into the new job instead of rereading
+and recompressing every file. Source mode ignores changes below ignored paths;
+exact mode observes them. A selected entry change invalidates and atomically
+replaces the cache. The per-job link remains immutable during submission, so a
+later cache replacement cannot change an in-flight job's bytes.
+
 SSH transports the archive with SFTP after `workspace-begin`. Account-device
 RPC uses bounded base64 chunks inside the existing end-to-end encrypted
 `HostInvoke` envelope. Neither transport puts source bytes in command-line
 arguments, process listings, logs, or the outbound observer record.
 
+After a target has fully verified and materialized a snapshot, it retains one
+owner-only archive keyed by the archive SHA-256. A later job with identical
+metadata attaches that immutable archive and reports the full retained offset,
+so both SSH and account-device controllers skip the source transfer. The
+temporary per-job archive link is removed after materialization; each job still
+gets its own writable `current/` directory, so cache reuse never makes jobs
+share writes. Cache entries expire after 30 days without a hit.
+
 ## Synchronization semantics
 
 A snapshot is an immutable input boundary, not a live shared folder:
@@ -135,6 +167,18 @@ Workspace upload uses the internal `workspace-begin`, `workspace-chunk`, and
 `workspace-commit` verbs. They are target data-plane operations and are not
 normal product or Peer Device Mode commands.
 
+`dispatch_worker_cli_profile` is a required execution-safety capability. It
+means every dispatch process selects `DeliveryProfile::Cli` before model/config
+inspection can lazily initialize product-full tool state. Controllers must
+check it both during target setup and immediately before submission; package
+version equality is not evidence of this behavior.
+
+CLI installation smoke-tests the same capability before replacing an existing
+target binary. An untagged Desktop development build may, after the normal
+explicit source-build confirmation, archive its clean current Git commit and
+build that exact source on the target. This avoids reinstalling an older
+same-semver release while keeping executable transfer an explicit user action.
+
 Account-device transport wraps target verbs in names reserved for detached
 dispatch, such as `dispatch_target_submit`. They are handled before the
 attach-shaped Peer Host bridge and never acquire an attached-controller lease.
@@ -187,6 +231,8 @@ originally submitted the job.
 ## Failure rules
 
 - A missing or offline target fails submit; the Relay does not queue jobs.
+- A target missing a required behavioral capability fails preflight before a
+  durable job is created, even when its CLI package version matches.
 - A lost submit response leaves `submission_unknown`; status or an idempotent
   retry reconciles the target's durable truth.
 - A live PID that no longer matches the exact worker command is never signaled

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

@@ -19,8 +19,8 @@ use protocol::{
     DispatchListRequest, DispatchProbeRequest, DispatchProbeResponse, DispatchStatusRequest,
     DispatchStatusResponse, DispatchSubmitRequest, DispatchSubmitResponse,
     DispatchWorkspaceBeginRequest, DispatchWorkspaceChunkRequest, DispatchWorkspaceCommitRequest,
-    DispatchWorkspaceResultChunkRequest, DispatchWorkspaceResultRequest,
-    DispatchWorkspaceProbe, DISPATCH_PROTOCOL_VERSION, MAX_DISPATCH_TEXT_BYTES,
+    DispatchWorkspaceProbe, DispatchWorkspaceResultChunkRequest, DispatchWorkspaceResultRequest,
+    DISPATCH_PROTOCOL_VERSION, MAX_DISPATCH_TEXT_BYTES,
 };
 use store::{CreateJobOutcome, DispatchStateRecord, DispatchStore};
 
@@ -76,10 +76,12 @@ 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"),
+        "workspace-result-chunk" => {
+            serde_json::to_value(workspace::result_chunk(parse::<
+                DispatchWorkspaceResultChunkRequest,
+            >(input)?)?)
+            .context("encode workspace result chunk response")
+        }
         _ => bail!("unsupported dispatch verb: {verb}"),
     }
 }
@@ -111,9 +113,17 @@ async fn probe(request: DispatchProbeRequest) -> Result<DispatchProbeResponse> {
         "event_log_completeness".to_string(),
         "workspace_snapshot_exact".to_string(),
         "workspace_snapshot_chunked".to_string(),
+        // A target may share the same package version while predating the
+        // dispatch entrypoint's early CLI-profile selection. Such a binary can
+        // accept a job but every detached worker then fails before execution.
+        // Advertise the behavioral fix explicitly so controllers fail closed.
+        "dispatch_worker_cli_profile".to_string(),
         // Optional on purpose: controllers must feature-detect this rather than
         // require it, so an older target stays usable for everything else.
         "workspace_result_bundle".to_string(),
+        // Identical snapshots from different jobs reuse one verified archive
+        // on the target. Jobs still receive independent writable workspaces.
+        "workspace_snapshot_cache".to_string(),
     ];
     if runner::is_supported() {
         capabilities.push("detached_worker".to_string());

src/apps/cli/src/dispatch/store.rs

@@ -39,6 +39,9 @@ const TERMINAL_JOB_RETENTION_DAYS: i64 = 30;
 const RETENTION_GC_INTERVAL_SECONDS: u64 = 24 * 60 * 60;
 const RETENTION_GC_MARKER: &str = ".retention-gc";
 const RETENTION_GC_LOCK: &str = ".retention-gc.lock";
+const WORKSPACE_SNAPSHOT_CACHE_DIR: &str = "workspace-cache";
+pub(super) const WORKSPACE_SNAPSHOT_CACHE_RECORD_FILE: &str = "cache.json";
+const WORKSPACE_SNAPSHOT_CACHE_RETENTION_DAYS: i64 = 30;
 
 #[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
@@ -128,6 +131,12 @@ struct StoredAppendMessage {
     created_at: String,
 }
 
+#[derive(Clone, Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct WorkspaceSnapshotCacheRetentionRecord {
+    last_used_at: String,
+}
+
 #[derive(Clone, Debug)]
 pub(crate) struct DispatchStore {
     root: PathBuf,
@@ -149,6 +158,7 @@ impl DispatchStore {
         create_private_dir(&root)?;
         create_private_dir(&root.join("jobs"))?;
         create_private_dir(&root.join("workspaces"))?;
+        create_private_dir(&root.join(WORKSPACE_SNAPSHOT_CACHE_DIR))?;
         Ok(Self {
             root,
             max_events_bytes: DEFAULT_MAX_EVENTS_BYTES,
@@ -837,6 +847,10 @@ impl DispatchStore {
         Ok(self.root.join("workspaces").join(job_id))
     }
 
+    pub(crate) fn workspace_snapshot_cache_root(&self) -> PathBuf {
+        self.root.join(WORKSPACE_SNAPSHOT_CACHE_DIR)
+    }
+
     fn maybe_collect_expired_terminal_jobs(&self) -> Result<()> {
         let marker = self.root.join(RETENTION_GC_MARKER);
         if fs::metadata(&marker)
@@ -1025,9 +1039,78 @@ impl DispatchStore {
             })?;
             removed += 1;
         }
+        self.collect_expired_workspace_snapshot_cache(now)?;
         Ok(removed)
     }
 
+    fn collect_expired_workspace_snapshot_cache(
+        &self,
+        now: chrono::DateTime<chrono::Utc>,
+    ) -> Result<()> {
+        let cache_root = self.workspace_snapshot_cache_root();
+        for entry in fs::read_dir(&cache_root)
+            .with_context(|| format!("read dispatch workspace cache {}", cache_root.display()))?
+        {
+            let entry = entry?;
+            let Some(digest) = entry.file_name().to_str().map(ToOwned::to_owned) else {
+                continue;
+            };
+            if digest.len() != 64 || !digest.bytes().all(|byte| byte.is_ascii_hexdigit()) {
+                continue;
+            }
+            let cache_dir = entry.path();
+            let metadata = fs::symlink_metadata(&cache_dir)?;
+            if metadata.file_type().is_symlink() || !metadata.is_dir() {
+                continue;
+            }
+            let lock_path = cache_root.join(format!(".{digest}.lock"));
+            let Some(_lock) = JobLock::try_exclusive(&lock_path)? else {
+                continue;
+            };
+            let record = match read_json::<WorkspaceSnapshotCacheRetentionRecord>(
+                &cache_dir.join(WORKSPACE_SNAPSHOT_CACHE_RECORD_FILE),
+            ) {
+                Ok(record) => record,
+                Err(error) => {
+                    tracing::warn!(
+                            "Skipping unreadable dispatch workspace cache entry: digest={} error={error:#}",
+                            digest
+                        );
+                    continue;
+                }
+            };
+            let Some(last_used_at) = chrono::DateTime::parse_from_rfc3339(&record.last_used_at)
+                .ok()
+                .map(|value| value.with_timezone(&chrono::Utc))
+            else {
+                continue;
+            };
+            if now.signed_duration_since(last_used_at).num_days()
+                < WORKSPACE_SNAPSHOT_CACHE_RETENTION_DAYS
+            {
+                continue;
+            }
+            let tombstone = cache_root.join(format!(
+                ".gc-{}-{}",
+                digest,
+                uuid::Uuid::new_v4().as_simple()
+            ));
+            fs::rename(&cache_dir, &tombstone).with_context(|| {
+                format!(
+                    "quarantine expired dispatch workspace cache {}",
+                    cache_dir.display()
+                )
+            })?;
+            fs::remove_dir_all(&tombstone).with_context(|| {
+                format!(
+                    "remove expired dispatch workspace cache {}",
+                    tombstone.display()
+                )
+            })?;
+        }
+        Ok(())
+    }
+
     fn load_state_unlocked(&self, job_dir: &Path) -> Result<DispatchStateRecord> {
         read_json(&job_dir.join(STATE_FILE))
     }
@@ -2379,6 +2462,24 @@ mod tests {
             &expired,
         )
         .expect("age terminal state");
+        let expired_digest = "a".repeat(64);
+        let recent_digest = "b".repeat(64);
+        for (digest, last_used_at) in [
+            (
+                &expired_digest,
+                (now - chrono::Duration::days(WORKSPACE_SNAPSHOT_CACHE_RETENTION_DAYS + 1))
+                    .to_rfc3339(),
+            ),
+            (&recent_digest, now.to_rfc3339()),
+        ] {
+            let cache_dir = store.workspace_snapshot_cache_root().join(digest);
+            create_private_dir(&cache_dir).expect("create cache entry");
+            atomic_write_json(
+                &cache_dir.join(WORKSPACE_SNAPSHOT_CACHE_RECORD_FILE),
+                &serde_json::json!({ "lastUsedAt": last_used_at }),
+            )
+            .expect("write cache record");
+        }
 
         assert_eq!(
             store
@@ -2392,6 +2493,14 @@ mod tests {
         assert!(store.root.join("workspaces/recent").exists());
         assert!(store.root.join("jobs/running").exists());
         assert!(store.root.join("workspaces/running").exists());
+        assert!(!store
+            .workspace_snapshot_cache_root()
+            .join(expired_digest)
+            .exists());
+        assert!(store
+            .workspace_snapshot_cache_root()
+            .join(recent_digest)
+            .exists());
     }
 
     #[test]