osobh/clawmates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

slice 9 cleanup: drop legacy research/loops backend + tables

Browse Source 
Retires the legacy research/loops backend after the missions arc
(slices 1-9) fully replaced it. Frontend cutover was 4663348; this
commit finishes the job on the backend + database.

Migration:
  - 0053_drop_legacy_research_loops.sql — drops the 8 legacy tables
    (research_topics, research_topic_agents, research_outcomes,
    research_publish_approvals, loops, loop_agents, loop_orgs,
    loop_teams) and the 3 topology_runs FK columns
    (research_topic_id, loop_id, iteration). parent_run_id stays;
    recursive_exec still uses it.

Files deleted (11):
  - crates/cm-api/src/routes/{research,loops,research_setup,
    research_pipeline,wizard_repo,probe}.rs
  - crates/cm-api/src/research_container.rs
  - crates/cm-db/src/repo/{research_topics,research_outcomes,
    research_publish_approvals,loops}.rs
  - crates/cm-runtime/src/loops.rs
  - crates/cm-api/tests/research_publish_role.rs

Files edited:
  - crates/cm-api/src/lib.rs — dropped 20 legacy route registrations
    (all /api/research/* + /api/loops/* + /webhooks/loops + probe)
    and module decls
  - crates/cm-api/src/topology_worker.rs — deleted legacy dispatch
    (freeze_research_outcome, advance_loop_after_completion,
    continue_initial_burst, maybe_transition_research_topic,
    parse_reorder_rationale, per-topic/loop gateway resolver).
    reap_stuck_runs now keys on mission_id (not topic_id).
    Executor path unconditionally uses ZeroClawDriveExecutor::from_env
    — mission_orchestrator provisions each claw as an agent inside
    the shared runtime via RuntimeProvisioner, so per-team gateway
    resolution is no longer applicable.
  - crates/cm-api/src/routes/topology.rs — deleted container-log SSE
    endpoint (research/loop-specific), dropped loop_id filter and
    iteration field from ListRunsQuery/RunSummary
  - crates/cm-api/src/routes/world.rs — removed
    active_research_topics/active_loops/preseed_repo_paths;
    World SSE no longer emits repo:{topic}/loop:{id} landmark orbs
    (follow-up task #21 tracks adding mission:{id} equivalents)
  - crates/cm-api/src/runtime_provision.rs — removed now-unused
    mint_workspace_service_token
  - crates/cm-db/src/repo/topology_runs.rs — removed 9 legacy
    helpers (research_topic_id lookup, loop_id_for_run,
    iteration_for_run, active_runs_for_research_topic, etc.)
  - crates/cm-db/src/repo/teams.rs — removed 4 dead helpers
    (team_for_loop, team_for_research_topic + setters)
  - crates/cm-api/tests/topology_jobs.rs — removed loop/topic
    tests, dropped enqueue_run_with_topic helper
  - crates/bins/clawmates-server/src/main.rs — removed
    spawn_loop_scheduler call
  - crates/cm-api/src/routes/mod.rs, crates/cm-db/src/repo/mod.rs,
    crates/cm-runtime/src/lib.rs — module decls stripped

sqlx cache: regenerated against post-migration schema
  (71 files changed, ~+70 / -8896 net)

Test/build: SQLX_OFFLINE=true cargo check --workspace clean;
cargo test --workspace --no-run clean.

Follow-up (task #21): World view lost the in-flight-work landmarks
when repo:{topic} / loop:{id} orbs disappeared. Add mission:{id}
orbs as the missions-era replacement.
This commit is contained in:
Omar Sobh
2026-07-19 18:37:24 -07:00
parent 56201a6985
commit fdb8cfeecc
71 changed files with 70 additions and 8896 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/cm-db/src/repo/loops.rs
-877
View File
@@ -1,877 +0,0 @@
//! Loops — durable recurring topology executions. The row holds the
//! definition (graph + task_template + triggers + repeat_policy) and a small
//! amount of scheduler state (enabled, next_fire_at, last_run_id).
//! Each fire produces a normal `topology_runs` row with loop_id + iteration
//! + parent_run_id set, so the run driver picks it up like any other job.
//!
//! See 0031 migration header for the state semantics and missed-window rule.
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::PgPool;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::DbError;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Loop {
pub id: Uuid,
pub workspace_id: Uuid,
pub title: String,
pub description: String,
pub graph: Value,
pub task_template: String,
pub triggers: Value,
pub repeat_policy: Value,
pub enabled: bool,
pub next_fire_at: Option<OffsetDateTime>,
pub last_run_id: Option<Uuid>,
pub webhook_token: Option<String>,
pub webhook_signing_key: Option<String>,
pub created_by: Uuid,
pub created_at: OffsetDateTime,
pub updated_at: OffsetDateTime,
}
/// Minimal fields the scheduler needs when it wakes up.
#[derive(Debug, Clone)]
pub struct DueLoop {
pub id: Uuid,
pub workspace_id: Uuid,
pub graph: Value,
pub task_template: String,
pub triggers: Value,
pub repeat_policy: Value,
pub last_run_id: Option<Uuid>,
}
pub struct NewLoop<'a> {
pub workspace_id: Uuid,
pub title: &'a str,
pub description: &'a str,
pub graph: &'a Value,
pub task_template: &'a str,
pub triggers: &'a Value,
pub repeat_policy: &'a Value,
pub enabled: bool,
pub next_fire_at: Option<OffsetDateTime>,
pub webhook_token: Option<&'a str>,
pub webhook_signing_key: Option<&'a str>,
pub created_by: Uuid,
}
pub async fn create(pool: &PgPool, input: NewLoop<'_>) -> Result<Uuid, DbError> {
let id = Uuid::now_v7();
sqlx::query!(
"INSERT INTO loops
(id, workspace_id, title, description, graph, task_template,
triggers, repeat_policy, enabled, next_fire_at,
webhook_token, webhook_signing_key, created_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)",
id,
input.workspace_id,
input.title,
input.description,
input.graph,
input.task_template,
input.triggers,
input.repeat_policy,
input.enabled,
input.next_fire_at,
input.webhook_token,
input.webhook_signing_key,
input.created_by,
)
.execute(pool)
.await?;
Ok(id)
}
pub async fn list(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<Loop>, DbError> {
let rows = sqlx::query_as!(
Loop,
"SELECT id, workspace_id, title, description, graph, task_template,
triggers, repeat_policy, enabled, next_fire_at, last_run_id,
webhook_token, webhook_signing_key, created_by, created_at, updated_at
FROM loops
WHERE workspace_id = $1
ORDER BY updated_at DESC",
workspace_id,
)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// Kind + source_research_topic_id + task_template — the minimum a
/// caller needs to compose the right iteration for a loop without
/// hydrating the whole Loop struct. Kind='exec' preserves today's
/// behavior; kind='research' builds a research prompt bound to the
/// source topic so freeze_research_outcome writes a new outcome
/// version.
pub async fn kind_and_binding(
pool: &PgPool,
loop_id: Uuid,
) -> Result<Option<(String, Option<Uuid>, String)>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"SELECT kind, source_research_topic_id, task_template
FROM loops
WHERE id = $1",
)
.bind(loop_id)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| {
(
r.get::<String, _>("kind"),
r.try_get::<Option<Uuid>, _>("source_research_topic_id")
.ok()
.flatten(),
r.get::<String, _>("task_template"),
)
}))
}
/// Cross-workspace fetch used by internal callers (topology_worker
/// completion hooks) where the run row is authoritative for the
/// workspace binding — no need for a second scoping check. Returns
/// None if the loop was deleted between run enqueue and completion.
pub async fn get_any_workspace(pool: &PgPool, id: Uuid) -> Result<Option<Loop>, DbError> {
// Dynamic query so this callsite doesn't require an offline sqlx
// cache regen — used from the completion hook, not on the hot path.
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"SELECT id, workspace_id, title, description, graph, task_template,
triggers, repeat_policy, enabled, next_fire_at, last_run_id,
webhook_token, webhook_signing_key, created_by, created_at, updated_at
FROM loops
WHERE id = $1",
)
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| Loop {
id: r.get("id"),
workspace_id: r.get("workspace_id"),
title: r.get("title"),
description: r.get("description"),
graph: r.get("graph"),
task_template: r.get("task_template"),
triggers: r.get("triggers"),
repeat_policy: r.get("repeat_policy"),
enabled: r.get("enabled"),
next_fire_at: r.try_get("next_fire_at").ok().flatten(),
last_run_id: r.try_get("last_run_id").ok().flatten(),
webhook_token: r.try_get("webhook_token").ok().flatten(),
webhook_signing_key: r.try_get("webhook_signing_key").ok().flatten(),
created_by: r.get("created_by"),
created_at: r.get("created_at"),
updated_at: r.get("updated_at"),
}))
}
pub async fn get(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<Option<Loop>, DbError> {
let row = sqlx::query_as!(
Loop,
"SELECT id, workspace_id, title, description, graph, task_template,
triggers, repeat_policy, enabled, next_fire_at, last_run_id,
webhook_token, webhook_signing_key, created_by, created_at, updated_at
FROM loops
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
)
.fetch_optional(pool)
.await?;
Ok(row)
}
/// Look up a loop by its webhook token — used only by the webhook receiver,
/// which has no session context. Returns the minimal shape needed to enqueue
/// an iteration and verify the HMAC signature.
pub async fn get_by_webhook_token(
pool: &PgPool,
token: &str,
) -> Result<Option<(Uuid, Uuid, String, DueLoop)>, DbError> {
let row = sqlx::query!(
"SELECT id, workspace_id, graph, task_template, triggers, repeat_policy,
last_run_id, webhook_signing_key
FROM loops
WHERE webhook_token = $1 AND enabled",
token,
)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| {
let key = r.webhook_signing_key?;
Some((
r.id,
r.workspace_id,
key,
DueLoop {
id: r.id,
workspace_id: r.workspace_id,
graph: r.graph,
task_template: r.task_template,
triggers: r.triggers,
repeat_policy: r.repeat_policy,
last_run_id: r.last_run_id,
},
))
}))
}
pub struct UpdateLoop<'a> {
pub title: &'a str,
pub description: &'a str,
pub graph: &'a Value,
pub task_template: &'a str,
pub triggers: &'a Value,
pub repeat_policy: &'a Value,
pub next_fire_at: Option<OffsetDateTime>,
}
pub async fn update(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
input: UpdateLoop<'_>,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE loops
SET title = $3, description = $4, graph = $5, task_template = $6,
triggers = $7, repeat_policy = $8, next_fire_at = $9,
updated_at = now()
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
input.title,
input.description,
input.graph,
input.task_template,
input.triggers,
input.repeat_policy,
input.next_fire_at,
)
.execute(pool)
.await?;
Ok(())
}
pub async fn set_enabled(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
enabled: bool,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE loops SET enabled = $3, updated_at = now()
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
enabled,
)
.execute(pool)
.await?;
Ok(())
}
pub async fn delete(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<(), DbError> {
sqlx::query!(
"DELETE FROM loops WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
)
.execute(pool)
.await?;
Ok(())
}
/// Persist the per-loop team container name + gateway URL after a
/// successful `spawn_loop` (P2). Dynamic query so the new columns don't
/// need a fresh .sqlx offline cache entry.
pub async fn set_zeroclaw_container(
pool: &PgPool,
loop_id: Uuid,
workspace_id: Uuid,
container: &str,
gateway_url: &str,
) -> Result<(), DbError> {
sqlx::query(
"UPDATE loops
SET zeroclaw_container = $3,
zeroclaw_gateway_url = $4,
updated_at = now()
WHERE id = $1 AND workspace_id = $2",
)
.bind(loop_id)
.bind(workspace_id)
.bind(container)
.bind(gateway_url)
.execute(pool)
.await?;
Ok(())
}
/// Append one reorder rationale event to the loop's reorder_events
/// jsonb array. Called from the topology_worker completion hook after
/// parsing REORDER: markers out of the run output. Each event carries
/// the iteration index, run_id, text, and now() timestamp so a
/// downstream mini-timeline can show WHEN the plan was adjusted and
/// WHY. Idempotent: appending a duplicate text/run_id combo is allowed
/// (rare — indicates the parser matched twice on the same line).
/// Set a loop's kind. Used by materialize_topic_loops right after
/// create() — the create path doesn't take a kind parameter (default
/// 'exec' matches every legacy loop), so research-kind loops flip the
/// column in a follow-up UPDATE.
pub async fn set_kind(pool: &PgPool, loop_id: Uuid, kind: &str) -> Result<(), DbError> {
sqlx::query("UPDATE loops SET kind = $2, updated_at = now() WHERE id = $1")
.bind(loop_id)
.bind(kind)
.execute(pool)
.await?;
Ok(())
}
/// Set the countdown for the triggers.initial_burst quota. On
/// create_loop we set this to `burst - 1` after firing the first
/// iteration inline; on each subsequent completion we decrement and,
/// while it's > 0, enqueue another iteration. Dynamic query so the
/// new column doesn't need an offline sqlx cache regen.
pub async fn set_initial_burst_remaining(
pool: &PgPool,
loop_id: Uuid,
remaining: i32,
) -> Result<(), DbError> {
sqlx::query("UPDATE loops SET initial_burst_remaining = $2 WHERE id = $1")
.bind(loop_id)
.bind(remaining)
.execute(pool)
.await?;
Ok(())
}
/// Atomically decrement initial_burst_remaining, returning the value
/// BEFORE decrement. Zero is a no-op (returns 0). Used by the
/// completion hook: caller enqueues a new iteration if the returned
/// value is > 0. CAS-safe: two workers can't race and both enqueue.
pub async fn take_initial_burst_slot(pool: &PgPool, loop_id: Uuid) -> Result<i32, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"UPDATE loops
SET initial_burst_remaining = GREATEST(initial_burst_remaining - 1, 0)
WHERE id = $1 AND initial_burst_remaining > 0
RETURNING initial_burst_remaining + 1 AS prev",
)
.bind(loop_id)
.fetch_optional(pool)
.await?;
Ok(row
.and_then(|r| r.try_get::<i32, _>("prev").ok())
.unwrap_or(0))
}
/// Enumerate loops that should wake up when a research topic gets a
/// new outcome. Filters to kind='exec', enabled, and
/// triggers.on_artifact_update = true. Called by the completion hook
/// after freeze_research_outcome inserts a new row. Returns
/// (loop_id, workspace_id, task_template, graph) so the caller can
/// enqueue directly without a second fetch.
/// Return the kind='research' loop that owns a topic's runs (there
/// should be at most one — created by the wizard's
/// materialize_topic_loops). Used by the topic detail endpoint to
/// tell the canvas that classic Start/Submit buttons should be
/// replaced with the loop-managed UI.
pub async fn research_loop_for_topic(
pool: &PgPool,
topic_id: Uuid,
) -> Result<
Option<(
Uuid,
String,
bool,
Option<time::OffsetDateTime>,
Option<Uuid>,
Value,
)>,
DbError,
> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"SELECT id, title, enabled, next_fire_at, last_run_id, triggers
FROM loops
WHERE source_research_topic_id = $1
AND kind = 'research'
ORDER BY created_at DESC
LIMIT 1",
)
.bind(topic_id)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| {
(
r.get::<Uuid, _>("id"),
r.get::<String, _>("title"),
r.get::<bool, _>("enabled"),
r.try_get::<Option<time::OffsetDateTime>, _>("next_fire_at")
.ok()
.flatten(),
r.try_get::<Option<Uuid>, _>("last_run_id").ok().flatten(),
r.get::<Value, _>("triggers"),
)
}))
}
pub async fn loops_awaiting_topic(
pool: &PgPool,
topic_id: Uuid,
) -> Result<Vec<(Uuid, Uuid, String, Value)>, DbError> {
use sqlx::Row;
let rows = sqlx::query(
"SELECT id, workspace_id, task_template, graph
FROM loops
WHERE source_research_topic_id = $1
AND kind = 'exec'
AND enabled = true
AND (triggers ->> 'on_artifact_update')::boolean = true",
)
.bind(topic_id)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| {
(
r.get::<Uuid, _>("id"),
r.get::<Uuid, _>("workspace_id"),
r.get::<String, _>("task_template"),
r.get::<Value, _>("graph"),
)
})
.collect())
}
/// Truthy when the loop currently has a queued or running iteration.
/// Used by triggers (on_artifact_update, on_completion chain) to
/// coalesce — no point enqueuing another iteration while one is
/// already pending.
pub async fn has_active_run(pool: &PgPool, loop_id: Uuid) -> Result<bool, DbError> {
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"SELECT 1 AS one FROM topology_runs
WHERE loop_id = $1 AND status IN ('queued', 'running')
LIMIT 1",
)
.bind(loop_id)
.fetch_optional(pool)
.await?;
Ok(row.is_some())
}
/// Read the reorder_events array for a loop, newest-first, capped at
/// `limit`. Used by the progress endpoint to surface a compact recent
/// history on the sidebar card. Empty array for standalone loops or
/// loops whose coordinator hasn't emitted any REORDER markers yet.
pub async fn recent_reorders(
pool: &PgPool,
loop_id: Uuid,
limit: i64,
) -> Result<Vec<Value>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> =
sqlx::query("SELECT reorder_events FROM loops WHERE id = $1")
.bind(loop_id)
.fetch_optional(pool)
.await?;
let arr: Vec<Value> = row
.and_then(|r| r.try_get::<Value, _>("reorder_events").ok())
.and_then(|v| v.as_array().cloned())
.unwrap_or_default();
// Appended in chronological order (oldest → newest); reversing then
// taking `limit` yields the newest N in newest-first order.
let recent: Vec<Value> = arr.into_iter().rev().take(limit as usize).collect();
Ok(recent)
}
pub async fn append_reorder_event(
pool: &PgPool,
loop_id: Uuid,
run_id: Uuid,
iteration: i32,
text: &str,
) -> Result<(), DbError> {
// Build the event server-side so `ts` uses postgres now() (canonical
// wall clock; avoids skew if callers had stale local clocks).
sqlx::query(
"UPDATE loops
SET reorder_events = reorder_events || jsonb_build_object(
'run_id', $2::text,
'iteration', $3::int,
'text', $4::text,
'ts', to_char(now() AT TIME ZONE 'UTC',
'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"')
),
updated_at = now()
WHERE id = $1",
)
.bind(loop_id)
.bind(run_id.to_string())
.bind(iteration)
.bind(text)
.execute(pool)
.await?;
Ok(())
}
/// Atomically append `completed` INT-XX ids to the loop's
/// `consumed_int_ids` array and bump `current_int_index` by the count
/// of NEW ids landed. Existing ids are not re-appended (idempotent on
/// re-runs). Called from the topology_worker completion hook after
/// parsing "COMPLETED: INT-XX" markers out of the run's final output.
pub async fn advance_after_completion(
pool: &PgPool,
loop_id: Uuid,
completed: &[String],
) -> Result<(), DbError> {
if completed.is_empty() {
return Ok(());
}
// Use array set semantics: append only ids not already present.
// The subquery computes the new list; length delta feeds the index bump.
sqlx::query(
"UPDATE loops
SET consumed_int_ids = (
SELECT ARRAY(
SELECT DISTINCT unnest(consumed_int_ids || $2::TEXT[])
)
),
current_int_index = current_int_index + (
SELECT count(*) FROM unnest($2::TEXT[]) AS n(v)
WHERE NOT (consumed_int_ids @> ARRAY[v])
),
updated_at = now()
WHERE id = $1",
)
.bind(loop_id)
.bind(completed)
.execute(pool)
.await?;
Ok(())
}
/// Bind (or unbind) a loop's source research topic. When set, the loop's
/// enqueue path prepends the topic's latest artifact + a "focus on the
/// next unconsumed INT" instruction to the coordinator task (option b,
/// order-sequential iteration).
pub async fn set_source_research_topic(
pool: &PgPool,
loop_id: Uuid,
workspace_id: Uuid,
source: Option<Uuid>,
) -> Result<(), DbError> {
sqlx::query(
"UPDATE loops
SET source_research_topic_id = $3, updated_at = now()
WHERE id = $1 AND workspace_id = $2",
)
.bind(loop_id)
.bind(workspace_id)
.bind(source)
.execute(pool)
.await?;
Ok(())
}
/// Read a loop's source research topic id + consumed INT ids +
/// current index. Used by the enqueue path when building the
/// coordinator task string. Missing rows / NULL columns return None
/// so the caller can fall back to the plain task_template.
pub async fn source_research_context(
pool: &PgPool,
loop_id: Uuid,
) -> Result<Option<(Uuid, Vec<String>, i32)>, DbError> {
use sqlx::Row;
let row = sqlx::query(
"SELECT source_research_topic_id, consumed_int_ids, current_int_index
FROM loops
WHERE id = $1",
)
.bind(loop_id)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| {
let topic = r
.try_get::<Option<Uuid>, _>("source_research_topic_id")
.ok()
.flatten()?;
let consumed = r
.try_get::<Vec<String>, _>("consumed_int_ids")
.unwrap_or_default();
let idx = r.try_get::<i32, _>("current_int_index").unwrap_or(0);
Some((topic, consumed, idx))
}))
}
/// Read the per-loop gateway URL (or None if the loop hasn't spawned a
/// container yet). Used by `topology_worker` to prefer the isolated
/// daemon over the workspace-wide one.
pub async fn zeroclaw_gateway_url(pool: &PgPool, loop_id: Uuid) -> Result<Option<String>, DbError> {
use sqlx::Row;
let row = sqlx::query("SELECT zeroclaw_gateway_url FROM loops WHERE id = $1")
.bind(loop_id)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| {
r.try_get::<Option<String>, _>("zeroclaw_gateway_url")
.ok()
.flatten()
}))
}
// --- Staffing ---------------------------------------------------------------
//
// Loops attach agents, teams, and/or orgs. The three join tables are
// parallel; a loop can mix modes (e.g. one team + a couple of specialist
// agents). Callers use the `set_*` replace-all shape so PATCH is a single
// transactional swap — simpler than diffing and cheap for the list sizes
// this UI generates.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentSlot {
pub agent_id: Uuid,
pub role_slot: Option<String>,
}
pub async fn set_agents(pool: &PgPool, loop_id: Uuid, slots: &[AgentSlot]) -> Result<(), DbError> {
let mut tx = pool.begin().await?;
sqlx::query!("DELETE FROM loop_agents WHERE loop_id = $1", loop_id)
.execute(&mut *tx)
.await?;
for s in slots {
sqlx::query!(
"INSERT INTO loop_agents (loop_id, agent_id, role_slot)
VALUES ($1, $2, $3)
ON CONFLICT (loop_id, agent_id) DO UPDATE
SET role_slot = EXCLUDED.role_slot",
loop_id,
s.agent_id,
s.role_slot,
)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
pub async fn agents(pool: &PgPool, loop_id: Uuid) -> Result<Vec<AgentSlot>, DbError> {
let rows = sqlx::query!(
"SELECT agent_id, role_slot FROM loop_agents WHERE loop_id = $1",
loop_id,
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| AgentSlot {
agent_id: r.agent_id,
role_slot: r.role_slot,
})
.collect())
}
pub async fn set_teams(pool: &PgPool, loop_id: Uuid, ids: &[Uuid]) -> Result<(), DbError> {
let mut tx = pool.begin().await?;
sqlx::query!("DELETE FROM loop_teams WHERE loop_id = $1", loop_id)
.execute(&mut *tx)
.await?;
for id in ids {
sqlx::query!(
"INSERT INTO loop_teams (loop_id, team_id) VALUES ($1, $2)
ON CONFLICT (loop_id, team_id) DO NOTHING",
loop_id,
id,
)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
pub async fn teams(pool: &PgPool, loop_id: Uuid) -> Result<Vec<Uuid>, DbError> {
let rows = sqlx::query!("SELECT team_id FROM loop_teams WHERE loop_id = $1", loop_id,)
.fetch_all(pool)
.await?;
Ok(rows.into_iter().map(|r| r.team_id).collect())
}
pub async fn set_orgs(pool: &PgPool, loop_id: Uuid, ids: &[Uuid]) -> Result<(), DbError> {
let mut tx = pool.begin().await?;
sqlx::query!("DELETE FROM loop_orgs WHERE loop_id = $1", loop_id)
.execute(&mut *tx)
.await?;
for id in ids {
sqlx::query!(
"INSERT INTO loop_orgs (loop_id, org_id) VALUES ($1, $2)
ON CONFLICT (loop_id, org_id) DO NOTHING",
loop_id,
id,
)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
pub async fn orgs(pool: &PgPool, loop_id: Uuid) -> Result<Vec<Uuid>, DbError> {
let rows = sqlx::query!("SELECT org_id FROM loop_orgs WHERE loop_id = $1", loop_id,)
.fetch_all(pool)
.await?;
Ok(rows.into_iter().map(|r| r.org_id).collect())
}
/// Loops the scheduler tick should fire NOW. Only reads what the enqueue
/// path needs, so the tick stays cheap even when the workspace has hundreds
/// of loops.
pub async fn due(pool: &PgPool) -> Result<Vec<DueLoop>, DbError> {
let rows = sqlx::query!(
"SELECT id, workspace_id, graph, task_template, triggers, repeat_policy,
last_run_id
FROM loops
WHERE enabled AND next_fire_at IS NOT NULL AND next_fire_at <= now()",
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| DueLoop {
id: r.id,
workspace_id: r.workspace_id,
graph: r.graph,
task_template: r.task_template,
triggers: r.triggers,
repeat_policy: r.repeat_policy,
last_run_id: r.last_run_id,
})
.collect())
}
/// Post-fire bookkeeping: bump last_run_id + advance next_fire_at (NULL when
/// the loop has no cron trigger). Called by the scheduler after a successful
/// enqueue_iteration.
pub async fn mark_fired(
pool: &PgPool,
id: Uuid,
run_id: Uuid,
next_fire_at: Option<OffsetDateTime>,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE loops
SET last_run_id = $2, next_fire_at = $3, updated_at = now()
WHERE id = $1",
id,
run_id,
next_fire_at,
)
.execute(pool)
.await?;
Ok(())
}
/// Next iteration number for a loop (1 if it has never fired).
pub async fn next_iteration(pool: &PgPool, loop_id: Uuid) -> Result<i32, DbError> {
let n: Option<i32> = sqlx::query_scalar!(
"SELECT MAX(iteration) FROM topology_runs WHERE loop_id = $1",
loop_id,
)
.fetch_one(pool)
.await?;
Ok(n.unwrap_or(0) + 1)
}
/// Enqueue an iteration as a normal `topology_runs` row. The scheduler,
/// on-completion hook, and webhook receiver all funnel through here so the
/// invariants (loop_id + iteration + parent_run_id all set together) stay
/// in one place.
pub async fn enqueue_iteration(
pool: &PgPool,
loop_id: Uuid,
workspace_id: Uuid,
task: &str,
graph: &Value,
iteration: i32,
parent_run_id: Option<Uuid>,
) -> Result<Uuid, DbError> {
enqueue_iteration_with_topic(
pool,
IterationEnqueue {
loop_id,
workspace_id,
task,
graph,
iteration,
parent_run_id,
research_topic_id: None,
},
)
.await
}
/// Batched arguments for `enqueue_iteration_with_topic`. Bundled so the
/// signature stays under clippy's 7-arg ceiling — the columns are
/// all conceptually one "run to enqueue for a loop", not free-floating
/// parameters.
pub struct IterationEnqueue<'a> {
pub loop_id: Uuid,
pub workspace_id: Uuid,
pub task: &'a str,
pub graph: &'a Value,
pub iteration: i32,
pub parent_run_id: Option<Uuid>,
pub research_topic_id: Option<Uuid>,
}
/// Variant of `enqueue_iteration` that also sets `research_topic_id` on
/// the topology_runs row. Used by kind='research' loops so
/// `freeze_research_outcome` writes a new outcome version each
/// iteration, and by any future flow that binds a run to both a loop
/// and a research topic.
pub async fn enqueue_iteration_with_topic(
pool: &PgPool,
args: IterationEnqueue<'_>,
) -> Result<Uuid, DbError> {
let IterationEnqueue {
loop_id,
workspace_id,
task,
graph,
iteration,
parent_run_id,
research_topic_id,
} = args;
let run_id = Uuid::now_v7();
// Dynamic query so the new column combination (loop_id +
// research_topic_id on the same row) doesn't require an offline
// sqlx cache regen — the enqueue path only runs on user actions,
// not the tight worker loop.
sqlx::query(
"INSERT INTO topology_runs
(id, workspace_id, task, kind, status, graph, tier,
loop_id, iteration, parent_run_id, research_topic_id)
VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7, $8)",
)
.bind(run_id)
.bind(workspace_id)
.bind(task)
.bind(graph)
.bind(loop_id)
.bind(iteration)
.bind(parent_run_id)
.bind(research_topic_id)
.execute(pool)
.await?;
Ok(run_id)
}
crates/cm-db/src/repo/mod.rs
-4
View File
@@ -10,7 +10,6 @@ pub mod files;
pub mod fleet_beszel;
pub mod fleet_tailscale;
pub mod level_up;
pub mod loops;
pub mod messages;
pub mod missions;
pub mod node_metrics;
@@ -21,9 +20,6 @@ pub mod orgs;
pub mod outbox;
pub mod repo_connections;
pub mod repos;
pub mod research_outcomes;
pub mod research_publish_approvals;
pub mod research_topics;
pub mod routine_runs;
pub mod routines;
pub mod run_events;
crates/cm-db/src/repo/research_outcomes.rs
-86
View File
@@ -1,86 +0,0 @@
//! Persisted research artifacts — one row per run's final synthesis. When
//! `topology_worker` completes a run tagged with a `research_topic_id`, it
//! extracts the orchestrator's `RunRecord.final_output` and calls
//! [`insert`] here. The frontend canvas then renders the latest outcome
//! instead of the topic description when the topic has moved past
//! `standby`, so reviewers see the actual draft.
//!
//! Version is per-topic and monotonically increasing so reject-with-
//! revision loops accumulate history rather than clobber prior drafts.
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::DbError;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Outcome {
pub id: Uuid,
pub topic_id: Uuid,
pub version: i32,
pub body_md: String,
pub produced_by_run_id: Option<Uuid>,
// RFC3339 on the wire so `new Date(...)` in the browser parses it
// instead of choking on the `time` crate's default `[y, ordinal,
// ...]` array format (surfaced as "Invalid Date" in the Draft
// header).
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
}
/// Insert a new outcome. Version is derived server-side as `max(version) + 1`
/// for the topic (starting at 1) so callers never need to know the current
/// count. Returns the persisted row.
pub async fn insert(
pool: &PgPool,
topic_id: Uuid,
body_md: &str,
produced_by_run_id: Option<Uuid>,
) -> Result<Outcome, DbError> {
let id = Uuid::now_v7();
let row = sqlx::query!(
"INSERT INTO research_outcomes (id, topic_id, version, body_md, produced_by_run_id)
SELECT $1, $2, coalesce(max(version), 0) + 1, $3, $4
FROM research_outcomes
WHERE topic_id = $2
RETURNING id, topic_id, version, body_md, produced_by_run_id, created_at",
id,
topic_id,
body_md,
produced_by_run_id,
)
.fetch_one(pool)
.await?;
Ok(Outcome {
id: row.id,
topic_id: row.topic_id,
version: row.version,
body_md: row.body_md,
produced_by_run_id: row.produced_by_run_id,
created_at: row.created_at,
})
}
/// Newest outcome for a topic, or `None` if no run has completed yet.
pub async fn latest(pool: &PgPool, topic_id: Uuid) -> Result<Option<Outcome>, DbError> {
let row = sqlx::query!(
"SELECT id, topic_id, version, body_md, produced_by_run_id, created_at
FROM research_outcomes
WHERE topic_id = $1
ORDER BY version DESC
LIMIT 1",
topic_id,
)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| Outcome {
id: r.id,
topic_id: r.topic_id,
version: r.version,
body_md: r.body_md,
produced_by_run_id: r.produced_by_run_id,
created_at: r.created_at,
}))
}
crates/cm-db/src/repo/research_publish_approvals.rs
-157
View File
@@ -1,157 +0,0 @@
//! Publish approval gate for research topics — see 0032 migration header.
//! Small table with a small state machine (pending → approved | rejected).
//! One pending row per topic at a time; enforced at the route layer by
//! looking up `pending_for_topic` before create.
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::DbError;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PublishApproval {
pub id: Uuid,
pub workspace_id: Uuid,
pub topic_id: Uuid,
pub requested_by: Uuid,
pub status: String,
pub decided_by: Option<Uuid>,
pub decided_at: Option<OffsetDateTime>,
pub created_at: OffsetDateTime,
}
pub async fn create(
pool: &PgPool,
workspace_id: Uuid,
topic_id: Uuid,
requested_by: Uuid,
) -> Result<Uuid, DbError> {
let id = Uuid::now_v7();
sqlx::query!(
"INSERT INTO research_publish_approvals
(id, workspace_id, topic_id, requested_by, status)
VALUES ($1, $2, $3, $4, 'pending')",
id,
workspace_id,
topic_id,
requested_by,
)
.execute(pool)
.await?;
Ok(id)
}
/// Pending approval for a topic, if any. The route layer uses this to
/// short-circuit before writing a duplicate request.
pub async fn pending_for_topic(
pool: &PgPool,
topic_id: Uuid,
) -> Result<Option<PublishApproval>, DbError> {
let row = sqlx::query_as!(
PublishApproval,
"SELECT id, workspace_id, topic_id, requested_by, status,
decided_by, decided_at, created_at
FROM research_publish_approvals
WHERE topic_id = $1 AND status = 'pending'
LIMIT 1",
topic_id,
)
.fetch_optional(pool)
.await?;
Ok(row)
}
pub async fn get(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
) -> Result<Option<PublishApproval>, DbError> {
let row = sqlx::query_as!(
PublishApproval,
"SELECT id, workspace_id, topic_id, requested_by, status,
decided_by, decided_at, created_at
FROM research_publish_approvals
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
)
.fetch_optional(pool)
.await?;
Ok(row)
}
pub async fn list_pending(
pool: &PgPool,
workspace_id: Uuid,
) -> Result<Vec<PublishApproval>, DbError> {
let rows = sqlx::query_as!(
PublishApproval,
"SELECT id, workspace_id, topic_id, requested_by, status,
decided_by, decided_at, created_at
FROM research_publish_approvals
WHERE workspace_id = $1 AND status = 'pending'
ORDER BY created_at DESC",
workspace_id,
)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// Atomically flip a pending row to approved/rejected. Returns whether the
/// caller was the one who won the race — false when the row was already
/// decided (idempotent). Optional `notes` are stashed on the row so a
/// subsequent `start_topic` can pick them up as revision guidance
/// (R2 — reject-with-revision).
pub async fn decide(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
decided_by: Uuid,
approve: bool,
notes: Option<&str>,
) -> Result<bool, DbError> {
let new_status = if approve { "approved" } else { "rejected" };
// Dynamic sqlx::query so the new `notes` column doesn't need a fresh
// .sqlx offline cache entry — the value is bound at runtime.
let result = sqlx::query(
"UPDATE research_publish_approvals
SET status = $4, decided_by = $3, decided_at = now(), notes = $5
WHERE id = $1 AND workspace_id = $2 AND status = 'pending'",
)
.bind(id)
.bind(workspace_id)
.bind(decided_by)
.bind(new_status)
.bind(notes)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Most-recent rejected-approval notes for a topic, or None. Used by
/// `start_topic` to prepend a reviewer's revision guidance to the next
/// coordinator task. Only returns non-empty strings; a rejection with
/// no notes reads the same as no rejection at all.
pub async fn latest_rejection_notes(
pool: &PgPool,
topic_id: Uuid,
) -> Result<Option<String>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"SELECT notes
FROM research_publish_approvals
WHERE topic_id = $1 AND status = 'rejected' AND notes IS NOT NULL
ORDER BY decided_at DESC NULLS LAST
LIMIT 1",
)
.bind(topic_id)
.fetch_optional(pool)
.await?;
Ok(row
.and_then(|r| r.try_get::<Option<String>, _>("notes").ok().flatten())
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty()))
}
crates/cm-db/src/repo/research_topics.rs
-357
View File
@@ -1,357 +0,0 @@
//! Research topics — user-scoped inquiry containers that group agents around
//! a shared question and drive them toward a named outcome (spec, prod_plan,
//! roadmap, paper). The status column is a small state machine; see the
//! 0030 migration header for the transitions. Runs are owned via
//! `topology_runs.research_topic_id`, so all durable execution state lives
//! there — this repo only manages the container + status + agent binding.
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::DbError;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResearchTopic {
pub id: Uuid,
pub workspace_id: Uuid,
pub title: String,
pub description: String,
pub outcome_kind: String,
pub status: String,
pub created_by: Uuid,
pub created_at: OffsetDateTime,
pub updated_at: OffsetDateTime,
pub published_at: Option<OffsetDateTime>,
/// Topology shape start_topic builds when firing this topic. Options:
/// hub_spoke, pipeline, hierarchical, star_moe. Defaults to hub_spoke.
pub topology_kind: String,
/// The workspace repo the wizard bound to this topic (optional). When
/// set, `start_topic` clones it and feeds the coordinator prompt with
/// the checkout path + a file-tree overview so the agents can reason
/// about the actual code.
pub repo_id: Option<Uuid>,
/// Absolute path on the API host where `start_topic` cloned the bound
/// repo. Written once on the first successful clone; subsequent starts
/// reuse it. Null until then.
pub repo_workspace_path: Option<String>,
/// Docker container name of the per-topic ZeroClaw team runtime, e.g.
/// "research-<topic_id>-team". Set by `research_container::spawn`;
/// cleared by teardown. Also used to look up the container for stop.
pub zeroclaw_container_name: Option<String>,
/// Reachable URL of the per-topic team's gateway, e.g.
/// `http://research-<topic_id>-team:42617`. Persisted so
/// topology_worker can point ZeroClawDriveExecutor at the isolated
/// endpoint for THIS topic's runs instead of the global env one.
pub zeroclaw_gateway_url: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct AgentSlot {
pub agent_id: Uuid,
pub role_slot: Option<String>,
}
/// Fields the wizard collected for a new research topic. Grouped into a
/// struct so `create` stays under the 7-argument clippy ceiling and future
/// wizard additions (repo commit-branch, etc.) don't cascade into every
/// call site.
pub struct NewTopic<'a> {
pub workspace_id: Uuid,
pub title: &'a str,
pub description: &'a str,
pub outcome_kind: &'a str,
pub topology_kind: &'a str,
pub repo_id: Option<Uuid>,
pub created_by: Uuid,
}
/// Creates a topic in `standby`. Returns the new row's id.
pub async fn create(pool: &PgPool, input: NewTopic<'_>) -> Result<Uuid, DbError> {
let id = Uuid::now_v7();
sqlx::query!(
"INSERT INTO research_topics
(id, workspace_id, title, description, outcome_kind, topology_kind, repo_id, status, created_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, 'standby', $8)",
id,
input.workspace_id,
input.title,
input.description,
input.outcome_kind,
input.topology_kind,
input.repo_id,
input.created_by,
)
.execute(pool)
.await?;
Ok(id)
}
/// Persist the per-topic ZeroClaw container coordinates. Called from
/// `research_container::spawn` after `docker start` succeeds. Pass `None`
/// on both to clear the fields during teardown.
pub async fn set_zeroclaw_container(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
container_name: Option<&str>,
gateway_url: Option<&str>,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE research_topics
SET zeroclaw_container_name = $3,
zeroclaw_gateway_url = $4,
updated_at = now()
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
container_name,
gateway_url,
)
.execute(pool)
.await?;
Ok(())
}
/// Persist the clone path for a topic's bound repo. Set once, on the first
/// successful clone; a re-start reads it back and skips re-cloning.
pub async fn set_repo_workspace_path(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
path: &str,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE research_topics
SET repo_workspace_path = $3, updated_at = now()
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
path,
)
.execute(pool)
.await?;
Ok(())
}
/// Workspace's topics, newest-updated first.
pub async fn list(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<ResearchTopic>, DbError> {
let rows = sqlx::query_as!(
ResearchTopic,
"SELECT id, workspace_id, title, description, outcome_kind, status,
created_by, created_at, updated_at, published_at, topology_kind,
repo_id, repo_workspace_path,
zeroclaw_container_name, zeroclaw_gateway_url
FROM research_topics
WHERE workspace_id = $1
ORDER BY updated_at DESC",
workspace_id,
)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// Conditional set_status — advance ONLY if the current status
/// matches `from`. Used by the fold hook that bumps standby →
/// processing when a research loop's first iteration goes out
/// without racing with later hooks that may have already advanced
/// the topic further. Returns silently on no match; the caller
/// treats it as best-effort.
pub async fn set_status_if(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
from: &str,
to: &str,
) -> Result<(), DbError> {
sqlx::query(
"UPDATE research_topics
SET status = $4,
published_at = CASE
WHEN $4 = 'publishing' AND published_at IS NULL THEN now()
ELSE published_at
END,
updated_at = now()
WHERE id = $1 AND workspace_id = $2 AND status = $3",
)
.bind(id)
.bind(workspace_id)
.bind(from)
.bind(to)
.execute(pool)
.await?;
Ok(())
}
/// Cross-workspace fetch used by internal callers (topology_worker
/// completion hooks, kind='research' loop iteration builders) where
/// the caller already has an authoritative workspace binding from the
/// linked loop row. Skip the workspace scope filter to avoid a second
/// hop.
pub async fn get_any_workspace(pool: &PgPool, id: Uuid) -> Result<Option<ResearchTopic>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"SELECT id, workspace_id, title, description, outcome_kind, status,
created_by, created_at, updated_at, published_at, topology_kind,
repo_id, repo_workspace_path,
zeroclaw_container_name, zeroclaw_gateway_url
FROM research_topics
WHERE id = $1",
)
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| ResearchTopic {
id: r.get("id"),
workspace_id: r.get("workspace_id"),
title: r.get("title"),
description: r.get("description"),
outcome_kind: r.get("outcome_kind"),
status: r.get("status"),
created_by: r.get("created_by"),
created_at: r.get("created_at"),
updated_at: r.get("updated_at"),
published_at: r.try_get("published_at").ok().flatten(),
topology_kind: r.get("topology_kind"),
repo_id: r.try_get("repo_id").ok().flatten(),
repo_workspace_path: r.try_get("repo_workspace_path").ok().flatten(),
zeroclaw_container_name: r.try_get("zeroclaw_container_name").ok().flatten(),
zeroclaw_gateway_url: r.try_get("zeroclaw_gateway_url").ok().flatten(),
}))
}
pub async fn get(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
) -> Result<Option<ResearchTopic>, DbError> {
let row = sqlx::query_as!(
ResearchTopic,
"SELECT id, workspace_id, title, description, outcome_kind, status,
created_by, created_at, updated_at, published_at, topology_kind,
repo_id, repo_workspace_path,
zeroclaw_container_name, zeroclaw_gateway_url
FROM research_topics
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
)
.fetch_optional(pool)
.await?;
Ok(row)
}
/// Hard-delete a topic and cascade every dependent row. FK cascades on
/// research_topic_agents, research_publish_approvals, and research_outcomes
/// clean themselves up; topology_runs.research_topic_id is SET NULL so
/// historical runs survive with the back-ref cleared.
pub async fn delete(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<(), DbError> {
sqlx::query!(
"DELETE FROM research_topics WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
)
.execute(pool)
.await?;
Ok(())
}
/// Non-status fields; the state machine transitions are their own endpoints.
pub async fn update_fields(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
title: &str,
description: &str,
outcome_kind: &str,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE research_topics
SET title = $3, description = $4, outcome_kind = $5, updated_at = now()
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
title,
description,
outcome_kind,
)
.execute(pool)
.await?;
Ok(())
}
/// State-machine transition. Caller enforces which transitions are valid;
/// this is the single write path so we can bump `updated_at` (and
/// `published_at` on landing in `publishing`).
pub async fn set_status(
pool: &PgPool,
id: Uuid,
workspace_id: Uuid,
status: &str,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE research_topics
SET status = $3,
updated_at = now(),
published_at = CASE
WHEN $3 = 'publishing' AND published_at IS NULL THEN now()
ELSE published_at
END
WHERE id = $1 AND workspace_id = $2",
id,
workspace_id,
status,
)
.execute(pool)
.await?;
Ok(())
}
pub async fn attach_agent(
pool: &PgPool,
topic_id: Uuid,
agent_id: Uuid,
role_slot: Option<&str>,
) -> Result<(), DbError> {
sqlx::query!(
"INSERT INTO research_topic_agents (topic_id, agent_id, role_slot)
VALUES ($1, $2, $3)
ON CONFLICT (topic_id, agent_id) DO UPDATE
SET role_slot = EXCLUDED.role_slot",
topic_id,
agent_id,
role_slot,
)
.execute(pool)
.await?;
Ok(())
}
pub async fn detach_agent(pool: &PgPool, topic_id: Uuid, agent_id: Uuid) -> Result<(), DbError> {
sqlx::query!(
"DELETE FROM research_topic_agents WHERE topic_id = $1 AND agent_id = $2",
topic_id,
agent_id,
)
.execute(pool)
.await?;
Ok(())
}
pub async fn agents(pool: &PgPool, topic_id: Uuid) -> Result<Vec<AgentSlot>, DbError> {
let rows = sqlx::query!(
"SELECT agent_id, role_slot FROM research_topic_agents WHERE topic_id = $1",
topic_id,
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| AgentSlot {
agent_id: r.agent_id,
role_slot: r.role_slot,
})
.collect())
}
crates/cm-db/src/repo/teams.rs
-55
View File
@@ -318,61 +318,6 @@ pub async fn set_team_runtime_config(
Ok(())
}
/// Resolve the team attached to a loop (via loops.team_id, added in
/// 0045). Returns `None` when the loop has no team bound — the runtime
/// then uses whatever fallback rules apply (paired research topic's
/// team, or the template default).
pub async fn team_for_loop(pool: &PgPool, loop_id: Uuid) -> Result<Option<Uuid>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query("SELECT team_id FROM loops WHERE id = $1")
.bind(loop_id)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| r.try_get::<Option<Uuid>, _>("team_id").ok().flatten()))
}
/// Symmetric to `team_for_loop` but for research topics.
pub async fn team_for_research_topic(
pool: &PgPool,
topic_id: Uuid,
) -> Result<Option<Uuid>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> =
sqlx::query("SELECT team_id FROM research_topics WHERE id = $1")
.bind(topic_id)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| r.try_get::<Option<Uuid>, _>("team_id").ok().flatten()))
}
/// Bind a team to a loop (or clear the binding by passing None).
pub async fn set_team_for_loop(
pool: &PgPool,
loop_id: Uuid,
team_id: Option<Uuid>,
) -> Result<(), DbError> {
sqlx::query("UPDATE loops SET team_id = $2 WHERE id = $1")
.bind(loop_id)
.bind(team_id)
.execute(pool)
.await?;
Ok(())
}
/// Bind a team to a research topic (or clear).
pub async fn set_team_for_research_topic(
pool: &PgPool,
topic_id: Uuid,
team_id: Option<Uuid>,
) -> Result<(), DbError> {
sqlx::query("UPDATE research_topics SET team_id = $2 WHERE id = $1")
.bind(topic_id)
.bind(team_id)
.execute(pool)
.await?;
Ok(())
}
/// 0046: read the team's per-container coordinates. Both fields NULL
/// means the team has never spawned; the runtime provisions on first
/// iteration.
crates/cm-db/src/repo/topology_runs.rs
+3 -252
View File
@@ -10,16 +10,14 @@ use uuid::Uuid;
use crate::DbError;
/// A row summary for the recent-runs list. `iteration` and `finished_at`
/// are populated for loop iterations and for terminal runs respectively;
/// `None` for compares or still-in-flight runs.
/// A row summary for the recent-runs list. `finished_at` is populated for
/// terminal runs; `None` for compares or still-in-flight runs.
pub struct TopologyRunSummary {
pub id: Uuid,
pub task: String,
pub status: String,
pub kind: String,
pub created_at: OffsetDateTime,
pub iteration: Option<i32>,
pub finished_at: Option<OffsetDateTime>,
}
@@ -147,170 +145,6 @@ pub async fn enqueue_run_for_team(
Ok(())
}
/// Count `queued` + `running` runs whose `research_topic_id` matches. The
/// research canvas polls this so it can show a spinner "the pipeline is
/// running" and suppress the manual "Submit for review" escape hatch
/// while any run is still in flight.
pub async fn active_runs_for_research_topic(
pool: &PgPool,
research_topic_id: Uuid,
) -> Result<i64, DbError> {
let row = sqlx::query!(
"SELECT count(*) AS n
FROM topology_runs
WHERE research_topic_id = $1
AND status IN ('queued', 'running')",
research_topic_id,
)
.fetch_one(pool)
.await?;
Ok(row.n.unwrap_or(0))
}
/// Live-run panel companion to `active_runs_for_research_topic`: return
/// the actual run ids (queued + running) so the UI can subscribe to
/// their SSE event streams. Ordered newest first — the freshest run is
/// the one the user just kicked off.
/// Batch run-count feeder for the research topic list. Returns a
/// (topic_id, in_flight, failed) tuple per topic in `topic_ids`,
/// omitting topics with zero runs. Used to render the errored-state
/// icon + "rerun" affordance on cards in the left sidebar.
///
/// `failed` counts runs that terminated in `failed` since the topic's
/// most recent successful run (or all-time if none have succeeded).
/// That way an old failure on a topic that later succeeded doesn't
/// keep the card flagged as broken.
pub async fn run_counts_by_research_topic(
pool: &PgPool,
topic_ids: &[Uuid],
) -> Result<Vec<(Uuid, i64, i64)>, DbError> {
use sqlx::Row;
if topic_ids.is_empty() {
return Ok(Vec::new());
}
let rows: Vec<sqlx::postgres::PgRow> = sqlx::query(
"WITH last_success AS (
SELECT research_topic_id, max(created_at) AS ts
FROM topology_runs
WHERE research_topic_id = ANY($1)
AND status = 'completed'
GROUP BY research_topic_id
)
SELECT r.research_topic_id AS topic_id,
count(*) FILTER (WHERE r.status IN ('queued','running')) AS in_flight,
count(*) FILTER (
WHERE r.status = 'failed'
AND r.created_at > coalesce(ls.ts, 'epoch'::timestamptz)
) AS failed
FROM topology_runs r
LEFT JOIN last_success ls
ON ls.research_topic_id = r.research_topic_id
WHERE r.research_topic_id = ANY($1)
GROUP BY r.research_topic_id",
)
.bind(topic_ids)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| {
(
r.get::<Uuid, _>("topic_id"),
r.get::<i64, _>("in_flight"),
r.get::<i64, _>("failed"),
)
})
.collect())
}
pub async fn active_run_ids_for_research_topic(
pool: &PgPool,
research_topic_id: Uuid,
) -> Result<Vec<Uuid>, DbError> {
use sqlx::Row;
// Dynamic query (not `sqlx::query!`) so cm-db builds air-gapped
// without a fresh `cargo sqlx prepare` round-trip. Schema shape is
// identical to `active_runs_for_research_topic` above.
let rows: Vec<sqlx::postgres::PgRow> = sqlx::query(
"SELECT id
FROM topology_runs
WHERE research_topic_id = $1
AND status IN ('queued', 'running')
ORDER BY created_at DESC",
)
.bind(research_topic_id)
.fetch_all(pool)
.await?;
Ok(rows.into_iter().map(|r| r.get::<Uuid, _>("id")).collect())
}
/// The research topic this run belongs to, if any. Used by the topology
/// worker's `freeze_research_outcome` post-hook to snapshot the run's
/// final synthesis into `research_outcomes`.
pub async fn research_topic_id(pool: &PgPool, id: Uuid) -> Result<Option<Uuid>, DbError> {
let row = sqlx::query!(
"SELECT research_topic_id FROM topology_runs WHERE id = $1",
id,
)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| r.research_topic_id))
}
/// The loop this run belongs to, if any. Mirror of `research_topic_id`.
/// Used by `topology_worker` to look up the per-loop gateway URL so a
/// loop's runs land on its isolated daemon (P2). Non-loop runs return
/// None.
pub async fn loop_id_for_run(pool: &PgPool, id: Uuid) -> Result<Option<Uuid>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> =
sqlx::query("SELECT loop_id FROM topology_runs WHERE id = $1")
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| r.try_get::<Option<Uuid>, _>("loop_id").ok().flatten()))
}
/// The iteration counter for a loop-bound run. Returns None for chat /
/// research runs (iteration column is nullable). Used by the reorder
/// rationale hook so the mini-timeline can order events by iteration.
pub async fn iteration_for_run(pool: &PgPool, id: Uuid) -> Result<Option<i32>, DbError> {
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> =
sqlx::query("SELECT iteration FROM topology_runs WHERE id = $1")
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|r| r.try_get::<Option<i32>, _>("iteration").ok().flatten()))
}
/// Enqueue a durable run bound to a research topic. `research_topic_id` is
/// stored so `notify_run_completed` can flip the owning topic
/// `processing → reviewing` when its last run terminates (see
/// `topology_worker::maybe_transition_research_topic`).
pub async fn enqueue_run_for_research_topic(
pool: &PgPool,
id: Uuid,
workspace_id: WorkspaceId,
task: &str,
graph: &Value,
research_topic_id: Uuid,
) -> Result<(), DbError> {
sqlx::query!(
"INSERT INTO topology_runs
(id, workspace_id, task, kind, status, graph, tier, research_topic_id)
VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5)",
id,
workspace_id.as_uuid(),
task,
graph,
research_topic_id,
)
.execute(pool)
.await?;
Ok(())
}
/// Result of `check_ephemeral_teardown` when this run's terminal completion
/// should tear down its team.
pub struct EphemeralTeardown {
@@ -423,54 +257,6 @@ pub async fn touch(pool: &PgPool, id: Uuid) -> Result<(), DbError> {
Ok(())
}
/// If this run belongs to a research topic AND no siblings of that topic
/// are still queued or running, transition the topic `processing → reviewing`.
/// Guarded by `status = 'processing'` so a repeat call (e.g. a retry) is a
/// no-op; a topic already reviewing/publishing/published stays put.
/// Returns `true` when the topic was transitioned.
pub async fn notify_run_completed(pool: &PgPool, id: Uuid) -> Result<bool, DbError> {
// One statement: subquery locates the topic id, subquery counts siblings
// still in flight (excluding *this* run — it's about to be flipped to
// completed/failed by the caller, but ordering isn't guaranteed here).
//
// Only advance the topic when it has AT LEAST ONE outcome — otherwise a
// failed run with no synthesis would push the topic into `reviewing`,
// the UI would offer "Request publish", the user would click Approve, and
// decide_publish would 409 on the "no outcome" guard. Stays in
// `processing` when zero outcomes exist so the loop's next iteration
// still has a chance to produce one.
// Dynamic query — the added EXISTS clause on research_outcomes
// doesn't have an entry in the offline sqlx cache, so we bind
// values by hand instead of using the `query!` macro.
use sqlx::Row;
let row: Option<sqlx::postgres::PgRow> = sqlx::query(
"UPDATE research_topics t
SET status = 'reviewing', updated_at = now()
WHERE t.id = (
SELECT research_topic_id FROM topology_runs
WHERE id = $1 AND research_topic_id IS NOT NULL
)
AND t.status = 'processing'
AND EXISTS (
SELECT 1 FROM research_outcomes
WHERE topic_id = t.id
)
AND NOT EXISTS (
SELECT 1 FROM topology_runs
WHERE research_topic_id = t.id
AND id <> $1
AND status IN ('queued', 'running')
)
RETURNING t.id",
)
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row
.map(|r| r.try_get::<Uuid, _>("id").is_ok())
.unwrap_or(false))
}
/// Mark a job completed and store its final result blob.
pub async fn complete(pool: &PgPool, id: Uuid, result: &Value) -> Result<(), DbError> {
sqlx::query!(
@@ -559,7 +345,7 @@ pub async fn list_recent(
limit: i64,
) -> Result<Vec<TopologyRunSummary>, DbError> {
let rows = sqlx::query!(
"SELECT id, task, status, kind, created_at, iteration, finished_at
"SELECT id, task, status, kind, created_at, finished_at
FROM topology_runs
WHERE workspace_id = $1 ORDER BY created_at DESC LIMIT $2",
workspace_id.as_uuid(),
@@ -575,41 +361,6 @@ pub async fn list_recent(
status: r.status,
kind: r.kind,
created_at: r.created_at,
iteration: r.iteration,
finished_at: r.finished_at,
})
.collect())
}
/// Iterations of a loop, newest first. Uses the partial index
/// `topology_runs_loop_idx` on `(loop_id, iteration DESC)`.
pub async fn list_by_loop(
pool: &PgPool,
workspace_id: WorkspaceId,
loop_id: Uuid,
limit: i64,
) -> Result<Vec<TopologyRunSummary>, DbError> {
let rows = sqlx::query!(
"SELECT id, task, status, kind, created_at, iteration, finished_at
FROM topology_runs
WHERE workspace_id = $1 AND loop_id = $2
ORDER BY iteration DESC NULLS LAST, created_at DESC
LIMIT $3",
workspace_id.as_uuid(),
loop_id,
limit,
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| TopologyRunSummary {
id: r.id,
task: r.task,
status: r.status,
kind: r.kind,
created_at: r.created_at,
iteration: r.iteration,
finished_at: r.finished_at,
})
.collect())