teams: ephemeral lifecycle for Scheduled + Triggered planner modes
Migration 0033: adds teams.lifecycle ('permanent' | 'ephemeral') and a
topology_runs.team_id back-ref with a partial index for the sibling-in-
flight check.
cm-db repo:
- teams::insert_team_with_lifecycle (insert_team keeps the permanent default)
- topology_runs::enqueue_run_for_team (populates team_id)
- topology_runs::check_ephemeral_teardown — atomic SELECT that only
returns Some when the team is ephemeral AND no siblings are still
queued/running; carries the workspace + bound claw ids for cleanup.
cm-api:
- topology_worker post-terminal hook maybe_teardown_ephemeral_team
runs deprovision_claw on each bound claw (best-effort; failures log
but don't block Postgres deletion), then hard_purge each agent row,
then delete_team.
- routes::teams::build_team_with_lifecycle (build_team keeps default);
run_team enqueues with team_id.
- planner ScaffoldRequest gains mode; lifecycle_for(mode) sets the team
to ephemeral for scheduled + triggered, permanent otherwise.
Frontend MasterPlannerModal passes mode in the scaffold payload so the
backend can derive lifecycle without duplicating the mode taxonomy.
Tests: 3 new (returns claws when no siblings, holds when siblings queued,
ignores permanent teams). 10/10 topology_jobs green; workspace clippy
--tests clean.
This commit is contained in:
+18
@@ -0,0 +1,18 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "INSERT INTO topology_runs\n (id, workspace_id, task, kind, status, graph, tier, team_id)\n VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5)",
|
||||||
|
"describe": {
|
||||||
|
"columns": [],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Uuid",
|
||||||
|
"Uuid",
|
||||||
|
"Text",
|
||||||
|
"Jsonb",
|
||||||
|
"Uuid"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": []
|
||||||
|
},
|
||||||
|
"hash": "3aec092bcc9501d5d525f3623758a83154db58776ad473bdb768649cc27a8edc"
|
||||||
|
}
|
||||||
+28
@@ -0,0 +1,28 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "SELECT t.id AS team_id, t.workspace_id\n FROM topology_runs r\n JOIN teams t ON t.id = r.team_id\n WHERE r.id = $1\n AND t.lifecycle = 'ephemeral'\n AND NOT EXISTS (\n SELECT 1 FROM topology_runs sib\n WHERE sib.team_id = t.id\n AND sib.id <> r.id\n AND sib.status IN ('queued', 'running')\n )",
|
||||||
|
"describe": {
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"ordinal": 0,
|
||||||
|
"name": "team_id",
|
||||||
|
"type_info": "Uuid"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"ordinal": 1,
|
||||||
|
"name": "workspace_id",
|
||||||
|
"type_info": "Uuid"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Uuid"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": [
|
||||||
|
false,
|
||||||
|
false
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"hash": "3d25bfd48bf920da8db9ced892ff99bac77372aeed3e69b32cf52d0100ba2771"
|
||||||
|
}
|
||||||
+22
@@ -0,0 +1,22 @@
|
|||||||
|
{
|
||||||
|
"db_name": "PostgreSQL",
|
||||||
|
"query": "SELECT claw_id FROM team_members WHERE team_id = $1",
|
||||||
|
"describe": {
|
||||||
|
"columns": [
|
||||||
|
{
|
||||||
|
"ordinal": 0,
|
||||||
|
"name": "claw_id",
|
||||||
|
"type_info": "Uuid"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"parameters": {
|
||||||
|
"Left": [
|
||||||
|
"Uuid"
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"nullable": [
|
||||||
|
false
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"hash": "84eeb89124aff5282a8185d527737954d7bb7bb194ff22c4c3fa14be33ec094e"
|
||||||
|
}
|
||||||
+4
-3
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"db_name": "PostgreSQL",
|
"db_name": "PostgreSQL",
|
||||||
"query": "INSERT INTO teams (id, workspace_id, name, kind, graph)\n VALUES ($1, $2, $3, $4, $5)",
|
"query": "INSERT INTO teams (id, workspace_id, name, kind, graph, lifecycle)\n VALUES ($1, $2, $3, $4, $5, $6)",
|
||||||
"describe": {
|
"describe": {
|
||||||
"columns": [],
|
"columns": [],
|
||||||
"parameters": {
|
"parameters": {
|
||||||
@@ -9,10 +9,11 @@
|
|||||||
"Uuid",
|
"Uuid",
|
||||||
"Text",
|
"Text",
|
||||||
"Text",
|
"Text",
|
||||||
"Jsonb"
|
"Jsonb",
|
||||||
|
"Text"
|
||||||
]
|
]
|
||||||
},
|
},
|
||||||
"nullable": []
|
"nullable": []
|
||||||
},
|
},
|
||||||
"hash": "c121bdbab5de2f82f58c6762bb90ebf08a45cb6e2766e4c69dddd765e168579a"
|
"hash": "f5930d9189b8aedc86109c8fea96e33dc502d8d5584518f8b83c24792d98508f"
|
||||||
}
|
}
|
||||||
@@ -11,7 +11,7 @@ use serde_json::{json, Value};
|
|||||||
use std::convert::Infallible;
|
use std::convert::Infallible;
|
||||||
|
|
||||||
use crate::routes::claws::{apply_reference_to_claw, enhance_and_publish, extract_json};
|
use crate::routes::claws::{apply_reference_to_claw, enhance_and_publish, extract_json};
|
||||||
use crate::routes::teams::{build_team, TeamMemberInput};
|
use crate::routes::teams::TeamMemberInput;
|
||||||
use crate::{AppState, Authed};
|
use crate::{AppState, Authed};
|
||||||
|
|
||||||
fn sse(v: Value) -> Result<Event, Infallible> {
|
fn sse(v: Value) -> Result<Event, Infallible> {
|
||||||
@@ -173,11 +173,23 @@ pub struct ScaffoldRequest {
|
|||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub schedule: Option<ScaffoldSchedule>,
|
pub schedule: Option<ScaffoldSchedule>,
|
||||||
pub members: Vec<ScaffoldMember>,
|
pub members: Vec<ScaffoldMember>,
|
||||||
|
/// Original planner mode — decides the team's lifecycle. `scheduled` and
|
||||||
|
/// `triggered` produce `ephemeral` teams (torn down by the topology_worker
|
||||||
|
/// after the last run terminates). Anything else is `permanent`.
|
||||||
|
#[serde(default)]
|
||||||
|
pub mode: String,
|
||||||
}
|
}
|
||||||
fn default_kind() -> String {
|
fn default_kind() -> String {
|
||||||
"hub_spoke".to_string()
|
"hub_spoke".to_string()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn lifecycle_for(mode: &str) -> &'static str {
|
||||||
|
match mode {
|
||||||
|
"scheduled" | "triggered" => "ephemeral",
|
||||||
|
_ => "permanent",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// `POST /api/planner/scaffold` — build the approved team (SSE progress): create
|
/// `POST /api/planner/scaffold` — build the approved team (SSE progress): create
|
||||||
/// agents + topology, refine+attach a brain per agent, set up the nightly loop.
|
/// agents + topology, refine+attach a brain per agent, set up the nightly loop.
|
||||||
pub async fn planner_scaffold(
|
pub async fn planner_scaffold(
|
||||||
@@ -200,7 +212,8 @@ pub async fn planner_scaffold(
|
|||||||
system_prompt: m.system_prompt.clone(),
|
system_prompt: m.system_prompt.clone(),
|
||||||
accent: String::new(),
|
accent: String::new(),
|
||||||
}).collect();
|
}).collect();
|
||||||
let (team_id, claw_ids) = match build_team(&state, user.workspace_id, user.user_id, &body.team_name, &body.topology_kind, &members).await {
|
let lifecycle = lifecycle_for(&body.mode);
|
||||||
|
let (team_id, claw_ids) = match crate::routes::teams::build_team_with_lifecycle(&state, user.workspace_id, user.user_id, &body.team_name, &body.topology_kind, &members, lifecycle).await {
|
||||||
Ok(r) => r,
|
Ok(r) => r,
|
||||||
Err(_) => { yield sse(json!({"stage":"error","pct":100,"label":"Team creation failed"})); return; }
|
Err(_) => { yield sse(json!({"stage":"error","pct":100,"label":"Team creation failed"})); return; }
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -56,6 +56,30 @@ pub(crate) async fn build_team(
|
|||||||
name: &str,
|
name: &str,
|
||||||
kind_str: &str,
|
kind_str: &str,
|
||||||
members: &[TeamMemberInput],
|
members: &[TeamMemberInput],
|
||||||
|
) -> Result<(Uuid, Vec<Uuid>), ApiError> {
|
||||||
|
build_team_with_lifecycle(
|
||||||
|
state,
|
||||||
|
workspace_id,
|
||||||
|
user_id,
|
||||||
|
name,
|
||||||
|
kind_str,
|
||||||
|
members,
|
||||||
|
"permanent",
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Same as `build_team` but with an explicit `lifecycle` (`permanent` |
|
||||||
|
/// `ephemeral`). Ephemeral teams are torn down by the topology_worker after
|
||||||
|
/// their last run terminates — used by the Scheduled + Triggered planner modes.
|
||||||
|
pub(crate) async fn build_team_with_lifecycle(
|
||||||
|
state: &AppState,
|
||||||
|
workspace_id: cm_domain::WorkspaceId,
|
||||||
|
user_id: cm_domain::UserId,
|
||||||
|
name: &str,
|
||||||
|
kind_str: &str,
|
||||||
|
members: &[TeamMemberInput],
|
||||||
|
lifecycle: &str,
|
||||||
) -> Result<(Uuid, Vec<Uuid>), ApiError> {
|
) -> Result<(Uuid, Vec<Uuid>), ApiError> {
|
||||||
if members.is_empty() {
|
if members.is_empty() {
|
||||||
return Err(ApiError::BadRequest);
|
return Err(ApiError::BadRequest);
|
||||||
@@ -101,13 +125,14 @@ pub(crate) async fn build_team(
|
|||||||
|
|
||||||
let team_id = Uuid::now_v7();
|
let team_id = Uuid::now_v7();
|
||||||
let graph_json = serde_json::to_value(&graph).map_err(|_| ApiError::Internal)?;
|
let graph_json = serde_json::to_value(&graph).map_err(|_| ApiError::Internal)?;
|
||||||
cm_db::repo::teams::insert_team(
|
cm_db::repo::teams::insert_team_with_lifecycle(
|
||||||
&state.pool,
|
&state.pool,
|
||||||
team_id,
|
team_id,
|
||||||
workspace_id,
|
workspace_id,
|
||||||
name,
|
name,
|
||||||
kind.as_str(),
|
kind.as_str(),
|
||||||
&graph_json,
|
&graph_json,
|
||||||
|
lifecycle,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
for (i, node) in graph.nodes.iter().enumerate() {
|
for (i, node) in graph.nodes.iter().enumerate() {
|
||||||
@@ -373,12 +398,13 @@ pub async fn run_team(
|
|||||||
let team = cm_db::repo::teams::get_team(&state.pool, id, user.workspace_id).await?;
|
let team = cm_db::repo::teams::get_team(&state.pool, id, user.workspace_id).await?;
|
||||||
crate::quota::enforce_new_run(&state, user.workspace_id).await?;
|
crate::quota::enforce_new_run(&state, user.workspace_id).await?;
|
||||||
let run_id = Uuid::now_v7();
|
let run_id = Uuid::now_v7();
|
||||||
cm_db::repo::topology_runs::enqueue_run(
|
cm_db::repo::topology_runs::enqueue_run_for_team(
|
||||||
&state.pool,
|
&state.pool,
|
||||||
run_id,
|
run_id,
|
||||||
user.workspace_id,
|
user.workspace_id,
|
||||||
&body.task,
|
&body.task,
|
||||||
&team.graph,
|
&team.graph,
|
||||||
|
id,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
Ok((
|
Ok((
|
||||||
|
|||||||
@@ -154,6 +154,51 @@ async fn maybe_transition_research_topic(pool: &PgPool, id: Uuid) {
|
|||||||
Ok(false) => {}
|
Ok(false) => {}
|
||||||
Err(e) => eprintln!("topology_worker: notify_run_completed({id}) failed: {e}"),
|
Err(e) => eprintln!("topology_worker: notify_run_completed({id}) failed: {e}"),
|
||||||
}
|
}
|
||||||
|
maybe_teardown_ephemeral_team(pool, id).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Post-terminal hook: if this run's team is `ephemeral` and no siblings are
|
||||||
|
/// still in flight, deprovision every bound claw on the ZeroClaw daemon,
|
||||||
|
/// delete the claw rows, and delete the team row. Best-effort — a failure to
|
||||||
|
/// tear down leaves the team intact and logs; a future sweep can retry.
|
||||||
|
async fn maybe_teardown_ephemeral_team(pool: &PgPool, id: Uuid) {
|
||||||
|
let teardown = match cm_db::repo::topology_runs::check_ephemeral_teardown(pool, id).await {
|
||||||
|
Ok(Some(t)) => t,
|
||||||
|
Ok(None) => return,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("topology_worker: check_ephemeral_teardown({id}) failed: {e}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
// Deprovision each claw on the daemon before deleting rows — if the daemon
|
||||||
|
// side fails we still delete our rows (the daemon can be swept for orphans
|
||||||
|
// by the fleet-reconcile timer). This is the trade cm-api owns everywhere:
|
||||||
|
// Postgres is authoritative, the daemon config is a cache.
|
||||||
|
if let Some(prov) = crate::runtime_provision::RuntimeProvisioner::from_env() {
|
||||||
|
for cid in &teardown.claw_ids {
|
||||||
|
if let Err(e) = prov.deprovision_claw(*cid).await {
|
||||||
|
eprintln!("topology_worker: deprovision_claw({cid}) failed: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for cid in &teardown.claw_ids {
|
||||||
|
if let Err(e) = cm_db::repo::agents::hard_purge(pool, cm_domain::AgentId::from(*cid)).await
|
||||||
|
{
|
||||||
|
eprintln!("topology_worker: agents::hard_purge({cid}) failed: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if let Err(e) = cm_db::repo::teams::delete_team(
|
||||||
|
pool,
|
||||||
|
teardown.team_id,
|
||||||
|
cm_domain::WorkspaceId::from(teardown.workspace_id),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
eprintln!(
|
||||||
|
"topology_worker: teams::delete_team({}) failed: {e}",
|
||||||
|
teardown.team_id
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Drive a graph to completion with the durable per-step checkpoint +
|
/// Drive a graph to completion with the durable per-step checkpoint +
|
||||||
|
|||||||
@@ -2,8 +2,10 @@
|
|||||||
//! (CAS) → checkpoint → complete, plus the stale-run resume sweep. This is the
|
//! (CAS) → checkpoint → complete, plus the stale-run resume sweep. This is the
|
||||||
//! foundation that lets long-horizon topology runs survive worker restarts.
|
//! foundation that lets long-horizon topology runs survive worker restarts.
|
||||||
|
|
||||||
use cm_db::repo::{loops, research_topics, topology_runs, users, workspaces};
|
use cm_db::repo::{loops, research_topics, teams, topology_runs, users, workspaces};
|
||||||
use cm_domain::{Role, User, UserId, Workspace, WorkspaceId};
|
use cm_domain::{
|
||||||
|
AccessPolicy, Agent, AgentId, AgentStatus, Role, User, UserId, Workspace, WorkspaceId,
|
||||||
|
};
|
||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
@@ -373,3 +375,123 @@ async fn notify_run_completed_ignores_runs_with_no_research_topic() {
|
|||||||
"no research_topic_id → nothing to transition"
|
"no research_topic_id → nothing to transition"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn seed_team(
|
||||||
|
pool: &sqlx::PgPool,
|
||||||
|
ws: WorkspaceId,
|
||||||
|
user_id: UserId,
|
||||||
|
lifecycle: &str,
|
||||||
|
claw_count: usize,
|
||||||
|
) -> (Uuid, Vec<Uuid>) {
|
||||||
|
let team_id = Uuid::now_v7();
|
||||||
|
let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
|
||||||
|
teams::insert_team_with_lifecycle(pool, team_id, ws, "T", "hub_spoke", &graph, lifecycle)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let mut claws = Vec::with_capacity(claw_count);
|
||||||
|
for i in 0..claw_count {
|
||||||
|
let agent = Agent {
|
||||||
|
id: AgentId::new(),
|
||||||
|
workspace_id: ws,
|
||||||
|
name: format!("claw{i}"),
|
||||||
|
job_title: "worker".into(),
|
||||||
|
system_prompt: String::new(),
|
||||||
|
avatar: String::new(),
|
||||||
|
accent: String::new(),
|
||||||
|
wallpaper: String::new(),
|
||||||
|
managed_by: user_id,
|
||||||
|
status: AgentStatus::Online,
|
||||||
|
};
|
||||||
|
cm_db::repo::agents::insert(pool, &agent, &AccessPolicy::default())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
teams::add_member(
|
||||||
|
pool,
|
||||||
|
team_id,
|
||||||
|
&format!("n{i}"),
|
||||||
|
agent.id.as_uuid(),
|
||||||
|
"worker",
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
claws.push(agent.id.as_uuid());
|
||||||
|
}
|
||||||
|
(team_id, claws)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn check_ephemeral_teardown_returns_claws_when_no_siblings_left() {
|
||||||
|
let pool = cm_testkit::test_pool().await;
|
||||||
|
let ws = seed_workspace(&pool).await;
|
||||||
|
let user_id = seed_user(&pool, ws, "[email protected]").await;
|
||||||
|
let (team_id, claws) = seed_team(&pool, ws, user_id, "ephemeral", 3).await;
|
||||||
|
|
||||||
|
let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
|
||||||
|
let run_id = Uuid::now_v7();
|
||||||
|
topology_runs::enqueue_run_for_team(&pool, run_id, ws, "task", &graph, team_id)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
topology_runs::complete(&pool, run_id, &json!({}))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let teardown = topology_runs::check_ephemeral_teardown(&pool, run_id)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.expect("ephemeral team, no siblings — should return teardown");
|
||||||
|
assert_eq!(teardown.team_id, team_id);
|
||||||
|
assert_eq!(teardown.workspace_id, ws.as_uuid());
|
||||||
|
let mut got = teardown.claw_ids.clone();
|
||||||
|
let mut want = claws.clone();
|
||||||
|
got.sort();
|
||||||
|
want.sort();
|
||||||
|
assert_eq!(got, want, "returns every bound claw");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn check_ephemeral_teardown_holds_when_siblings_in_flight() {
|
||||||
|
let pool = cm_testkit::test_pool().await;
|
||||||
|
let ws = seed_workspace(&pool).await;
|
||||||
|
let user_id = seed_user(&pool, ws, "[email protected]").await;
|
||||||
|
let (team_id, _) = seed_team(&pool, ws, user_id, "ephemeral", 2).await;
|
||||||
|
|
||||||
|
let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
|
||||||
|
let done = Uuid::now_v7();
|
||||||
|
let still_queued = Uuid::now_v7();
|
||||||
|
topology_runs::enqueue_run_for_team(&pool, done, ws, "first", &graph, team_id)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
topology_runs::enqueue_run_for_team(&pool, still_queued, ws, "second", &graph, team_id)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
topology_runs::complete(&pool, done, &json!({}))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let teardown = topology_runs::check_ephemeral_teardown(&pool, done)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(teardown.is_none(), "sibling still queued → hold teardown");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn check_ephemeral_teardown_ignores_permanent_teams() {
|
||||||
|
let pool = cm_testkit::test_pool().await;
|
||||||
|
let ws = seed_workspace(&pool).await;
|
||||||
|
let user_id = seed_user(&pool, ws, "[email protected]").await;
|
||||||
|
let (team_id, _) = seed_team(&pool, ws, user_id, "permanent", 1).await;
|
||||||
|
|
||||||
|
let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
|
||||||
|
let run_id = Uuid::now_v7();
|
||||||
|
topology_runs::enqueue_run_for_team(&pool, run_id, ws, "task", &graph, team_id)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
topology_runs::complete(&pool, run_id, &json!({}))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let teardown = topology_runs::check_ephemeral_teardown(&pool, run_id)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(teardown.is_none(), "permanent teams are never torn down");
|
||||||
|
}
|
||||||
|
|||||||
@@ -37,6 +37,8 @@ pub struct TeamMember {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Insert a team (the topology graph). Members are added separately.
|
/// Insert a team (the topology graph). Members are added separately.
|
||||||
|
/// Lifecycle defaults to `permanent`; use `insert_ephemeral_team` for the
|
||||||
|
/// scheduled / triggered flows.
|
||||||
pub async fn insert_team(
|
pub async fn insert_team(
|
||||||
pool: &PgPool,
|
pool: &PgPool,
|
||||||
id: Uuid,
|
id: Uuid,
|
||||||
@@ -44,15 +46,31 @@ pub async fn insert_team(
|
|||||||
name: &str,
|
name: &str,
|
||||||
kind: &str,
|
kind: &str,
|
||||||
graph: &Value,
|
graph: &Value,
|
||||||
|
) -> Result<(), DbError> {
|
||||||
|
insert_team_with_lifecycle(pool, id, workspace_id, name, kind, graph, "permanent").await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Insert a team with an explicit `lifecycle` (`permanent` | `ephemeral`).
|
||||||
|
/// Ephemeral teams are torn down after the last in-flight run terminates —
|
||||||
|
/// see cm-api::topology_worker::maybe_teardown_ephemeral_team.
|
||||||
|
pub async fn insert_team_with_lifecycle(
|
||||||
|
pool: &PgPool,
|
||||||
|
id: Uuid,
|
||||||
|
workspace_id: WorkspaceId,
|
||||||
|
name: &str,
|
||||||
|
kind: &str,
|
||||||
|
graph: &Value,
|
||||||
|
lifecycle: &str,
|
||||||
) -> Result<(), DbError> {
|
) -> Result<(), DbError> {
|
||||||
sqlx::query!(
|
sqlx::query!(
|
||||||
"INSERT INTO teams (id, workspace_id, name, kind, graph)
|
"INSERT INTO teams (id, workspace_id, name, kind, graph, lifecycle)
|
||||||
VALUES ($1, $2, $3, $4, $5)",
|
VALUES ($1, $2, $3, $4, $5, $6)",
|
||||||
id,
|
id,
|
||||||
workspace_id.as_uuid(),
|
workspace_id.as_uuid(),
|
||||||
name,
|
name,
|
||||||
kind,
|
kind,
|
||||||
graph,
|
graph,
|
||||||
|
lifecycle,
|
||||||
)
|
)
|
||||||
.execute(pool)
|
.execute(pool)
|
||||||
.await?;
|
.await?;
|
||||||
|
|||||||
@@ -122,6 +122,78 @@ pub async fn enqueue_run_tier(
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Enqueue a durable run bound to a specific team. `team_id` is stored so the
|
||||||
|
/// post-terminal ephemeral-teardown hook can find the team from a completed run.
|
||||||
|
pub async fn enqueue_run_for_team(
|
||||||
|
pool: &PgPool,
|
||||||
|
id: Uuid,
|
||||||
|
workspace_id: WorkspaceId,
|
||||||
|
task: &str,
|
||||||
|
graph: &Value,
|
||||||
|
team_id: Uuid,
|
||||||
|
) -> Result<(), DbError> {
|
||||||
|
sqlx::query!(
|
||||||
|
"INSERT INTO topology_runs
|
||||||
|
(id, workspace_id, task, kind, status, graph, tier, team_id)
|
||||||
|
VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5)",
|
||||||
|
id,
|
||||||
|
workspace_id.as_uuid(),
|
||||||
|
task,
|
||||||
|
graph,
|
||||||
|
team_id,
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Result of `check_ephemeral_teardown` when this run's terminal completion
|
||||||
|
/// should tear down its team.
|
||||||
|
pub struct EphemeralTeardown {
|
||||||
|
pub team_id: Uuid,
|
||||||
|
pub workspace_id: Uuid,
|
||||||
|
pub claw_ids: Vec<Uuid>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// If this run belongs to an ephemeral team AND no siblings are still queued /
|
||||||
|
/// running, return the team's teardown context (team id + workspace + bound
|
||||||
|
/// claws). Callers then deprovision each claw and delete the team. Returns
|
||||||
|
/// `None` if there's nothing to do (permanent team, still-running siblings, or
|
||||||
|
/// no team back-ref at all).
|
||||||
|
pub async fn check_ephemeral_teardown(
|
||||||
|
pool: &PgPool,
|
||||||
|
run_id: Uuid,
|
||||||
|
) -> Result<Option<EphemeralTeardown>, DbError> {
|
||||||
|
let row = sqlx::query!(
|
||||||
|
"SELECT t.id AS team_id, t.workspace_id
|
||||||
|
FROM topology_runs r
|
||||||
|
JOIN teams t ON t.id = r.team_id
|
||||||
|
WHERE r.id = $1
|
||||||
|
AND t.lifecycle = 'ephemeral'
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM topology_runs sib
|
||||||
|
WHERE sib.team_id = t.id
|
||||||
|
AND sib.id <> r.id
|
||||||
|
AND sib.status IN ('queued', 'running')
|
||||||
|
)",
|
||||||
|
run_id,
|
||||||
|
)
|
||||||
|
.fetch_optional(pool)
|
||||||
|
.await?;
|
||||||
|
let Some(row) = row else { return Ok(None) };
|
||||||
|
let claws = sqlx::query!(
|
||||||
|
"SELECT claw_id FROM team_members WHERE team_id = $1",
|
||||||
|
row.team_id,
|
||||||
|
)
|
||||||
|
.fetch_all(pool)
|
||||||
|
.await?;
|
||||||
|
Ok(Some(EphemeralTeardown {
|
||||||
|
team_id: row.team_id,
|
||||||
|
workspace_id: row.workspace_id,
|
||||||
|
claw_ids: claws.into_iter().map(|r| r.claw_id).collect(),
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
/// Atomically claim the oldest queued job, flipping it to `running`. Uses
|
/// Atomically claim the oldest queued job, flipping it to `running`. Uses
|
||||||
/// `FOR UPDATE SKIP LOCKED` so multiple workers never claim the same job.
|
/// `FOR UPDATE SKIP LOCKED` so multiple workers never claim the same job.
|
||||||
/// Returns `None` when the queue is empty.
|
/// Returns `None` when the queue is empty.
|
||||||
|
|||||||
@@ -290,9 +290,14 @@ export function MasterPlannerModal({ onClose }: { onClose: () => void }) {
|
|||||||
if (!proposal || building) return;
|
if (!proposal || building) return;
|
||||||
// If the user picked a topology from the gallery, force it into the proposal
|
// If the user picked a topology from the gallery, force it into the proposal
|
||||||
// — the planner is instructed to honor it but this is the belt-and-braces.
|
// — the planner is instructed to honor it but this is the belt-and-braces.
|
||||||
const effectiveProposal: Proposal = topologyKind && (mode === "team" || mode === "swarm")
|
const effectiveProposal: Proposal & { mode: Mode } = {
|
||||||
? { ...proposal, topology_kind: topologyKind }
|
...proposal,
|
||||||
: proposal;
|
...(topologyKind && (mode === "team" || mode === "swarm")
|
||||||
|
? { topology_kind: topologyKind }
|
||||||
|
: {}),
|
||||||
|
// Backend derives lifecycle from mode (scheduled/triggered → ephemeral).
|
||||||
|
mode,
|
||||||
|
};
|
||||||
setBuilding(true); setError(null); setBuildProg({ pct: 2, label: "Starting…" });
|
setBuilding(true); setError(null); setBuildProg({ pct: 2, label: "Starting…" });
|
||||||
try {
|
try {
|
||||||
await readSse("/api/planner/scaffold", effectiveProposal, async (e) => {
|
await readSse("/api/planner/scaffold", effectiveProposal, async (e) => {
|
||||||
|
|||||||
@@ -0,0 +1,28 @@
|
|||||||
|
-- Ephemeral team lifecycle: a `lifecycle` column on teams + a back-ref from
|
||||||
|
-- topology_runs so the post-terminal hook in cm-api::topology_worker knows
|
||||||
|
-- which team an ephemeral run belongs to.
|
||||||
|
--
|
||||||
|
-- permanent — the default; team + claws persist across runs (Specialists / Team modes).
|
||||||
|
-- ephemeral — team is torn down (claws deprovisioned, rows deleted) after the last
|
||||||
|
-- in-flight run terminates. Scheduled + Triggered modes use this.
|
||||||
|
--
|
||||||
|
-- Teardown policy: on a run's terminal transition (completed/failed/cancelled),
|
||||||
|
-- if the run's team is ephemeral AND no siblings are still queued/running for
|
||||||
|
-- that team, deprovision every bound claw and delete the team row. The FK on
|
||||||
|
-- team_members (ON DELETE CASCADE from 0010) cleans the bindings; agents rows
|
||||||
|
-- are removed by the worker after deprovision_claw() succeeds.
|
||||||
|
|
||||||
|
ALTER TABLE teams
|
||||||
|
ADD COLUMN lifecycle TEXT NOT NULL DEFAULT 'permanent'
|
||||||
|
CHECK (lifecycle IN ('permanent', 'ephemeral'));
|
||||||
|
|
||||||
|
-- Back-ref from a durable run to its team, when the run was fired by
|
||||||
|
-- `POST /api/teams/:id/run` or a scheduled/webhook fire of a team's graph.
|
||||||
|
-- Nullable — ad-hoc topology runs + swarm/research runs leave this NULL.
|
||||||
|
ALTER TABLE topology_runs
|
||||||
|
ADD COLUMN team_id UUID REFERENCES teams (id) ON DELETE SET NULL;
|
||||||
|
|
||||||
|
-- Fast lookup for "any siblings of this team's run still in flight?".
|
||||||
|
CREATE INDEX topology_runs_team_active_idx
|
||||||
|
ON topology_runs (team_id, status)
|
||||||
|
WHERE team_id IS NOT NULL AND status IN ('queued', 'running');
|
||||||
Reference in New Issue
Block a user