feat(missions): capture runs automatically, and once more before teardown
Wires diff capture into the two sweeps that matter. `phase_runner::sweep_once` gains `capture_finished_coding_phases`, guarded by `NOT EXISTS (code_diff for this phase)`. Deliberately a separate step rather than a hook on `close_finished_phases` or `evaluate_finished_phases`: a phase reaches `completed` through one or the other depending on whether it declared a `done_when`, so hanging capture off either would silently skip half the missions. The guard also makes it retryable — a capture that errors is simply re-selected next tick. `mission_runtime::sweep_once` captures anything still outstanding immediately before `teardown_container`, which deletes the checkout. This covers what the phase sweep structurally cannot: a mission that ended `failed` mid-coding still has real work on disk, and reaping it unexamined destroys the only evidence of what the agents actually did. Applies to coding, benchmark and security_scan phases — all three operate on a repo. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 5
parent
716ee9a304
commit
322c1be89c
@@ -699,6 +699,14 @@ async fn sweep_once(pool: &sqlx::PgPool, grace: std::time::Duration) -> Result<(
|
|||||||
for row in rows {
|
for row in rows {
|
||||||
let id: Uuid = row.get("id");
|
let id: Uuid = row.get("id");
|
||||||
let workspace_id: Uuid = row.get("workspace_id");
|
let workspace_id: Uuid = row.get("workspace_id");
|
||||||
|
// Last chance. `teardown_container` deletes the checkout, so anything
|
||||||
|
// not captured by now is gone for good. The phase sweep should have
|
||||||
|
// handled this minutes ago; this covers the cases it cannot — a phase
|
||||||
|
// that ended `failed` rather than `completed`, or a capture that kept
|
||||||
|
// erroring until the grace window ran out.
|
||||||
|
if let Err(e) = capture_outstanding_phases(pool, id).await {
|
||||||
|
eprintln!("mission_runtime::sweeper: last-chance capture for {id}: {e}");
|
||||||
|
}
|
||||||
if let Err(e) = prov.teardown_container(id).await {
|
if let Err(e) = prov.teardown_container(id).await {
|
||||||
// A not-found is expected when the container was already
|
// A not-found is expected when the container was already
|
||||||
// reaped by a docker restart or a manual op; log at info
|
// reaped by a docker restart or a manual op; log at info
|
||||||
@@ -716,6 +724,46 @@ async fn sweep_once(pool: &sqlx::PgPool, grace: std::time::Duration) -> Result<(
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Capture any phase of `mission_id` that has a repo and no `code_diff` yet,
|
||||||
|
/// regardless of how the phase ended.
|
||||||
|
///
|
||||||
|
/// The phase sweep only captures `completed` phases. A mission that failed
|
||||||
|
/// mid-coding still has real work in its checkout, and deleting it
|
||||||
|
/// unexamined is how a debugging session loses the only evidence of what the
|
||||||
|
/// agents actually did.
|
||||||
|
async fn capture_outstanding_phases(pool: &sqlx::PgPool, mission_id: Uuid) -> Result<(), String> {
|
||||||
|
use sqlx::Row;
|
||||||
|
let rows = sqlx::query(
|
||||||
|
"SELECT mp.id
|
||||||
|
FROM mission_phases mp
|
||||||
|
JOIN missions m ON m.id = mp.mission_id
|
||||||
|
WHERE mp.mission_id = $1
|
||||||
|
AND m.repo_id IS NOT NULL
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM mission_artifacts a
|
||||||
|
WHERE a.mission_id = mp.mission_id
|
||||||
|
AND a.phase_id = mp.id
|
||||||
|
AND a.kind = 'code_diff'
|
||||||
|
)",
|
||||||
|
)
|
||||||
|
.bind(mission_id)
|
||||||
|
.fetch_all(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("select uncaptured phases: {e}"))?;
|
||||||
|
|
||||||
|
for row in rows {
|
||||||
|
let phase_id: Uuid = row.get("id");
|
||||||
|
if let Err(e) =
|
||||||
|
crate::mission_delivery::capture_phase_diff(pool, mission_id, phase_id).await
|
||||||
|
{
|
||||||
|
eprintln!(
|
||||||
|
"mission_runtime::sweeper: capture mission {mission_id} phase {phase_id}: {e}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|||||||
@@ -55,10 +55,67 @@ async fn sweep_once(pool: &PgPool, runtime: &cm_runtime::Runtime) -> Result<(),
|
|||||||
// Between "all runs finished" and "phase done" sits the completion
|
// Between "all runs finished" and "phase done" sits the completion
|
||||||
// evaluation, for phases that declare a condition.
|
// evaluation, for phases that declare a condition.
|
||||||
evaluate_finished_phases(pool, runtime).await?;
|
evaluate_finished_phases(pool, runtime).await?;
|
||||||
|
// Capture before the mission closes and long before the sweeper reaps the
|
||||||
|
// checkout. Idempotent, so a failure here is retried on the next tick
|
||||||
|
// rather than losing the phase's work.
|
||||||
|
capture_finished_coding_phases(pool).await?;
|
||||||
close_finished_missions(pool).await?;
|
close_finished_missions(pool).await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// How many phases to capture per tick. Capture shells out to git against a
|
||||||
|
/// working tree, so a backlog should be worked through steadily rather than
|
||||||
|
/// all at once.
|
||||||
|
const CAPTURE_BATCH: i64 = 5;
|
||||||
|
|
||||||
|
/// Write out the diff for coding phases that have finished and not yet been
|
||||||
|
/// captured.
|
||||||
|
///
|
||||||
|
/// Deliberately not hung off `close_finished_phases` or
|
||||||
|
/// `evaluate_finished_phases`: a phase reaches `completed` through either
|
||||||
|
/// path depending on whether it declared a `done_when`, and bolting capture
|
||||||
|
/// onto one of them would silently skip the other. Driving it from the sweep
|
||||||
|
/// with a `NOT EXISTS` guard covers both and is retryable by construction.
|
||||||
|
async fn capture_finished_coding_phases(pool: &PgPool) -> Result<(), String> {
|
||||||
|
let rows = sqlx::query(
|
||||||
|
"SELECT mp.id, mp.mission_id
|
||||||
|
FROM mission_phases mp
|
||||||
|
JOIN missions m ON m.id = mp.mission_id
|
||||||
|
WHERE mp.status = 'completed'
|
||||||
|
AND mp.kind IN ('coding', 'benchmark', 'security_scan')
|
||||||
|
AND m.repo_id IS NOT NULL
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM mission_artifacts a
|
||||||
|
WHERE a.mission_id = mp.mission_id
|
||||||
|
AND a.phase_id = mp.id
|
||||||
|
AND a.kind = 'code_diff'
|
||||||
|
)
|
||||||
|
ORDER BY mp.completed_at NULLS LAST
|
||||||
|
LIMIT $1",
|
||||||
|
)
|
||||||
|
.bind(CAPTURE_BATCH)
|
||||||
|
.fetch_all(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("select phases to capture: {e}"))?;
|
||||||
|
|
||||||
|
for row in rows {
|
||||||
|
use sqlx::Row;
|
||||||
|
let phase_id: Uuid = row.get("id");
|
||||||
|
let mission_id: Uuid = row.get("mission_id");
|
||||||
|
if let Err(e) =
|
||||||
|
crate::mission_delivery::capture_phase_diff(pool, mission_id, phase_id).await
|
||||||
|
{
|
||||||
|
// Left uncaptured on purpose: the guard above re-selects it next
|
||||||
|
// tick. Only a permanently broken checkout keeps failing, and that
|
||||||
|
// is worth the recurring log line.
|
||||||
|
eprintln!(
|
||||||
|
"phase_runner: capturing diff for mission {mission_id} phase {phase_id}: {e}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// Enqueue topology_runs for every phase whose predecessors are done.
|
/// Enqueue topology_runs for every phase whose predecessors are done.
|
||||||
async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
|
async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
|
||||||
// Eligible = pending phase, mission running, all lower-order phases
|
// Eligible = pending phase, mission running, all lower-order phases
|
||||||
|
|||||||
Reference in New Issue
Block a user