vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor(#2416): remove the non-pr config-change flow (request_apply_commit / applycommit)

Browse source
This commit is contained in:
damocles 2026-07-15 20:45:29 +02:00 • committed by mara
commit c2bd7db998
34 changed files with 293 additions and 1635 deletions
Download patch file Download diff file Expand all files Collapse all files

317
hive-c0re/src/actions.rs
View file

@ -15,10 +15,10 @@ use crate::lifecycle;
/// 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 `ApplyCommit`).
/// (operator no longer blocks on a 30-90s spinner for `MergeConfigPr`).
///
/// Dispatch:
/// - `ApplyCommit` / `MergeConfigPr` → a single-node `ApprovalDeploy`
/// - `MergeConfigPr` → a single-node `ApprovalDeploy`
/// DAG (the two-phase meta deploy stays opaque in v1; ~30-90s)
/// - `UpdateMetaInputs` → a `MetaUpdate` DAG (fan-out on completion)
/// - `Spawn` → a `Spawn` DAG (`Create → WriteDropin → Reconcile`)
@ -46,15 +46,6 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
let notes_dir = Coordinator::agent_notes_dir(&approval.agent);
run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await
}
ApprovalKind::ApplyCommit => {
enqueue_approval_rebuild(
&coord,
&approval.agent,
id,
format!("approval #{id} apply commit"),
);
Ok(())
}
ApprovalKind::UpdateMetaInputs => {
// Inputs JSON-encoded into commit_ref by the manager's
// submit path — surface them on the DAG so the dashboard
@ -109,11 +100,11 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
result
}
ApprovalKind::MergeConfigPr => {
// Like ApplyCommit, the work ends in a container rebuild, so
// route it through the rebuild queue. The queue worker
// dispatches MergeConfigPr approvals to `run_merge_config_pr`
// (verify the reviewed PR head, ff the forge config repo's
// main to it, mark merged, then the shared deploy tail).
// The work ends in a container rebuild, so route it through the
// rebuild queue. The queue worker dispatches MergeConfigPr
// approvals to `run_merge_config_pr` (verify the reviewed PR head,
// ff the forge config repo's main to it, mark merged, then the
// deploy tail).
enqueue_approval_rebuild(
&coord,
&approval.agent,
@ -126,10 +117,9 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
}
/// Submit the single-node `ApprovalDeploy` DAG tied to an approval id.
/// Shared by the `ApplyCommit` and `MergeConfigPr` dispatch arms — both
/// end in a container rebuild routed through the queue, differing only
/// in the `reason`. The node executor branches on the approval's kind
/// to pick the right handler.
/// Used by the `MergeConfigPr` dispatch arm — the work ends in a container
/// rebuild routed through the queue; the node executor runs
/// `run_merge_config_pr`.
fn enqueue_approval_rebuild(
coord: &Arc<Coordinator>,
agent: &str,
@ -149,40 +139,9 @@ fn enqueue_approval_rebuild(
coord.emit_rebuild_queue_snapshot();
}
/// Worker entry point for `ApprovalKind::ApplyCommit` queue entries.
/// Re-fetches the approval row, runs the commit pipeline, and fires
/// `ApprovalResolved` + the lifecycle event (`Rebuilt` / `Spawned`
/// for first-spawn).
pub async fn run_approval_apply_commit(
coord: &Arc<Coordinator>,
queue_entry_id: Option<u64>,
approval_id: i64,
) -> Result<()> {
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::ApplyCommit)?;
// Runtime dir creation is handled inside lifecycle::rebuild_no_meta's
// spawn path (first-spawn) or is already present for rebuilds.
let agent_dir = crate::paths::agent_runtime_dir(&approval.agent);
let applied_dir = crate::paths::applied_dir(&approval.agent);
coord.set_queue_step(queue_entry_id, "apply commit");
let (result, terminal_tag, is_first_spawn) =
run_apply_commit(coord, &approval, &agent_dir, &applied_dir, queue_entry_id).await;
coord.set_queue_step(queue_entry_id, "forge push");
if let Err(e) = crate::forge::push_config(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after apply failed");
}
if is_first_spawn && result.is_ok() {
coord.set_queue_step(queue_entry_id, "first-spawn forge bootstrap");
forge_after_first_spawn(coord, &approval.agent).await;
}
// `finish_approval` returns the original `result` so the queue
// worker sees Ok/Err and marks the queue entry accordingly. The
// approval row + helper events have already been fanned out.
finish_approval(coord, &approval, result, terminal_tag, is_first_spawn)
}
/// Worker entry point for `ApprovalKind::MergeConfigPr` queue entries —
/// the PR-based config flow's counterpart to `run_approval_apply_commit`.
/// Re-fetches the approval row, runs the merge pipeline, and fires
/// Worker entry point for `ApprovalKind::MergeConfigPr` queue entries — the
/// config-change flow's deploy worker. Re-fetches the approval row, runs the
/// merge pipeline, and fires
/// `ApprovalResolved` + the `Rebuilt` lifecycle event via `finish_approval`.
/// `run_merge_config_pr` already fast-forwarded the forge repo's `main` to the
/// reviewed head (that IS the merge), so `push_config`'s `main` refspec is a
@ -218,7 +177,7 @@ pub async fn run_approval_merge_config_pr(
if let Err(e) = &result {
post_merge_failure_to_pr(coord, &approval, since_ts, e).await;
}
finish_approval(coord, &approval, result, terminal_tag, false)
finish_approval(coord, &approval, result, terminal_tag)
}
/// Max stderr bytes to inline in a PR failure comment. Keeps the comment
@ -298,7 +257,7 @@ fn tail_bytes(s: &str, max_bytes: usize) -> String {
/// 2. fetch the reviewed head into the applied repo so later git ops resolve
/// it locally;
/// 3. eval-verify the reviewed commit against the meta flake BEFORE the
/// irreversible push (same gate `run_apply_commit` uses);
/// irreversible push;
/// 4. fast-forward the forge repo's `main` to the reviewed head — THE merge;
/// 5. mark the PR merged (best-effort: `main` is already at the head, so a
/// failure here is logged, not fatal);
@ -399,8 +358,7 @@ async fn run_merge_config_pr(
Err(e) => return (Err(anyhow::anyhow!("ff-merge PR #{pr}: {e}")), None),
}
// 5. Shared deploy tail. target == finalize == the reviewed head;
// never a first spawn (the agent already exists).
// 5. Deploy tail. target == finalize == the reviewed head.
deploy_applied_target(
coord,
&approval.agent,
@ -410,7 +368,6 @@ async fn run_merge_config_pr(
&reviewed,
id,
&prev_main_sha,
false,
queue_entry_id,
)
.await
@ -443,7 +400,7 @@ async fn run_approval_schedule_prompt(
.context("insert scheduled prompt")
}
.await;
finish_approval(coord, &approval, result, None, false)
finish_approval(coord, &approval, result, None)
}
/// Terminal hook for approval-carrying DAGs — the job queue's
@ -497,7 +454,7 @@ pub(crate) async fn resolve_approval_dag(
crate::dashboard::emit_tombstones_snapshot(coord).await;
}
}
if let Err(e) = finish_approval(coord, &approval, result, None, false) {
if let Err(e) = finish_approval(coord, &approval, result, None) {
tracing::warn!(approval_id, error = ?e, "approval dag resolved with failure");
}
}
@ -590,7 +547,7 @@ async fn run_approval_init_config(
{
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed");
}
finish_approval(coord, &approval, result, None, false)
finish_approval(coord, &approval, result, None)
}
fn finish_approval(
@ -598,7 +555,6 @@ fn finish_approval(
approval: &hive_sh4re::Approval,
result: Result<()>,
terminal_tag: Option<String>,
is_first_spawn: bool,
) -> Result<()> {
let (status, note, ok) = match &result {
Ok(()) => (ApprovalStatus::Approved, None, true),
@ -665,22 +621,10 @@ fn finish_approval(
sha: approval.fetched_sha.clone(),
},
),
ApprovalKind::ApplyCommit if is_first_spawn => {
coord.notify_submitter(
approval.id,
&HelperEvent::Spawned {
agent: approval.agent.clone(),
ok,
note,
sha: approval.fetched_sha.clone(),
},
);
}
// MergeConfigPr ends in a container rebuild just like a
// non-first-spawn ApplyCommit, so both surface the same Rebuilt
// lifecycle event. (MergeConfigPr is never a first spawn — the
// agent already exists — so it never hits the Spawned arm above.)
ApprovalKind::ApplyCommit | ApprovalKind::MergeConfigPr => {
// 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 => {
coord.notify_submitter(
approval.id,
&HelperEvent::Rebuilt {
@ -699,148 +643,25 @@ fn finish_approval(
result
}
/// Tag-driven `ApplyCommit` handler. Walks the approval through the tag
/// state machine documented in `docs/approvals.md`: stamp
/// `approved/<id>` and `building/<id>` first so the audit trail
/// captures intent, then drop the candidate tree into the working dir
/// without moving HEAD, run the rebuild, and either fast-forward
/// `applied/main` to the proposal commit on success
/// (`deployed/<id>`) or annotate `failed/<id>` with the build error
/// and reset the working tree back to the last known-good main. main
/// never advances on a failed build, so a crash-and-recover doesn't
/// leave the agent pointing at a tree it can't evaluate. The shared
/// ff/deploy/rebuild/finalize tail lives in `deploy_applied_target`.
async fn run_apply_commit(
coord: &Arc<Coordinator>,
approval: &hive_sh4re::Approval,
agent_dir: &std::path::Path,
applied_dir: &std::path::Path,
queue_entry_id: Option<u64>,
) -> (Result<()>, Option<String>, bool) {
let id = approval.id;
let proposal_ref = format!("refs/tags/proposal/{id}");
// Detect first spawn before we touch anything so we can branch on it
// throughout this function.
let is_first_spawn = !lifecycle::container_exists(&approval.agent).await;
// Defensive: submit-time should have planted proposal/<id>, but if
// the row was migrated from an older schema or the tag got pruned
// we fail early with a clear note rather than building a stale
// tree.
if let Err(e) = lifecycle::git_rev_parse(applied_dir, &proposal_ref).await {
return (
Err(anyhow::anyhow!(
"missing proposal tag {proposal_ref}: {e:#}"
)),
None,
is_first_spawn,
);
}
// Capture the currently-deployed sha so we can roll applied/main
// (and the meta lock indirectly) back if the build fails.
let prev_main_sha = match lifecycle::git_rev_parse(applied_dir, "refs/heads/main").await {
Ok(s) => s,
Err(e) => {
return (
Err(anyhow::anyhow!("read applied/main: {e:#}")),
None,
is_first_spawn,
);
}
};
// Pre-flight eval-verify the proposal commit against the meta flake
// WITHOUT mutating applied/main or the meta lock, so an evaluation
// error (bad nix, missing module option, unresolvable lock) fails
// fast here instead of after we've fast-forwarded main and have to
// roll it back. Skipped on first spawn: the agent has no
// `agent-<name>` meta input to override yet (sync_agents adds it
// below). This is the reusable verify primitive the PR-based config
// flow gates its irreversible ff-push on.
if !is_first_spawn {
let proposal_sha = match lifecycle::git_rev_parse(applied_dir, &proposal_ref).await {
Ok(s) => s,
Err(e) => {
return (
Err(anyhow::anyhow!("rev-parse {proposal_ref}: {e:#}")),
None,
is_first_spawn,
);
}
};
coord.set_queue_step(queue_entry_id, "verify proposal (eval)");
if let Err(e) =
crate::meta::verify_commit(&approval.agent, applied_dir, &proposal_sha).await
{
return (
Err(anyhow::anyhow!("verify proposal {proposal_ref}: {e:#}")),
None,
is_first_spawn,
);
}
}
coord.set_queue_step(queue_entry_id, "plant tags");
if let Err(e) = lifecycle::git_tag(applied_dir, &format!("approved/{id}"), &proposal_ref).await
{
return (
Err(anyhow::anyhow!("plant approved/{id}: {e:#}")),
None,
is_first_spawn,
);
}
if let Err(e) = lifecycle::git_tag(applied_dir, &format!("building/{id}"), &proposal_ref).await
{
return (
Err(anyhow::anyhow!("plant building/{id}: {e:#}")),
None,
is_first_spawn,
);
}
// Fast-forward applied/main to the proposal, run the meta deploy +
// container rebuild, and finalize/roll-back — the tail shared with the
// PR-merge flow. ApplyCommit's target == finalize sha source is
// `fetched_sha` (or the proposal ref when unset), matching the prior
// inline behavior exactly.
let (result, tag) = deploy_applied_target(
coord,
&approval.agent,
agent_dir,
applied_dir,
&proposal_ref,
approval.fetched_sha.as_deref().unwrap_or(&proposal_ref),
id,
&prev_main_sha,
is_first_spawn,
queue_entry_id,
)
.await;
(result, tag, is_first_spawn)
}
/// Shared deploy tail for config-applying approvals (`ApplyCommit` + the
/// PR-merge flow). Fast-forwards `applied/main` to `target_ref`, syncs the
/// working tree, runs the meta two-phase deploy + container rebuild, and
/// plants the `deployed/<tag_base>` / `failed/<tag_base>` bookkeeping tags.
/// On build failure it rolls `applied/main` back to `prev_main_sha` and aborts
/// the staged meta lock so the agent stays on its last-good tree. Returns the
/// build result + the terminal tag name.
/// Deploy tail for the config-PR merge flow. Fast-forwards `applied/main` to
/// `target_ref`, syncs the working tree, runs the meta two-phase deploy +
/// container rebuild, and plants the `deployed/<tag_base>` /
/// `failed/<tag_base>` bookkeeping tags. On build failure it rolls
/// `applied/main` back to `prev_main_sha` and aborts the staged meta lock so
/// the agent stays on its last-good tree. Returns the build result + the
/// terminal tag name.
///
/// Caller-specific bits stay OUT of here: the source fetch (proposal tag vs
/// forge fetch), the `approved/building` tags, `verify_commit`, and any forge
/// ff-push / mark-merged. `is_first_spawn` gates the one-time meta
/// `sync_agents` step (only `ApplyCommit`'s first spawn passes `true`;
/// the PR-merge flow always passes `false` — the agent already exists).
/// `finalize_sha` is the sha recorded by `meta::finalize_deploy`; `target_ref`
/// is what `applied/main` fast-forwards to (a proposal ref or a commit sha).
/// Caller-specific bits stay OUT of here: fetching the PR head, the
/// `verify_commit` gate, the ff-merge, and forge mark-merged. `finalize_sha`
/// is the sha recorded by `meta::finalize_deploy`; `target_ref` is what
/// `applied/main` fast-forwards to. 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.
#[allow(
clippy::too_many_arguments,
clippy::too_many_lines,
reason = "one sequential ff/deploy/rebuild/finalize pipeline shared by both \
config-apply callers; splitting it would obscure the linear flow"
reason = "one sequential ff/deploy/rebuild/finalize pipeline; splitting it \
would obscure the linear flow"
)]
async fn deploy_applied_target(
coord: &Arc<Coordinator>,
@ -851,7 +672,6 @@ async fn deploy_applied_target(
finalize_sha: &str,
tag_base: i64,
prev_main_sha: &str,
is_first_spawn: bool,
queue_entry_id: Option<u64>,
) -> (Result<()>, Option<String>) {
let id = tag_base;
@ -872,33 +692,6 @@ async fn deploy_applied_target(
return (Err(anyhow::anyhow!("read-tree to main: {e:#}")), None);
}
// First spawn: sync_agents must add this agent to the meta flake
// before prepare_deploy can update its input lock (which won't
// exist yet if this is the agent's first deploy).
if is_first_spawn {
coord.set_queue_step(queue_entry_id, "meta sync_agents (first spawn)");
let agents = match lifecycle::agents_for_meta_listing_with(agent).await {
Ok(a) => a,
Err(e) => {
let _ =
lifecycle::git_update_ref(applied_dir, "refs/heads/main", prev_main_sha).await;
let _ = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await;
return (
Err(anyhow::anyhow!("agents_for_meta_listing_with: {e:#}")),
None,
);
}
};
if let Err(e) = crate::meta::sync_agents(&coord.hive_env(), &agents).await {
let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", prev_main_sha).await;
let _ = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await;
return (
Err(anyhow::anyhow!("meta sync_agents for first spawn: {e:#}")),
None,
);
}
}
coord.set_queue_step(queue_entry_id, "meta prepare_deploy");
// Phase 1 of the meta two-phase deploy: relock without committing.
if let Err(e) = crate::meta::prepare_deploy(agent).await {
@ -1066,42 +859,14 @@ async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> {
crate::meta::sync_agents(&coord.hive_env(), &agents).await
}
pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()> {
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");
let mut tag = None;
// MergeConfigPr denials carry no git tag — the PR stays open on the forge.
let tag: Option<String> = None;
if let Some(a) = approval {
let sha = a.fetched_sha.clone();
// ApplyCommit approvals leave a `denied/<id>` tag on the
// proposal commit so rejected configs are first-class git
// objects — `git show denied/<id>` in the manager's applied
// mount yields both the tree the operator rejected and (in
// the annotated body) the reason. Spawn approvals have no
// commit to tag, so they fall through unannotated.
if matches!(a.kind, ApprovalKind::ApplyCommit) {
let applied_dir = crate::paths::applied_dir(&a.agent);
let proposal_ref = format!("refs/tags/proposal/{id}");
if lifecycle::git_rev_parse(&applied_dir, &proposal_ref)
.await
.is_ok()
{
let tag_name = format!("denied/{id}");
let body = note.unwrap_or("").to_owned();
if let Err(e) =
lifecycle::git_tag_annotated(&applied_dir, &tag_name, &proposal_ref, &body)
.await
{
tracing::warn!(%id, error = ?e, "plant denied tag failed");
} else {
tag = Some(tag_name);
}
}
// Mirror the denied/<id> tag to the forge.
if let Err(e) = crate::forge::push_config(&a.agent).await {
tracing::warn!(%id, agent = %a.agent, error = ?e, "forge: push_config after deny failed");
}
}
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();