missions: sweeper + socket-proxy NETWORKS grant + mount ordering (C3 slice 4-5)
- mission_runtime::spawn_sweeper: force-removes runtime containers for missions terminal for >=30 min, clears runtime_endpoint. Wired into clawmates-server main(). - docker-compose socket-proxy: NETWORKS=1 so bollard.connect_network can attach containers to clawmates_edge for provider egress. - phase_runner ordering: ensure_checkout BEFORE ensure_container so the mission dir exists before docker mounts it. - provisioner: mkdir_p the mission dir defensively for research-only missions that skip checkout entirely.
This commit is contained in:
@@ -289,6 +289,10 @@ async fn run() -> Result<(), String> {
|
|||||||
// rows so the canvas renders a live status timeline.
|
// rows so the canvas renders a live status timeline.
|
||||||
cm_api::task_card_worker::spawn(pool.clone());
|
cm_api::task_card_worker::spawn(pool.clone());
|
||||||
cm_api::phase_runner::spawn(pool.clone());
|
cm_api::phase_runner::spawn(pool.clone());
|
||||||
|
// Per-mission runtime container sweeper (C3): tears down mission
|
||||||
|
// runtime containers 30 min after the mission reaches a terminal
|
||||||
|
// state so operators have a window to pull final artifacts.
|
||||||
|
cm_api::mission_runtime::spawn_sweeper(pool.clone(), std::time::Duration::from_secs(30 * 60));
|
||||||
// PDF renderer worker (Slice 6): watches mission_artifacts for
|
// PDF renderer worker (Slice 6): watches mission_artifacts for
|
||||||
// MD entries with render_pdf_status='pending', calls the
|
// MD entries with render_pdf_status='pending', calls the
|
||||||
// configured LLM (default Gemini 2.5 Flash) for styled HTML,
|
// configured LLM (default Gemini 2.5 Flash) for styled HTML,
|
||||||
|
|||||||
@@ -125,8 +125,11 @@ impl MissionRuntimeProvisioner {
|
|||||||
.await;
|
.await;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Create fresh.
|
// Create fresh. Ensure the bind source exists first — research-
|
||||||
|
// only missions (no repo checkout) still need the directory
|
||||||
|
// present or docker start fails with EACCES/ENOENT.
|
||||||
let mission_dir = format!("{MISSIONS_HOST_ROOT}/{mission_id}");
|
let mission_dir = format!("{MISSIONS_HOST_ROOT}/{mission_id}");
|
||||||
|
let _ = tokio::fs::create_dir_all(&mission_dir).await;
|
||||||
let mounts = vec![
|
let mounts = vec![
|
||||||
// Mount just this mission's directory. Agents can navigate
|
// Mount just this mission's directory. Agents can navigate
|
||||||
// its `/repo` subdir but never see other missions'.
|
// its `/repo` subdir but never see other missions'.
|
||||||
@@ -230,6 +233,75 @@ impl MissionRuntimeProvisioner {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Background sweeper: force-remove runtime containers for missions
|
||||||
|
/// that reached a terminal state ≥ `grace` ago. Keeps the container
|
||||||
|
/// around briefly after `completed`/`failed`/`cancelled` so the
|
||||||
|
/// operator can re-open the UI and pull the last checkpoint before
|
||||||
|
/// the daemon disappears. Runs on the same cadence as the phase
|
||||||
|
/// runner (10s) with a much longer per-mission grace.
|
||||||
|
pub fn spawn_sweeper(pool: sqlx::PgPool, grace: std::time::Duration) {
|
||||||
|
tokio::spawn(async move {
|
||||||
|
// Wait past the phase_runner boot so we don't fight over
|
||||||
|
// just-launched missions.
|
||||||
|
tokio::time::sleep(std::time::Duration::from_secs(30)).await;
|
||||||
|
let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
|
||||||
|
ticker.tick().await;
|
||||||
|
loop {
|
||||||
|
ticker.tick().await;
|
||||||
|
if let Err(e) = sweep_once(&pool, grace).await {
|
||||||
|
eprintln!("mission_runtime::sweeper: sweep failed: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn sweep_once(pool: &sqlx::PgPool, grace: std::time::Duration) -> Result<(), String> {
|
||||||
|
use sqlx::Row;
|
||||||
|
let grace_secs = grace.as_secs() as f64;
|
||||||
|
let rows = sqlx::query(
|
||||||
|
"SELECT id, workspace_id, runtime_container_name
|
||||||
|
FROM missions
|
||||||
|
WHERE status IN ('completed', 'failed', 'cancelled')
|
||||||
|
AND runtime_endpoint IS NOT NULL
|
||||||
|
AND completed_at IS NOT NULL
|
||||||
|
AND completed_at < now() - make_interval(secs => $1::float)
|
||||||
|
LIMIT 20",
|
||||||
|
)
|
||||||
|
.bind(grace_secs)
|
||||||
|
.fetch_all(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("query terminal missions: {e}"))?;
|
||||||
|
if rows.is_empty() {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let Some(prov) = MissionRuntimeProvisioner::from_env() else {
|
||||||
|
return Ok(());
|
||||||
|
};
|
||||||
|
for row in rows {
|
||||||
|
let id: Uuid = row.get("id");
|
||||||
|
let workspace_id: Uuid = row.get("workspace_id");
|
||||||
|
if let Err(e) = prov.teardown_container(id).await {
|
||||||
|
// A not-found is expected when the container was already
|
||||||
|
// reaped by a docker restart or a manual op; log at info
|
||||||
|
// level (via eprintln) and clear the binding anyway so the
|
||||||
|
// sweeper doesn't retry forever.
|
||||||
|
eprintln!("mission_runtime::sweeper: teardown mission {id}: {e}");
|
||||||
|
}
|
||||||
|
if let Err(e) = cm_db::repo::missions::set_runtime_binding(
|
||||||
|
pool,
|
||||||
|
id,
|
||||||
|
workspace_id,
|
||||||
|
None,
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
eprintln!("mission_runtime::sweeper: clear binding for {id}: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|||||||
@@ -142,11 +142,31 @@ async fn launch_phase(
|
|||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
// Provision the per-mission runtime container if it isn't already
|
// Ensure the mission's repo is checked out first — the runtime
|
||||||
// bound. Idempotent — on_launch sets this on initial launch, but
|
// container bind-mounts /var/lib/clawmates-missions/{id}, which
|
||||||
// pre-C3 missions or retries against a torn-down container land
|
// must exist before docker start or the mount fails.
|
||||||
// here. Non-fatal: if docker is unreachable the run falls back to
|
// Idempotent: fetch+reset on existing clones, clone on missing.
|
||||||
// the shared runtime.
|
// Non-fatal: research-only missions have no repo and skip cleanly.
|
||||||
|
match crate::mission_workspace::ensure_checkout(
|
||||||
|
pool,
|
||||||
|
cm_domain::WorkspaceId::from(workspace_id),
|
||||||
|
mission_id,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
Ok(Some(path)) => eprintln!(
|
||||||
|
"phase_runner: repo checked out at {} for mission {mission_id} phase {phase_id}",
|
||||||
|
path.display()
|
||||||
|
),
|
||||||
|
Ok(None) => {}
|
||||||
|
Err(e) => eprintln!(
|
||||||
|
"phase_runner: repo checkout for mission {mission_id} phase {phase_id} failed (continuing): {e}"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
// Provision the per-mission runtime container if not bound.
|
||||||
|
// Idempotent — on_launch sets this on initial launch, but pre-C3
|
||||||
|
// missions or retries against a torn-down container land here.
|
||||||
let mission = cm_db::repo::missions::get(pool, mission_id, workspace_id)
|
let mission = cm_db::repo::missions::get(pool, mission_id, workspace_id)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| format!("load mission for runtime binding: {e}"))?;
|
.map_err(|e| format!("load mission for runtime binding: {e}"))?;
|
||||||
@@ -176,28 +196,6 @@ async fn launch_phase(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Ensure the mission's repo is checked out before firing any run.
|
|
||||||
// Idempotent: fetch+reset on existing clones, clone on missing.
|
|
||||||
// Runs on EVERY phase launch — including retries — so a retry
|
|
||||||
// after a failed clone (permissions/auth fixed) will now succeed.
|
|
||||||
// Non-fatal: research-only missions have no repo and skip cleanly.
|
|
||||||
match crate::mission_workspace::ensure_checkout(
|
|
||||||
pool,
|
|
||||||
cm_domain::WorkspaceId::from(workspace_id),
|
|
||||||
mission_id,
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
Ok(Some(path)) => eprintln!(
|
|
||||||
"phase_runner: repo checked out at {} for mission {mission_id} phase {phase_id}",
|
|
||||||
path.display()
|
|
||||||
),
|
|
||||||
Ok(None) => {}
|
|
||||||
Err(e) => eprintln!(
|
|
||||||
"phase_runner: repo checkout for mission {mission_id} phase {phase_id} failed (continuing): {e}"
|
|
||||||
),
|
|
||||||
}
|
|
||||||
|
|
||||||
let task = phase_task_text(kind, title, description);
|
let task = phase_task_text(kind, title, description);
|
||||||
|
|
||||||
// Purge prior failed / cancelled runs for this phase so the card
|
// Purge prior failed / cancelled runs for this phase so the card
|
||||||
|
|||||||
@@ -104,6 +104,11 @@ services:
|
|||||||
EXEC: 1
|
EXEC: 1
|
||||||
DELETE: 1
|
DELETE: 1
|
||||||
VERSION: 1
|
VERSION: 1
|
||||||
|
# NETWORKS grant (C3): mission_runtime::ensure_container needs
|
||||||
|
# to attach per-mission runtime containers to the clawmates_edge
|
||||||
|
# network for provider egress, in addition to creating them on
|
||||||
|
# clawmates_core.
|
||||||
|
NETWORKS: 1
|
||||||
volumes:
|
volumes:
|
||||||
- /var/run/docker.sock:/var/run/docker.sock:ro
|
- /var/run/docker.sock:/var/run/docker.sock:ro
|
||||||
networks: [engine_net]
|
networks: [engine_net]
|
||||||
|
|||||||
Reference in New Issue
Block a user