Fix the finalize budget, hand the writer to the coordinator, drop scaffolding

Review follow-ups on the AgentDriver wiring.

1. Derive the finalize budget instead of passing `upload_timeout`.

The coordinator's floor is `script_timeout + upload_timeout`, so passing
`upload_timeout` alone made `remaining > floor` false for every possible
configuration (defaults: 120s > 60s + 120s). Because this path `return`s
past the legacy one-shot upload, enabling `PeriodicHandoffCheckpoints`
meant no end-of-run snapshot at all: runs shorter than one checkpoint
interval uploaded nothing, and longer runs lost everything since the last
periodic tick.

Use `checkpoint_coordinator::finalize_budget(script_timeout,
upload_timeout)`, which owns the floor so the two cannot drift apart.

2. Give the coordinator the declarations writer.

`run_snapshot_upload` flushes queued driver-side `file` appends before
running the declarations script so none is in flight when the bash script
starts appending to the same JSONL. The coordinator path skipped that, so
checkpoints could miss the agent's most recent edits. The coordinator is
gated on a superset of the writer's own conditions, so the handle is
always present when the coordinator exists.

3. Remove the temporary `allow(dead_code)` scaffolding.

Both earlier PRs in this stack noted their allows were "removable once
that PR is merged on top of this one" — this is that PR. Removes the
module-wide `#![allow(dead_code)]` in `checkpoint_coordinator` (which
would otherwise permanently blind the lint over new async/networking
code) plus the per-item allows in `snapshot` and `harness_support`, along
with the now-stale comments explaining them.

Verified with `cargo clippy -p warp --all-targets --tests -- -D warnings`.

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

Author: joeywangzr

app/src/ai/agent_sdk/driver.rs

@@ -785,6 +785,10 @@ impl AgentDriver {
                     client,
                     id,
                     working_dir.clone(),
+                    // Shared with the history subscription so every attempt can drain
+                    // queued `file` appends before the declarations script runs, exactly
+                    // as `run_snapshot_upload` does on the legacy path.
+                    snapshot_file_writer.clone(),
                     ctx.spawner(),
                     checkpoint_interval
                         .unwrap_or(checkpoint_coordinator::DEFAULT_CHECKPOINT_INTERVAL),
@@ -3855,12 +3859,22 @@ impl AgentDriver {
         }
 
         // 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
+        // end-of-run path: `finalize` drains the declarations writer, regenerates
+        // declarations, runs one last best-effort attempt, 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.
+        //
+        // The budget must come from `finalize_budget`, not from `upload_timeout` alone:
+        // the coordinator's floor is `script_timeout + upload_timeout`, so a smaller
+        // budget silently skips the final attempt — and since this path `return`s past
+        // the legacy upload below, that would mean no end-of-run snapshot at all.
         if let Some(coordinator) = checkpoint_coordinator {
-            coordinator.finalize(upload_timeout).await;
+            coordinator
+                .finalize(checkpoint_coordinator::finalize_budget(
+                    script_timeout,
+                    upload_timeout,
+                ))
+                .await;
             return;
         }
 

app/src/ai/agent_sdk/driver/checkpoint_coordinator.rs

@@ -16,14 +16,6 @@
 //! This trades a small amount of latency (up to [`SAFE_BOUNDARY_POLL_INTERVAL`]) for
 //! avoiding new push-subscription wiring through the UI model graph.
 
-// This module has no production caller yet -- `AgentDriver` wires
-// `CheckpointCoordinatorHandle::new` into its spawn/finalize lifecycle in a
-// follow-up, stacked PR. Until then, `CheckpointCoordinatorHandle::new` (and
-// everything it reaches) is unreachable outside `#[cfg(test)]`, which reaches the
-// same code through `new_for_test`. This module-level allow is temporary and should
-// be removable once that PR merges on top of this one.
-#![allow(dead_code)]
-
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;

app/src/ai/agent_sdk/driver/snapshot.rs

@@ -665,9 +665,6 @@ struct SnapshotOutcome {
 
 /// Outcome of one checkpoint attempt, where [`SnapshotOutcome`] only covers per-entry upload
 /// results within that attempt.
-// The checkpoint pipeline below has no production caller until the periodic coordinator
-// lands in a follow-up, stacked PR; the `allow(dead_code)`s go away with it.
-#[allow(dead_code)]
 #[derive(Debug)]
 pub(super) enum CheckpointResult {
     /// Every required object uploaded and the exact-set commit succeeded, so `generation` is
@@ -692,13 +689,10 @@ pub(super) enum CheckpointResult {
 /// [`SnapshotUploadMode`] for the server-side semantics.
 enum PipelineMode {
     Legacy,
-    // Not constructed until the coordinator PR.
-    #[allow(dead_code)]
     Checkpoint(CheckpointGeneration),
 }
 
 /// Disambiguates [`mint_generation`] calls landing in the same millisecond.
-#[allow(dead_code)]
 static GENERATION_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
 
 /// Mint a `<millis-since-epoch>-<counter>` generation identifier, which satisfies
@@ -707,7 +701,6 @@ static GENERATION_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::Ato
 /// Call this exactly once per attempt, after that attempt's payload has been gathered.
 /// Re-uploading an already-gathered payload must reuse its generation; enforcing that is the
 /// caller's job (see the coordinator).
-#[allow(dead_code)]
 pub(super) fn mint_generation() -> CheckpointGeneration {
     let millis = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -723,7 +716,6 @@ pub(super) fn mint_generation() -> CheckpointGeneration {
 /// Only needed to assemble the exact-set [`CommitSnapshotRequest`]: everything earlier in the
 /// pipeline speaks logical names, and the server derives the storage name for each presigned
 /// target itself.
-#[allow(dead_code)]
 fn storage_name(generation: &CheckpointGeneration, logical: &str) -> String {
     format!("checkpoint_{}__{logical}", generation.as_str())
 }
@@ -981,7 +973,6 @@ async fn run_pipeline(
 /// Unlike [`upload_snapshot_from_declarations_file`], an unusable declarations file is
 /// [`CheckpointResult::Skipped`] rather than `None`, because the coordinator's state machine
 /// distinguishes "nothing to do" from "tried and failed".
-#[allow(dead_code)]
 pub(super) async fn run_checkpoint_from_declarations_file(
     path: &Path,
     client: Arc<dyn HarnessSupportClient>,
@@ -1003,7 +994,6 @@ pub(super) async fn run_checkpoint_from_declarations_file(
 
 /// Upload and commit an already-gathered payload under `generation`. Split out so a caller
 /// re-running the exact same attempt can reuse both the payload and the generation.
-#[allow(dead_code)]
 async fn run_checkpoint_pipeline(
     client: Arc<dyn HarnessSupportClient>,
     generation: CheckpointGeneration,

app/src/server/server_api/harness_support.rs

@@ -132,8 +132,6 @@ impl CheckpointGeneration {
     /// Construct from a string the caller has already shaped to [`Self::is_valid`].
     /// `snapshot::mint_generation` is the only production caller and satisfies it by
     /// construction, so the invariant is a debug assertion rather than a fallible return.
-    // Unused until the periodic checkpoint coordinator (a follow-up, stacked PR) lands.
-    #[allow(dead_code)]
     pub(crate) fn from_validated(value: String) -> Self {
         debug_assert!(
             Self::is_valid(&value),
@@ -142,7 +140,6 @@ impl CheckpointGeneration {
         Self(value)
     }
 
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &str {
         &self.0
     }
@@ -162,16 +159,13 @@ impl std::fmt::Display for CheckpointGeneration {
 ///
 /// Exact-set: the server persists `objects` verbatim as the commit marker and selection
 /// later returns exactly that set, not everything sharing the generation prefix.
-// Not constructed until the periodic checkpoint coordinator (a follow-up, stacked PR) lands.
-#[allow(dead_code)]
 #[derive(Debug, Clone, serde::Serialize)]
 pub struct CommitSnapshotRequest {
     pub generation: String,
     pub manifest_object: String,
     pub objects: Vec<String>,
 }
 
-#[allow(dead_code)]
 #[derive(Debug, Clone, serde::Deserialize)]
 pub struct CommitSnapshotResponse {
     pub generation: String,
@@ -347,8 +341,6 @@ pub trait HarnessSupportClient: 'static + Send + Sync {
     /// Only call this once every object in `request.objects` (including
     /// `request.manifest_object`) has uploaded successfully; the server verifies existence
     /// and per-attempt size limits and rejects the whole commit otherwise.
-    // Not called until the periodic checkpoint coordinator (a follow-up, stacked PR) lands.
-    #[allow(dead_code)]
     async fn commit_snapshot(
         &self,
         request: &CommitSnapshotRequest,