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

wip(#1838): extract deploy_applied_target shared deploy tail + dedup approve dispatch

Browse source
This commit is contained in:
damocles 2026-06-23 11:05:54 +02:00
commit 5f5626456a
1 changed files with 139 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -46,20 +46,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await
}
ApprovalKind::ApplyCommit => {
coord
.rebuild_queue
.enqueue_full(crate::rebuild_queue::FullEnqueue {
kind: crate::rebuild_queue::QueueKind::Rebuild,
agent: approval.agent.clone(),
source: crate::rebuild_queue::QueueSource::Approval,
reason: format!("approval #{id} apply commit"),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(id),
perm_payload: None,
depends_on: Vec::new(),
});
coord.emit_rebuild_queue_snapshot();
enqueue_approval_rebuild(
&coord,
&approval.agent,
id,
format!("approval #{id} apply commit"),
);
Ok(())
}
ApprovalKind::UpdateMetaInputs => {
@ -134,25 +126,43 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// 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).
coord
.rebuild_queue
.enqueue_full(crate::rebuild_queue::FullEnqueue {
kind: crate::rebuild_queue::QueueKind::Rebuild,
agent: approval.agent.clone(),
source: crate::rebuild_queue::QueueSource::Approval,
reason: format!("approval #{id} merge config pr"),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(id),
perm_payload: None,
depends_on: Vec::new(),
});
coord.emit_rebuild_queue_snapshot();
enqueue_approval_rebuild(
&coord,
&approval.agent,
id,
format!("approval #{id} merge config pr"),
);
Ok(())
}
}
}
/// Enqueue a `Rebuild` queue entry 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 queue `reason`.
/// The queue worker branches on the approval's kind to pick the right handler.
fn enqueue_approval_rebuild(
coord: &Arc<Coordinator>,
agent: &str,
approval_id: i64,
reason: String,
) {
coord
.rebuild_queue
.enqueue_full(crate::rebuild_queue::FullEnqueue {
kind: crate::rebuild_queue::QueueKind::Rebuild,
agent: agent.to_owned(),
source: crate::rebuild_queue::QueueSource::Approval,
reason,
parent_id: None,
inputs: Vec::new(),
approval_id: Some(approval_id),
perm_payload: None,
depends_on: Vec::new(),
});
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`
@ -446,22 +456,19 @@ fn finish_approval(
sha: approval.fetched_sha.clone(),
});
}
ApprovalKind::ApplyCommit => coord.notify_manager(&HelperEvent::Rebuilt {
agent: approval.agent.clone(),
ok,
note,
sha: approval.fetched_sha.clone(),
tag: terminal_tag,
}),
// MergeConfigPr ends in a container rebuild like ApplyCommit, so
// surface the same Rebuilt lifecycle event.
ApprovalKind::MergeConfigPr => coord.notify_manager(&HelperEvent::Rebuilt {
agent: approval.agent.clone(),
ok,
note,
sha: approval.fetched_sha.clone(),
tag: terminal_tag,
}),
// 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 => {
coord.notify_manager(&HelperEvent::Rebuilt {
agent: approval.agent.clone(),
ok,
note,
sha: approval.fetched_sha.clone(),
tag: terminal_tag,
});
}
// UpdateMetaInputs / SchedulePrompt: ApprovalResolved already
// carries the result. No separate lifecycle event needed.
ApprovalKind::UpdateMetaInputs | ApprovalKind::SchedulePrompt => {}
@ -478,12 +485,8 @@ fn finish_approval(
/// (`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.
#[allow(
clippy::too_many_lines,
reason = "one sequential build/tag/notify pipeline; splitting the steps \
across helpers would obscure the linear flow without shrinking it"
)]
/// 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,
@ -574,28 +577,76 @@ async fn run_apply_commit(
);
}
// 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.
///
/// 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).
#[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"
)]
async fn deploy_applied_target(
coord: &Arc<Coordinator>,
agent: &str,
agent_dir: &std::path::Path,
applied_dir: &std::path::Path,
target_ref: &str,
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;
coord.set_queue_step(queue_entry_id, "fast-forward applied/main");
// Fast-forward applied/main to proposal/<id> + sync the working
// tree. Meta input pins `?ref=main`, so this is what makes nix
// re-lock to the proposal commit on the prepare_deploy step
// below. On build failure we roll main back to prev_main_sha so
// a crash leaves the agent on its last-good tree.
if let Err(e) = lifecycle::git_update_ref(applied_dir, "refs/heads/main", &proposal_ref).await {
return (
Err(anyhow::anyhow!("ff main to {proposal_ref}: {e:#}")),
None,
is_first_spawn,
);
// Fast-forward applied/main to target_ref + 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. On build
// failure we roll main back to prev_main_sha so a crash leaves the
// agent on its last-good tree.
if let Err(e) = lifecycle::git_update_ref(applied_dir, "refs/heads/main", target_ref).await {
return (Err(anyhow::anyhow!("ff main to {target_ref}: {e:#}")), None);
}
if let Err(e) = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await {
// main is ahead; working tree didn't sync. Roll main back to
// keep the two consistent before bailing.
let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await;
return (
Err(anyhow::anyhow!("read-tree to main: {e:#}")),
None,
is_first_spawn,
);
let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", prev_main_sha).await;
return (Err(anyhow::anyhow!("read-tree to main: {e:#}")), None);
}
// First spawn: sync_agents must add this agent to the meta flake
@ -603,40 +654,34 @@ async fn run_apply_commit(
// 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(&approval.agent).await {
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;
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,
is_first_spawn,
);
}
};
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_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,
is_first_spawn,
);
}
}
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(&approval.agent).await {
let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await;
if let Err(e) = crate::meta::prepare_deploy(agent).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 prepare_deploy: {e:#}")),
None,
is_first_spawn,
);
return (Err(anyhow::anyhow!("meta prepare_deploy: {e:#}")), None);
}
// Container-level rebuild (or first-time create) against meta#<name>.
@ -644,9 +689,9 @@ async fn run_apply_commit(
// the dashboard reflects actual phase progress rather than a static
// "nixos-container update" label for the whole multi-minute window.
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(&approval.agent, agent_dir.to_path_buf());
let paths = Coordinator::agent_paths(agent, agent_dir.to_path_buf());
let build_result = lifecycle::rebuild_no_meta(
&approval.agent,
agent,
&hive,
&paths,
&|step| coord.set_queue_step(queue_entry_id, step),
@ -664,54 +709,47 @@ async fn run_apply_commit(
Ok(()) => {
coord.set_queue_step(queue_entry_id, "finalize deploy");
let tag = format!("deployed/{id}");
if let Err(e) = lifecycle::git_tag(applied_dir, &tag, &proposal_ref).await {
tracing::warn!(agent = %approval.agent, %id, error = ?e, "plant deployed tag failed");
if let Err(e) = lifecycle::git_tag(applied_dir, &tag, target_ref).await {
tracing::warn!(%agent, %id, error = ?e, "plant deployed tag failed");
}
if let Err(e) = crate::meta::finalize_deploy(
&approval.agent,
approval.fetched_sha.as_deref().unwrap_or(&proposal_ref),
&tag,
)
.await
{
if let Err(e) = crate::meta::finalize_deploy(agent, finalize_sha, &tag).await {
// The build itself succeeded — meta lock landed but
// couldn't be committed. Surface as a soft warn so the
// operator can git-commit by hand if they care.
tracing::warn!(agent = %approval.agent, %id, error = ?e, "meta finalize_deploy failed");
tracing::warn!(%agent, %id, error = ?e, "meta finalize_deploy failed");
}
// Wake the agent on its next turn so claude sees the
// config change took effect. Same hint pattern as
// auto_update::rebuild_agent — manager approved a
// proposal, agent picks up where it left off with the
// new env / packages.
coord.kick_agent(&approval.agent, "config update applied");
(Ok(()), Some(tag), is_first_spawn)
coord.kick_agent(agent, "config update applied");
(Ok(()), Some(tag))
}
Err(e) => {
let tag = format!("failed/{id}");
let body = format!("{e:#}");
if let Err(te) =
lifecycle::git_tag_annotated(applied_dir, &tag, &proposal_ref, &body).await
lifecycle::git_tag_annotated(applied_dir, &tag, target_ref, &body).await
{
tracing::warn!(agent = %approval.agent, %id, error = ?te, "annotate failed tag failed");
tracing::warn!(%agent, %id, error = ?te, "annotate failed tag failed");
}
// Roll main back to last known-good so the on-disk state
// matches what nixos-container last successfully built.
if let Err(re) =
lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await
lifecycle::git_update_ref(applied_dir, "refs/heads/main", prev_main_sha).await
{
tracing::warn!(agent = %approval.agent, %id, error = ?re, "main rollback failed");
tracing::warn!(%agent, %id, error = ?re, "main rollback failed");
}
if let Err(re) = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await {
tracing::warn!(agent = %approval.agent, %id, error = ?re, "rollback read-tree failed");
tracing::warn!(%agent, %id, error = ?re, "rollback read-tree failed");
}
// Drop the staged meta lock change so the deploy log
// only ever shows successes.
if let Err(ae) = crate::meta::abort_deploy().await {
tracing::warn!(agent = %approval.agent, %id, error = ?ae, "meta abort_deploy failed");
tracing::warn!(%agent, %id, error = ?ae, "meta abort_deploy failed");
}
let _ = coord;
(Err(e), Some(tag), is_first_spawn)
(Err(e), Some(tag))
}
}
}