research: build real topology graph in materialize_topic_loops
The wizard-created research loop's first iteration failed with
"missing or invalid graph" — materialize_topic_loops was writing
`{nodes: [], edges: []}` as a placeholder, which the topology worker
rejects. Also explains why prepare_topic_runtime hadn't cloned the
repo or spawned the container: the run failed before
compose_and_enqueue_iteration got to call it.
Fix: extract build_topic_graph_json() into research_setup.rs — same
shape start_topic uses (hydrate roster, promote a role_slot-tagged
coordinator to index 0 for hub_spoke/hierarchical/star_moe, build
via cm_topology::build, serialize via cm_topology::to_json). Called
from materialize_topic_loops instead of the empty placeholder.
Fully best-effort. Any DB/topology failure falls back to a
single-node hub graph so the loop still runs (degraded, but not
silently broken).
The next wizard-created topic should now:
1. Materialize the research loop with a valid graph
2. Fire the initial burst
3. compose_and_enqueue_iteration calls prepare_topic_runtime →
clones the repo, spawns the container
4. Enqueues a run whose graph parse succeeds
5. Run drives to completion, produces an outcome
This commit is contained in:
@@ -209,6 +209,99 @@ pub async fn ensure_repo_workspace(
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Build a topology graph JSON for a research topic — same shape
|
||||||
|
/// start_topic uses (roster with coordinator promotion, topology-kind
|
||||||
|
/// aware role labeling, cm_topology::build). Called from
|
||||||
|
/// materialize_topic_loops so wizard-created research loops carry a
|
||||||
|
/// valid graph on their topology_run rows; without this the topology
|
||||||
|
/// worker rejects the run with `missing or invalid graph`.
|
||||||
|
///
|
||||||
|
/// Best-effort — returns a minimal fallback (single-node hub) on any
|
||||||
|
/// DB / topology-build failure so the loop still runs (degraded, but
|
||||||
|
/// not silently broken).
|
||||||
|
pub async fn build_topic_graph_json(pool: &PgPool, topic_id: Uuid) -> serde_json::Value {
|
||||||
|
use serde_json::json;
|
||||||
|
let topic = match cm_db::repo::research_topics::get_any_workspace(pool, topic_id).await {
|
||||||
|
Ok(Some(t)) => t,
|
||||||
|
_ => {
|
||||||
|
return json!({ "nodes": [{"id": "hub", "role": "coordinator", "attrs": {}}], "edges": [] })
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let slots = cm_db::repo::research_topics::agents(pool, topic_id)
|
||||||
|
.await
|
||||||
|
.unwrap_or_default();
|
||||||
|
if slots.is_empty() {
|
||||||
|
return json!({ "nodes": [{"id": "hub", "role": "coordinator", "attrs": {}}], "edges": [] });
|
||||||
|
}
|
||||||
|
let mut roster: Vec<(cm_db::repo::research_topics::AgentSlot, cm_domain::Agent)> = Vec::new();
|
||||||
|
for s in &slots {
|
||||||
|
if let Ok(agent) =
|
||||||
|
cm_db::repo::agents::get(pool, cm_domain::AgentId::from(s.agent_id)).await
|
||||||
|
{
|
||||||
|
roster.push((s.clone(), agent));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if roster.is_empty() {
|
||||||
|
return json!({ "nodes": [{"id": "hub", "role": "coordinator", "attrs": {}}], "edges": [] });
|
||||||
|
}
|
||||||
|
let topo: cm_topology::TopologyKind =
|
||||||
|
serde_json::from_value(json!(topic.topology_kind.as_str()))
|
||||||
|
.unwrap_or(cm_topology::TopologyKind::HubSpoke);
|
||||||
|
let is_pipeline = matches!(topo, cm_topology::TopologyKind::Pipeline);
|
||||||
|
if !is_pipeline {
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let head_label = if is_pipeline {
|
||||||
|
"stage 1"
|
||||||
|
} else {
|
||||||
|
"coordinator"
|
||||||
|
};
|
||||||
|
let roles: Vec<String> = roster
|
||||||
|
.iter()
|
||||||
|
.enumerate()
|
||||||
|
.map(|(i, (s, a))| {
|
||||||
|
if i == 0 {
|
||||||
|
head_label.to_string()
|
||||||
|
} else if let Some(r) = &s.role_slot {
|
||||||
|
r.clone()
|
||||||
|
} else if !a.job_title.is_empty() {
|
||||||
|
a.job_title.clone()
|
||||||
|
} else if is_pipeline {
|
||||||
|
format!("stage {}", i + 1)
|
||||||
|
} else {
|
||||||
|
"specialist".to_string()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
let role_refs: Vec<&str> = roles.iter().map(|s| s.as_str()).collect();
|
||||||
|
let graph = match cm_topology::build(topo, &role_refs) {
|
||||||
|
Ok(g) => g,
|
||||||
|
Err(_) => {
|
||||||
|
return json!({ "nodes": [{"id": "hub", "role": "coordinator", "attrs": {}}], "edges": [] })
|
||||||
|
}
|
||||||
|
};
|
||||||
|
match cm_topology::to_json(&graph)
|
||||||
|
.ok()
|
||||||
|
.and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
|
||||||
|
{
|
||||||
|
Some(v) => v,
|
||||||
|
None => {
|
||||||
|
json!({ "nodes": [{"id": "hub", "role": "coordinator", "attrs": {}}], "edges": [] })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Creates the paired research + optional coding loops for a topic
|
/// Creates the paired research + optional coding loops for a topic
|
||||||
/// (D1 fold). Fails soft — logs and returns, letting the topic land
|
/// (D1 fold). Fails soft — logs and returns, letting the topic land
|
||||||
/// even if loop creation stumbles. Skips the empty-roster gate
|
/// even if loop creation stumbles. Skips the empty-roster gate
|
||||||
@@ -223,7 +316,10 @@ pub async fn materialize_topic_loops(
|
|||||||
also_coding: bool,
|
also_coding: bool,
|
||||||
) {
|
) {
|
||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
let graph = json!({ "nodes": [], "edges": [] });
|
// Build a valid topology graph up front — an empty {nodes: [],
|
||||||
|
// edges: []} placeholder was rejected by the topology worker with
|
||||||
|
// "missing or invalid graph".
|
||||||
|
let graph = build_topic_graph_json(pool, topic_id).await;
|
||||||
|
|
||||||
let (r_triggers, next_fire_at) = match mode {
|
let (r_triggers, next_fire_at) = match mode {
|
||||||
"nightly" => (
|
"nightly" => (
|
||||||
|
|||||||
Reference in New Issue
Block a user