research: split pipeline diagnostics into research_pipeline.rs
Gates step failed on run#211 because research.rs grew to 1313 lines (over the 1250 budget) after adding the pipeline_state handler in the previous commit. Move the diagnostic into its own module — pure extraction, no behavior change. research.rs: 1313 → 1126 lines research_pipeline.rs: new, 187 lines lib.rs: route now points at routes::research_pipeline::pipeline_state
This commit is contained in:
Omar Sobh
@@ -19,6 +19,7 @@ pub mod orgs;
|
||||
pub mod planner;
|
||||
pub mod repos;
|
||||
pub mod research;
|
||||
pub mod research_pipeline;
|
||||
pub mod routines;
|
||||
pub mod sessions;
|
||||
pub mod skills;
|
||||
|
||||
@@ -1027,196 +1027,9 @@ pub async fn get_artifact(
|
||||
))
|
||||
}
|
||||
|
||||
// ── pipeline diagnostics ──────────────────────────────────────────────────
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct PipelineStage {
|
||||
/// Machine-readable stage id: staffing / repo / container / runs /
|
||||
/// outcomes / approval. Frontend uses this to key the checklist.
|
||||
pub key: String,
|
||||
/// User-facing one-line summary.
|
||||
pub label: String,
|
||||
/// ok | warn | fail | skip — drives the pill color in the UI.
|
||||
pub status: &'static str,
|
||||
/// Optional error text (last-known failure reason from the underlying
|
||||
/// row) so the user can see WHY a stage failed instead of a green tick
|
||||
/// with no artifact behind it.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub detail: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct PipelineState {
|
||||
pub topic_id: Uuid,
|
||||
pub status: String,
|
||||
pub stages: Vec<PipelineStage>,
|
||||
}
|
||||
|
||||
/// `GET /api/research/:id/pipeline-state` — read-only report that walks
|
||||
/// the pipeline stages for a topic and returns per-stage status + any
|
||||
/// captured error text. Purpose: give users (and diagnostics tooling)
|
||||
/// end-to-end visibility so silent failures like "run failed with 0
|
||||
/// outcomes but topic auto-transitioned" are surfaced instead of buried
|
||||
/// in an empty artifact download.
|
||||
///
|
||||
/// Every stage runs in isolation and never fails the endpoint — this is
|
||||
/// a diagnostic, not a workflow gate. Missing rows show as skip/warn so
|
||||
/// the frontend can render the whole chain even when the topic is
|
||||
/// mid-pipeline.
|
||||
pub async fn pipeline_state(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(id): Path<Uuid>,
|
||||
) -> Result<Json<PipelineState>, ApiError> {
|
||||
let topic = cm_db::repo::research_topics::get(&state.pool, id, user.workspace_id.as_uuid())
|
||||
.await?
|
||||
.ok_or(ApiError::NotFound)?;
|
||||
let mut stages = Vec::new();
|
||||
|
||||
// 1. staffing — the workspace needs agents assigned to this topic.
|
||||
let agents = cm_db::repo::research_topics::agents(&state.pool, id)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
stages.push(PipelineStage {
|
||||
key: "staffing".into(),
|
||||
label: format!("{} agent(s) assigned", agents.len()),
|
||||
status: if agents.is_empty() { "fail" } else { "ok" },
|
||||
detail: None,
|
||||
});
|
||||
|
||||
// 2. repo — optional, but if repo_id is set we care whether the
|
||||
// clone landed. topic.repo_workspace_path is populated by
|
||||
// start_topic after `git clone` succeeds.
|
||||
if topic.repo_id.is_some() {
|
||||
let cloned = topic
|
||||
.repo_workspace_path
|
||||
.as_ref()
|
||||
.is_some_and(|p| !p.is_empty());
|
||||
stages.push(PipelineStage {
|
||||
key: "repo".into(),
|
||||
label: if cloned {
|
||||
format!(
|
||||
"Repo cloned at {}",
|
||||
topic.repo_workspace_path.as_deref().unwrap_or("")
|
||||
)
|
||||
} else {
|
||||
"Repo bound but never cloned".into()
|
||||
},
|
||||
status: if cloned { "ok" } else { "fail" },
|
||||
detail: None,
|
||||
});
|
||||
} else {
|
||||
stages.push(PipelineStage {
|
||||
key: "repo".into(),
|
||||
label: "No repo bound (optional)".into(),
|
||||
status: "skip",
|
||||
detail: None,
|
||||
});
|
||||
}
|
||||
|
||||
// 3. container — the per-topic team runtime. Populated by spawn().
|
||||
let container_ok =
|
||||
topic.zeroclaw_container_name.is_some() && topic.zeroclaw_gateway_url.is_some();
|
||||
stages.push(PipelineStage {
|
||||
key: "container".into(),
|
||||
label: if container_ok {
|
||||
format!(
|
||||
"Container: {}",
|
||||
topic.zeroclaw_container_name.as_deref().unwrap_or("")
|
||||
)
|
||||
} else {
|
||||
"Container not spawned (falling back to shared gateway)".into()
|
||||
},
|
||||
status: if container_ok { "ok" } else { "warn" },
|
||||
detail: None,
|
||||
});
|
||||
|
||||
// 4. runs — every research run this topic has produced, with each
|
||||
// one's terminal status + error. This is the diagnostic that
|
||||
// catches "run failed, no outcome" — the prior downstream stages
|
||||
// would otherwise look fine.
|
||||
use sqlx::Row;
|
||||
let run_rows = sqlx::query(
|
||||
"SELECT id, status, error, created_at
|
||||
FROM topology_runs
|
||||
WHERE research_topic_id = $1
|
||||
ORDER BY created_at DESC",
|
||||
)
|
||||
.bind(id)
|
||||
.fetch_all(&state.pool)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
let n_runs = run_rows.len();
|
||||
let n_failed = run_rows
|
||||
.iter()
|
||||
.filter(|r| r.try_get::<String, _>("status").ok().as_deref() == Some("failed"))
|
||||
.count();
|
||||
let latest_error = run_rows
|
||||
.iter()
|
||||
.find_map(|r| r.try_get::<Option<String>, _>("error").ok().flatten())
|
||||
.filter(|s| !s.is_empty());
|
||||
let run_status = if n_runs == 0 {
|
||||
"warn"
|
||||
} else if n_failed == n_runs {
|
||||
"fail"
|
||||
} else if n_failed > 0 {
|
||||
"warn"
|
||||
} else {
|
||||
"ok"
|
||||
};
|
||||
stages.push(PipelineStage {
|
||||
key: "runs".into(),
|
||||
label: format!("{n_runs} run(s), {n_failed} failed"),
|
||||
status: run_status,
|
||||
detail: latest_error,
|
||||
});
|
||||
|
||||
// 5. outcomes — the actual artifact rows. This is what
|
||||
// get_artifact reads; a 0-outcome topic that reached 'published'
|
||||
// is the silent-failure the diagnostic is meant to surface.
|
||||
let outcome_count: i64 =
|
||||
sqlx::query_scalar("SELECT count(*) FROM research_outcomes WHERE topic_id = $1")
|
||||
.bind(id)
|
||||
.fetch_one(&state.pool)
|
||||
.await
|
||||
.unwrap_or(0);
|
||||
stages.push(PipelineStage {
|
||||
key: "outcomes".into(),
|
||||
label: format!("{outcome_count} outcome(s) written"),
|
||||
status: if outcome_count > 0 { "ok" } else { "fail" },
|
||||
detail: if outcome_count == 0 {
|
||||
Some("No outcome produced yet — check the runs stage for the failure reason.".into())
|
||||
} else {
|
||||
None
|
||||
},
|
||||
});
|
||||
|
||||
// 6. approval — a pending publish approval is a normal state; the
|
||||
// diagnostic just flags it as a pending signal, not a failure.
|
||||
let pending = cm_db::repo::research_publish_approvals::pending_for_topic(&state.pool, id)
|
||||
.await
|
||||
.ok()
|
||||
.flatten();
|
||||
if let Some(a) = pending {
|
||||
stages.push(PipelineStage {
|
||||
key: "approval".into(),
|
||||
label: format!(
|
||||
"Approval pending (requested {})",
|
||||
a.created_at
|
||||
.format(&time::format_description::well_known::Rfc3339)
|
||||
.unwrap_or_default()
|
||||
),
|
||||
status: "warn",
|
||||
detail: None,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(Json(PipelineState {
|
||||
topic_id: id,
|
||||
status: topic.status,
|
||||
stages,
|
||||
}))
|
||||
}
|
||||
// Pipeline diagnostics moved to `research_pipeline.rs` to keep this file
|
||||
// under the 1250-line budget. Route registration in lib.rs points at
|
||||
// `routes::research_pipeline::pipeline_state`.
|
||||
|
||||
// ── wizard refine ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
//! Pipeline diagnostics for a research topic.
|
||||
//!
|
||||
//! Walks the pipeline stages (staffing, repo, container, runs, outcomes,
|
||||
//! approval) and returns a per-stage report. Read-only — every stage is
|
||||
//! evaluated in isolation and any lookup failure downgrades to warn/skip
|
||||
//! rather than failing the endpoint. Purpose: give users end-to-end
|
||||
//! visibility so silent failures (a run that dies before writing an
|
||||
//! outcome) are surfaced instead of buried in an empty artifact
|
||||
//! download.
|
||||
|
||||
use axum::extract::{Path, State};
|
||||
use axum::Json;
|
||||
use serde::Serialize;
|
||||
use sqlx::Row;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{ApiError, AppState, Authed};
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct PipelineStage {
|
||||
/// Machine-readable stage id: staffing / repo / container / runs /
|
||||
/// outcomes / approval. Frontend uses this to key the checklist.
|
||||
pub key: String,
|
||||
/// User-facing one-line summary.
|
||||
pub label: String,
|
||||
/// ok | warn | fail | skip — drives the pill color in the UI.
|
||||
pub status: &'static str,
|
||||
/// Optional error text (last-known failure reason from the underlying
|
||||
/// row) so the user can see WHY a stage failed instead of a green tick
|
||||
/// with no artifact behind it.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub detail: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct PipelineState {
|
||||
pub topic_id: Uuid,
|
||||
pub status: String,
|
||||
pub stages: Vec<PipelineStage>,
|
||||
}
|
||||
|
||||
/// `GET /api/research/:id/pipeline-state`.
|
||||
pub async fn pipeline_state(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(id): Path<Uuid>,
|
||||
) -> Result<Json<PipelineState>, ApiError> {
|
||||
let topic = cm_db::repo::research_topics::get(&state.pool, id, user.workspace_id.as_uuid())
|
||||
.await?
|
||||
.ok_or(ApiError::NotFound)?;
|
||||
let mut stages = Vec::new();
|
||||
|
||||
// 1. staffing.
|
||||
let agents = cm_db::repo::research_topics::agents(&state.pool, id)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
stages.push(PipelineStage {
|
||||
key: "staffing".into(),
|
||||
label: format!("{} agent(s) assigned", agents.len()),
|
||||
status: if agents.is_empty() { "fail" } else { "ok" },
|
||||
detail: None,
|
||||
});
|
||||
|
||||
// 2. repo — optional. When bound, we check the clone actually landed.
|
||||
if topic.repo_id.is_some() {
|
||||
let cloned = topic
|
||||
.repo_workspace_path
|
||||
.as_ref()
|
||||
.is_some_and(|p| !p.is_empty());
|
||||
stages.push(PipelineStage {
|
||||
key: "repo".into(),
|
||||
label: if cloned {
|
||||
format!(
|
||||
"Repo cloned at {}",
|
||||
topic.repo_workspace_path.as_deref().unwrap_or("")
|
||||
)
|
||||
} else {
|
||||
"Repo bound but never cloned".into()
|
||||
},
|
||||
status: if cloned { "ok" } else { "fail" },
|
||||
detail: None,
|
||||
});
|
||||
} else {
|
||||
stages.push(PipelineStage {
|
||||
key: "repo".into(),
|
||||
label: "No repo bound (optional)".into(),
|
||||
status: "skip",
|
||||
detail: None,
|
||||
});
|
||||
}
|
||||
|
||||
// 3. container — per-topic team runtime.
|
||||
let container_ok =
|
||||
topic.zeroclaw_container_name.is_some() && topic.zeroclaw_gateway_url.is_some();
|
||||
stages.push(PipelineStage {
|
||||
key: "container".into(),
|
||||
label: if container_ok {
|
||||
format!(
|
||||
"Container: {}",
|
||||
topic.zeroclaw_container_name.as_deref().unwrap_or("")
|
||||
)
|
||||
} else {
|
||||
"Container not spawned (falling back to shared gateway)".into()
|
||||
},
|
||||
status: if container_ok { "ok" } else { "warn" },
|
||||
detail: None,
|
||||
});
|
||||
|
||||
// 4. runs — catches the failure with the actual error text.
|
||||
let run_rows = sqlx::query(
|
||||
"SELECT id, status, error, created_at
|
||||
FROM topology_runs
|
||||
WHERE research_topic_id = $1
|
||||
ORDER BY created_at DESC",
|
||||
)
|
||||
.bind(id)
|
||||
.fetch_all(&state.pool)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
let n_runs = run_rows.len();
|
||||
let n_failed = run_rows
|
||||
.iter()
|
||||
.filter(|r| r.try_get::<String, _>("status").ok().as_deref() == Some("failed"))
|
||||
.count();
|
||||
let latest_error = run_rows
|
||||
.iter()
|
||||
.find_map(|r| r.try_get::<Option<String>, _>("error").ok().flatten())
|
||||
.filter(|s| !s.is_empty());
|
||||
let run_status = if n_runs == 0 {
|
||||
"warn"
|
||||
} else if n_failed == n_runs {
|
||||
"fail"
|
||||
} else if n_failed > 0 {
|
||||
"warn"
|
||||
} else {
|
||||
"ok"
|
||||
};
|
||||
stages.push(PipelineStage {
|
||||
key: "runs".into(),
|
||||
label: format!("{n_runs} run(s), {n_failed} failed"),
|
||||
status: run_status,
|
||||
detail: latest_error,
|
||||
});
|
||||
|
||||
// 5. outcomes — the artifact rows get_artifact reads.
|
||||
let outcome_count: i64 =
|
||||
sqlx::query_scalar("SELECT count(*) FROM research_outcomes WHERE topic_id = $1")
|
||||
.bind(id)
|
||||
.fetch_one(&state.pool)
|
||||
.await
|
||||
.unwrap_or(0);
|
||||
stages.push(PipelineStage {
|
||||
key: "outcomes".into(),
|
||||
label: format!("{outcome_count} outcome(s) written"),
|
||||
status: if outcome_count > 0 { "ok" } else { "fail" },
|
||||
detail: if outcome_count == 0 {
|
||||
Some("No outcome produced yet — check the runs stage for the failure reason.".into())
|
||||
} else {
|
||||
None
|
||||
},
|
||||
});
|
||||
|
||||
// 6. approval — pending publish-approval, if any.
|
||||
let pending = cm_db::repo::research_publish_approvals::pending_for_topic(&state.pool, id)
|
||||
.await
|
||||
.ok()
|
||||
.flatten();
|
||||
if let Some(a) = pending {
|
||||
stages.push(PipelineStage {
|
||||
key: "approval".into(),
|
||||
label: format!(
|
||||
"Approval pending (requested {})",
|
||||
a.created_at
|
||||
.format(&time::format_description::well_known::Rfc3339)
|
||||
.unwrap_or_default()
|
||||
),
|
||||
status: "warn",
|
||||
detail: None,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(Json(PipelineState {
|
||||
topic_id: id,
|
||||
status: topic.status,
|
||||
stages,
|
||||
}))
|
||||
}
|
||||
Reference in New Issue
Block a user