Wire the periodic checkpoint coordinator into AgentDriver

Instantiates and drives CheckpointCoordinatorHandle from AgentDriver's
own lifecycle, completing REMOTE-2111 Phase 3:

- AgentDriverOptions gains checkpoint_interval (override for the
  default 5-minute-plus-jitter cadence; primarily for tests).
- AgentDriver spawns a coordinator when a cloud task_id is present,
  snapshot uploads aren't disabled, and both FeatureFlag::OzHandoff and
  the new FeatureFlag::PeriodicHandoffCheckpoints are enabled.
- Driver finalization routes through the coordinator's finalize()
  (bounded by the existing snapshot upload timeout) when a coordinator
  is active, instead of the legacy one-shot end-of-run snapshot path.
- warp_features: add PeriodicHandoffCheckpoints, off by default while
  the coordinator rolls out.

Co-Authored-By: Oz <oz-agent@warp.dev>

Author: joeywangzr

app/src/ai/agent_sdk/driver.rs

@@ -313,6 +313,10 @@ pub struct AgentDriverOptions {
     pub snapshot_upload_timeout: Option<Duration>,
     /// Declarations script timeout override.
     pub snapshot_script_timeout: Option<Duration>,
+    /// Periodic checkpoint cadence override. Only used when
+    /// `FeatureFlag::PeriodicHandoffCheckpoints` is enabled; deliberately separate
+    /// from `snapshot_upload_timeout`/`snapshot_script_timeout` per-attempt budgets.
+    pub checkpoint_interval: Option<Duration>,
     /// Skip the initial `StartFromAmbientRunPrompt` so the agent waits for a
     /// follow-up instead of hallucinating an empty turn. Sourced from the
     /// `--skip-initial-turn` CLI flag, which the worker emits when the
@@ -378,6 +382,13 @@ pub struct AgentDriver {
     snapshot_upload_timeout: Duration,
     snapshot_script_timeout: Duration,
 
+    /// Periodic workspace-handoff checkpoint coordinator. `Some` only when
+    /// `FeatureFlag::OzHandoff` and `FeatureFlag::PeriodicHandoffCheckpoints` are both
+    /// enabled, the run has a cloud task id, and `--no-snapshot` was not set;
+    /// `None` otherwise, in which case `run_snapshot_upload` falls back to the legacy
+    /// one-shot upload path unchanged.
+    checkpoint_coordinator: Option<checkpoint_coordinator::CheckpointCoordinatorHandle>,
+
     /// Conversation ID this driver is running. Set at construction for
     /// resumed runs and on `ConversationServerTokenAssigned` for fresh
     /// runs; consumed by `unregister_streamer_consumer` at end of run.
@@ -651,6 +662,7 @@ impl AgentDriver {
             snapshot_disabled,
             snapshot_upload_timeout,
             snapshot_script_timeout,
+            checkpoint_interval,
             skip_initial_turn,
             strict_mcp_startup,
             mcp_startup_timeout,
@@ -759,6 +771,32 @@ impl AgentDriver {
             _ => None,
         };
 
+        // Spawn the periodic checkpoint coordinator under the same gates as the
+        // declarations writer above, plus the dedicated rollout flag. `None` keeps
+        // `run_snapshot_upload` on the legacy one-shot upload path unchanged.
+        let checkpoint_coordinator = match task_id {
+            Some(id)
+                if FeatureFlag::OzHandoff.is_enabled()
+                    && FeatureFlag::PeriodicHandoffCheckpoints.is_enabled()
+                    && !snapshot_disabled_value =>
+            {
+                let client = ServerApiProvider::as_ref(ctx).get_harness_support_client();
+                Some(checkpoint_coordinator::CheckpointCoordinatorHandle::new(
+                    client,
+                    id,
+                    working_dir.clone(),
+                    ctx.spawner(),
+                    checkpoint_interval
+                        .unwrap_or(checkpoint_coordinator::DEFAULT_CHECKPOINT_INTERVAL),
+                    snapshot_script_timeout
+                        .unwrap_or(snapshot::DEFAULT_DECLARATIONS_SCRIPT_TIMEOUT),
+                    snapshot_upload_timeout.unwrap_or(snapshot::DEFAULT_SNAPSHOT_UPLOAD_TIMEOUT),
+                    ctx.background_executor(),
+                ))
+            }
+            _ => None,
+        };
+
         Ok(Self {
             terminal_driver,
             working_dir,
@@ -778,6 +816,7 @@ impl AgentDriver {
                 .unwrap_or(snapshot::DEFAULT_SNAPSHOT_UPLOAD_TIMEOUT),
             snapshot_script_timeout: snapshot_script_timeout
                 .unwrap_or(snapshot::DEFAULT_DECLARATIONS_SCRIPT_TIMEOUT),
+            checkpoint_coordinator,
             run_conversation_id,
             parent_run_id: parent_run_id_for_self,
             third_party_harness_model_config,
@@ -821,6 +860,7 @@ impl AgentDriver {
             snapshot_disabled: false,
             snapshot_upload_timeout: snapshot::DEFAULT_SNAPSHOT_UPLOAD_TIMEOUT,
             snapshot_script_timeout: snapshot::DEFAULT_DECLARATIONS_SCRIPT_TIMEOUT,
+            checkpoint_coordinator: None,
             run_conversation_id: None,
             parent_run_id: None,
             third_party_harness_model_config: None,
@@ -3789,13 +3829,20 @@ impl AgentDriver {
 
         // Snapshot upload is only meaningful for cloud task runs, so short-circuit before
         // pulling the rest of the context onto this task.
-        let Ok((Some(task_id), snapshot_disabled, upload_timeout, script_timeout)) = spawner
+        let Ok((
+            Some(task_id),
+            snapshot_disabled,
+            upload_timeout,
+            script_timeout,
+            checkpoint_coordinator,
+        )) = spawner
             .spawn(|me, _| {
                 (
                     me.task_id,
                     me.snapshot_disabled,
                     me.snapshot_upload_timeout,
                     me.snapshot_script_timeout,
+                    me.checkpoint_coordinator.clone(),
                 )
             })
             .await
@@ -3807,6 +3854,16 @@ impl AgentDriver {
             return;
         }
 
+        // When the periodic checkpoint coordinator is active, it owns the entire
+        // end-of-run path: `finalize` regenerates declarations, runs one last
+        // best-effort attempt bounded by `upload_timeout`, and commits it as the
+        // selected checkpoint. This replaces the legacy one-shot upload below so
+        // there is exactly one end-of-run snapshot path, not two.
+        if let Some(coordinator) = checkpoint_coordinator {
+            coordinator.finalize(upload_timeout).await;
+            return;
+        }
+
         let Ok((working_dir, client)) = spawner
             .spawn(|me, ctx| {
                 let client = ServerApiProvider::as_ref(ctx).get_harness_support_client();

app/src/ai/agent_sdk/mod.rs

@@ -1069,6 +1069,7 @@ impl AgentDriverRunner {
                         .snapshot
                         .snapshot_script_timeout
                         .map(|duration| duration.into()),
+                    checkpoint_interval: None,
                     skip_initial_turn: args.skip_initial_turn,
                     strict_mcp_startup: args.strict_mcp_startup,
                     mcp_startup_timeout: args.mcp_startup_timeout.map(|duration| duration.into()),

crates/warp_features/src/lib.rs

@@ -932,6 +932,12 @@ pub enum FeatureFlag {
     /// authenticated with the logged-in user's session token. No manual MCP
     /// setup or API key required.
     FactoryMcp,
+
+    /// Enables periodic workspace-handoff checkpoints during a cloud agent run,
+    /// rather than only uploading a workspace snapshot once at end-of-run.
+    /// Requires `OzHandoff` to also be enabled; a no-op for local runs and when
+    /// `--no-snapshot` is set. Off by default while the coordinator rolls out.
+    PeriodicHandoffCheckpoints,
 }
 
 static FLAG_STATES: [AtomicBool; cardinality::<FeatureFlag>()] =