research: wire start_topic to actually run the pipeline
ci / gates (push) Successful in 7s
ci / frontend (push) Successful in 25s
ci / rust (push) Successful in 3m4s
ci / e2e (push) Has been skipped
ci / publish (push) Successful in 2m34s

The prior start_topic only flipped the status column — no work was
enqueued. Now clicking "Start research" actually launches the
assigned agents through the orchestrator.

- start_topic loads the topic + its research_topic_agents, picks a
  coordinator (first slot with role_slot containing "coordinator";
  else the first slot), and swaps it to index 0.
- Builds a hub_spoke topology graph via cm_topology::build with
  roles = [coordinator, spoke1, spoke2, …]. hub_spoke wires edges
  from the hub to every spoke and back, so the coordinator can
  address any specialist per turn.
- Assembles a coordinator prompt from the topic's title,
  description, outcome_kind, and a roster line for each teammate —
  so the coordinator knows who's on the team and what each does.
- Enqueues via a new topology_runs helper
  enqueue_run_for_research_topic that stores research_topic_id on
  the run row. `topology_worker::maybe_transition_research_topic`
  → `notify_run_completed` already picks up on that back-ref and
  flips the topic processing → reviewing when the last run
  terminates — that path was dead code until now.
- Per-agent activity streams into each claw's card for free:
  the orchestrator journals turn events into run_events; the
  existing /api/world/live SSE normalizer emits
  agent.reasoning.delta / agent.tool.call / agent.task.update
  keyed by agent id, which ClawCommandCenter is already
  subscribed to.

The `published` terminal state is still unreached (that's the
"publishing → published + artifact" step from the earlier
walkthrough — separate follow-up).
This commit is contained in:
Omar Sobh
2026-07-08 16:25:24 -07:00
parent 81a436d221
commit 7e2b02d8bb
2 changed files with 130 additions and 3 deletions
+103 -3
View File
@@ -232,9 +232,15 @@ pub async fn detach_agent(
Ok(StatusCode::NO_CONTENT)
}
/// `POST /api/research/:id/start` — flips status from `standby` to
/// `processing`. Only valid transition on this endpoint; the reviewing +
/// publishing hops come from the orchestrator and the publish approval gate.
/// `POST /api/research/:id/start` — enqueue the research pipeline and flip
/// status `standby → processing`. Builds a hub_spoke topology graph over the
/// topic's assigned agents (first agent with role_slot="coordinator" becomes
/// the hub; if none is designated the first assigned agent takes that role).
/// The task string is a coordinator prompt derived from the topic's title,
/// description, outcome_kind, and roster. The orchestrator drives the run;
/// `topology_worker::maybe_transition_research_topic` flips the topic to
/// `reviewing` when the last run terminates. Per-agent activity streams via
/// `run_events` and lands on each claw's card through `/api/world/live`.
pub async fn start_topic(
State(state): State<AppState>,
Authed(user): Authed,
@@ -246,6 +252,100 @@ pub async fn start_topic(
if topic.status != "standby" {
return Err(ApiError::Conflict);
}
let slots = cm_db::repo::research_topics::agents(&state.pool, id).await?;
if slots.is_empty() {
return Err(ApiError::BadRequest);
}
// Hydrate names + job titles so the coordinator prompt names each teammate.
let mut roster = Vec::with_capacity(slots.len());
for s in &slots {
let agent =
cm_db::repo::agents::get(&state.pool, cm_domain::AgentId::from(s.agent_id)).await?;
roster.push((s.clone(), agent));
}
// Coordinator = first slot whose role_slot mentions "coordinator" (case-
// insensitive), else the first slot. The coordinator becomes the hub of
// the hub_spoke graph, so it can talk to every other agent per-turn.
let coord_ix = roster
.iter()
.position(|(s, _)| {
s.role_slot
.as_deref()
.map(|r| r.to_ascii_lowercase().contains("coordinator"))
.unwrap_or(false)
})
.unwrap_or(0);
if coord_ix != 0 {
roster.swap(0, coord_ix);
}
// Build a role list where element 0 is the coordinator (hub) and the rest
// are spokes. cm_topology's hub_spoke builder wires edges hub↔every spoke.
let roles: Vec<String> = roster
.iter()
.enumerate()
.map(|(i, (s, a))| {
if i == 0 {
"coordinator".to_string()
} else if let Some(r) = &s.role_slot {
r.clone()
} else if !a.job_title.is_empty() {
a.job_title.clone()
} else {
"specialist".to_string()
}
})
.collect();
let role_refs: Vec<&str> = roles.iter().map(|s| s.as_str()).collect();
let graph = cm_topology::build(cm_topology::TopologyKind::HubSpoke, &role_refs)
.map_err(|_| ApiError::BadRequest)?;
let graph_json_str = cm_topology::to_json(&graph).map_err(|_| ApiError::BadRequest)?;
let graph_value: serde_json::Value =
serde_json::from_str(&graph_json_str).map_err(|_| ApiError::BadRequest)?;
// Coordinator prompt: topic framing + roster + delegation instruction.
let roster_lines = roster
.iter()
.enumerate()
.map(|(i, (s, a))| {
let role = if i == 0 {
"coordinator (you)".to_string()
} else if let Some(r) = &s.role_slot {
r.clone()
} else if !a.job_title.is_empty() {
a.job_title.clone()
} else {
"specialist".to_string()
};
format!("- {}{}", a.name, role)
})
.collect::<Vec<_>>()
.join("\n");
let task = format!(
"RESEARCH TOPIC: {title}\n\
OUTCOME KIND: {outcome} (spec / prod_plan / roadmap / paper)\n\n\
DESCRIPTION:\n{description}\n\n\
TEAM (hub_spoke — you are the hub, the rest are spokes you can address per-turn):\n{roster}\n\n\
YOUR JOB (coordinator):\n\
1. Break the topic into concrete sub-tasks and assign each to the best-fit spoke.\n\
2. Delegate turn-by-turn: each spoke's response feeds your next dispatch.\n\
3. Synthesize their outputs into a single {outcome} that satisfies the description.\n\
4. Cite each spoke's contribution where it lands in the final artifact.",
title = topic.title,
outcome = topic.outcome_kind,
description = topic.description,
roster = roster_lines,
);
let run_id = uuid::Uuid::now_v7();
cm_db::repo::topology_runs::enqueue_run_for_research_topic(
&state.pool,
run_id,
user.workspace_id,
&task,
&graph_value,
id,
)
.await?;
cm_db::repo::research_topics::set_status(
&state.pool,
id,