research: fix wizard loops never firing + missing clone/spawn (regression)
Two bugs surfaced by the first end-to-end wizard run — the pipeline diagnostic showed "0 run(s)", "Repo bound but never cloned", and "Container not spawned" for a topic that had been created with a nightly research loop. Bug 1: materialize_topic_loops bypassed initial_burst firing. The wizard-materialized loops path calls cm_db::repo::loops::create directly (a plain INSERT). The initial_burst-fires-first-iteration logic lived inside the create_loop HTTP handler, so wizard loops landed in the DB but never fired their initial iteration. Fix: extract routes::loops::fire_initial_burst_if_set as a pub helper that read triggers, ensures the loop container, and calls the kind-aware compose_and_enqueue_iteration. Both create_loop and materialize_topic_loops now call it. Bug 2: research-kind loop iterations skipped clone/spawn. compose_research_iteration_task only built the coordinator prompt; the repo clone and topic container spawn lived only in start_topic. So the first research iteration ran against a nonexistent clone directory and a stale gateway, and every run failed. Fix: extract routes::research::prepare_topic_runtime as a pub helper that runs ensure_repo_workspace + research_container::spawn. Idempotent — second iteration reattaches. Called from compose_and_enqueue_iteration before enqueuing a research iteration. Topics without a repo bound are a no-op. Both fixes ship as one commit because they surface together on the same user path (wizard → research loop → first iteration) — you can't hit one without the other manifesting. Follow-up: start_topic still runs its own inline clone/spawn code (now duplicated with prepare_topic_runtime). Next commit collapses start_topic to just call prepare_topic_runtime + build_task like the loop path does, so the one-shot and loop paths agree on setup.
This commit is contained in:
@@ -86,6 +86,11 @@ pub async fn compose_and_enqueue_iteration(
|
|||||||
let Some(topic_id) = source_topic else {
|
let Some(topic_id) = source_topic else {
|
||||||
return Err(cm_db::DbError::NotFound);
|
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::prepare_topic_runtime(pool, workspace_id, topic_id).await;
|
||||||
let task = compose_research_iteration_task(pool, topic_id, template_ref).await;
|
let task = compose_research_iteration_task(pool, topic_id, template_ref).await;
|
||||||
cm_db::repo::loops::enqueue_iteration_with_topic(
|
cm_db::repo::loops::enqueue_iteration_with_topic(
|
||||||
pool,
|
pool,
|
||||||
@@ -400,47 +405,21 @@ pub async fn create_loop(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// initial_burst — fire the first iteration inline, so every loop
|
// Fire the initial burst if the triggers request it. Extracted so
|
||||||
// runs at least once when the trigger requests it (D1). The burst
|
// materialize_topic_loops (the wizard-materialized loops path) can
|
||||||
// continues via the on_completion trigger up to `initial_burst`
|
// reuse the same logic — previously the burst logic lived only in
|
||||||
// (tracked in loops.initial_burst_remaining, decremented on each
|
// this handler and wizard-created loops never fired their first
|
||||||
// completion). Best-effort: a docker/DB failure on the FIRST fire
|
// iteration.
|
||||||
// logs but the loop still lives with next_fire_at set for its
|
fire_initial_burst_if_set(
|
||||||
// cron.
|
&state.pool,
|
||||||
let parsed = parse_triggers(&body.triggers);
|
user.workspace_id.as_uuid(),
|
||||||
let initial_burst = parsed.as_ref().map(|t| t.initial_burst).unwrap_or(0);
|
id,
|
||||||
let chain_on_completion = parsed.as_ref().map(|t| t.on_completion).unwrap_or(false);
|
&body.triggers,
|
||||||
if initial_burst > 0 {
|
body.task_template.trim(),
|
||||||
ensure_loop_container(&state.pool, user.workspace_id.as_uuid(), id).await;
|
&body.graph,
|
||||||
// Kind-aware compose + enqueue — research-kind loops get a
|
next_fire_at,
|
||||||
// research prompt and research_topic_id set on the run.
|
)
|
||||||
let template = body.task_template.trim();
|
.await;
|
||||||
match compose_and_enqueue_iteration(
|
|
||||||
&state.pool,
|
|
||||||
id,
|
|
||||||
user.workspace_id.as_uuid(),
|
|
||||||
&body.graph,
|
|
||||||
None,
|
|
||||||
Some(template),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
Ok(run_id) => {
|
|
||||||
let _ = cm_db::repo::loops::mark_fired(&state.pool, id, run_id, next_fire_at).await;
|
|
||||||
// Log the burst remaining so on_completion can advance it.
|
|
||||||
// Stored on loops.initial_burst_remaining (added by the
|
|
||||||
// same migration as loops.kind). Falls back silently on
|
|
||||||
// schema absence so we don't break dev DBs mid-rollout.
|
|
||||||
let remaining = initial_burst.saturating_sub(1) as i32;
|
|
||||||
if remaining > 0 || chain_on_completion {
|
|
||||||
let _ =
|
|
||||||
cm_db::repo::loops::set_initial_burst_remaining(&state.pool, id, remaining)
|
|
||||||
.await;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Err(e) => eprintln!("loops::create: initial_burst enqueue failed: {e:?}"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok((
|
Ok((
|
||||||
StatusCode::CREATED,
|
StatusCode::CREATED,
|
||||||
@@ -452,6 +431,52 @@ pub async fn create_loop(
|
|||||||
))
|
))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// 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 = 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(
|
async fn apply_staffing(
|
||||||
pool: &sqlx::PgPool,
|
pool: &sqlx::PgPool,
|
||||||
loop_id: Uuid,
|
loop_id: Uuid,
|
||||||
|
|||||||
@@ -329,6 +329,68 @@ fn research_workspace_root() -> std::path::PathBuf {
|
|||||||
std::env::temp_dir().join("clawmates-research")
|
std::env::temp_dir().join("clawmates-research")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Set up the on-disk workspace + container for a research topic —
|
||||||
|
/// clone repo (idempotent) + spawn ZeroClaw team container (idempotent).
|
||||||
|
/// Callable from both the one-shot `start_topic` handler and the
|
||||||
|
/// kind='research' loop iteration path in routes::loops. Fully
|
||||||
|
/// best-effort: any failure (docker unreachable, no clone_url) logs
|
||||||
|
/// and returns, letting the caller enqueue the run against the
|
||||||
|
/// workspace-wide gateway instead.
|
||||||
|
pub async fn prepare_topic_runtime(pool: &sqlx::PgPool, workspace_id: Uuid, topic_id: Uuid) {
|
||||||
|
let topic = match cm_db::repo::research_topics::get_any_workspace(pool, topic_id).await {
|
||||||
|
Ok(Some(t)) => t,
|
||||||
|
_ => return,
|
||||||
|
};
|
||||||
|
// Repo binding is optional; without it we just skip clone + spawn.
|
||||||
|
let Some(repo_id) = topic.repo_id else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let repo =
|
||||||
|
match cm_db::repo::repos::get(pool, repo_id, cm_domain::WorkspaceId::from(workspace_id))
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
Ok(r) => r,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("prepare_topic_runtime({topic_id}): repo fetch failed: {e:?}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let ctx = match ensure_repo_workspace(pool, topic_id, workspace_id, &repo, &topic).await {
|
||||||
|
Ok(c) => c,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("prepare_topic_runtime({topic_id}): clone failed: {e}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let repo_path = std::path::PathBuf::from(&ctx.path);
|
||||||
|
let state_root = research_workspace_root()
|
||||||
|
.join(topic_id.to_string())
|
||||||
|
.join("state");
|
||||||
|
let docker = match crate::research_container::connect() {
|
||||||
|
Ok(d) => d,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("prepare_topic_runtime({topic_id}): docker connect failed: {e}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
match crate::research_container::spawn(&docker, topic_id, &repo_path, &state_root).await {
|
||||||
|
Ok(spawned) => {
|
||||||
|
if let Err(e) = cm_db::repo::research_topics::set_zeroclaw_container(
|
||||||
|
pool,
|
||||||
|
topic_id,
|
||||||
|
workspace_id,
|
||||||
|
Some(&spawned.name),
|
||||||
|
Some(&spawned.gateway_url),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
eprintln!("prepare_topic_runtime({topic_id}): persist container failed: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(e) => eprintln!("prepare_topic_runtime({topic_id}): spawn failed: {e}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Clone the bound repo (shallow, single branch) into a per-topic
|
/// Clone the bound repo (shallow, single branch) into a per-topic
|
||||||
/// workspace and gather a tree preview for the coordinator prompt.
|
/// workspace and gather a tree preview for the coordinator prompt.
|
||||||
/// Persists the clone path on the topic so a re-start reuses it instead
|
/// Persists the clone path on the topic so a re-start reuses it instead
|
||||||
@@ -583,6 +645,21 @@ async fn materialize_topic_loops(
|
|||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
let _ = cm_db::repo::loops::set_kind(pool, loop_id, "research").await;
|
let _ = cm_db::repo::loops::set_kind(pool, loop_id, "research").await;
|
||||||
|
// Fire the initial burst NOW (this is what create_loop's
|
||||||
|
// handler does after inserting the row; materialize_topic_loops
|
||||||
|
// bypasses that handler). Without this call the loop lands
|
||||||
|
// in the DB but its initial_burst=1 never fires and the
|
||||||
|
// diagnostic shows "0 runs".
|
||||||
|
crate::routes::loops::fire_initial_burst_if_set(
|
||||||
|
pool,
|
||||||
|
workspace_id,
|
||||||
|
loop_id,
|
||||||
|
&r_triggers,
|
||||||
|
"Refresh the topic's research per the outcome kind.",
|
||||||
|
&graph,
|
||||||
|
next_fire_at,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
eprintln!("materialize_topic_loops: research loop create failed: {e:?}");
|
eprintln!("materialize_topic_loops: research loop create failed: {e:?}");
|
||||||
@@ -621,6 +698,20 @@ async fn materialize_topic_loops(
|
|||||||
Some(topic_id),
|
Some(topic_id),
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
|
// Fire the coding loop's initial burst too. It's
|
||||||
|
// typically 1 (single wake to consume the first
|
||||||
|
// artifact) and on_artifact_update handles subsequent
|
||||||
|
// waves via the fan-out hook.
|
||||||
|
crate::routes::loops::fire_initial_burst_if_set(
|
||||||
|
pool,
|
||||||
|
workspace_id,
|
||||||
|
loop_id,
|
||||||
|
&c_triggers,
|
||||||
|
"Execute the next unconsumed INT-XX from the artifact.",
|
||||||
|
&graph,
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
eprintln!("materialize_topic_loops: coding loop create failed: {e:?}");
|
eprintln!("materialize_topic_loops: coding loop create failed: {e:?}");
|
||||||
|
|||||||
Reference in New Issue
Block a user