Omar Sobh
bfcdca0583
Closes the UX gap the fold introduced: the topic canvas was still showing "Start research" for standby-state topics even when a scheduled loop already owned the runs. Clicking it would 409 (or worse: race the loop into a duplicate run). Topic status stayed at standby forever because the loop path bypassed start_topic's set_status transition. Four changes: 1. **Backend status transition** — compose_and_enqueue_iteration for kind='research' now calls set_status_if(standby, processing) on the topic before the run is enqueued. New DB helper set_status_if only advances when the current status matches the "from" arg — safe against races and re-invocations. Later iterations no-op since the topic is already past standby. 2. **has_managed_loop on TopicDetail** — get_topic hydrates a new ManagedLoop struct (loop_id, title, enabled, next_fire_at, last_run_id, schedule_summary) when a kind='research' loop is bound to the topic. summarize_schedule() derives a human string from the loop's triggers jsonb (e.g. "cron: 0 3 * * * · on new artifact", "one-shot", "manual"). New DB helper loops::research_loop_for_topic returns the row. 3. **Canvas branch** — nextAction takes a managedByLoop flag; when set + status=standby, returns null (no button). The canvas renders a "MANAGED BY LOOP" strip below the topic title showing loop name, schedule summary, next fire time, and enabled dot. Reviewer buttons (Request publish / Approve / Reject) still show normally in later states — reviewers should still promote outcomes even when a loop is producing them. 4. **start_topic guard** — refuses with 409 when a research loop already owns the topic. Closes the direct-POST hole for anyone bypassing the frontend. TS type + summarize_schedule live in the same commit so an old client hitting a new backend just ignores the extra field (no breakage), and a new client hitting an old backend renders the classic buttons (managed_by_loop is optional).
882 lines
33 KiB
Rust
882 lines
33 KiB
Rust
//! Loop endpoints — CRUD, enable/disable, immediate-run, and the public
|
|
//! webhook receiver.
|
|
//!
|
|
//! GET /api/loops list workspace's loops
|
|
//! POST /api/loops create
|
|
//! GET /api/loops/:id detail
|
|
//! PATCH /api/loops/:id update definition
|
|
//! DELETE /api/loops/:id delete
|
|
//! POST /api/loops/:id/run trigger one iteration NOW (bypass schedule)
|
|
//! POST /api/loops/:id/enable set enabled=true; recomputes next_fire_at
|
|
//! POST /api/loops/:id/disable set enabled=false
|
|
//! POST /webhooks/loops/:token public; HMAC-SHA256-verified via
|
|
//! X-Loop-Signature: sha256=<hex>
|
|
|
|
use axum::body::Bytes;
|
|
use axum::extract::{Path, State};
|
|
use axum::http::{HeaderMap, StatusCode};
|
|
use axum::Json;
|
|
use base64::Engine;
|
|
use cm_runtime::scheduling::next_occurrence;
|
|
use hmac::{Hmac, Mac};
|
|
use serde::{Deserialize, Serialize};
|
|
use serde_json::Value;
|
|
use sqlx::PgPool;
|
|
use time::OffsetDateTime;
|
|
use uuid::Uuid;
|
|
|
|
/// Root of per-loop state dirs on the host. Same overridable env pattern
|
|
/// as research_workspace_root — prod points at the bind-mounted volume
|
|
/// `/var/lib/clawmates-loops` on gw-04.
|
|
fn loop_state_root() -> std::path::PathBuf {
|
|
std::env::var("CLAWMATES_LOOPS_STATE_ROOT")
|
|
.map(std::path::PathBuf::from)
|
|
.unwrap_or_else(|_| std::path::PathBuf::from("/var/lib/clawmates-loops"))
|
|
}
|
|
|
|
/// Best-effort spawn of the per-loop team container before an iteration
|
|
/// is enqueued. Idempotent — an already-running container is just
|
|
/// reattached. Failures (docker unreachable, image missing) log and
|
|
/// return without blocking the run; the topology_worker will fall back
|
|
/// to the workspace-wide gateway. Records the container name + URL on
|
|
/// the loop row on first success so subsequent fires skip re-writing.
|
|
/// Build the task string an iteration will actually run.
|
|
///
|
|
/// - Standalone loops (no source research topic bound): returns
|
|
/// `task_template` verbatim, matching legacy behavior.
|
|
/// - Loops bound to a research topic: fetches the topic's latest
|
|
/// research_outcome and prepends a block of the shape:
|
|
///
|
|
/// ```text
|
|
/// RESEARCH ARTIFACT (integration plan you're executing):
|
|
/// <markdown>
|
|
/// ITERATION FOCUS: next unconsumed INT-XX in order. If prereqs are
|
|
/// unmet, work on the smallest unblocking INT-XX. Log
|
|
/// COMPLETED: INT-<NN> at the end so the loop can advance.
|
|
/// ORIGINAL TASK:
|
|
/// <task_template>
|
|
/// ```
|
|
///
|
|
/// The topology_worker's completion hook (P3) parses the COMPLETED
|
|
/// marker to update `consumed_int_ids`.
|
|
/// One-shot compose + enqueue for a loop iteration. Reads the loop's
|
|
/// kind from the DB and dispatches: kind='exec' uses
|
|
/// compose_iteration_task (INT-consumption prepend); kind='research'
|
|
/// uses compose_research_iteration_task AND sets research_topic_id on
|
|
/// the topology_run so freeze_research_outcome writes a new outcome
|
|
/// version at completion. Returns the run id. Standalone exec loops
|
|
/// (no source topic) still work — the compose helper returns the
|
|
/// task_template verbatim.
|
|
pub async fn compose_and_enqueue_iteration(
|
|
pool: &sqlx::PgPool,
|
|
loop_id: Uuid,
|
|
workspace_id: Uuid,
|
|
graph: &Value,
|
|
parent_run_id: Option<Uuid>,
|
|
task_template_override: Option<&str>,
|
|
) -> Result<Uuid, cm_db::DbError> {
|
|
// Kind is the source of truth — task_template alone isn't enough
|
|
// to know whether to write to research_outcomes.
|
|
let (kind, source_topic, template) =
|
|
match cm_db::repo::loops::kind_and_binding(pool, loop_id).await? {
|
|
Some(t) => t,
|
|
None => return Err(cm_db::DbError::NotFound),
|
|
};
|
|
let template_ref = task_template_override.unwrap_or(&template);
|
|
let iter = cm_db::repo::loops::next_iteration(pool, loop_id).await?;
|
|
if kind == "research" {
|
|
// Research-kind requires a bound topic (schema-level constraint
|
|
// isn't enforced yet — surface the misconfiguration explicitly).
|
|
let Some(topic_id) = source_topic else {
|
|
return Err(cm_db::DbError::NotFound);
|
|
};
|
|
// Clone + spawn container BEFORE enqueuing so the run has real
|
|
// repo files + an isolated daemon to hit. Idempotent — the
|
|
// second iteration reattaches to the existing container. Runs
|
|
// even when the topic has no repo (harmless no-op).
|
|
crate::routes::research_setup::prepare_topic_runtime(pool, workspace_id, topic_id).await;
|
|
// D1 fold — advance the topic's status column when a fresh
|
|
// research iteration goes out so the canvas's classic state-
|
|
// machine card reflects reality. Only fire the standby →
|
|
// processing transition; later iterations already sit in
|
|
// processing/reviewing/publishing and set_status is a no-op
|
|
// when the status is already the target.
|
|
let _ = cm_db::repo::research_topics::set_status_if(
|
|
pool,
|
|
topic_id,
|
|
workspace_id,
|
|
"standby",
|
|
"processing",
|
|
)
|
|
.await;
|
|
let task = compose_research_iteration_task(pool, topic_id, template_ref).await;
|
|
cm_db::repo::loops::enqueue_iteration_with_topic(
|
|
pool,
|
|
cm_db::repo::loops::IterationEnqueue {
|
|
loop_id,
|
|
workspace_id,
|
|
task: &task,
|
|
graph,
|
|
iteration: iter,
|
|
parent_run_id,
|
|
research_topic_id: Some(topic_id),
|
|
},
|
|
)
|
|
.await
|
|
} else {
|
|
let task = compose_iteration_task(pool, loop_id, template_ref).await;
|
|
cm_db::repo::loops::enqueue_iteration(
|
|
pool,
|
|
loop_id,
|
|
workspace_id,
|
|
&task,
|
|
graph,
|
|
iter,
|
|
parent_run_id,
|
|
)
|
|
.await
|
|
}
|
|
}
|
|
|
|
/// Build the coordinator prompt for a kind='research' loop iteration.
|
|
/// Wraps the topic's description + outcome_kind + prior artifact
|
|
/// version pointer into an instruction that asks the team to refresh
|
|
/// the plan (survey new sources, revise existing INTs, add new ones)
|
|
/// and emit the updated artifact using the same section shape. The
|
|
/// completion hook's `freeze_research_outcome` will insert a new
|
|
/// versioned row automatically because the topology_run carries
|
|
/// research_topic_id.
|
|
pub async fn compose_research_iteration_task(
|
|
pool: &PgPool,
|
|
topic_id: Uuid,
|
|
task_template: &str,
|
|
) -> String {
|
|
let (title, description, outcome_kind, prior_version) =
|
|
match cm_db::repo::research_topics::get_any_workspace(pool, topic_id).await {
|
|
Ok(Some(t)) => {
|
|
let prior = cm_db::repo::research_outcomes::latest(pool, topic_id)
|
|
.await
|
|
.unwrap_or(None)
|
|
.map(|o| o.version)
|
|
.unwrap_or(0);
|
|
(t.title, t.description, t.outcome_kind, prior)
|
|
}
|
|
_ => return task_template.to_string(),
|
|
};
|
|
format!(
|
|
"RESEARCH LOOP ITERATION\n\
|
|
=======================\n\
|
|
Topic: {title}\n\
|
|
Outcome kind: {outcome_kind}\n\
|
|
Prior artifact version: v{prior_version} (0 = fresh)\n\n\
|
|
DESCRIPTION:\n{description}\n\n\
|
|
AUTONOMY CONTRACT (READ FIRST):\n\
|
|
- This is a scheduled autonomous run. NO HUMAN WILL ANSWER YOU.\n\
|
|
- Do NOT ask 'Should I proceed?' or 'Which approach?' — proceed with\n\
|
|
your best judgment and produce the artifact.\n\
|
|
- You MUST emit the completed artifact as your final message.\n\
|
|
Failure to emit = the entire loop iteration is wasted.\n\n\
|
|
YOUR JOB THIS ITERATION:\n\
|
|
- Refresh the research — pull in any new papers / findings since v{prior_version}.\n\
|
|
- Update the artifact using the SAME section structure the outcome_kind\n\
|
|
requires (e.g. integrations kind = executive summary + INT-XX cards).\n\
|
|
- Preserve stable ids (INT-01 stays INT-01 across versions). If an item\n\
|
|
is superseded, mark it {{deprecated: <reason>}} rather than deleting so\n\
|
|
downstream coding loops that already consumed it don't lose context.\n\
|
|
- Add NEW items with new ids continuing from the last used number.\n\
|
|
- Cite what you can verify. When you can't cite a specific paper or\n\
|
|
benchmark, write `[claim needs verification]` inline and MOVE ON — do\n\
|
|
not stall the loop asking a human for permission. The next iteration\n\
|
|
can strengthen citations; a written v{} with rough citations beats a\n\
|
|
blocked v{} waiting for approval.\n\
|
|
- Do NOT fabricate concrete paper titles, author names, or DOIs.\n\
|
|
Vague-but-honest ('a 2024 HNSW improvement paper') beats invented specifics.\n\n\
|
|
The workspace's final synthesis is captured as research_outcomes v{}. \
|
|
Downstream on_artifact_update loops will wake up on this write.\n\n\
|
|
LOOP OPERATOR NOTES:\n{task_template}\n",
|
|
prior_version + 1,
|
|
prior_version + 1,
|
|
prior_version + 1
|
|
)
|
|
}
|
|
|
|
pub async fn compose_iteration_task(pool: &PgPool, loop_id: Uuid, task_template: &str) -> String {
|
|
let ctx = cm_db::repo::loops::source_research_context(pool, loop_id)
|
|
.await
|
|
.unwrap_or(None);
|
|
let Some((topic_id, consumed, current_idx)) = ctx else {
|
|
return task_template.to_string();
|
|
};
|
|
let outcome = match cm_db::repo::research_outcomes::latest(pool, topic_id).await {
|
|
Ok(Some(o)) => o,
|
|
_ => return task_template.to_string(),
|
|
};
|
|
let consumed_list = if consumed.is_empty() {
|
|
"(none yet)".to_string()
|
|
} else {
|
|
consumed.join(", ")
|
|
};
|
|
format!(
|
|
"RESEARCH ARTIFACT (integration plan you're executing, v{}):\n\
|
|
--- BEGIN ARTIFACT ---\n{}\n--- END ARTIFACT ---\n\n\
|
|
ITERATION FOCUS:\n\
|
|
- You are on iteration index {}.\n\
|
|
- Already completed: {}.\n\
|
|
- Address the NEXT unconsumed INT-XX item in the artifact, in order.\n\
|
|
- If the next item has unmet prerequisites, work on the smallest\n\
|
|
unblocking INT-XX instead. When you reorder, emit a line\n\
|
|
`REORDER: <one-sentence rationale>` at the top of your first\n\
|
|
substantive turn — the loop indexes these for a review timeline.\n\
|
|
- Emit `COMPLETED: INT-<NN>` on its own line at the end of the run\n\
|
|
when the item is done — the loop advances on that marker.\n\
|
|
- Both markers must appear literally with the colon (no bold, no\n\
|
|
code fence); the parser is line-based.\n\n\
|
|
ORIGINAL TASK TEMPLATE:\n{}\n",
|
|
outcome.version, outcome.body_md, current_idx, consumed_list, task_template
|
|
)
|
|
}
|
|
|
|
async fn ensure_loop_container(pool: &PgPool, workspace_id: Uuid, loop_id: Uuid) {
|
|
let docker = match crate::research_container::connect() {
|
|
Ok(d) => d,
|
|
Err(e) => {
|
|
eprintln!("loops::ensure_loop_container({loop_id}): docker connect failed: {e}");
|
|
return;
|
|
}
|
|
};
|
|
let state_root = loop_state_root().join(loop_id.to_string()).join("state");
|
|
let spawned = match crate::research_container::spawn_loop(&docker, loop_id, &state_root).await {
|
|
Ok(s) => s,
|
|
Err(e) => {
|
|
eprintln!("loops::ensure_loop_container({loop_id}): spawn failed: {e}");
|
|
return;
|
|
}
|
|
};
|
|
if let Err(e) = cm_db::repo::loops::set_zeroclaw_container(
|
|
pool,
|
|
loop_id,
|
|
workspace_id,
|
|
&spawned.name,
|
|
&spawned.gateway_url,
|
|
)
|
|
.await
|
|
{
|
|
eprintln!("loops::ensure_loop_container({loop_id}): persist failed: {e:?}");
|
|
}
|
|
}
|
|
|
|
use crate::{ApiError, AppState, Authed};
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct CreateLoopRequest {
|
|
pub title: String,
|
|
pub description: String,
|
|
pub graph: Value,
|
|
pub task_template: String,
|
|
/// {cron?: '0 */6 * * *', on_completion?: bool, webhook_enabled?: bool}
|
|
#[serde(default)]
|
|
pub triggers: Value,
|
|
/// {kind: 'infinite' | 'iters' | 'until', n?: int}
|
|
#[serde(default = "default_repeat")]
|
|
pub repeat_policy: Value,
|
|
#[serde(default)]
|
|
pub agents: Vec<AgentSlotInput>,
|
|
#[serde(default)]
|
|
pub teams: Vec<Uuid>,
|
|
#[serde(default)]
|
|
pub orgs: Vec<Uuid>,
|
|
/// Optional research topic id. When set, each iteration prepends the
|
|
/// topic's latest research_outcome markdown + a "focus on next
|
|
/// unconsumed INT" instruction to the coordinator task. Migration
|
|
/// 0042 added the pointer column + consumed_int_ids tracking.
|
|
#[serde(default)]
|
|
pub source_research_topic_id: Option<Uuid>,
|
|
}
|
|
fn default_repeat() -> Value {
|
|
serde_json::json!({"kind": "infinite"})
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct AgentSlotInput {
|
|
pub agent_id: Uuid,
|
|
#[serde(default)]
|
|
pub role_slot: Option<String>,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct LoopCreated {
|
|
pub id: Uuid,
|
|
/// Set when `triggers.webhook_enabled == true`. The full URL is
|
|
/// `<origin>/webhooks/loops/<webhook_token>`; the signing key is
|
|
/// returned exactly once at creation and never surfaced again.
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
pub webhook_token: Option<String>,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
pub webhook_signing_key: Option<String>,
|
|
}
|
|
|
|
fn parse_triggers(v: &Value) -> Option<Triggers> {
|
|
serde_json::from_value(v.clone()).ok()
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct Triggers {
|
|
#[serde(default)]
|
|
cron: Option<String>,
|
|
#[serde(default)]
|
|
#[allow(dead_code)]
|
|
on_completion: bool,
|
|
#[serde(default)]
|
|
webhook_enabled: bool,
|
|
/// NEW — number of iterations to fire back-to-back at loop-create
|
|
/// time. Enqueue path: fire once immediately, then chain each
|
|
/// subsequent one via on_completion until the burst quota is
|
|
/// exhausted (tracked in run metadata). Defaults to 0 for existing
|
|
/// loops (no auto-fire); new wizards typically set 1 (D1: every
|
|
/// runnable thing runs at least once).
|
|
#[serde(default)]
|
|
initial_burst: u32,
|
|
/// NEW — when this loop is bound to a source_research_topic and
|
|
/// that topic gets a fresh research_outcomes row (via
|
|
/// freeze_research_outcome), enqueue one iteration on this loop.
|
|
/// Coalesced with any in-flight run (D3: coordinator resolves;
|
|
/// no race, just one wake per artifact update). Read directly
|
|
/// from the loops.triggers jsonb by loops_awaiting_topic — no
|
|
/// need for the Rust parser to hold it after the fact.
|
|
#[serde(default)]
|
|
#[allow(dead_code)]
|
|
on_artifact_update: bool,
|
|
}
|
|
|
|
fn make_webhook_material() -> (String, String) {
|
|
// 24 bytes ≈ 192 bits of entropy each; URL-safe base64 for the token,
|
|
// standard base64 for the signing key.
|
|
let mut token_buf = [0u8; 24];
|
|
let mut key_buf = [0u8; 24];
|
|
// getrandom is already in the dep tree via base64/hmac/etc; failure
|
|
// (broken kernel RNG) is fatal enough that unwrapping is fine here.
|
|
getrandom::getrandom(&mut token_buf).expect("OS RNG");
|
|
getrandom::getrandom(&mut key_buf).expect("OS RNG");
|
|
let token = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(token_buf);
|
|
let key = base64::engine::general_purpose::STANDARD_NO_PAD.encode(key_buf);
|
|
(token, key)
|
|
}
|
|
|
|
fn compute_next_fire(triggers: &Value) -> Option<OffsetDateTime> {
|
|
let t = parse_triggers(triggers)?;
|
|
let pattern = t.cron?;
|
|
if pattern.trim().is_empty() {
|
|
return None;
|
|
}
|
|
next_occurrence(pattern.trim(), OffsetDateTime::now_utc()).ok()
|
|
}
|
|
|
|
pub async fn create_loop(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Json(body): Json<CreateLoopRequest>,
|
|
) -> Result<(StatusCode, Json<LoopCreated>), ApiError> {
|
|
if body.title.trim().is_empty() || body.task_template.trim().is_empty() {
|
|
return Err(ApiError::BadRequest);
|
|
}
|
|
// Empty-roster gate: a workspace with zero agents has nothing to staff
|
|
// the loop with — refuse before any DB writes. Frontend already
|
|
// disables the create button in this state; this closes the direct-POST
|
|
// hole so we don't materialize orphan loops that never fire.
|
|
if cm_db::repo::agents::count_active(&state.pool, user.workspace_id).await? == 0 {
|
|
return Err(ApiError::Conflict);
|
|
}
|
|
let webhook_enabled = parse_triggers(&body.triggers)
|
|
.map(|t| t.webhook_enabled)
|
|
.unwrap_or(false);
|
|
let (webhook_token, webhook_signing_key) = if webhook_enabled {
|
|
let (t, k) = make_webhook_material();
|
|
(Some(t), Some(k))
|
|
} else {
|
|
(None, None)
|
|
};
|
|
let next_fire_at = compute_next_fire(&body.triggers);
|
|
|
|
let id = cm_db::repo::loops::create(
|
|
&state.pool,
|
|
cm_db::repo::loops::NewLoop {
|
|
workspace_id: user.workspace_id.as_uuid(),
|
|
title: body.title.trim(),
|
|
description: body.description.trim(),
|
|
graph: &body.graph,
|
|
task_template: body.task_template.trim(),
|
|
triggers: &body.triggers,
|
|
repeat_policy: &body.repeat_policy,
|
|
enabled: true,
|
|
next_fire_at,
|
|
webhook_token: webhook_token.as_deref(),
|
|
webhook_signing_key: webhook_signing_key.as_deref(),
|
|
created_by: user.user_id.as_uuid(),
|
|
},
|
|
)
|
|
.await?;
|
|
|
|
apply_staffing(&state.pool, id, &body.agents, &body.teams, &body.orgs).await?;
|
|
|
|
// Bridge to research (option C — snapshot in task_template + save
|
|
// pointer so a refresh can pull latest artifact into subsequent
|
|
// iterations). Ownership-checked via research_topics::get so we
|
|
// can't be tricked into pointing at another workspace's topic.
|
|
if let Some(topic_id) = body.source_research_topic_id {
|
|
let topic =
|
|
cm_db::repo::research_topics::get(&state.pool, topic_id, user.workspace_id.as_uuid())
|
|
.await?
|
|
.ok_or(ApiError::NotFound)?;
|
|
if let Err(e) = cm_db::repo::loops::set_source_research_topic(
|
|
&state.pool,
|
|
id,
|
|
user.workspace_id.as_uuid(),
|
|
Some(topic.id),
|
|
)
|
|
.await
|
|
{
|
|
eprintln!("loops::create: bind source research topic failed: {e:?}");
|
|
}
|
|
}
|
|
|
|
// Fire the initial burst if the triggers request it. Extracted so
|
|
// materialize_topic_loops (the wizard-materialized loops path) can
|
|
// reuse the same logic — previously the burst logic lived only in
|
|
// this handler and wizard-created loops never fired their first
|
|
// iteration.
|
|
fire_initial_burst_if_set(
|
|
&state.pool,
|
|
user.workspace_id.as_uuid(),
|
|
id,
|
|
&body.triggers,
|
|
body.task_template.trim(),
|
|
&body.graph,
|
|
next_fire_at,
|
|
)
|
|
.await;
|
|
|
|
Ok((
|
|
StatusCode::CREATED,
|
|
Json(LoopCreated {
|
|
id,
|
|
webhook_token,
|
|
webhook_signing_key,
|
|
}),
|
|
))
|
|
}
|
|
|
|
/// Fire the initial_burst if the loop's triggers request one. On the
|
|
/// first fire, ensures the per-loop container is spawned and (for
|
|
/// kind='research' loops) that the topic's repo is cloned and the
|
|
/// topic container is up. Sets `initial_burst_remaining` to
|
|
/// `burst - 1` so the completion hook can continue the chain.
|
|
/// Best-effort: a docker or DB hiccup on the FIRST fire logs but the
|
|
/// loop row still lives — cron / on_artifact_update / webhook can
|
|
/// still fire it later.
|
|
pub async fn fire_initial_burst_if_set(
|
|
pool: &sqlx::PgPool,
|
|
workspace_id: Uuid,
|
|
loop_id: Uuid,
|
|
triggers: &Value,
|
|
task_template: &str,
|
|
graph: &Value,
|
|
next_fire_at: Option<OffsetDateTime>,
|
|
) {
|
|
let parsed = parse_triggers(triggers);
|
|
let initial_burst = parsed.as_ref().map(|t| t.initial_burst).unwrap_or(0);
|
|
let chain_on_completion = parsed.as_ref().map(|t| t.on_completion).unwrap_or(false);
|
|
if initial_burst == 0 {
|
|
return;
|
|
}
|
|
ensure_loop_container(pool, workspace_id, loop_id).await;
|
|
match compose_and_enqueue_iteration(
|
|
pool,
|
|
loop_id,
|
|
workspace_id,
|
|
graph,
|
|
None,
|
|
Some(task_template),
|
|
)
|
|
.await
|
|
{
|
|
Ok(run_id) => {
|
|
let _ = cm_db::repo::loops::mark_fired(pool, loop_id, run_id, next_fire_at).await;
|
|
let remaining = initial_burst.saturating_sub(1) as i32;
|
|
if remaining > 0 || chain_on_completion {
|
|
let _ =
|
|
cm_db::repo::loops::set_initial_burst_remaining(pool, loop_id, remaining).await;
|
|
}
|
|
}
|
|
Err(e) => eprintln!("fire_initial_burst_if_set({loop_id}): enqueue failed: {e:?}"),
|
|
}
|
|
}
|
|
|
|
async fn apply_staffing(
|
|
pool: &sqlx::PgPool,
|
|
loop_id: Uuid,
|
|
agents: &[AgentSlotInput],
|
|
teams: &[Uuid],
|
|
orgs: &[Uuid],
|
|
) -> Result<(), ApiError> {
|
|
let slots: Vec<cm_db::repo::loops::AgentSlot> = agents
|
|
.iter()
|
|
.map(|a| cm_db::repo::loops::AgentSlot {
|
|
agent_id: a.agent_id,
|
|
role_slot: a.role_slot.clone(),
|
|
})
|
|
.collect();
|
|
cm_db::repo::loops::set_agents(pool, loop_id, &slots).await?;
|
|
cm_db::repo::loops::set_teams(pool, loop_id, teams).await?;
|
|
cm_db::repo::loops::set_orgs(pool, loop_id, orgs).await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct LoopWithStaffing {
|
|
#[serde(flatten)]
|
|
pub inner: cm_db::repo::loops::Loop,
|
|
pub agents: Vec<cm_db::repo::loops::AgentSlot>,
|
|
pub teams: Vec<Uuid>,
|
|
pub orgs: Vec<Uuid>,
|
|
}
|
|
|
|
async fn hydrate_staffing(
|
|
pool: &sqlx::PgPool,
|
|
inner: cm_db::repo::loops::Loop,
|
|
) -> Result<LoopWithStaffing, ApiError> {
|
|
let id = inner.id;
|
|
let agents = cm_db::repo::loops::agents(pool, id).await?;
|
|
let teams = cm_db::repo::loops::teams(pool, id).await?;
|
|
let orgs = cm_db::repo::loops::orgs(pool, id).await?;
|
|
Ok(LoopWithStaffing {
|
|
inner,
|
|
agents,
|
|
teams,
|
|
orgs,
|
|
})
|
|
}
|
|
|
|
pub async fn list_loops(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
) -> Result<Json<Vec<LoopWithStaffing>>, ApiError> {
|
|
let loops = cm_db::repo::loops::list(&state.pool, user.workspace_id.as_uuid()).await?;
|
|
let mut out = Vec::with_capacity(loops.len());
|
|
for l in loops {
|
|
out.push(hydrate_staffing(&state.pool, l).await?);
|
|
}
|
|
Ok(Json(out))
|
|
}
|
|
|
|
pub async fn get_loop(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<Json<LoopWithStaffing>, ApiError> {
|
|
let inner = cm_db::repo::loops::get(&state.pool, id, user.workspace_id.as_uuid())
|
|
.await?
|
|
.ok_or(ApiError::NotFound)?;
|
|
Ok(Json(hydrate_staffing(&state.pool, inner).await?))
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct LoopProgress {
|
|
pub loop_id: Uuid,
|
|
pub source_topic_id: Uuid,
|
|
pub source_topic_title: String,
|
|
pub source_outcome_version: i32,
|
|
pub consumed_count: usize,
|
|
pub total_int_count: usize,
|
|
pub current_int_index: i32,
|
|
/// Recent coordinator-issued reorders on this loop — newest first,
|
|
/// capped at 5 so the sidebar card stays compact. Full history is
|
|
/// on the loop row's reorder_events column.
|
|
pub recent_reorders: Vec<serde_json::Value>,
|
|
}
|
|
|
|
/// `GET /api/loops/progress` — bulk progress read for every loop in the
|
|
/// workspace that's bound to a research topic. Skips standalone loops
|
|
/// entirely (empty entry). Parses INT-XX ids from the source outcome's
|
|
/// markdown to compute the total; consumed count comes straight from
|
|
/// `consumed_int_ids`. Used by the loops sidebar to render an
|
|
/// "N/M INTs" pill on each source-bound card.
|
|
///
|
|
/// Cost: one query for the loops list + one outcome fetch per unique
|
|
/// source topic (memoized in the loop below). No N+1 on the topic
|
|
/// lookup when many loops share a source.
|
|
pub async fn list_progress(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
) -> Result<Json<Vec<LoopProgress>>, ApiError> {
|
|
let loops = cm_db::repo::loops::list(&state.pool, user.workspace_id.as_uuid()).await?;
|
|
let mut by_topic: std::collections::HashMap<Uuid, (String, i32, usize)> =
|
|
std::collections::HashMap::new();
|
|
let mut out = Vec::new();
|
|
for l in &loops {
|
|
let ctx = match cm_db::repo::loops::source_research_context(&state.pool, l.id).await {
|
|
Ok(Some(c)) => c,
|
|
_ => continue,
|
|
};
|
|
let (topic_id, consumed, current_idx) = ctx;
|
|
let (title, version, total) = match by_topic.get(&topic_id) {
|
|
Some(cached) => cached.clone(),
|
|
None => {
|
|
// Ownership check via get + then count INTs in the latest
|
|
// outcome. Any failure downgrades to (title, 0, 0) so the
|
|
// pill still renders — showing 3/0 is better than 500ing
|
|
// the whole list.
|
|
let topic = match cm_db::repo::research_topics::get(
|
|
&state.pool,
|
|
topic_id,
|
|
user.workspace_id.as_uuid(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Some(t)) => t,
|
|
_ => continue,
|
|
};
|
|
let outcome = cm_db::repo::research_outcomes::latest(&state.pool, topic_id)
|
|
.await
|
|
.unwrap_or(None);
|
|
let (version, total) = match &outcome {
|
|
Some(o) => (o.version, count_int_ids(&o.body_md)),
|
|
None => (0, 0),
|
|
};
|
|
let cached = (topic.title.clone(), version, total);
|
|
by_topic.insert(topic_id, cached.clone());
|
|
cached
|
|
}
|
|
};
|
|
let recent = cm_db::repo::loops::recent_reorders(&state.pool, l.id, 5)
|
|
.await
|
|
.unwrap_or_default();
|
|
out.push(LoopProgress {
|
|
loop_id: l.id,
|
|
source_topic_id: topic_id,
|
|
source_topic_title: title,
|
|
source_outcome_version: version,
|
|
consumed_count: consumed.len(),
|
|
total_int_count: total,
|
|
current_int_index: current_idx,
|
|
recent_reorders: recent,
|
|
});
|
|
}
|
|
Ok(Json(out))
|
|
}
|
|
|
|
/// Count unique INT-<number> ids in a markdown blob. Case-insensitive,
|
|
/// tolerates prefixes like `### INT-01` and inline references. Same
|
|
/// permissive matcher used by the completion-marker parser, so what the
|
|
/// pill counts matches what the completion path can advance against.
|
|
fn count_int_ids(text: &str) -> usize {
|
|
let upper = text.to_ascii_uppercase();
|
|
let mut seen = std::collections::HashSet::new();
|
|
let mut i = 0;
|
|
while let Some(pos) = upper[i..].find("INT-") {
|
|
let start = i + pos + 4;
|
|
let end = start
|
|
+ upper[start..]
|
|
.chars()
|
|
.take_while(|c| c.is_ascii_digit())
|
|
.count();
|
|
if end > start {
|
|
seen.insert(upper[start..end].parse::<u32>().ok());
|
|
}
|
|
i = end.max(i + pos + 4);
|
|
}
|
|
seen.into_iter().flatten().count()
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct UpdateLoopRequest {
|
|
pub title: String,
|
|
pub description: String,
|
|
pub graph: Value,
|
|
pub task_template: String,
|
|
pub triggers: Value,
|
|
pub repeat_policy: Value,
|
|
#[serde(default)]
|
|
pub agents: Vec<AgentSlotInput>,
|
|
#[serde(default)]
|
|
pub teams: Vec<Uuid>,
|
|
#[serde(default)]
|
|
pub orgs: Vec<Uuid>,
|
|
}
|
|
|
|
pub async fn patch_loop(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<UpdateLoopRequest>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
if body.title.trim().is_empty() || body.task_template.trim().is_empty() {
|
|
return Err(ApiError::BadRequest);
|
|
}
|
|
cm_db::repo::loops::get(&state.pool, id, user.workspace_id.as_uuid())
|
|
.await?
|
|
.ok_or(ApiError::NotFound)?;
|
|
let next_fire_at = compute_next_fire(&body.triggers);
|
|
cm_db::repo::loops::update(
|
|
&state.pool,
|
|
id,
|
|
user.workspace_id.as_uuid(),
|
|
cm_db::repo::loops::UpdateLoop {
|
|
title: body.title.trim(),
|
|
description: body.description.trim(),
|
|
graph: &body.graph,
|
|
task_template: body.task_template.trim(),
|
|
triggers: &body.triggers,
|
|
repeat_policy: &body.repeat_policy,
|
|
next_fire_at,
|
|
},
|
|
)
|
|
.await?;
|
|
apply_staffing(&state.pool, id, &body.agents, &body.teams, &body.orgs).await?;
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
|
|
pub async fn delete_loop(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
cm_db::repo::loops::delete(&state.pool, id, user.workspace_id.as_uuid()).await?;
|
|
// Tear down the per-loop container (P2). Fire-and-forget: the row
|
|
// is gone, so any Docker failure is a log-line, not an API failure.
|
|
crate::research_container::teardown_loop(id).await;
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
|
|
pub async fn enable_loop(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
cm_db::repo::loops::set_enabled(&state.pool, id, user.workspace_id.as_uuid(), true).await?;
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
|
|
pub async fn disable_loop(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<StatusCode, ApiError> {
|
|
cm_db::repo::loops::set_enabled(&state.pool, id, user.workspace_id.as_uuid(), false).await?;
|
|
// Stop the per-loop container while disabled — re-enabling later will
|
|
// spawn a fresh one on the next `run_now` / webhook fire. Keeps
|
|
// paused loops from holding a docker slot.
|
|
crate::research_container::teardown_loop(id).await;
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct RunTriggered {
|
|
pub run_id: Uuid,
|
|
pub iteration: i32,
|
|
}
|
|
|
|
/// `POST /api/loops/:id/run` — enqueue one iteration NOW, bypassing the
|
|
/// scheduler and any trigger config. Iteration counter continues from
|
|
/// wherever it was; parent_run_id chains to whatever last_run_id points at.
|
|
pub async fn run_now(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<Json<RunTriggered>, ApiError> {
|
|
let l = cm_db::repo::loops::get(&state.pool, id, user.workspace_id.as_uuid())
|
|
.await?
|
|
.ok_or(ApiError::NotFound)?;
|
|
// P2: spawn the per-loop container before enqueue so topology_worker
|
|
// resolves its gateway URL when it picks up the run. Best-effort;
|
|
// never blocks the enqueue on Docker being unreachable.
|
|
ensure_loop_container(&state.pool, l.workspace_id, l.id).await;
|
|
// Kind-aware — research loops write to research_outcomes.
|
|
let iter = cm_db::repo::loops::next_iteration(&state.pool, l.id).await?;
|
|
let run_id = compose_and_enqueue_iteration(
|
|
&state.pool,
|
|
l.id,
|
|
l.workspace_id,
|
|
&l.graph,
|
|
l.last_run_id,
|
|
Some(&l.task_template),
|
|
)
|
|
.await
|
|
.map_err(|_| ApiError::Internal)?;
|
|
cm_db::repo::loops::mark_fired(&state.pool, l.id, run_id, l.next_fire_at).await?;
|
|
Ok(Json(RunTriggered {
|
|
run_id,
|
|
iteration: iter,
|
|
}))
|
|
}
|
|
|
|
/// `POST /webhooks/loops/:token` — public, HMAC-verified. Enqueues one
|
|
/// iteration on the loop that owns `token`. Returns 202 + `{run_id}` on
|
|
/// success, 401 on missing/bad signature, 404 on unknown token.
|
|
pub async fn webhook_receive(
|
|
State(state): State<AppState>,
|
|
Path(token): Path<String>,
|
|
headers: HeaderMap,
|
|
body: Bytes,
|
|
) -> (StatusCode, Json<Value>) {
|
|
let Ok(Some((_id, _ws, key, l))) =
|
|
cm_db::repo::loops::get_by_webhook_token(&state.pool, &token).await
|
|
else {
|
|
return (StatusCode::NOT_FOUND, Json(Value::Null));
|
|
};
|
|
|
|
let Some(sig_header) = headers
|
|
.get("X-Loop-Signature")
|
|
.and_then(|v| v.to_str().ok())
|
|
else {
|
|
return (StatusCode::UNAUTHORIZED, Json(Value::Null));
|
|
};
|
|
let Some(provided) = sig_header.strip_prefix("sha256=") else {
|
|
return (StatusCode::UNAUTHORIZED, Json(Value::Null));
|
|
};
|
|
if !verify_hmac(&key, &body, provided) {
|
|
return (StatusCode::UNAUTHORIZED, Json(Value::Null));
|
|
}
|
|
|
|
ensure_loop_container(&state.pool, l.workspace_id, l.id).await;
|
|
let iter = match cm_db::repo::loops::next_iteration(&state.pool, l.id).await {
|
|
Ok(n) => n,
|
|
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(Value::Null)),
|
|
};
|
|
let run_id = match compose_and_enqueue_iteration(
|
|
&state.pool,
|
|
l.id,
|
|
l.workspace_id,
|
|
&l.graph,
|
|
l.last_run_id,
|
|
Some(&l.task_template),
|
|
)
|
|
.await
|
|
{
|
|
Ok(r) => r,
|
|
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(Value::Null)),
|
|
};
|
|
let _ = cm_db::repo::loops::mark_fired(&state.pool, l.id, run_id, None).await;
|
|
|
|
(
|
|
StatusCode::ACCEPTED,
|
|
Json(serde_json::json!({"run_id": run_id, "iteration": iter})),
|
|
)
|
|
}
|
|
|
|
fn verify_hmac(key: &str, body: &[u8], provided_hex: &str) -> bool {
|
|
let Ok(mut mac) = Hmac::<sha2::Sha256>::new_from_slice(key.as_bytes()) else {
|
|
return false;
|
|
};
|
|
mac.update(body);
|
|
let expected = hex::encode(mac.finalize().into_bytes());
|
|
if expected.len() != provided_hex.len() {
|
|
return false;
|
|
}
|
|
// Constant-time compare.
|
|
expected
|
|
.bytes()
|
|
.zip(provided_hex.bytes())
|
|
.fold(0u8, |acc, (a, b)| acc | (a ^ b))
|
|
== 0
|
|
}
|