hyperhive/hive-c0re/src/actions.rs
atlas 6338939657 refactor(#2916): destroy submits a DAG instead of an imperative teardown
Destroy was a straight-line async fn with no queue node behind it, so
nothing in the graph could answer "is this container going down on
purpose?". That gap is why an imperative crash-watch suppression guard
existed: an RAII handle held for the operation's duration, a second way
to say what every other lifecycle op already says through its node.

Reuse the existing Stop node rather than teaching a new node to stop
things:

    Stop -> DestroyContainer -> (PurgeState) -> DestroyBookkeeping

Stop already declares takes_container_down honestly, so the suppression
is now derived from the graph like every other op's. It also turns the
precondition into an edge: DestroyContainer runs only under a completed
Stop, so it operates on an already-stopped container and carries
takes_container_down = false permanently. A container still alive at
that point is a real bug and stays loud instead of being absorbed by a
flag -- which matters because a wrong true silently swallows a crash
while a wrong false only costs a spurious event.

Removes suppress_crash_watch, CrashWatchSuppression, crash_suppressed,
crash_watch_suppressed and NO_NODE_LABEL. The migration call sites went
with the obsolete startup migrations, so destroy was the last caller and
intent now has exactly one home.

destroy() becomes a submit-and-return, matching every sibling endpoint
(rebuild, kill, restart, start, pause, resume) -- it was the only
lifecycle op that awaited its work. The container rescan moves into the
bookkeeping tail, so ContainerRemoved now arrives after the 200 rather
than before it.

Also drops an orphaned doc-comment in coordinator.rs: two stacked blocks
where only the second described crash_suppressed, the first documenting
a field that no longer exists. Removing the field would have re-pointed
it at recent_transient.
2026-08-14 00:24:45 +02:00

952 lines
42 KiB
Rust

//! Operations that are exposed through more than one surface (the host admin
//! socket *and* the dashboard's POST endpoints). Each function takes a
//! `&Coordinator` and the request parameters; callers stitch the response
//! shape they want (HTTP redirect vs JSON).
use std::sync::Arc;
use anyhow::{Context as _, Result, bail};
use hive_sh4re::approvals::{ApprovalKind, ApprovalStatus};
use hive_sh4re::manager::HelperEvent;
use crate::coordinator::Coordinator;
use crate::lifecycle;
/// Approve a pending request. Marks the approval row durably, then
/// either runs the work inline (`InitConfig`, sub-second git ops) or
/// submits it to the job queue so the dashboard POST returns
/// immediately while the long-running pipeline runs off-thread
/// (operator no longer blocks on a 30-90s spinner for `MergeConfigPr`).
///
/// Dispatch:
/// - `MergeConfigPr` → a `DeployWindow` DAG (`MergeVerify → DeployApply →
/// FinalizeDeploy`, plus an `AfterAny` `DeployTail`, under a
/// resource-holding root; ~30-90s)
/// - `UpdateMetaInputs` → a `MetaUpdate` DAG (fan-out on completion)
/// - `Spawn` → a `Spawn` DAG (`Create → WriteDropin → Reconcile`)
/// - `InitConfig` → inline (<1s; queue card would be noise)
///
/// Every queued kind — deploys included — resolves its approval row via
/// [`resolve_approval_dag`] when the DAG settles terminal.
pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
let approval = coord.approvals.mark_approved(id)?;
tracing::info!(
%approval.id,
%approval.agent,
kind = ?approval.kind,
%approval.commit_ref,
"approval: dispatching",
);
match approval.kind {
ApprovalKind::InitConfig => {
// Sub-second git seed + forge-remote wire. Routing through
// the queue would surface a queue card that's gone before
// the operator's eyes refocus. Run inline.
let proposed_dir = Coordinator::agent_proposed_dir(&approval.agent);
let claude_dir = Coordinator::agent_claude_dir(&approval.agent);
let notes_dir = Coordinator::agent_notes_dir(&approval.agent);
run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await
}
ApprovalKind::UpdateMetaInputs => {
// Inputs JSON-encoded into commit_ref by the manager's
// submit path — surface them on the DAG so the dashboard
// can show *which* inputs are about to bump. The cascade
// rebuilds fan out when the lock bump lands (so they build
// against the post-bump lock, and a failed bump fans out
// nothing).
let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let inserted = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::meta_update(b, inputs, Some(id));
Vec::new()
});
if let Err(e) = inserted {
return Err(e.context("insert meta-update dag"));
}
coord.emit_rebuild_queue_snapshot();
Ok(())
}
ApprovalKind::Spawn => {
// The spawn's tail `Reconcile` starts the container, so the
// new agent's power intent is `Up` from the outset.
if let Err(e) = coord
.power
.set(approval.agent.as_str(), crate::power::Wanted::Up)
{
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
}
let inserted = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::spawn(b, approval.agent.as_str(), id);
Vec::new()
});
if let Err(e) = inserted {
return Err(e.context("insert spawn dag"));
}
coord.emit_rebuild_queue_snapshot();
Ok(())
}
ApprovalKind::SchedulePrompt => {
// No queue card for SchedulePrompt — the work is a single
// sqlite insert, the actual "running" lifetime lives on
// the scheduled-prompts surface itself (worker fires it
// at the scheduled time). Run inline + fire
// `ApprovalResolved` so the approval row leaves Pending
// immediately.
let result = run_approval_schedule_prompt(&coord, approval).await;
if result.is_ok() {
coord.emit_schedules_snapshot();
}
result
}
ApprovalKind::MergeConfigPr => {
// The work ends in a container rebuild, so route it through the
// rebuild queue. The queue worker dispatches the deploy DAG's
// nodes to `run_deploy_merge_verify` (drift gate + eval),
// `run_deploy_apply` (ff-merge, then grows the rebuild subgraph),
// `run_finalize_deploy` (deploy tag + lock commit) and
// `run_deploy_tail` (compensation + forge mirror).
enqueue_approval_rebuild(&coord, approval.agent.as_str(), id);
Ok(())
}
}
}
/// Submit the deploy DAG tied to an approval id. Used by the `MergeConfigPr`
/// dispatch arm — the work ends in a container rebuild routed through the
/// queue. See [`crate::job_queue::templates::approval_deploy`] for the node
/// shape; the executor dispatches each node to the `run_deploy_*` bodies below.
fn enqueue_approval_rebuild(coord: &Arc<Coordinator>, agent: &str, approval_id: i64) {
if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::approval_deploy(b, agent, approval_id);
Vec::new()
}) {
tracing::error!(%agent, approval_id, error = ?e, "insert approval deploy dag failed");
}
coord.emit_rebuild_queue_snapshot();
}
/// Ref under which [`run_deploy_apply`] parks the pre-merge `applied/main`
/// sha, for [`run_deploy_tail`] to compensate with.
///
/// Deliberately a *git ref in the applied repo* rather than an in-memory value
/// handed between nodes: hive-c0re can restart between the apply and the tail,
/// and the whole point of splitting the deploy is that the tail still knows
/// what to undo when it does. The ref's existence IS the "a merge landed but
/// hasn't been confirmed good yet" flag — [`run_finalize_deploy`] drops it the
/// moment the rebuild has come up clean.
fn rollback_ref(approval_id: i64) -> String {
format!("refs/hyperhive/rollback/{approval_id}")
}
/// Everything a deploy node needs, re-derived from sqlite on each node rather
/// than cached across the DAG. Nothing here is *computed* by an earlier node —
/// `pr` and `reviewed` are fields of the approval row the operator signed off
/// on — so re-reading is both cheap and the authoritative source of truth.
struct DeployCtx {
approval: hive_sh4re::approvals::Approval,
/// PR number, parsed from `approval.commit_ref`.
pr: u64,
/// The PR head sha the operator reviewed (`approval.fetched_sha`).
reviewed: String,
applied_dir: std::path::PathBuf,
/// The agent's forge config repo (`<owner>/<name>`).
repo: String,
}
fn deploy_ctx(coord: &Coordinator, approval_id: i64) -> Result<DeployCtx> {
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::MergeConfigPr)?;
let pr: u64 = approval.commit_ref.parse().map_err(|e| {
anyhow::anyhow!(
"parse PR number from commit_ref {:?}: {e}",
approval.commit_ref
)
})?;
let reviewed = approval.fetched_sha.clone().ok_or_else(|| {
anyhow::anyhow!("merge config pr approval {approval_id} has no reviewed head sha")
})?;
Ok(DeployCtx {
pr,
reviewed,
applied_dir: crate::paths::applied_dir(approval.agent.as_str()),
repo: crate::forge::config_repo(approval.agent.as_str()),
approval,
})
}
/// `MergeVerify` node body — everything that can say "no" before anything is
/// mutated. `approval.commit_ref` is the PR number; `approval.fetched_sha` is
/// the PR head the operator reviewed. Steps:
/// 1. drift gate — re-read the live PR head; if it moved since review, abort
/// (the operator must re-review the new head);
/// 2. fetch the reviewed head into the applied repo so later git ops resolve
/// it locally;
/// 3. ancestry gate — the reviewed head must descend from `applied/main`;
/// 4. eval-verify the reviewed commit against the meta flake.
///
/// Steps 1 and 3 ask different questions and both are load-bearing. The drift
/// gate asks whether the *head* is still what was reviewed; the ancestry gate
/// asks whether the *base* is still underneath it. A PR opened from a stale base
/// passes the drift gate untouched and then rewinds `main` when it lands,
/// silently dropping every commit made in between.
///
/// Nothing here needs undoing on failure: the fetch only adds objects, and
/// `main` doesn't move. That's the whole reason this is its own node — a
/// failure at this stage leaves [`run_deploy_tail`] with no ref to compensate.
///
/// # Errors
///
/// Returns an error if the approval can't be loaded, if the live PR head has
/// drifted from the reviewed sha, if fetching that head into the applied repo
/// fails, if the reviewed head does not descend from `applied/main`, or if the
/// eval-verify of the reviewed commit fails. Every one of these leaves the forge
/// and `main` untouched, so the node is safely retryable.
pub async fn run_deploy_merge_verify(
coord: &Arc<Coordinator>,
approval_id: i64,
node_id: Option<u64>,
) -> Result<()> {
let ctx = deploy_ctx(coord, approval_id)?;
let pr = ctx.pr;
let reviewed = ctx.reviewed.as_str();
// 1. Drift gate: the live PR head must still equal what was reviewed.
let head = crate::forge::pr_head_sha(&ctx.repo, pr)
.await
.map_err(|e| anyhow::anyhow!("read PR #{pr} head: {e}"))?;
if head != reviewed {
bail!(
"PR #{pr} head drifted since review (reviewed {reviewed}, now {head}); re-review before merging"
);
}
// 2. Fetch the reviewed head into applied so ff/verify/deploy resolve it.
crate::forge::fetch_pr_head_into_applied(&ctx.repo, pr)
.await
.map_err(|e| anyhow::anyhow!("fetch PR #{pr} head into applied: {e}"))?;
// 3. Ancestry gate: `main` must be reachable from the reviewed head, or the
// "fast-forward" in prepare_applied_target is really a rewind that drops
// every commit between the PR's base and where `main` actually is now.
let current_main = lifecycle::git_rev_parse(&ctx.applied_dir, "refs/heads/main")
.await
.map_err(|e| anyhow::anyhow!("read applied/main: {e:#}"))?;
if !lifecycle::git_is_ancestor(&ctx.applied_dir, &current_main, reviewed)
.await
.map_err(|e| anyhow::anyhow!("ancestry check {current_main}..{reviewed}: {e:#}"))?
{
bail!(
"PR #{pr} does not descend from applied/main (main {current_main}, reviewed {reviewed}); \
merging it would discard commits — rebase the PR onto main and re-review"
);
}
// 4. Eval-verify BEFORE the irreversible merge (bad nix fails fast here).
crate::meta::verify_commit(
ctx.approval.agent.as_str(),
&ctx.applied_dir,
reviewed,
node_id,
)
.await
.map_err(|e| anyhow::anyhow!("verify merge head {reviewed}: {e:#}"))?;
Ok(())
}
/// `DeployApply` node body — the irreversible half. Parks the rollback ref,
/// fast-forward-merges the PR (THE merge), then opens the deploy.
///
/// The ref is parked *before* the merge, so a hive-c0re crash anywhere from
/// here on still leaves [`run_deploy_tail`] enough to undo. If the merge itself
/// fails, `main` never moved and the tail's compensation is a no-op against the
/// same sha — harmless, and cheaper than trying to be clever about it.
///
/// Returning `Ok` is the signal for the caller to grow the rebuild subgraph into
/// this DAG under this node; the build itself does not happen here.
///
/// # Errors
///
/// Returns an error if the approval can't be loaded, if reading or parking the
/// pre-merge `main` sha fails, if the forge refuses the fast-forward merge
/// (including a head that drifted between verify and merge), or if
/// fast-forwarding `applied/main` / `meta::prepare_deploy` fails. From the merge
/// onward a failure is *not* retryable on its own — [`run_deploy_tail`] runs
/// `AfterAny` to compensate.
pub async fn run_deploy_apply(
coord: &Arc<Coordinator>,
approval_id: i64,
node_id: Option<u64>,
) -> Result<()> {
let ctx = deploy_ctx(coord, approval_id)?;
let agent = ctx.approval.agent.as_str();
let pr = ctx.pr;
let prev_main = lifecycle::git_rev_parse(&ctx.applied_dir, "refs/heads/main")
.await
.map_err(|e| anyhow::anyhow!("read applied/main: {e:#}"))?;
lifecycle::git_update_ref(&ctx.applied_dir, &rollback_ref(approval_id), &prev_main)
.await
.map_err(|e| anyhow::anyhow!("park rollback ref for approval {approval_id}: {e:#}"))?;
// THE merge: fast-forward-only merge the reviewed head to `main` via the
// forge API, pinned to the reviewed sha (`head_commit_id`). This one call
// both advances `main` to the reviewed head and marks the PR merged — no
// direct push to the protected branch. A failure here means `main` was NOT
// advanced, so it's fatal: we must not deploy a head the forge didn't merge.
match crate::forge::merge_config_pr_ff(&ctx.repo, pr, &ctx.reviewed).await {
Ok(()) => {}
Err(crate::forge::ForgeMergeError::HeadDrift { expected, actual }) => bail!(
"PR #{pr} head drifted before merge (reviewed {expected}, now {actual}); re-review before merging"
),
Err(e) => bail!("ff-merge PR #{pr}: {e}"),
}
prepare_applied_target(agent, &ctx.applied_dir, &ctx.reviewed, &prev_main, node_id).await
}
/// `DeployTail` node body — compensation + bookkeeping, `AfterAny` the apply
/// node so it runs on every outcome including a cancel-cascade. Infallible by
/// construction: it is the recovery step, so it has nothing to hand a failure
/// to. Every fallible call inside warns and continues.
///
/// 1. If the rollback ref survived, the deploy did not confirm good: tag the
/// merged commit `failed/<id>` (annotated with the DAG's first error, while
/// that sha is still reachable), then roll `applied/main` back to the parked
/// sha, resync the working tree, and drop the staged meta lock so the deploy
/// log only ever shows successes.
/// 2. Mirror the agent's config repo to the forge. `main` is already ff'd by
/// the merge, so only the `deployed/<id>` / `failed/<id>` tag refspec
/// actually lands — that's what gives the merged commit a forge-visible
/// deploy marker.
///
/// Takes `agent` from the node payload rather than the approval row so it still
/// works if the row vanished underneath the DAG (deny race, purge).
pub async fn run_deploy_tail(
coord: &Arc<Coordinator>,
queue_entry_id: Option<u64>,
agent: &str,
approval_id: i64,
) {
let applied_dir = crate::paths::applied_dir(agent);
let rollback = rollback_ref(approval_id);
if let Ok(prev_main) = lifecycle::git_rev_parse(&applied_dir, &rollback).await {
// Belt and braces: `run_deploy_apply` drops the ref before it plants
// `deployed/<id>`, so seeing both means the *delete* failed on an
// otherwise-successful deploy. Rolling back there would be the worst
// outcome this node can produce, so the tag wins.
if lifecycle::git_rev_parse(&applied_dir, &format!("deployed/{approval_id}"))
.await
.is_ok()
{
tracing::warn!(
%agent, approval_id,
"deploy tail: rollback ref outlived a successful deploy; dropping it without compensating"
);
} else {
// Mark the commit that failed to deploy, before undoing the merge
// that put it on `main`. After the rollback below, that sha is only
// reachable through this tag.
//
// Gated on `main` having actually moved: the rollback ref is parked
// *before* the merge, so its existence alone doesn't mean a merge
// happened. A pre-merge rejection (drift gate, eval failure, or the
// ff-merge itself failing) has no deployed commit to blame, and
// tagging the previous — innocent — head would point the operator at
// a commit that never got near a container.
//
// The annotation is read off the DAG rather than passed down from
// the node that failed: this node runs `AfterAny` its subject, so by
// now that node has settled `Failed` with its error recorded.
if let Ok(merged) = lifecycle::git_rev_parse(&applied_dir, "refs/heads/main").await
&& merged != prev_main
{
let tag = format!("failed/{approval_id}");
let body = queue_entry_id
.and_then(|dag_id| coord.job_queue.first_error(dag_id))
.unwrap_or_else(|| "deploy failed".to_owned());
if let Err(e) =
lifecycle::git_tag_annotated(&applied_dir, &tag, &merged, &body).await
{
tracing::warn!(%agent, approval_id, error = ?e, "deploy tail: annotate failed tag failed");
}
}
if let Err(e) =
lifecycle::git_update_ref(&applied_dir, "refs/heads/main", &prev_main).await
{
tracing::warn!(%agent, approval_id, error = ?e, "deploy tail: main rollback failed");
}
if let Err(e) = lifecycle::git_read_tree_reset(&applied_dir, "refs/heads/main").await {
tracing::warn!(%agent, approval_id, error = ?e, "deploy tail: rollback read-tree failed");
}
if let Err(e) = crate::meta::abort_deploy().await {
tracing::warn!(%agent, approval_id, error = ?e, "deploy tail: meta abort_deploy failed");
}
}
if let Err(e) = lifecycle::git_delete_ref(&applied_dir, &rollback).await {
tracing::warn!(%agent, approval_id, error = ?e, "deploy tail: drop rollback ref failed");
}
}
if let Err(e) = crate::forge::push_config(agent).await {
tracing::warn!(%agent, error = ?e, "forge: push_config after merge failed");
}
}
/// Max stderr bytes to inline in a PR failure comment. Keeps the comment
/// readable and under forge's size limits while still carrying the tail
/// where the nix/build error actually surfaces.
const PR_FAIL_LOG_TAIL_BYTES: usize = 4000;
/// On a failed `MergeConfigPr` deploy, post the failing build log back to the
/// config PR as a comment so the manager sees the rejection reason on the PR
/// itself. Best-effort: any error here is logged, never allowed to disturb the
/// approval-resolution path.
///
/// The failing `build_log` row is located heuristically: the most recent `fail`
/// row for this agent that started at/after the approval was decided (i.e. when
/// its deploy DAG was submitted). Because deploys are serialised per agent
/// through the queue, that is the step which just failed — `verify`,
/// `prepare-deploy`, `prebuild`, or the container rebuild. Pre-build failures
/// (drift gate, fetch) create no `build_log` row, so the comment then carries
/// only the error text.
async fn post_merge_failure_to_pr(
coord: &Arc<Coordinator>,
approval: &hive_sh4re::approvals::Approval,
err: &anyhow::Error,
) {
let Ok(pr) = approval.commit_ref.parse::<u64>() else {
return;
};
let repo = crate::forge::config_repo(approval.agent.as_str());
let since_ts = approval
.resolved_at
.unwrap_or(approval.requested_at)
.timestamp();
let log_section = coord
.build_logs
.list_recent_for_agent(approval.agent.as_str(), 10)
.ok()
.and_then(|rows| {
rows.into_iter()
.find(|r| r.status.as_deref() == Some("fail") && r.started_at >= since_ts)
})
.and_then(|row| coord.build_logs.get_full(row.id).ok().flatten())
.map(|full| {
let tail = tail_bytes(full.stderr.trim_end(), PR_FAIL_LOG_TAIL_BYTES);
format!(
"\n\n**Failing step:** `{}` (build log #{})\n\n```\n{tail}\n```",
full.header.kind, full.header.id
)
})
.unwrap_or_default();
let body = format!(
"## ⚠️ config deploy failed\n\n\
Approval #{} to merge this PR could not be deployed:\n\n\
```\n{err:#}\n```{log_section}",
approval.id
);
if let Err(e) = crate::forge::post_pr_comment(&repo, pr, &body).await {
tracing::warn!(agent = %approval.agent, %pr, error = ?e, "post merge-failure comment to PR failed");
}
}
/// Return the last `max_bytes` of `s`, snapped to a char boundary, prefixed
/// with an elision marker when truncated.
fn tail_bytes(s: &str, max_bytes: usize) -> String {
if s.len() <= max_bytes {
return s.to_owned();
}
let mut start = s.len() - max_bytes;
while start < s.len() && !s.is_char_boundary(start) {
start += 1;
}
format!("[… truncated …]\n{}", &s[start..])
}
/// Inline (non-queued) handler for `ApprovalKind::SchedulePrompt`.
/// On approve, decode the `SchedulePromptPayload` JSON from the
/// approval's `commit_ref`, insert a row into `scheduled_prompts`
/// (with `source = Approval { id }`), and fire `ApprovalResolved`.
/// The worker takes over from here — fan-out at fire time.
async fn run_approval_schedule_prompt(
coord: &Coordinator,
approval: hive_sh4re::approvals::Approval,
) -> Result<()> {
let result: Result<()> = async {
let payload: hive_sh4re::manager::SchedulePromptPayload =
serde_json::from_str(&approval.commit_ref)
.context("decode SchedulePromptPayload from approval.commit_ref")?;
coord
.scheduled_prompts
.submit(&crate::scheduled_prompts::NewSchedule {
owner: approval.agent.to_string(),
targets: payload.targets,
body: payload.body,
first_fire_at_unix: payload.first_fire_at_unix,
interval_seconds: payload.interval_seconds,
description: payload.description,
source: crate::scheduled_prompts::ScheduleSource::Approval { id: approval.id },
})
.map(|_| ())
.context("insert scheduled prompt")
}
.await;
finish_approval(coord, &approval, result, None).await
}
/// Resolve an approval row from how its DAG's work ended — the body of the
/// [`NodeKind::ResolveApproval`] tail node. Every approval-carrying template
/// resolves here, deploys included: the deploy pipeline is ordinary queue nodes,
/// so the work's terminal state is the authoritative outcome and there's no
/// in-node resolution to skip around.
///
/// `outcome` is the one the calling node was built to report — a template emits
/// one tail per outcome, so success, failure and cancel each arrive here from
/// their own node rather than from one node branching.
///
/// [`NodeKind::ResolveApproval`]: crate::job_queue::NodeKind::ResolveApproval
pub(crate) async fn resolve_approval_dag(
coord: &Arc<Coordinator>,
approval_id: i64,
outcome: crate::job_queue::TerminalState,
error: Option<&str>,
) {
use crate::job_queue::TerminalState;
let approval = match coord.approvals.get(approval_id) {
Ok(Some(a)) => a,
Ok(None) => {
tracing::warn!(approval_id, "approval dag terminal: row no longer exists");
return;
}
Err(e) => {
tracing::warn!(approval_id, error = ?e, "approval dag terminal: row read failed");
return;
}
};
let result: Result<()> = match outcome {
TerminalState::Done => Ok(()),
// `Skipped` never reaches here — no template emits a tail for it — but it
// reads the same to an operator either way: the work did not happen.
TerminalState::Cancelled | TerminalState::Skipped => {
Err(anyhow::anyhow!("cancelled before completion"))
}
TerminalState::Failed => Err(anyhow::anyhow!("{}", error.unwrap_or("job dag failed"))),
};
let mut terminal_tag = None;
match approval.kind {
ApprovalKind::Spawn => {
// Post-spawn forge bookkeeping (user, config repo mirror, meta
// access) — warn-only, then the resolution events + a rescan so
// the dashboard reflects the post-spawn state either way.
if result.is_ok() {
forge_after_first_spawn(coord, approval.agent.as_str()).await;
} else {
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(coord).await;
}
}
ApprovalKind::MergeConfigPr => {
terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, outcome).await;
// On a failed deploy, surface the failing build log back onto the
// PR so the manager sees why it was rejected without leaving the
// forge. Posted here rather than inside a node because this is the
// one place that holds the DAG's definitive error — a `MergeVerify`
// rejection and a `DeployApply` build failure both land here.
if let Err(e) = &result {
post_merge_failure_to_pr(coord, &approval, e).await;
}
}
_ => {}
}
if let Err(e) = finish_approval(coord, &approval, result, terminal_tag).await {
tracing::warn!(approval_id, error = ?e, "approval dag resolved with failure");
}
}
/// Which bookkeeping tag a settled deploy DAG actually planted, for the
/// `Rebuilt` event's `tag` field. The state picks the candidate name, but the
/// applied repo has the final say: a pre-merge rejection (`MergeVerify` drift
/// gate, eval failure) fails the DAG without ever planting `failed/<id>`, and
/// tag plants are best-effort. Reporting a tag that isn't there would send the
/// manager looking for a ref that doesn't exist.
async fn deploy_terminal_tag(
agent: &str,
approval_id: i64,
outcome: crate::job_queue::TerminalState,
) -> Option<String> {
use crate::job_queue::TerminalState;
let candidate = match outcome {
TerminalState::Done => format!("deployed/{approval_id}"),
// Nothing ran, so nothing was planted.
TerminalState::Cancelled | TerminalState::Skipped => return None,
TerminalState::Failed => format!("failed/{approval_id}"),
};
lifecycle::git_rev_parse(&crate::paths::applied_dir(agent), &candidate)
.await
.ok()
.map(|_| candidate)
}
/// Re-fetch an approval row from sqlite for a queue-worker dispatch.
/// Bails if the row is gone (deny race), if its kind doesn't match,
/// or if the lookup itself fails. The kind check is defensive — the
/// queue's `dispatch` already routes by `QueueKind`, but the approval
/// kind is the authoritative source of truth and a mismatch points
/// at a deeper bug we'd want to surface.
fn fetch_approval_for_worker(
coord: &Coordinator,
approval_id: i64,
expected_kind: ApprovalKind,
) -> Result<hive_sh4re::approvals::Approval> {
let approval = coord
.approvals
.get(approval_id)
.map_err(|e| anyhow::anyhow!("read approval {approval_id}: {e:#}"))?
.ok_or_else(|| anyhow::anyhow!("approval {approval_id} no longer exists"))?;
if approval.kind != expected_kind {
bail!(
"approval {approval_id} kind mismatch: queue expected {expected_kind:?}, row is {actual:?}",
actual = approval.kind
);
}
Ok(approval)
}
/// Forge bookkeeping run once after the very first container spawn:
/// create the per-agent forge user, mirror the applied repo, and grant
/// read access to core/meta. Also rescans containers so the dashboard
/// reflects the post-spawn state.
async fn forge_after_first_spawn(coord: &Arc<Coordinator>, agent: &str) {
if let Err(e) = crate::forge::ensure_user_for(agent).await {
tracing::warn!(%agent, error = ?e, "forge: ensure_user after first spawn failed");
}
if let Err(e) = crate::forge::ensure_config_repo(agent).await {
tracing::warn!(%agent, error = ?e, "forge: ensure_config_repo after first spawn failed");
}
if let Some(core_token) = crate::forge::core_token()
&& let Err(e) = crate::forge::meta_read_access(agent, &core_token).await
{
tracing::warn!(%agent, error = ?e, "forge: meta_read_access after first spawn failed");
}
if let Err(e) = crate::forge::ensure_meta_remote(agent).await {
tracing::warn!(%agent, error = ?e, "forge: ensure_meta_remote after first spawn failed");
}
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(coord).await;
}
/// Inline (non-queued) handler for `ApprovalKind::InitConfig`. Just
/// seeds the proposed git repo + the per-agent dirs — sub-second
/// work that doesn't justify a queue card.
async fn run_approval_init_config(
coord: &Coordinator,
approval: hive_sh4re::approvals::Approval,
proposed_dir: std::path::PathBuf,
claude_dir: std::path::PathBuf,
notes_dir: std::path::PathBuf,
) -> Result<()> {
let result: Result<()> = async {
// Place the new child under its requesting parent (carried in
// commit_ref by submit_init_config). An empty commit_ref means
// no explicit parent was named (privileged manager socket, or an
// approval queued before the new-child feature) — write no edge
// and let `topology::reconcile` assign the default position on
// first spawn, so this path never names a specific root agent.
if !approval.commit_ref.is_empty() {
crate::topology::add_child(approval.agent.as_str(), &approval.commit_ref)
.map_err(|e| anyhow::anyhow!("topology add_child: {e}"))?;
}
// Create the agent's state root as a btrfs subvolume FIRST, before
// any dir-seed touches it. `ensure_agent_state_subvolume` is
// progressive ("root exists → skip"), and `setup_proposed` does a
// `create_dir_all` on the proposed-config path which would
// materialise the state root as a plain directory — after which
// the subvolume create is silently skipped and the agent never
// lands on a subvolume (no quota, no snapshot). Order matters.
lifecycle::ensure_agent_state_subvolume(approval.agent.as_str()).await?;
lifecycle::setup_proposed(&proposed_dir, approval.agent.as_str()).await?;
lifecycle::ensure_claude_dir(&claude_dir)?;
lifecycle::ensure_state_dir(&notes_dir)?;
Ok(())
}
.await;
if result.is_ok()
&& let Err(e) = crate::forge::ensure_meta_remote(approval.agent.as_str()).await
{
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed");
}
finish_approval(coord, &approval, result, None).await
}
async fn finish_approval(
coord: &Coordinator,
approval: &hive_sh4re::approvals::Approval,
result: Result<()>,
terminal_tag: Option<String>,
) -> Result<()> {
let (status, note, ok) = match &result {
Ok(()) => (ApprovalStatus::Approved, None, true),
Err(e) => {
let note = format!("{e:#}");
let _ = coord.approvals.mark_failed(approval.id, &note);
(ApprovalStatus::Failed, Some(note), false)
}
};
coord.notify_submitter(
approval.id,
&HelperEvent::ApprovalResolved {
id: approval.id,
agent: approval.agent.to_string(),
commit_ref: approval.commit_ref.clone(),
status,
note: note.clone(),
sha: approval.fetched_sha.clone(),
tag: terminal_tag.clone(),
},
);
// Phase 5b: also fire on the dashboard event channel so the
// browser moves the row out of pending into history without a
// snapshot refetch. `approved` rows that succeed get the
// approval's logged resolved_at indirectly via `Utc::now()`;
// failures already wrote it via mark_failed above.
let approval_kind = approval.kind.as_str();
let sha_short = approval
.fetched_sha
.as_deref()
.map(|s| s[..s.len().min(12)].to_owned());
let status_str = if ok { "approved" } else { "failed" };
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id: approval.id,
agent: approval.agent.as_str(),
approval_kind,
sha_short,
status: status_str,
note: note.clone(),
description: approval.description.clone(),
});
// For spawn/rebuild/init_config approvals, also surface the underlying
// action so the manager knows whether the lifecycle step succeeded.
// The ApprovalResolved event already carries the same `ok` signal but
// separating it lets the manager react to the lifecycle change
// without having to special-case approvals.
match approval.kind {
ApprovalKind::InitConfig => {
if ok {
let _ = coord
.push_todo_submitter(
approval.id,
"core",
Some(format!("config_ready:{}", approval.agent)),
format!(
"agent '{}' config repo ready — edit + apply",
approval.agent
),
None,
)
.await;
}
}
ApprovalKind::Spawn => {
let summary = if ok {
format!("agent '{}' spawned", approval.agent)
} else {
format!(
"agent '{}' spawn FAILED: {}",
approval.agent,
note.as_deref().unwrap_or("unknown error")
)
};
let _ = coord
.push_todo_submitter(
approval.id,
"core",
Some(format!("spawned:{}", approval.agent)),
summary,
None,
)
.await;
}
// MergeConfigPr ends in a container rebuild — surface a Rebuilt
// lifecycle event. (It is never a first spawn — the agent already
// exists — so it never needs the Spawned arm above.)
ApprovalKind::MergeConfigPr => {
let summary = crate::coordinator::rebuilt_todo_summary(
approval.agent.as_str(),
ok,
note.as_deref(),
approval.fetched_sha.as_deref(),
terminal_tag.as_deref(),
);
let _ = coord
.push_todo_submitter(
approval.id,
"core",
Some(format!("rebuilt:{}", approval.agent)),
summary,
None,
)
.await;
}
// UpdateMetaInputs / SchedulePrompt: ApprovalResolved already
// carries the result. No separate lifecycle event needed.
ApprovalKind::UpdateMetaInputs | ApprovalKind::SchedulePrompt => {}
}
result
}
/// Open the deploy for the config-PR flow: fast-forward `applied/main` to
/// `target`, sync the working tree, and run phase 1 of the meta two-phase
/// deploy. The container rebuild that used to run inline here is now the
/// subgraph [`run_deploy_apply`]'s node grows into the DAG, and the closing half
/// is [`run_finalize_deploy`].
///
/// **Undo is not this function's job.** Every early return here leaves the
/// applied repo dirty on purpose — [`run_deploy_tail`] owns compensation, and
/// it runs whether this returns `Err`, panics, or never returns at all because
/// hive-c0re was restarted underneath it. That's the whole point of parking the
/// pre-merge sha in a git ref instead of a local variable.
///
/// Caller-specific bits stay OUT of here: fetching the PR head, the
/// `verify_commit` gate, and the ff-merge. The agent always already exists here
/// (a merge is never a first spawn), so there's no `sync_agents` step — the
/// operator `Spawn` flow owns first-time meta registration.
async fn prepare_applied_target(
agent: &str,
applied_dir: &std::path::Path,
target: &str,
expected_main: &str,
node_id: Option<u64>,
) -> Result<()> {
// Fast-forward applied/main to target + sync the working tree. Meta input
// pins `?ref=main`, so this is what makes nix re-lock to the target commit
// on the prepare_deploy step below.
//
// Compare-and-swap, not a bare set: `main` must still be the sha the caller
// read before the merge. A plain `update-ref` here moves the branch to
// `target` whatever it currently points at, which turns "fast-forward" into
// "discard anything that landed in the meantime" — the ancestry gate in
// run_deploy_merge_verify only proves the target is safe against the `main`
// observed *then*, so this is what makes that proof still true *now*.
lifecycle::git_update_ref_cas(applied_dir, "refs/heads/main", target, expected_main)
.await
.map_err(|e| {
anyhow::anyhow!("ff main {expected_main} -> {target} (concurrent move?): {e:#}")
})?;
lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main")
.await
.map_err(|e| anyhow::anyhow!("read-tree to main: {e:#}"))?;
// Phase 1 of the meta two-phase deploy: relock without committing. The
// staged lock then stays uncommitted across the whole appended rebuild —
// which is why the `MetaWindow` is held by the deploy root, not by a node.
crate::meta::prepare_deploy(agent, node_id)
.await
.map_err(|e| anyhow::anyhow!("meta prepare_deploy: {e:#}"))
}
/// `FinalizeDeploy` node body — phase 2 of the meta two-phase deploy, run once
/// the appended rebuild subgraph has built, swapped, and brought the container
/// back up.
///
/// Drops the rollback ref *first*: from here the deploy is good and
/// [`run_deploy_tail`] must not roll `main` back. Ordering that ahead of the tag
/// plant is what makes the tail's `deployed/<id>` cross-check a second line of
/// defence rather than the only one. No agent kick — the rebuild's own
/// `RebuildBookkeeping` already did it.
///
/// # Errors
///
/// Returns an error if the approval can't be loaded, if dropping the rollback
/// ref fails, or if planting the `deployed/<id>` tag fails. Those two git writes
/// *are* the deploy's "confirmed good" signal, so warning past them would let
/// this node report success while leaving the tail looking at the git state of a
/// failure — and the tail would then compensate a good deploy. A failing
/// `meta::finalize_deploy` is deliberately *not* an error: the container is
/// already running the new config by then, and the staged `flake.lock` it
/// couldn't commit is something the operator can land by hand.
pub async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) -> Result<()> {
let ctx = deploy_ctx(coord, approval_id)?;
let agent = ctx.approval.agent.as_str();
let target = ctx.reviewed.as_str();
lifecycle::git_delete_ref(&ctx.applied_dir, &rollback_ref(approval_id))
.await
.map_err(|e| anyhow::anyhow!("drop rollback ref for approval {approval_id}: {e:#}"))?;
let tag = format!("deployed/{approval_id}");
lifecycle::git_tag(&ctx.applied_dir, &tag, target)
.await
.map_err(|e| anyhow::anyhow!("plant {tag}: {e:#}"))?;
if let Err(e) = crate::meta::finalize_deploy(agent, target, &tag).await {
tracing::warn!(%agent, approval_id, error = ?e, "meta finalize_deploy failed");
}
Ok(())
}
/// Tear down a sub-agent container. By default this is non-destructive to
/// persistent state: the proposed/applied config repos and the Claude
/// credentials dir under `/var/lib/hyperhive/{agents,applied}/<name>/` are
/// kept, so recreating an agent of the same name reuses prior config + creds
/// (no re-login). The ephemeral runtime dir under `/run/hyperhive/agents/`
/// is cleared because its contents (the mcp socket) don't survive restarts
/// anyway. With `purge=true` the persistent trees are also wiped — config
/// history, claude creds, notes — there is no undo.
/// The bootstrap/root container is destroyable like any other: it's
/// imperative infra that `auto_update::ensure_root_agent` recreates on the
/// next hive-c0re startup if absent, so destroying it is transient rather
/// than something to refuse at the API.
///
/// Submits the teardown DAG and returns — it does not wait for the container to
/// go away. Same contract as every other lifecycle op (`rebuild`, `kill`,
/// `restart`, `start`): the queue owns the work, the caller gets an
/// acknowledgement. Progress is visible as real nodes on the dashboard.
pub fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) {
tracing::info!(%name, purge, "destroy");
if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::destroy(b, name, purge);
Vec::new()
}) {
tracing::error!(agent = %name, error = ?e, "destroy: insert failed");
}
coord.emit_rebuild_queue_snapshot();
}
pub fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()> {
let approval = coord.approvals.get(id)?;
coord.approvals.mark_denied(id, note)?;
tracing::info!(%id, note, "approval denied");
if let Some(a) = approval {
let sha = a.fetched_sha.clone();
let approval_kind = a.kind.as_str();
let sha_short = sha.as_deref().map(|s| s[..s.len().min(12)].to_owned());
let description = a.description.clone();
let agent_owned = a.agent.clone();
coord.notify_submitter(
a.id,
&HelperEvent::ApprovalResolved {
id: a.id,
agent: a.agent.to_string(),
commit_ref: a.commit_ref,
status: ApprovalStatus::Denied,
note: note.map(String::from),
sha,
// A denied config PR carries no git tag — it stays open on the forge.
tag: None,
},
);
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id,
agent: agent_owned.as_str(),
approval_kind,
sha_short,
status: "denied",
note: note.map(String::from),
description,
});
}
Ok(())
}