diff --git a/scripts/smoke/dispatch-claude.sh b/scripts/smoke/dispatch-claude.sh new file mode 100755 index 0000000..2487f8e --- /dev/null +++ b/scripts/smoke/dispatch-claude.sh @@ -0,0 +1,83 @@ +#!/usr/bin/env bash +# Live e2e smoke for `pillbox dispatch --agent claude` — the one-shot CLI-harness +# drive path. claude is `claude -p PROMPT` (one-shot): its prompt must ride the +# LAUNCH argv, not a later `session send` (the server/opencode model). dispatch now +# forks CLI workers WITH the prompt baked in (`run --detach -- ""`) and just +# waits+grades — the fix for the detach-then-send model that booted a VM but never +# ran a turn (→ silent 0/0). The unit tests cover the policy on a mock; this covers +# the REAL claude/Opus drive on a booted VM (the GHOST-004 analog for CLI agents). +# Also exercises 2 CONCURRENT claude forks through the vault (the degraded-lease / +# coalesce path), so it doubles as a fork-k-claude vault check. +# +# Usage: scripts/smoke/dispatch-claude.sh [runner-image] +# Prereqs: codesigned libkrun binary (scripts/lk-build.sh), claude authed +# (`pillbox auth login --agent claude`), the runner image present, Opus reachable. +set -euo pipefail +cd "$(dirname "$0")/../.." || exit 1 + +IMAGE="${1:-pillbox-runner:dev}" +BACKEND="${PILLBOX_BACKEND:-libkrun}" +PB="$(pwd)/target/debug/pillbox" +export PILLBOX_BACKEND="$BACKEND" PILLBOX_RUNNER_IMAGE="$IMAGE" + +fail() { echo " ✗ dispatch-claude: $1"; exit 1; } + +# libkrun-only (same grade-path constraint as dispatch.sh): the grader resolves each +# worker's live workspace via `session info --json` → `.session.workspace`, libkrun +# only. Skip (don't fail the suite) on docker. +if [ "$BACKEND" != libkrun ]; then + echo " · dispatch-claude smoke skipped — backend=$BACKEND not wired (libkrun-only grade path)" + exit 0 +fi +[ "$(nm "$PB" 2>/dev/null | grep -c LibkrunBackend)" -ge 1 ] \ + || fail "binary lacks the libkrun feature — run scripts/lk-build.sh" + +WS="$(mktemp -d)" +cleanup() { + if cd "$WS" 2>/dev/null; then + for s in $("$PB" session list --json 2>/dev/null | jq -r '.sessions[].id' 2>/dev/null); do + "$PB" session rm "$s" >/dev/null 2>&1 + done + "$PB" rm dispatch-claude-smoke >/dev/null 2>&1 + fi + cd / && rm -rf "$WS" +} +trap cleanup EXIT + +cd "$WS" || fail "cd into workspace $WS failed" +"$PB" new --name dispatch-claude-smoke >/dev/null 2>&1 || fail "pillbox new failed" +echo seed >seed.txt +"$PB" push --bookmark base >/dev/null 2>&1 || fail "push --bookmark failed" + +echo "▶ dispatch-claude smoke (image=$IMAGE, k=2 claude/Opus, prompt-at-launch)" +# k=2, no --model → claude's subscription default (Opus). A trivial, deterministic +# task; `--cmd` grades the pulled workspace (the fix is the DRIVE, not cleverness). +OUT="$(PILLBOX_DISPATCH_TURN_TIMEOUT=600 "$PB" dispatch --from-bookmark base \ + -k 2 --agent claude \ + --cmd 'test -f result.txt && grep -qi done result.txt' --json \ + -- 'Create a file named result.txt containing exactly the word DONE' 2>/tmp/dispatch-claude.err)" +RC=$? + +echo "$OUT" | jq . >/dev/null 2>&1 || { tail -6 /tmp/dispatch-claude.err; fail "verdict is not valid JSON: $OUT"; } +[ "$(echo "$OUT" | jq -r '.version')" = 1 ] || fail "verdict version != 1" +N="$(echo "$OUT" | jq -r '.dispatch.workers | length')" +[ "$N" -eq 2 ] || fail "expected 2 workers, got $N" + +# The crux: a CLI agent that was actually DRIVEN produces a gradeable result. Before +# the fix every worker errored/0 (booted, never ran a turn). Assert ≥1 scored. +SCORED="$(echo "$OUT" | jq -r '[.dispatch.workers[] | select(.status=="scored")] | length')" +echo " · $SCORED/2 workers scored" +[ "$SCORED" -ge 1 ] || { echo "$OUT" | jq .; tail -6 /tmp/dispatch-claude.err; fail "no worker drove a turn (the pre-fix 0/0 symptom)"; } + +WINNER="$(echo "$OUT" | jq -r '.dispatch.winner // empty')" +[ -n "$WINNER" ] || { echo "$OUT" | jq .; fail "no winner (rc=$RC)"; } +echo " ✓ winner selected: $WINNER" + +PULLED="$(echo "$OUT" | jq -r '.dispatch.pulled_to // empty')" +{ [ -n "$PULLED" ] && [ -d "$PULLED" ]; } || fail "winner not pulled (pulled_to=$PULLED)" +grep -qi 'done' "$PULLED/result.txt" 2>/dev/null \ + || fail "pulled winner lacks result.txt with DONE — got: $(cat "$PULLED/result.txt" 2>/dev/null | tr '\n' ' ')" +echo " ✓ winner pulled → result.txt recovered (claude/Opus actually ran a turn)" + +[ "$RC" -eq 0 ] || fail "winner found but exit code was $RC (want 0)" +echo " ✓✓ dispatch-claude PASS" diff --git a/src/commands/dispatch.rs b/src/commands/dispatch.rs index 76ac209..99586f0 100644 --- a/src/commands/dispatch.rs +++ b/src/commands/dispatch.rs @@ -313,7 +313,10 @@ trait WorkerDriver { /// Fork a new detached worker (the `i`-th, 0-based) from the bookmark → its /// session id. The index picks this worker's roster row (`--workers-spec`); /// without a roster it's ignored and every fork is identical. - fn fork(&self, i: usize) -> Result; + fn fork(&self, i: usize, first_turn: &str) -> Result; + /// True when this worker's first turn is consumed by the launch argv instead + /// of the in-session prompt API / PTY send path. + fn first_turn_driven_on_fork(&self, i: usize) -> bool; /// Block until the worker's current turn goes idle (or terminates). fn wait_idle(&self, id: &str) -> Result<()>; /// Grade the worker's current workspace against `grader` → the parsed verdict. @@ -344,6 +347,29 @@ fn resolve_worker(opts: &DispatchOpts, i: usize) -> (Option, Option String { + resolve_worker(opts, i) + .0 + .unwrap_or_else(|| default_agent.to_string()) +} + +fn default_agent(resolved: &Pillbox) -> String { + crate::config::resolve_run_config(resolved) + .agent + .unwrap_or_else(|| "claude".into()) +} + +fn agent_first_turn_driven_on_fork(agent: &str) -> bool { + crate::agents::lookup("dispatch", agent) + .map(|spec| spec.server.is_none()) + .unwrap_or(false) +} + +fn worker_first_turn_driven_on_fork(opts: &DispatchOpts, i: usize, default_agent: &str) -> bool { + let agent = effective_worker_agent(opts, i, default_agent); + agent_first_turn_driven_on_fork(&agent) +} + /// The distilled failure summary fed back as the next prompt on a retry — the /// structured signal (which checks failed + why), NOT the raw grader log, per /// the Parallel-Distill-Refine finding (a model acts better on a distilled @@ -570,26 +596,30 @@ fn run_dispatch( segments: Option<&[ResolvedSegment]>, ) -> DispatchVerdict { // Fork all k up front, THEN drive each. Forking first overlaps the k VM - // BOOTS (each `--detach` worker boots in the background); the agent turns - // themselves are driven SERIALLY below — a `--detach` worker comes up idle and - // does nothing until its first `send`, so turns do not overlap. (True - // turn-level concurrency would need driving workers on separate threads; the - // subprocess `WorkerDriver` calls are independent, so that's a safe future - // change — not done here.) A fork that fails becomes an `Errored` worker rather - // than aborting the batch — the successes are still driven, and no forked worker - // is left unrecorded (the orphan a `collect::()?` would leak). + // BOOTS (each `--detach` worker boots in the background). Server-mode worker + // turns are driven SERIALLY below; one-shot CLI workers receive their only + // prompt at launch and are merely awaited/graded below. True turn-level + // concurrency would need driving workers on separate threads; the subprocess + // `WorkerDriver` calls are independent, so that's a safe future change — not + // done here. A fork that fails becomes an `Errored` worker rather than aborting + // the batch — the successes are still driven, and no forked worker is left + // unrecorded (the orphan a `collect::()?` would leak). // // Each worker runs EITHER the segment chain (`--segments`, one session) OR the // single-prompt + retry loop (fork-`k`). With both `-k>1` and `--segments`, the // k workers each run the full chain → best-of-k OVER segmented chains. + let fork_first_turn = if segments.is_some() { "" } else { prompt }; let workers: Vec = (0..k) - .map(|i| driver.fork(i as usize)) + .map(|i| { + let i = i as usize; + (i, driver.fork(i, fork_first_turn)) + }) .collect::>() .into_iter() - .map(|forked| match forked { + .map(|(i, forked)| match forked { Ok(id) => match segments { - Some(segs) => drive_segments(driver, id, prompt, segs, reward, retries), - None => drive_one(driver, id, prompt, retries, reward), + Some(segs) => drive_segments(driver, i, id, prompt, segs, reward, retries), + None => drive_one(driver, i, id, prompt, retries, reward), }, Err(e) => { eprintln!("pillbox: worker fork failed: {e:#}"); @@ -636,12 +666,13 @@ fn run_dispatch( /// outcome (not propagated) so one worker's failure doesn't sink the others. fn drive_one( driver: &dyn WorkerDriver, + i: usize, id: String, prompt: &str, retries: u32, reward: &Grader, ) -> WorkerOutcome { - drive_one_inner(driver, &id, prompt, retries, reward).unwrap_or_else(|e| errored(id, e)) + drive_one_inner(driver, i, &id, prompt, retries, reward).unwrap_or_else(|e| errored(id, e)) } /// The shared `Errored` outcome for a worker whose drive raised (boot/drive/grade @@ -658,20 +689,26 @@ fn errored(id: String, e: anyhow::Error) -> WorkerOutcome { } } -/// The send → wait-idle → grade-against-`grader` → retry loop: re-drive with the -/// distilled failure summary until the grade passes or the `retries` budget is -/// spent → `(final grade, retries used)`. The shared core of both drive paths -/// (fork-`k` grades by the reward; each segment by its gate). The first turn must -/// be a `send` — a `--detach` fork comes up idle and does nothing until driven, -/// and a server agent treats a fork-baked positional as a pre-fill hint, not an -/// executed turn. +/// Drive a worker to a grade. One-shot CLI agents already consumed turn 1 at +/// fork, so they only wait + grade. Server-mode agents use the send → wait-idle +/// → grade-against-`grader` → retry loop: re-drive with the distilled failure +/// summary until the grade passes or the `retries` budget is spent → `(final +/// grade, retries used)`. A server agent treats a fork-baked positional as a +/// pre-fill hint, not an executed turn. fn drive_to_grade( driver: &dyn WorkerDriver, + i: usize, id: &str, first_turn: &str, grader: &Grader, retries: u32, ) -> Result<(Scored, u32)> { + if driver.first_turn_driven_on_fork(i) { + driver.wait_idle(id)?; + let grade = driver.grade(id, grader)?; + return Ok((grade, 0)); + } + let mut turn = first_turn.to_string(); let mut used = 0u32; loop { @@ -700,12 +737,13 @@ fn status_of(grade: &Scored) -> WorkerStatus { /// outcome. fn drive_one_inner( driver: &dyn WorkerDriver, + i: usize, id: &str, prompt: &str, retries: u32, reward: &Grader, ) -> Result { - let (grade, used) = drive_to_grade(driver, id, prompt, reward, retries)?; + let (grade, used) = drive_to_grade(driver, i, id, prompt, reward, retries)?; Ok(WorkerOutcome { session: id.to_string(), score: Some(grade.score), @@ -720,13 +758,14 @@ fn drive_one_inner( /// `Errored` outcome, like [`drive_one`]). fn drive_segments( driver: &dyn WorkerDriver, + i: usize, id: String, context: &str, segments: &[ResolvedSegment], reward: &Grader, retries: u32, ) -> WorkerOutcome { - drive_segments_inner(driver, &id, context, segments, reward, retries) + drive_segments_inner(driver, i, &id, context, segments, reward, retries) .unwrap_or_else(|e| errored(id, e)) } @@ -740,6 +779,7 @@ fn drive_segments( /// worker's `retries_used` is the sum across segments. fn drive_segments_inner( driver: &dyn WorkerDriver, + worker_i: usize, id: &str, context: &str, segments: &[ResolvedSegment], @@ -754,7 +794,7 @@ fn drive_segments_inner( } else { seg.prompt.clone() }; - let (grade, used) = drive_to_grade(driver, id, &turn, &seg.gate, retries)?; + let (grade, used) = drive_to_grade(driver, worker_i, id, &turn, &seg.gate, retries)?; seg_outcomes.push(SegmentOutcome { name: seg.name.clone(), passed: grade.passed, @@ -785,23 +825,33 @@ fn drive_segments_inner( struct CliDriver<'a> { exe: PathBuf, opts: &'a DispatchOpts, + default_agent: String, /// Durable dir the winner is pulled into (a TempDir would drop it). rundir: PathBuf, } impl<'a> CliDriver<'a> { - fn new(opts: &'a DispatchOpts) -> Result { + fn new(opts: &'a DispatchOpts, default_agent: String) -> Result { let exe = std::env::current_exe().context("locate the pillbox binary")?; let rundir = std::env::temp_dir().join(format!( "pillbox-dispatch-{}", uuid::Uuid::now_v7().simple() )); - Ok(Self { exe, opts, rundir }) + Ok(Self { + exe, + opts, + default_agent, + rundir, + }) + } + + fn effective_agent(&self, i: usize) -> String { + effective_worker_agent(self.opts, i, &self.default_agent) } } impl WorkerDriver for CliDriver<'_> { - fn fork(&self, i: usize) -> Result { + fn fork(&self, i: usize, first_turn: &str) -> Result { // Per-worker agent/model/temperature from the roster (`--workers-spec`) // falling back to the run-level scalars; without a roster this is the // scalar opts for every worker. The argv assembly below is otherwise @@ -831,8 +881,13 @@ impl WorkerDriver for CliDriver<'_> { // `expires_at` so `session prune` can later reap this worker. args.extend(["--ttl".into(), t.clone()]); } - // No positional prompt: a `--detach` worker comes up idle; the segment - // prompt is driven as turn 1 via `session send` (see `drive_one_inner`). + if self.first_turn_driven_on_fork(i) { + args.extend(["--".into(), first_turn.into()]); + } + // Server-mode workers get no positional prompt: a `--detach` server comes + // up idle; the segment prompt is driven as turn 1 via `session send` (see + // `drive_one_inner`). A server treats a baked positional as a pre-fill + // hint, not a turn. let out = self.capture(&args)?; let v: serde_json::Value = serde_json::from_str(&out) .with_context(|| format!("parse `run --json` output: {out:?}"))?; @@ -842,6 +897,10 @@ impl WorkerDriver for CliDriver<'_> { .context("`run --json` had no session.id") } + fn first_turn_driven_on_fork(&self, i: usize) -> bool { + agent_first_turn_driven_on_fork(&self.effective_agent(i)) + } + fn wait_idle(&self, id: &str) -> Result<()> { // Bounded so one stuck worker becomes an Errored worker (the caller maps // this Err → Errored), not a whole-dispatch hang. `wait-idle` exits 0 on @@ -1132,16 +1191,6 @@ pub(crate) fn dispatch(resolved: &Pillbox, opts: DispatchOpts) -> Result<()> { })?; } - // The run-level reward — the authoritative final grade, distinct from any - // per-segment gate. - let reward = Grader::from_opts(&opts)?; - // Parse + validate the segment spec up front (exit 2 on a bad spec) — before - // forking any worker. - let segments = match &opts.segments { - Some(p) => Some(load_segments(p)?), - None => None, - }; - // With a `--workers-spec` roster, its length is the authoritative `k`. An // explicit `-k N` that disagrees with the roster is a usage error (exit 2) // before any fork — but `-k` left at its default just derives `k` from the @@ -1165,7 +1214,28 @@ pub(crate) fn dispatch(resolved: &Pillbox, opts: DispatchOpts) -> Result<()> { None => opts.workers, }; - let driver = CliDriver::new(&opts)?; + let default_agent_id = default_agent(resolved); + if opts.segments.is_some() + && (0..k).any(|i| worker_first_turn_driven_on_fork(&opts, i as usize, &default_agent_id)) + { + return Err(PillboxError::usage( + "dispatch", + "`--segments` requires a server-mode agent (opencode); claude/codex are one-shot — use best-of-k (`-k`) instead.", + ) + .into()); + } + + // The run-level reward — the authoritative final grade, distinct from any + // per-segment gate. + let reward = Grader::from_opts(&opts)?; + // Parse + validate the segment spec up front (exit 2 on a bad spec) — before + // forking any worker. + let segments = match &opts.segments { + Some(p) => Some(load_segments(p)?), + None => None, + }; + + let driver = CliDriver::new(&opts, default_agent_id)?; let verdict = run_dispatch( &driver, k, @@ -1312,6 +1382,7 @@ mod tests { /// The fork-index passed to each `fork`, in call order — asserts the loop /// hands each worker its own 0-based roster index. fork_indices: RefCell>, + first_turn_on_fork: bool, } impl MockDriver { @@ -1328,19 +1399,27 @@ mod tests { sends: RefCell::new(0), pulls: RefCell::new(Vec::new()), fork_indices: RefCell::new(Vec::new()), + first_turn_on_fork: false, } } fn failing_grade(mut self, id: &str) -> Self { self.grade_errs.insert(id.to_string()); self } + fn first_turn_on_fork(mut self) -> Self { + self.first_turn_on_fork = true; + self + } } impl WorkerDriver for MockDriver { - fn fork(&self, i: usize) -> Result { + fn fork(&self, i: usize, _first_turn: &str) -> Result { self.fork_indices.borrow_mut().push(i); Ok(self.ids.borrow_mut().pop_front().expect("fork over budget")) } + fn first_turn_driven_on_fork(&self, _i: usize) -> bool { + self.first_turn_on_fork + } fn wait_idle(&self, _id: &str) -> Result<()> { Ok(()) } @@ -1417,6 +1496,18 @@ mod tests { assert!(d.pulls.borrow().is_empty()); } + #[test] + fn one_shot_drive_one_grades_fork_result_without_send_or_retry() { + let d = MockDriver::new(vec![("w0", vec![(false, 0.3)])]).first_turn_on_fork(); + let w = drive_one(&d, 0, "w0".into(), "task", 3, &reward()); + assert_eq!(w.session, "w0"); + assert_eq!(w.status, WorkerStatus::Failed); + assert_eq!(w.retries_used, 0); + assert_eq!(w.score, Some(0.3)); + assert_eq!(*d.sends.borrow(), 0); + assert!(d.pulls.borrow().is_empty()); + } + #[test] fn loop_forks_each_worker_with_its_index() { // The loop must hand each fork its own 0-based index (the roster row a