QUALITY-1759: require a genuine checkpoint for hibernation, avoid double-uploading it (review findings 6, 7)

Finding 6: run_snapshot_upload returned true for a skipped upload
(OzHandoff disabled, no cloud task id, or --no-snapshot), which is the
right contract for its ordinary best-effort callers but let a
wait_for_events hibernation claim a checkpoint it never produced. Add
a require_genuine_checkpoint parameter: when true, those same skip
conditions now return false, so the hibernation path reports
YieldedForEventsCheckpointFailed (-> ERROR) instead of a clean yield
whenever checkpointing capability isn't actually available.

Finding 7: a successful hibernation uploaded the final checkpoint once
from the WaitForEventsYielded handler, then again, unconditionally,
from the end-of-run cleanup in run(), breaking the one-checkpoint
invariant. Added final_checkpoint_uploaded_for_yield on AgentDriver,
set once the yield path has attempted its upload (success or
failure), and checked before the end-of-run cleanup's own upload so
it never re-uploads.

Adds run_snapshot_upload_fails_genuine_requirement_when_oz_handoff_disabled
to cover the new gate directly.

Author: warp-agent-staging[bot]

app/src/ai/agent_sdk/driver.rs

@@ -588,6 +588,12 @@ pub struct AgentDriver {
     /// pure no-op for local and disabled runs.
     snapshot_file_writer: Option<snapshot::DeclarationsWriterHandle>,
 
+    /// Set once the `WaitForEventsYielded` handler has already attempted the final
+    /// checkpoint upload for this run's clean exit (see `SDKConversationOutputStatus::
+    /// YieldedForEvents` / `YieldedForEventsCheckpointFailed`), so the unconditional
+    /// end-of-run cleanup in `run` does not redundantly upload a second final checkpoint.
+    final_checkpoint_uploaded_for_yield: bool,
+
     /// Whether the driver should skip dispatching the initial
     /// `StartFromAmbientRunPrompt`. Mirror of `AgentDriverOptions::skip_initial_turn`,
     /// sourced from the `--skip-initial-turn` CLI flag. Read by `execute_run`
@@ -1059,6 +1065,7 @@ impl AgentDriver {
             parent_run_id: parent_run_id_for_self,
             third_party_harness_model_config,
             snapshot_file_writer,
+            final_checkpoint_uploaded_for_yield: false,
             skip_initial_turn,
             strict_mcp_startup,
             mcp_startup_timeout: mcp_startup_timeout.unwrap_or(MCP_SERVER_STARTUP_TIMEOUT),
@@ -1108,6 +1115,7 @@ impl AgentDriver {
             parent_run_id: None,
             third_party_harness_model_config: None,
             snapshot_file_writer: None,
+            final_checkpoint_uploaded_for_yield: false,
             skip_initial_turn: false,
             strict_mcp_startup: false,
             mcp_startup_timeout: MCP_SERVER_STARTUP_TIMEOUT,
@@ -1327,7 +1335,13 @@ impl AgentDriver {
                          (reason={actual_reason:?}): {finalization_result:?}"
                     );
                 }
-                let _ = Self::run_snapshot_upload(&foreground).await;
+                let already_uploaded_for_yield = foreground
+                    .spawn(|me, _| me.final_checkpoint_uploaded_for_yield)
+                    .await
+                    .unwrap_or(false);
+                if !already_uploaded_for_yield {
+                    let _ = Self::run_snapshot_upload(&foreground, false).await;
+                }
 
                 if tx.send(result).is_err() {
                     report_error!("Caller did not wait for agent driver to finish");
@@ -4078,11 +4092,23 @@ impl AgentDriver {
                     let checkpoint_task_id = me.task_id;
                     let checkpoint_foreground = ctx.spawner();
                     ctx.spawn(
-                        async move { Self::run_snapshot_upload(&checkpoint_foreground).await },
-                        move |_me, checkpoint_succeeded, _ctx| {
+                        async move {
+                            // `require_genuine_checkpoint=true`: a hibernation may only
+                            // claim success when a resumable checkpoint was actually
+                            // produced. Unlike most callers of `run_snapshot_upload`, a
+                            // skipped upload (checkpointing disabled, no cloud task id,
+                            // or `--no-snapshot`) is not "nothing to fail" here -- it
+                            // means the closed wait tool call has no persisted result to
+                            // rehydrate from, so it must not be reported as success.
+                            Self::run_snapshot_upload(&checkpoint_foreground, true).await
+                        },
+                        move |me, checkpoint_succeeded, _ctx| {
                             log::info!(
                                 "Ambient agent idle lifecycle: event=run_completion_immediate task_id={checkpoint_task_id:?} terminal_view_id={terminal_id:?} outcome=yielded_for_events checkpoint_succeeded={checkpoint_succeeded}"
                             );
+                            // Attempted (successfully or not) here, so the unconditional
+                            // end-of-run cleanup must not upload a second final checkpoint.
+                            me.final_checkpoint_uploaded_for_yield = true;
                             let status = if checkpoint_succeeded {
                                 SDKConversationOutputStatus::YieldedForEvents
                             } else {
@@ -4492,22 +4518,30 @@ impl AgentDriver {
     /// Invoke the end-of-run snapshot upload pipeline if the feature flag is enabled and this
     /// driver is associated with a cloud task. Returns whether the upload is known to have
     /// succeeded. Most callers treat this as best-effort and ignore the result; a caller that
-    /// must prove resumability (e.g. the `wait_for_events` warm-wait-window yield) inspects it.
+    /// must prove resumability (e.g. the `wait_for_events` warm-wait-window yield) inspects it
+    /// and passes `require_genuine_checkpoint=true`.
     ///
-    /// A skipped upload (feature disabled, no task id, `--no-snapshot`) counts as success:
-    /// there was nothing to fail. The periodic checkpoint coordinator is itself best-effort and
-    /// does not report per-attempt success, so reaching its `finalize` call also counts as
-    /// success; only the legacy one-shot upload's timeout is currently observable as failure.
+    /// When `require_genuine_checkpoint` is `false` (the default for ordinary end-of-run
+    /// cleanup), a skipped upload (feature disabled, no task id, `--no-snapshot`) counts as
+    /// success: there was nothing to fail. When it is `true`, those same skip conditions
+    /// return `false` instead: a hibernation cannot claim a checkpoint it never produced, so
+    /// the caller must treat that as a failed yield rather than a clean one. The periodic
+    /// checkpoint coordinator is itself best-effort and does not report per-attempt success,
+    /// so reaching its `finalize` call also counts as success; only the legacy one-shot
+    /// upload's timeout is currently observable as failure.
     #[tracing::instrument(skip_all, fields(tags.cloud_agent = true))]
-    async fn run_snapshot_upload(spawner: &ModelSpawner<Self>) -> bool {
+    async fn run_snapshot_upload(
+        spawner: &ModelSpawner<Self>,
+        require_genuine_checkpoint: bool,
+    ) -> bool {
         if !FeatureFlag::OzHandoff.is_enabled() {
-            return true;
+            return !require_genuine_checkpoint;
         }
 
         // 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),
+            maybe_task_id,
             snapshot_disabled,
             upload_timeout,
             script_timeout,
@@ -4524,11 +4558,14 @@ impl AgentDriver {
             })
             .await
         else {
-            return true;
+            return !require_genuine_checkpoint;
+        };
+        let Some(task_id) = maybe_task_id else {
+            return !require_genuine_checkpoint;
         };
         if snapshot_disabled {
             log::info!("Skipping snapshot upload because --no-snapshot was specified");
-            return true;
+            return !require_genuine_checkpoint;
         }
 
         // An active coordinator replaces the legacy upload below. Budget must come from

app/src/ai/agent_sdk/driver_tests.rs

@@ -1016,6 +1016,50 @@ fn sandbox_deadline_reports_error_before_exit() {
     );
 }
 
+// ── QUALITY-1759 revision: genuine-checkpoint gate and no double upload ────
+
+/// `run_snapshot_upload` with `require_genuine_checkpoint=true` must not claim success
+/// when checkpointing is not actually available (here: `FeatureFlag::OzHandoff` disabled,
+/// its default state in tests unless explicitly overridden). A hibernation cannot claim a
+/// checkpoint it never produced.
+#[test]
+fn run_snapshot_upload_fails_genuine_requirement_when_oz_handoff_disabled() {
+    App::test((), |mut app| async move {
+        initialize_app_for_terminal_view(&mut app);
+        let temp = TempDir::new().unwrap();
+        let working_dir = dunce::canonicalize(temp.path()).unwrap();
+        let terminal_view = add_window_with_terminal(&mut app, None);
+        let driver_handle = app.add_model(|ctx| {
+            let terminal_driver =
+                super::terminal::TerminalDriver::create_from_existing_view(terminal_view, ctx);
+            AgentDriver::new_for_test(working_dir.clone(), terminal_driver, ctx)
+        });
+
+        let (tx, rx) = futures::channel::oneshot::channel::<(bool, bool)>();
+        driver_handle.update(&mut app, |_, ctx| {
+            let spawner = ctx.spawner();
+            ctx.spawn(
+                async move {
+                    let genuine = AgentDriver::run_snapshot_upload(&spawner, true).await;
+                    let best_effort = AgentDriver::run_snapshot_upload(&spawner, false).await;
+                    let _ = tx.send((genuine, best_effort));
+                },
+                |_, _, _| {},
+            );
+        });
+        let (genuine_result, best_effort_result) = rx.await.unwrap();
+
+        assert!(
+            !genuine_result,
+            "a hibernation must not claim checkpoint success when checkpointing is unavailable"
+        );
+        assert!(
+            best_effort_result,
+            "ordinary best-effort cleanup must still treat a skipped upload as success"
+        );
+    });
+}
+
 #[test]
 fn task_env_vars_include_parent_run_id_when_present() {
     let task_id: AmbientAgentTaskId = "550e8400-e29b-41d4-a716-446655440000".parse().unwrap();