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

@@ -670,11 +670,6 @@ struct SnapshotOutcome {
 
 /// Outcome of one checkpoint attempt, as opposed to [`SnapshotOutcome`] which only tracks
 /// per-entry upload results within a single attempt.
-// The whole checkpoint pipeline below (through `run_checkpoint_pipeline`) has no
-// production caller yet -- the periodic checkpoint coordinator that drives it lands
-// in a follow-up, stacked PR. `#[allow(dead_code)]` is temporary and should be
-// removable once that PR is merged on top of this one.
-#[allow(dead_code)]
 #[derive(Debug)]
 pub(super) enum CheckpointResult {
     /// Every required object (blobs plus manifest) for `generation` uploaded successfully
@@ -710,14 +705,11 @@ enum PipelineMode {
     Legacy,
     /// Periodic or finalization checkpoint attempt: the server stores each requested file
     /// as `checkpoint_<generation>__<filename>` and does not charge the cumulative quota.
-    // Not constructed until the coordinator PR (see the allow(dead_code) note above).
-    #[allow(dead_code)]
     Checkpoint(CheckpointGeneration),
 }
 
 /// Monotonic disambiguator for [`mint_generation`] so two attempts minted within the same
 /// millisecond (e.g. in tests, or on a very fast retry) never collide.
-#[allow(dead_code)]
 static GENERATION_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
 
 /// Mint a new checkpoint generation identifier.
@@ -731,7 +723,6 @@ static GENERATION_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::Ato
 ///
 /// Format: `<millis-since-epoch>-<counter>`. This satisfies the server's
 /// `[A-Za-z0-9._-]{1,128}` charset and never contains the reserved `__` separator.
-#[allow(dead_code)]
 pub(super) fn mint_generation() -> CheckpointGeneration {
     let millis = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -749,7 +740,6 @@ pub(super) fn mint_generation() -> CheckpointGeneration {
 /// building, and the upload-targets request; the server itself derives the storage name for
 /// each presigned upload target from that logical filename plus the request's `generation`
 /// field, so no client-side renaming is needed before that point.
-#[allow(dead_code)]
 fn storage_name(generation: &CheckpointGeneration, logical: &str) -> String {
     debug_assert!(
         !logical.contains("__"),
@@ -1012,7 +1002,6 @@ async fn run_pipeline(
 /// Unlike [`upload_snapshot_from_declarations_file`], a missing/empty/unusable declarations
 /// file is reported as [`CheckpointResult::Skipped`] rather than `None`, since the coordinator
 /// needs to distinguish "nothing to do" from "tried and failed" to drive its state machine.
-#[allow(dead_code)]
 pub(super) async fn run_checkpoint_from_declarations_file(
     path: &Path,
     client: Arc<dyn HarnessSupportClient>,
@@ -1036,7 +1025,6 @@ pub(super) async fn run_checkpoint_from_declarations_file(
 /// Upload and commit an already-gathered payload under `generation`. Split out from
 /// [`run_checkpoint_from_declarations_file`] so a caller retrying the exact same attempt (as
 /// opposed to gathering fresh) 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

@@ -150,9 +150,6 @@ impl CheckpointGeneration {
     /// enforced it, so "validated" was aspirational. `snapshot::mint_generation` is the only
     /// production caller and provably satisfies it, hence a `debug_assert!` rather than a
     /// fallible constructor.
-    // Only called by `snapshot::mint_generation`, which is itself 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),
@@ -161,7 +158,6 @@ impl CheckpointGeneration {
         Self(value)
     }
 
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &str {
         &self.0
     }
@@ -181,17 +177,13 @@ impl std::fmt::Display for CheckpointGeneration {
 /// server persists `objects` verbatim as the commit marker and later selection returns
 /// exactly that set, never every object sharing the generation prefix. See
 /// `docs/remote-2111-checkpoint-spec.md` (warp-server) for the full protocol.
-// Not constructed until the periodic checkpoint coordinator (a follow-up, stacked PR)
-// starts calling `HarnessSupportClient::commit_snapshot`.
-#[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,
@@ -367,9 +359,6 @@ pub trait HarnessSupportClient: 'static + Send + Sync {
     /// (including `request.manifest_object`) has itself uploaded successfully; the
     /// server verifies existence and per-attempt size limits before this becomes the
     /// selected checkpoint.
-    // Not called until the periodic checkpoint coordinator (a follow-up, stacked PR)
-    // lands.
-    #[allow(dead_code)]
     async fn commit_snapshot(
         &self,
         request: &CommitSnapshotRequest,