refactor(#1456): extract dashboard approval + diff endpoints into dashboard/approvals.rs
This commit is contained in:
damocles
mara
parent
55705f17d3
commit
ec3ca216c5
2 changed files with 240 additions and 209 deletions
|
|
@ -25,11 +25,11 @@ use tokio_stream::wrappers::BroadcastStream;
|
|||
use tokio_stream::{Stream, StreamExt};
|
||||
use tower_http::services::ServeDir;
|
||||
|
||||
use crate::actions;
|
||||
use crate::container_view::{ContainerView, claude_has_session};
|
||||
use crate::coordinator::Coordinator;
|
||||
use crate::lifecycle::{self, MANAGER_NAME};
|
||||
|
||||
mod approvals;
|
||||
mod build_logs;
|
||||
mod journal;
|
||||
mod lifecycle_ops;
|
||||
|
|
@ -40,6 +40,12 @@ mod schedules;
|
|||
mod topology;
|
||||
mod webhook;
|
||||
|
||||
// Pre-computed at approval-submit time by the manager-socket handler
|
||||
// (`manager_server.rs`) and embedded in the `ApprovalAdded` event, so
|
||||
// re-exported at the module root to preserve the `crate::dashboard::approval_diff`
|
||||
// path across the submodule split.
|
||||
pub(crate) use approvals::approval_diff;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct AppState {
|
||||
coord: Arc<Coordinator>,
|
||||
|
|
@ -68,8 +74,8 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
|||
tracing::info!(static_dir = %static_dir.display(), "dashboard static dir resolved");
|
||||
let app = Router::new()
|
||||
.route("/api/state", get(api_state))
|
||||
.route("/approve/{id}", post(post_approve))
|
||||
.route("/deny/{id}", post(post_deny))
|
||||
.route("/approve/{id}", post(approvals::post_approve))
|
||||
.route("/deny/{id}", post(approvals::post_deny))
|
||||
.route("/destroy/{name}", post(lifecycle_ops::post_destroy))
|
||||
.route("/kill/{name}", post(lifecycle_ops::post_kill))
|
||||
.route("/restart/{name}", post(lifecycle_ops::post_restart))
|
||||
|
|
@ -87,7 +93,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
|||
.route("/purge-tombstone/{name}", post(post_purge_tombstone))
|
||||
.route("/api/journal/{name}", get(journal::get_journal))
|
||||
.route("/api/journal-host", get(journal::get_journal_host))
|
||||
.route("/api/approval-diff/{id}", get(get_approval_diff))
|
||||
.route("/api/approval-diff/{id}", get(approvals::get_approval_diff))
|
||||
.route("/api/state-file", get(get_state_file))
|
||||
.route("/api/reminders", get(reminders::api_reminders))
|
||||
.route("/api/operator-inbox", get(api_operator_inbox))
|
||||
|
|
@ -476,7 +482,7 @@ async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::J
|
|||
let containers = state.coord.containers_snapshot().await;
|
||||
let any_stale = containers.iter().any(|c| c.needs_update);
|
||||
let transient_snapshot = state.coord.transient_snapshot();
|
||||
let pending_approvals = gc_orphans(
|
||||
let pending_approvals = approvals::gc_orphans(
|
||||
&state.coord,
|
||||
log_default("approvals.pending", state.coord.approvals.pending()),
|
||||
);
|
||||
|
|
@ -1070,39 +1076,6 @@ async fn dashboard_stream(
|
|||
Sse::new(stream).keep_alive(KeepAlive::default())
|
||||
}
|
||||
|
||||
async fn post_approve(State(state): State<AppState>, AxumPath(id): AxumPath<i64>) -> Response {
|
||||
match actions::approve(state.coord.clone(), id).await {
|
||||
// 200 instead of 303 — `actions::approve` fires
|
||||
// `ApprovalResolved` (success path) or the eventual failure
|
||||
// event, both of which the dashboard's derived store applies
|
||||
// live. The matching form carries `data-no-refresh`.
|
||||
Ok(()) => (StatusCode::OK, "ok").into_response(),
|
||||
Err(e) => error_response(&format!("approve {id} failed: {e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Default)]
|
||||
struct DenyForm {
|
||||
#[serde(default)]
|
||||
note: Option<String>,
|
||||
}
|
||||
|
||||
async fn post_deny(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
Form(form): Form<DenyForm>,
|
||||
) -> Response {
|
||||
let note = form
|
||||
.note
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty());
|
||||
match actions::deny(&state.coord, id, note).await {
|
||||
Ok(()) => (StatusCode::OK, "ok").into_response(),
|
||||
Err(e) => error_response(&format!("deny {id} failed: {e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RequestSpawnForm {
|
||||
name: String,
|
||||
|
|
@ -1891,174 +1864,3 @@ fn error_response(message: &str) -> Response {
|
|||
// wrapping would just clutter the message.
|
||||
(StatusCode::INTERNAL_SERVER_ERROR, message.to_owned()).into_response()
|
||||
}
|
||||
|
||||
/// Filter out approvals whose agent state dir was wiped out from under us
|
||||
/// (e.g. by a test script's cleanup). Marks them failed so they fall out of
|
||||
/// `pending` on next render.
|
||||
fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<Approval> {
|
||||
approvals
|
||||
.into_iter()
|
||||
.filter(|a| {
|
||||
// Spawn and InitConfig approvals are for not-yet-existent agents;
|
||||
// the proposed dir is supposed to be missing.
|
||||
if matches!(
|
||||
a.kind,
|
||||
hive_sh4re::ApprovalKind::Spawn | hive_sh4re::ApprovalKind::InitConfig
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
if Coordinator::agent_proposed_dir(&a.agent).exists() {
|
||||
true
|
||||
} else {
|
||||
let note = "agent state dir missing";
|
||||
let _ = coord.approvals.mark_failed(a.id, note);
|
||||
tracing::info!(id = a.id, agent = %a.agent, "auto-failed orphan approval");
|
||||
let sha_short = a
|
||||
.fetched_sha
|
||||
.as_deref()
|
||||
.map(|s| s[..s.len().min(12)].to_owned());
|
||||
coord.emit_approval_resolved(
|
||||
a.id,
|
||||
&a.agent,
|
||||
"apply_commit",
|
||||
sha_short,
|
||||
"failed",
|
||||
Some(note.to_owned()),
|
||||
a.description.clone(),
|
||||
);
|
||||
false
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Multi-file unified diff between the currently-deployed tree and
|
||||
/// the proposal for this approval. Runs against the applied repo
|
||||
/// since the canonical proposal commit lives there (manager-side
|
||||
/// amendments don't move it). Empty output means proposal == main —
|
||||
/// a no-op approval.
|
||||
///
|
||||
/// `pub(crate)` so the manager-socket handler can pre-compute the
|
||||
/// diff once at submission time and embed it in the `ApprovalAdded`
|
||||
/// dashboard event (instead of forcing the dashboard to wait a
|
||||
/// `/api/state` cycle to see the diff for newly-queued approvals).
|
||||
pub(crate) async fn approval_diff(agent: &str, approval_id: i64) -> String {
|
||||
let applied = Coordinator::agent_applied_dir(agent);
|
||||
if !applied.join(".git").exists() {
|
||||
return format!("(no applied git repo at {})", applied.display());
|
||||
}
|
||||
let proposal_ref = format!("refs/tags/proposal/{approval_id}");
|
||||
match git_diff_refs(&applied, "refs/heads/main", &proposal_ref).await {
|
||||
Ok(s) if s.is_empty() => "(proposal matches currently-deployed tree)".to_owned(),
|
||||
Ok(s) => s,
|
||||
Err(e) => format!("(error: {e:#})"),
|
||||
}
|
||||
}
|
||||
|
||||
async fn git_diff_refs(applied_dir: &Path, base_ref: &str, target_ref: &str) -> Result<String> {
|
||||
let out = lifecycle::git_command()
|
||||
.current_dir(applied_dir)
|
||||
.args(["diff", &format!("{base_ref}..{target_ref}")])
|
||||
.output()
|
||||
.await
|
||||
.with_context(|| format!("spawn `git diff` in {}", applied_dir.display()))?;
|
||||
if !out.status.success() {
|
||||
anyhow::bail!(
|
||||
"git diff {base_ref}..{target_ref} failed: {}",
|
||||
String::from_utf8_lossy(&out.stderr).trim()
|
||||
);
|
||||
}
|
||||
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
|
||||
}
|
||||
|
||||
/// Numeric ids of `<prefix>/<n>` tags in the applied repo (e.g.
|
||||
/// `proposal/3` → `3`). Unparseable suffixes are skipped. Used to
|
||||
/// resolve the `approved` / `previous` diff bases for an approval.
|
||||
async fn tag_ids(applied_dir: &Path, prefix: &str) -> Vec<i64> {
|
||||
let Ok(out) = lifecycle::git_command()
|
||||
.current_dir(applied_dir)
|
||||
.args(["tag", "-l", &format!("{prefix}/*")])
|
||||
.output()
|
||||
.await
|
||||
else {
|
||||
return Vec::new();
|
||||
};
|
||||
if !out.status.success() {
|
||||
return Vec::new();
|
||||
}
|
||||
let strip = format!("{prefix}/");
|
||||
String::from_utf8_lossy(&out.stdout)
|
||||
.lines()
|
||||
.filter_map(|l| l.trim().strip_prefix(&strip))
|
||||
.filter_map(|s| s.parse::<i64>().ok())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct DiffBaseQuery {
|
||||
/// `applied` (running tree — default), `approved` (most recent
|
||||
/// earlier approved proposal), or `previous` (the prior queued
|
||||
/// proposal for this agent).
|
||||
base: Option<String>,
|
||||
}
|
||||
|
||||
/// On-demand unified diff for one `ApplyCommit` approval against a
|
||||
/// chosen base. `applied` = `applied/main` (what's running);
|
||||
/// `approved` = the most recent earlier `approved/<n>` tag (the last
|
||||
/// proposal the operator OK'd, even if its build then failed);
|
||||
/// `previous` = the prior queued `proposal/<n>` (the incremental
|
||||
/// delta when the manager chains proposals). Returns the raw diff
|
||||
/// text — the dashboard classifies lines client-side.
|
||||
async fn get_approval_diff(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
axum::extract::Query(q): axum::extract::Query<DiffBaseQuery>,
|
||||
) -> Response {
|
||||
let base = q.base.as_deref().unwrap_or("applied");
|
||||
let approval = match state.coord.approvals.get(id) {
|
||||
Ok(Some(a)) => a,
|
||||
Ok(None) => return error_response(&format!("approval {id} not found")),
|
||||
Err(e) => return error_response(&format!("approval {id}: {e:#}")),
|
||||
};
|
||||
if !matches!(approval.kind, hive_sh4re::ApprovalKind::ApplyCommit) {
|
||||
return error_response("spawn approvals carry no commit to diff");
|
||||
}
|
||||
let applied = Coordinator::agent_applied_dir(&approval.agent);
|
||||
if !applied.join(".git").exists() {
|
||||
return plain_text(format!("(no applied git repo at {})", applied.display()));
|
||||
}
|
||||
let target = format!("refs/tags/proposal/{id}");
|
||||
let base_ref = match base {
|
||||
"applied" => Some("refs/heads/main".to_owned()),
|
||||
"approved" => {
|
||||
let ids = tag_ids(&applied, "approved").await;
|
||||
ids.into_iter()
|
||||
.filter(|&n| n != id)
|
||||
.max()
|
||||
.map(|n| format!("refs/tags/approved/{n}"))
|
||||
}
|
||||
"previous" => {
|
||||
let ids = tag_ids(&applied, "proposal").await;
|
||||
ids.into_iter()
|
||||
.filter(|&n| n < id)
|
||||
.max()
|
||||
.map(|n| format!("refs/tags/proposal/{n}"))
|
||||
}
|
||||
other => return error_response(&format!("unknown diff base {other:?}")),
|
||||
};
|
||||
let Some(base_ref) = base_ref else {
|
||||
return plain_text(match base {
|
||||
"approved" => "(no earlier approved proposal to diff against)".to_owned(),
|
||||
_ => "(no previous proposal to diff against)".to_owned(),
|
||||
});
|
||||
};
|
||||
match git_diff_refs(&applied, &base_ref, &target).await {
|
||||
Ok(s) if s.is_empty() => plain_text("(identical — no changes vs this base)".to_owned()),
|
||||
Ok(s) => plain_text(s),
|
||||
Err(e) => error_response(&format!("git diff: {e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
fn plain_text(body: String) -> Response {
|
||||
(StatusCode::OK, body).into_response()
|
||||
}
|
||||
|
|
|
|||
229
hive-c0re/src/dashboard/approvals.rs
Normal file
229
hive-c0re/src/dashboard/approvals.rs
Normal file
|
|
@ -0,0 +1,229 @@
|
|||
//! Approval endpoints + diff machinery for the dashboard.
|
||||
//!
|
||||
//! Approve/deny actions, the orphan-approval GC sweep used by the
|
||||
//! `/api/state` builder, and the unified-diff endpoints (on-demand
|
||||
//! `/api/approval-diff/{id}` against a chosen base, plus the `pub(crate)`
|
||||
//! `approval_diff` the manager-socket handler pre-computes at submit time).
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use axum::{
|
||||
extract::{Form, Path as AxumPath, State},
|
||||
http::StatusCode,
|
||||
response::{IntoResponse, Response},
|
||||
};
|
||||
use hive_sh4re::Approval;
|
||||
use serde::Deserialize;
|
||||
|
||||
use super::{AppState, error_response};
|
||||
use crate::actions;
|
||||
use crate::coordinator::Coordinator;
|
||||
use crate::lifecycle;
|
||||
|
||||
pub(super) async fn post_approve(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
) -> Response {
|
||||
match actions::approve(state.coord.clone(), id).await {
|
||||
// 200 instead of 303 — `actions::approve` fires
|
||||
// `ApprovalResolved` (success path) or the eventual failure
|
||||
// event, both of which the dashboard's derived store applies
|
||||
// live. The matching form carries `data-no-refresh`.
|
||||
Ok(()) => (StatusCode::OK, "ok").into_response(),
|
||||
Err(e) => error_response(&format!("approve {id} failed: {e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Default)]
|
||||
pub(super) struct DenyForm {
|
||||
#[serde(default)]
|
||||
note: Option<String>,
|
||||
}
|
||||
|
||||
pub(super) async fn post_deny(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
Form(form): Form<DenyForm>,
|
||||
) -> Response {
|
||||
let note = form
|
||||
.note
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty());
|
||||
match actions::deny(&state.coord, id, note).await {
|
||||
Ok(()) => (StatusCode::OK, "ok").into_response(),
|
||||
Err(e) => error_response(&format!("deny {id} failed: {e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// Filter out approvals whose agent state dir was wiped out from under us
|
||||
/// (e.g. by a test script's cleanup). Marks them failed so they fall out of
|
||||
/// `pending` on next render.
|
||||
pub(super) fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<Approval> {
|
||||
approvals
|
||||
.into_iter()
|
||||
.filter(|a| {
|
||||
// Spawn and InitConfig approvals are for not-yet-existent agents;
|
||||
// the proposed dir is supposed to be missing.
|
||||
if matches!(
|
||||
a.kind,
|
||||
hive_sh4re::ApprovalKind::Spawn | hive_sh4re::ApprovalKind::InitConfig
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
if Coordinator::agent_proposed_dir(&a.agent).exists() {
|
||||
true
|
||||
} else {
|
||||
let note = "agent state dir missing";
|
||||
let _ = coord.approvals.mark_failed(a.id, note);
|
||||
tracing::info!(id = a.id, agent = %a.agent, "auto-failed orphan approval");
|
||||
let sha_short = a
|
||||
.fetched_sha
|
||||
.as_deref()
|
||||
.map(|s| s[..s.len().min(12)].to_owned());
|
||||
coord.emit_approval_resolved(
|
||||
a.id,
|
||||
&a.agent,
|
||||
"apply_commit",
|
||||
sha_short,
|
||||
"failed",
|
||||
Some(note.to_owned()),
|
||||
a.description.clone(),
|
||||
);
|
||||
false
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Multi-file unified diff between the currently-deployed tree and
|
||||
/// the proposal for this approval. Runs against the applied repo
|
||||
/// since the canonical proposal commit lives there (manager-side
|
||||
/// amendments don't move it). Empty output means proposal == main —
|
||||
/// a no-op approval.
|
||||
///
|
||||
/// `pub(crate)` so the manager-socket handler can pre-compute the
|
||||
/// diff once at submission time and embed it in the `ApprovalAdded`
|
||||
/// dashboard event (instead of forcing the dashboard to wait a
|
||||
/// `/api/state` cycle to see the diff for newly-queued approvals).
|
||||
pub(crate) async fn approval_diff(agent: &str, approval_id: i64) -> String {
|
||||
let applied = Coordinator::agent_applied_dir(agent);
|
||||
if !applied.join(".git").exists() {
|
||||
return format!("(no applied git repo at {})", applied.display());
|
||||
}
|
||||
let proposal_ref = format!("refs/tags/proposal/{approval_id}");
|
||||
match git_diff_refs(&applied, "refs/heads/main", &proposal_ref).await {
|
||||
Ok(s) if s.is_empty() => "(proposal matches currently-deployed tree)".to_owned(),
|
||||
Ok(s) => s,
|
||||
Err(e) => format!("(error: {e:#})"),
|
||||
}
|
||||
}
|
||||
|
||||
async fn git_diff_refs(applied_dir: &Path, base_ref: &str, target_ref: &str) -> Result<String> {
|
||||
let out = lifecycle::git_command()
|
||||
.current_dir(applied_dir)
|
||||
.args(["diff", &format!("{base_ref}..{target_ref}")])
|
||||
.output()
|
||||
.await
|
||||
.with_context(|| format!("spawn `git diff` in {}", applied_dir.display()))?;
|
||||
if !out.status.success() {
|
||||
anyhow::bail!(
|
||||
"git diff {base_ref}..{target_ref} failed: {}",
|
||||
String::from_utf8_lossy(&out.stderr).trim()
|
||||
);
|
||||
}
|
||||
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
|
||||
}
|
||||
|
||||
/// Numeric ids of `<prefix>/<n>` tags in the applied repo (e.g.
|
||||
/// `proposal/3` → `3`). Unparseable suffixes are skipped. Used to
|
||||
/// resolve the `approved` / `previous` diff bases for an approval.
|
||||
async fn tag_ids(applied_dir: &Path, prefix: &str) -> Vec<i64> {
|
||||
let Ok(out) = lifecycle::git_command()
|
||||
.current_dir(applied_dir)
|
||||
.args(["tag", "-l", &format!("{prefix}/*")])
|
||||
.output()
|
||||
.await
|
||||
else {
|
||||
return Vec::new();
|
||||
};
|
||||
if !out.status.success() {
|
||||
return Vec::new();
|
||||
}
|
||||
let strip = format!("{prefix}/");
|
||||
String::from_utf8_lossy(&out.stdout)
|
||||
.lines()
|
||||
.filter_map(|l| l.trim().strip_prefix(&strip))
|
||||
.filter_map(|s| s.parse::<i64>().ok())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub(super) struct DiffBaseQuery {
|
||||
/// `applied` (running tree — default), `approved` (most recent
|
||||
/// earlier approved proposal), or `previous` (the prior queued
|
||||
/// proposal for this agent).
|
||||
base: Option<String>,
|
||||
}
|
||||
|
||||
/// On-demand unified diff for one `ApplyCommit` approval against a
|
||||
/// chosen base. `applied` = `applied/main` (what's running);
|
||||
/// `approved` = the most recent earlier `approved/<n>` tag (the last
|
||||
/// proposal the operator OK'd, even if its build then failed);
|
||||
/// `previous` = the prior queued `proposal/<n>` (the incremental
|
||||
/// delta when the manager chains proposals). Returns the raw diff
|
||||
/// text — the dashboard classifies lines client-side.
|
||||
pub(super) async fn get_approval_diff(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
axum::extract::Query(q): axum::extract::Query<DiffBaseQuery>,
|
||||
) -> Response {
|
||||
let base = q.base.as_deref().unwrap_or("applied");
|
||||
let approval = match state.coord.approvals.get(id) {
|
||||
Ok(Some(a)) => a,
|
||||
Ok(None) => return error_response(&format!("approval {id} not found")),
|
||||
Err(e) => return error_response(&format!("approval {id}: {e:#}")),
|
||||
};
|
||||
if !matches!(approval.kind, hive_sh4re::ApprovalKind::ApplyCommit) {
|
||||
return error_response("spawn approvals carry no commit to diff");
|
||||
}
|
||||
let applied = Coordinator::agent_applied_dir(&approval.agent);
|
||||
if !applied.join(".git").exists() {
|
||||
return plain_text(format!("(no applied git repo at {})", applied.display()));
|
||||
}
|
||||
let target = format!("refs/tags/proposal/{id}");
|
||||
let base_ref = match base {
|
||||
"applied" => Some("refs/heads/main".to_owned()),
|
||||
"approved" => {
|
||||
let ids = tag_ids(&applied, "approved").await;
|
||||
ids.into_iter()
|
||||
.filter(|&n| n != id)
|
||||
.max()
|
||||
.map(|n| format!("refs/tags/approved/{n}"))
|
||||
}
|
||||
"previous" => {
|
||||
let ids = tag_ids(&applied, "proposal").await;
|
||||
ids.into_iter()
|
||||
.filter(|&n| n < id)
|
||||
.max()
|
||||
.map(|n| format!("refs/tags/proposal/{n}"))
|
||||
}
|
||||
other => return error_response(&format!("unknown diff base {other:?}")),
|
||||
};
|
||||
let Some(base_ref) = base_ref else {
|
||||
return plain_text(match base {
|
||||
"approved" => "(no earlier approved proposal to diff against)".to_owned(),
|
||||
_ => "(no previous proposal to diff against)".to_owned(),
|
||||
});
|
||||
};
|
||||
match git_diff_refs(&applied, &base_ref, &target).await {
|
||||
Ok(s) if s.is_empty() => plain_text("(identical — no changes vs this base)".to_owned()),
|
||||
Ok(s) => plain_text(s),
|
||||
Err(e) => error_response(&format!("git diff: {e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
fn plain_text(body: String) -> Response {
|
||||
(StatusCode::OK, body).into_response()
|
||||
}
|
||||
Loading…
Reference in a new issue