loops: bridge research artifact into loop iterations (option C + b)
ci / gates (push) Successful in 7s
ci / frontend (push) Successful in 37s
ci / rust (push) Failing after 13m41s
ci / e2e (push) Has been skipped
ci / publish (push) Has been skipped

The bridge lets a coding loop "consume" an integrations research
artifact one INT-XX item per iteration. Options b (order-sequential
iteration) and C (snapshot in task_template + save the pointer for
future refresh) from the design discussion.

Migration 0042 — three new loops columns:
- source_research_topic_id — nullable pointer to research_topics.
- consumed_int_ids TEXT[]   — INT-XX ids the loop has completed.
  Advances when topology_worker parses "COMPLETED: INT-<NN>" markers
  from the run's final output (wired in a follow-up commit).
- current_int_index INT     — monotonic pointer for order-sequential
  iteration. Coordinator addresses INT-<current+1> unless prereqs are
  unmet, in which case it works on the smallest unblocking INT-XX and
  logs the reorder rationale.

Backend:
- cm_db::repo::loops::set_source_research_topic — bind/unbind pointer.
- cm_db::repo::loops::source_research_context   — read pointer + state.
- routes::loops::compose_iteration_task         — new caller-side helper
  that reads the pointer, fetches the topic's latest research_outcome,
  and prepends the artifact + focus instruction to task_template.
- run_now + webhook_receive both pass task_template through
  compose_iteration_task before enqueue. Standalone loops (no pointer)
  behave identically to before.
- CreateLoopRequest accepts `source_research_topic_id`, ownership-
  checked via research_topics::get before persist.

Frontend:
- New ResearchArtifactPicker modal — lists published topics, fetches
  the artifact on pick, returns (topic_id, markdown) to caller.
- LoopsWizard task_template step gains "Import from research artifact"
  button (right-aligned). Click opens the picker. On pick: task
  populates with the artifact markdown, pointer saved, textarea
  expands to 8 rows, small info strip shows "Loop is bound to topic
  <id>. Each iteration will focus on the next unconsumed INT-XX."
- Unlink button reverts to standalone loop mode.

Follow-up (next commit):
- topology_worker completion hook — parse "COMPLETED: INT-<NN>" out
  of the run's final output + update consumed_int_ids +
  current_int_index atomically. Without this, current_int_index stays
  at 0 forever and every iteration works on the same INT.
- Loop card refresh button — re-read source topic's latest outcome
  (useful after a reject-with-revision cycle on the source topic).
This commit is contained in:
Omar Sobh
2026-07-09 18:47:17 -07:00
parent 3ed1d03d2b
commit ce73abe5ab
6 changed files with 428 additions and 4 deletions
+79 -2
View File
@@ -40,6 +40,54 @@ fn loop_state_root() -> std::path::PathBuf {
/// 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
/// 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`.
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 AND log the reorder rationale in your\n\
opening turn so the reviewer can trace it.\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\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,
@@ -89,6 +137,12 @@ pub struct CreateLoopRequest {
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"})
@@ -198,6 +252,27 @@ pub async fn create_loop(
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:?}");
}
}
Ok((
StatusCode::CREATED,
Json(LoopCreated {
@@ -380,11 +455,12 @@ pub async fn run_now(
// never blocks the enqueue on Docker being unreachable.
ensure_loop_container(&state.pool, l.workspace_id, l.id).await;
let iter = cm_db::repo::loops::next_iteration(&state.pool, l.id).await?;
let task = compose_iteration_task(&state.pool, l.id, &l.task_template).await;
let run_id = cm_db::repo::loops::enqueue_iteration(
&state.pool,
l.id,
l.workspace_id,
&l.task_template,
&task,
&l.graph,
iter,
l.last_run_id,
@@ -430,11 +506,12 @@ pub async fn webhook_receive(
Ok(n) => n,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(Value::Null)),
};
let task = compose_iteration_task(&state.pool, l.id, &l.task_template).await;
let run_id = match cm_db::repo::loops::enqueue_iteration(
&state.pool,
l.id,
l.workspace_id,
&l.task_template,
&task,
&l.graph,
iter,
l.last_run_id,