diff --git a/crates/buzz-acp/src/checkpoint.rs b/crates/buzz-acp/src/checkpoint.rs index 9413c008fcb..b593b57c8c4 100644 --- a/crates/buzz-acp/src/checkpoint.rs +++ b/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 + 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 SaveFuture + Send>, last_started: Option, running: Option>, + 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)] diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 90b71f0f361..9e03681d30a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/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)); diff --git a/voltage/images/buzz-agent/checkpoint.py b/voltage/images/buzz-agent/checkpoint.py index 9466b23b804..6e516520985 100644 --- a/voltage/images/buzz-agent/checkpoint.py +++ b/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,16 +85,49 @@ 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): @@ -92,8 +135,11 @@ def pack(workspace, output): 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]]) diff --git a/voltage/images/buzz-agent/test-checkpoint.py b/voltage/images/buzz-agent/test-checkpoint.py index 91bafa819f5..c3e18dd4deb 100644 --- a/voltage/images/buzz-agent/test-checkpoint.py +++ b/voltage/images/buzz-agent/test-checkpoint.py @@ -89,6 +89,65 @@ def git(repo, *args): (repo/'leak.txt').unlink() print('PASS: untracked GitHub or Claude credential causes checkpoint refusal') + # A credential already on the remote does not block the checkpoint; one the + # agent adds, in a file or in an unpushed commit, does. + token = 'github_token = "ghp_'+'aB3dE6gH9jK2mN5pQ8sT1vW4yZ7cF0iL3oR6'+'"\n' + upstream = root/'upstream' + upstream.mkdir() + git(upstream,'init','-b','main') + git(upstream,'config','user.name','Checkpoint Fixture') + git(upstream,'config','user.email','fixture@example.invalid') + (upstream/'values.yaml').write_text(token) + git(upstream,'add','.') + git(upstream,'commit','-m','chore: commit a credential upstream') + clone = workspace/'clone' + subprocess.check_call(['git','clone','-q',str(upstream),str(clone)],stderr=subprocess.DEVNULL) + git(clone,'remote','set-url','origin','https://github.com/example/upstream.git') + git(clone,'config','user.name','Checkpoint Fixture') + git(clone,'config','user.email','fixture@example.invalid') + checkpoint.pack(workspace,root/'upstream.tar.gz') + # Editing other lines of a file that holds an upstream credential is allowed, + # and the checkpoint restores. + (clone/'values.yaml').write_text(token+'replicas: 2\n') + checkpoint.pack(workspace,root/'upstream-edited.tar.gz') + checkpoint.restore(root/'upstream-edited.tar.gz',root/'restored-upstream') + assert (root/'restored-upstream/clone/values.yaml').read_text() == token+'replicas: 2\n' + leaked = token+'replicas: 2\ngithub_token_2 = "ghp_'+'Zy9Xw8Vu7Ts6Rq5Po4Nm3Lk2Ji1Hg0Fe9Dc8'+'"\n' + (clone/'values.yaml').write_text(leaked) + try: + checkpoint.pack(workspace,root/'edited.tar.gz') + raise AssertionError('secret scanner admitted a credential added to a tracked file') + except ValueError: + pass + # Restore applies the same scope: an archive whose files add a credential + # is refused, and leaves no destination behind. + forged = root/'forged' + with tarfile.open(root/'upstream-edited.tar.gz') as archive: + archive.extractall(forged, filter='data') + index = [r['path'] for r in json.loads((forged/'snapshot/manifest.json').read_text())['repositories']].index('clone') + (forged/'snapshot/repositories'/str(index)/'files/values.yaml').write_text(leaked) + with tarfile.open(root/'forged.tar.gz','w:gz') as archive: + archive.add(forged/'snapshot',arcname='snapshot') + try: + checkpoint.restore(root/'forged.tar.gz',root/'restored-forged') + raise AssertionError('restore admitted a credential the archive added') + except ValueError: + pass + assert not (root/'restored-forged').exists() + git(clone,'commit','-am','chore: commit a credential locally') + try: + checkpoint.pack(workspace,root/'committed.tar.gz') + raise AssertionError('secret scanner admitted a credential in an unpushed commit') + except ValueError: + pass + # The exit code names the step, because buzz-acp discards helper output. + cli = subprocess.run(['buzz-agent-checkpoint','pack','--workspace',str(workspace), + '--archive',str(root/'cli.tar.gz')],capture_output=True,text=True) + assert cli.returncode == 11 and 'while scanning for secrets' in cli.stderr, cli.returncode + git(clone,'reset','-q','--hard','origin/main') + checkpoint.pack(workspace,root/'reset.tar.gz') + print('PASS: credentials already on the remote are allowed and restore; one added in a file, an unpushed commit or an archive is refused with the scan step') + (root/'private').write_text('outside workspace') (repo/'escape').symlink_to(root/'private') try: