fix(agents): checkpoint Sandboxes whose repositories have credentials upstream

crates/buzz-acp/src/checkpoint.rs

@@ -46,15 +46,40 @@ impl CheckpointError {
         }
     }
 
+    /// The helper step that failed, from its exit code.
+    fn step(&self) -> Option<&'static str> {
+        match self {
+            Self::ExitStatus(status) => helper_step(status.code()?),
+            _ => None,
+        }
+    }
+
     fn details(&self) -> serde_json::Value {
         let mut details = serde_json::json!({"reason": self.reason()});
         if let Self::ExitStatus(status) = self {
             details["exit_code"] = status.code().into();
         }
+        if let Some(step) = self.step() {
+            details["step"] = step.into();
+        }
         details
     }
 }
 
+/// `buzz-agent-checkpoint` exits with one code per step (`STEPS` in
+/// `voltage/images/buzz-agent/checkpoint.py`), so a failure names its step
+/// without the helper output.
+fn helper_step(code: i32) -> Option<&'static str> {
+    Some(match code {
+        10 => "reading the checkpoint settings",
+        11 => "scanning for secrets",
+        12 => "packing the workspace",
+        13 => "uploading the checkpoint",
+        14 => "restoring the checkpoint",
+        _ => return None,
+    })
+}
+
 /// Only these fields reach activity; anything else the helper adds is ignored.
 #[derive(Deserialize)]
 struct Receipt {
@@ -127,12 +152,14 @@ type SaveFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
 /// sends no SIGTERM, so the save on stop never runs; these bound the loss to
 /// the work since the last one. Saves start only at idle points, one runs at
 /// a time, at most one starts per `interval`, and the save on stop still runs
-/// after the adapters exit.
+/// after the adapters exit. A completed turn whose save cannot start yet stays
+/// pending until [`TurnCheckpoints::at_idle`] finds an idle point.
 pub(crate) struct TurnCheckpoints {
     interval: Duration,
     start: Box<dyn FnMut() -> SaveFuture + Send>,
     last_started: Option<tokio::time::Instant>,
     running: Option<tokio::task::JoinHandle<()>>,
+    pending: bool,
 }
 
 /// What the harness still has in progress when a turn completes.
@@ -174,6 +201,7 @@ impl TurnCheckpoints {
             start,
             last_started: None,
             running: None,
+            pending: false,
         }
     }
 
@@ -184,10 +212,24 @@ impl TurnCheckpoints {
     /// turn that starts while the save runs can still race it, so only the
     /// final save on stop attests the workspace.
     pub(crate) fn after_turn(&mut self, busy: Busy) {
-        if self.interval.is_zero()
-            || busy.any()
-            || self.running.as_ref().is_some_and(|t| !t.is_finished())
-        {
+        if self.interval.is_zero() {
+            return;
+        }
+        self.pending = true;
+        self.at_idle(busy);
+    }
+
+    /// Whether a completed turn still waits for its save.
+    pub(crate) fn is_pending(&self) -> bool {
+        self.pending
+    }
+
+    /// Start the save a completed turn is still waiting for, under the same
+    /// rules as [`TurnCheckpoints::after_turn`]. A turn's result arrives
+    /// before its prompt task is joined, so the save can find the pool busy
+    /// at that turn's own completion.
+    pub(crate) fn at_idle(&mut self, busy: Busy) {
+        if !self.pending || busy.any() || self.running.as_ref().is_some_and(|t| !t.is_finished()) {
             return;
         }
         let now = tokio::time::Instant::now();
@@ -197,6 +239,7 @@ impl TurnCheckpoints {
         {
             return;
         }
+        self.pending = false;
         self.last_started = Some(now);
         self.running = Some(tokio::spawn((self.start)()));
     }
@@ -300,6 +343,7 @@ async fn save_with_command(
         Err(error) => {
             tracing::error!(
                 reason = error.reason(),
+                step = error.step(),
                 "workspace checkpoint failed: {error}"
             );
             emit("checkpoint_failed", error.details());
@@ -429,6 +473,10 @@ mod tests {
                 "exit 3",
                 serde_json::json!({"reason":"exit_status","exit_code":3}),
             ),
+            (
+                "exit 11",
+                serde_json::json!({"reason":"exit_status","exit_code":11,"step":"scanning for secrets"}),
+            ),
             (
                 "printf 'not a receipt'",
                 serde_json::json!({"reason":"receipt_invalid"}),
@@ -714,12 +762,53 @@ mod tests {
         }
     }
 
+    /// A completed turn whose save could not start saves at the next idle
+    /// point without waiting for another turn, once the interval allows it.
+    #[tokio::test(start_paused = true)]
+    async fn a_deferred_turn_checkpoint_saves_at_the_next_idle_point() {
+        use std::sync::atomic::Ordering::SeqCst;
+        let (mut checkpoints, started) = counting(Duration::from_secs(600), Duration::from_secs(1));
+        // The turn's own prompt task is still unjoined when its result arrives.
+        checkpoints.after_turn(Busy {
+            prompt_tasks: true,
+            ..Busy::default()
+        });
+        settle_tasks().await;
+        assert_eq!(started.load(SeqCst), 0);
+        assert!(checkpoints.is_pending());
+        checkpoints.at_idle(Busy::default());
+        settle_tasks().await;
+        assert_eq!(started.load(SeqCst), 1);
+        assert!(!checkpoints.is_pending());
+        // Nothing pending: an idle point does not save.
+        tokio::time::advance(Duration::from_secs(601)).await;
+        checkpoints.at_idle(Busy::default());
+        settle_tasks().await;
+        assert_eq!(started.load(SeqCst), 1);
+        // A turn inside the interval stays pending until the interval ends.
+        checkpoints.after_turn(Busy::default());
+        settle_tasks().await;
+        assert_eq!(started.load(SeqCst), 2);
+        checkpoints.after_turn(Busy::default());
+        checkpoints.at_idle(Busy::default());
+        settle_tasks().await;
+        assert_eq!(started.load(SeqCst), 2);
+        assert!(checkpoints.is_pending());
+        tokio::time::advance(Duration::from_secs(600)).await;
+        settle_tasks().await;
+        checkpoints.at_idle(Busy::default());
+        settle_tasks().await;
+        assert_eq!(started.load(SeqCst), 3);
+    }
+
     #[tokio::test(start_paused = true)]
     async fn zero_interval_disables_turn_checkpoints() {
         let (mut checkpoints, started) = counting(Duration::ZERO, Duration::ZERO);
         checkpoints.after_turn(Busy::default());
+        checkpoints.at_idle(Busy::default());
         settle_tasks().await;
         assert_eq!(started.load(std::sync::atomic::Ordering::SeqCst), 0);
+        assert!(!checkpoints.is_pending());
     }
 
     #[tokio::test(start_paused = true)]

crates/buzz-acp/src/lib.rs

@@ -3344,7 +3344,8 @@ async fn run_harness(
             }
         }
 
-        if pool_ready && last_maintenance.elapsed() >= maintenance_interval {
+        let maintenance_due = last_maintenance.elapsed() >= maintenance_interval;
+        if pool_ready && maintenance_due {
             last_maintenance = std::time::Instant::now();
             queue.compact_expired_state();
 
@@ -3390,6 +3391,34 @@ async fn run_harness(
             }
         }
 
+        // A turn's save that found its own prompt task still unjoined, or that
+        // fell inside the interval, starts at the first idle tick. This tick
+        // also runs while a lazy pool sleeps, so the save does not wait for
+        // the next event.
+        if maintenance_due && turn_checkpoints.is_pending() {
+            last_maintenance = std::time::Instant::now();
+            if drain_ready_join_results(
+                &mut pool,
+                &mut queue,
+                config,
+                &mut heartbeat_in_flight,
+                &removed_channels,
+                &mut typing_channels,
+                &mut crash_history,
+                &respawn_tx,
+                &mut respawn_tasks,
+                observer.clone(),
+            ) == LoopAction::Exit
+            {
+                break;
+            }
+            turn_checkpoints.at_idle(checkpoint::Busy {
+                turn_in_flight: queue.has_in_flight() || heartbeat_in_flight,
+                prompt_tasks: !pool.join_set.is_empty(),
+                work_queued: queue.has_undispatched_work(),
+            });
+        }
+
         // Reap completed respawn handles from the JoinSet. Payloads are
         // delivered out-of-band through `respawn_rx` (selected below), so the
         // JoinSet is never joined by the normal flow — Tokio retains finished
@@ -3408,7 +3437,8 @@ async fn run_harness(
         } else {
             None
         };
-        let maintenance_at = pool_ready.then_some(last_maintenance + maintenance_interval);
+        let maintenance_at = (pool_ready || turn_checkpoints.is_pending())
+            .then_some(last_maintenance + maintenance_interval);
 
         // Borrow result_rx and join_set simultaneously via split-borrow helper.
         pool.retain_held_scopes(|scope| queue.has_pending_scope(scope));

voltage/images/buzz-agent/checkpoint.py

@@ -23,6 +23,16 @@
             'node_modules', 'target', '__pycache__', 'dist', 'build', '.terraform'}
 PRIVATE_NAMES = {'credentials', 'credentials.json', 'auth.json', 'id_rsa', 'id_ed25519',
                  '.npmrc', '.netrc', '.pypirc', '.git-credentials'}
+# buzz-acp discards this helper's output, so the exit code names the failed
+# step (helper_step in crates/buzz-acp/src/checkpoint.rs).
+STEPS = {'reading the checkpoint settings': 10, 'scanning for secrets': 11,
+         'packing the workspace': 12, 'uploading the checkpoint': 13,
+         'restoring the checkpoint': 14}
+STEP = ['packing the workspace']
+
+
+def step(name):
+    STEP[0] = name
 
 
 def run(*args, cwd=None, timeout=60, env=None):
@@ -75,25 +85,61 @@ def public_origin(repo):
 
 
 def scan(directory, *, history=False):
+    """Refuse a credential the agent added: commits not on any remote, or files.
+
+    History and files already on a remote are on the forge, where the agent's
+    token can read them, so the scan leaves them out."""
     # Immutable image configuration prevents workspace allowlists from bypassing
     # the scan. Scanner output can contain secrets even on errors, so suppress it.
+    previous = STEP[0]
+    step('scanning for secrets')
     env = {k: v for k, v in os.environ.items() if not k.startswith('GITLEAKS_')}
     command = ['gitleaks', 'git' if history else 'dir', str(directory),
                '--config', '/opt/agent-tools/checkpoint-gitleaks.toml',
                '--gitleaks-ignore-path', '/dev/null', '--ignore-gitleaks-allow',
                '--no-banner', '--redact', '--timeout', '60']
     if history:
-        command += ['--log-opts=--all HEAD']
+        command += ['--log-opts=--all HEAD --not --remotes']
     run(*command, timeout=65, env=env)
+    step(previous)
+
+
+def review_changes(repo, review):
+    """Write what the working tree adds to HEAD into review, for the secret scan.
+
+    Untracked files are copied whole. A tracked file contributes only its added
+    lines, so a credential already upstream in an edited file is not a finding."""
+    def write(path, content):
+        target = review / path
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_bytes(content)
+
+    for raw in git(repo, 'ls-files', '-z', '--others', '--exclude-standard').split(b'\0'):
+        path = safe_path(os.fsdecode(raw)) if raw else None
+        if path and not excluded(path) and (repo / path).is_file() and not (repo / path).is_symlink():
+            write(path, (repo / path).read_bytes())
+    for raw in git(repo, 'diff', '--name-only', '--no-renames', '-z', 'HEAD').split(b'\0'):
+        path = safe_path(os.fsdecode(raw)) if raw else None
+        if not path or excluded(path):
+            continue
+        patch = git(repo, 'diff', '--no-ext-diff', '--no-color', '--no-renames', '-U0',
+                    'HEAD', '--', str(path))
+        hunks = patch.split(b'\n@@', 1)[1] if b'\n@@' in patch else b''
+        added = [line[1:] for line in hunks.split(b'\n') if line.startswith(b'+')]
+        if added:
+            write(path, b'\n'.join(added) + b'\n')
 
 
 def pack(workspace, output):
     workspace = workspace.resolve(strict=True)
     manifest = {'version': 1, 'repositories': []}
     total = 0
     count = 0
-    with tempfile.TemporaryDirectory(prefix='agent-checkpoint-') as temp:
+    with tempfile.TemporaryDirectory(prefix='agent-checkpoint-') as temp, \
+            tempfile.TemporaryDirectory(prefix='agent-checkpoint-review-') as review_temp:
         stage = Path(temp)
+        # What the agent added to each repository, for the secret scan.
+        review = Path(review_temp)
         for index, repo in enumerate(repositories(workspace)):
             relative = repo.relative_to(workspace).as_posix()
             if relative == '.':
@@ -105,6 +151,7 @@ def pack(workspace, output):
             destination = stage / 'repositories' / str(index)
             (destination / 'files').mkdir(parents=True)
             scan(repo, history=True)
+            review_changes(repo, review / str(index))
             git(repo, 'bundle', 'create', str(destination / 'history.bundle'), '--all', 'HEAD')
             total += (destination / 'history.bundle').stat().st_size
             names = git(repo, 'ls-files', '-z', '--cached', '--others', '--exclude-standard')
@@ -146,7 +193,8 @@ def pack(workspace, output):
         if total > MAX_BYTES:
             raise ValueError('Checkpoint history exceeds the 256 MiB source budget')
         (stage / 'manifest.json').write_text(json.dumps(manifest))
-        scan(stage)
+        shutil.copyfile(stage / 'manifest.json', review / 'manifest.json')
+        scan(review)
         with tarfile.open(output, 'w:gz', dereference=False) as archive:
             archive.add(stage, arcname='snapshot')
     os.chmod(output, 0o600)
@@ -175,8 +223,21 @@ def restore(archive_path, destination):
         manifest = json.loads((stage / 'manifest.json').read_text())
         if manifest.get('version') != 1:
             raise ValueError('Unsupported checkpoint version')
-        scan(stage)
         destination.mkdir(mode=0o700, parents=True)
+        try:
+            restore_repositories(stage, manifest, destination)
+        except BaseException:
+            # The destination is new, so a refused restore leaves nothing behind.
+            shutil.rmtree(destination, ignore_errors=True)
+            raise
+        return manifest
+
+
+def restore_repositories(stage, manifest, destination):
+    with tempfile.TemporaryDirectory(prefix='agent-restore-review-') as review_temp:
+        # The same scope as pack(): what each repository adds to its HEAD.
+        review = Path(review_temp)
+        (review / 'manifest.json').write_text(json.dumps(manifest))
         for index, record in enumerate(manifest['repositories']):
             relative = safe_path(record['path'])
             if excluded(relative):
@@ -201,7 +262,8 @@ def restore(archive_path, destination):
                     if not excluded(path) and (target.is_file() or target.is_symlink()):
                         target.unlink()
             shutil.copytree(source / 'files', repo, dirs_exist_ok=True, symlinks=True)
-        return manifest
+            review_changes(repo, review / str(index))
+        scan(review)
 
 
 def storage():
@@ -218,10 +280,13 @@ def storage():
 
 
 def save(workspace):
+    step('reading the checkpoint settings')
     bucket, prefix = storage()
     with tempfile.TemporaryDirectory(prefix='agent-upload-') as temp:
         artifact = Path(temp) / 'checkpoint.tar.gz'
+        step('packing the workspace')
         manifest = pack(workspace, artifact)
+        step('uploading the checkpoint')
         checksum = base64.b64encode(digest(artifact)).decode()
         key = prefix + str(uuid.uuid4()) + '.tar.gz'
         run('aws', 's3api', 'put-object', '--bucket', bucket, '--key', key,
@@ -311,6 +376,8 @@ def main():
     parser.add_argument('--destination', type=Path)
     parser.add_argument('--key')
     args = parser.parse_args()
+    if args.action in ('restore', 'recover', 'restore-once'):
+        step('restoring the checkpoint')
     if args.action == 'save':
         result = save(args.workspace)
     elif args.action == 'pack' and args.archive:
@@ -331,5 +398,6 @@ def main():
     try:
         main()
     except (ValueError, OSError, subprocess.TimeoutExpired, tarfile.TarError):
-        print('Checkpoint failed; retain the workspace and inspect it before cleanup', file=sys.stderr)
-        sys.exit(1)
+        print('Checkpoint failed while '+STEP[0]+'; retain the workspace and inspect it before cleanup',
+              file=sys.stderr)
+        sys.exit(STEPS[STEP[0]])