Compare commits

...
Author SHA1 Message Date
atlas
ddc017f01b wip(#3001): convert tests off the container id; drop Source + insert_group
The last of the DAG-container removal. `tests.rs` navigated by the id
`submit` returned, so removing the container removed the tests' way of
finding what they inserted; they name the roots they assert on now, which
is the same handle production uses.

Three findings the port surfaced, each a behaviour change rather than a
test fix:

- Cancelling a rebuild's head no longer drops the job. `Reconcile`'s edge
  accepts a skipped brace, and a cancel-cascade skips rather than cancels,
  so the tail stays claimable. Dropping a job means cancelling every id the
  insert returned.
- A directly-cancelled group root reads terminal while a spared tail still
  runs; the cancel used to land on a node above it, which rolled up
  Finishing instead.
- "One DAG per hive-wide op" is not expressible without a container. The
  three tests asserting it now assert that every named root is top-level,
  which is what makes the per-agent subgraphs concurrent.

Deletes two tests: one asserted only that two containers get distinct ids,
the other re-ran an existing case under a second name.

`Source`, `insert_group` and the stop path's `reason` string went dead with
the container and are removed with it.
2026-08-04 19:57:32 +02:00
atlas
aef7ead0bc wip(#3001): delete NodeKind::Dag, the container this issue is about
The variant, its label, its agent-accessor arm and its no-op executor arm are
gone, along with the module prose describing a job as "a single container node
whose subtree is the work". A job is now just its nodes: a template declares
them and names the roots it wants back.

`dag_of` becomes `root_of`. It always wrapped the graph's `root_of` and still
returns the same thing, but the old name asserted a concept that no longer
exists — with no container, the parent chain ends at whichever root the template
declared, so the honest question is "which root owns this node", not "which DAG
is this in".

One comment kept its old wording on purpose: `visible_roots` explains that the
projection it replaced keyed on the container kind rather than selecting
structurally. That is a statement about the past and stays true; it now says
"the since-removed container kind" rather than naming a type that is not there
to look up.
2026-08-04 19:57:32 +02:00
atlas
0523b4f7de wip(#3001): convert the last submit call sites; the binary compiles again
`server.rs`'s five sites move to `power::{stop,start,restart}_many` and direct
template inserts. `submit_single` routes through the `*_many` builders with a
one-element slice rather than keeping a parallel single-target shape.

`templates::rebuild` and `templates::reparent` now return the guids of the roots
they declare, so a caller that has to wait on them can name them; previously
only the void-returning form existed and every caller got an empty id list.

Two comments corrected while converting, both contradicted by the code they sit
above:

* `templates::rebuild` said its tail is edged onto "(MetaSync, Prebuild,
  Reconcile)" and that "Prebuild's roll-up carries the subtree" — the brace has
  been the middle root since the AgentWindow change.
* the restart handler described the per-agent shape as starting with SetWanted,
  while `restart_chain`'s own doc says a restart never rewrites `wanted` — that
  is the difference between restart and stop/start.

Error handling is no longer swallowed: a failed insert becomes a reported error
rather than a silently-absent id.
2026-08-04 19:57:32 +02:00
atlas
02e916feee wip(#3001): power chains name their group roots
The `*_many` entry points returned `insert_job`'s result while their closures
ended in `Vec::new()` — naming nothing, so the returned id list was always
empty. `queued_dags` would have shipped `Some([])` and hivectl's wait loop would
have had nothing to poll. Silent: it compiles, the op still runs, and no test in
isolation looks.

Each `*_chain` now returns its group root's guid and the `*_nodes` collectors
gather them, so the ids a caller gets back are the roots it can actually wait on.

`start_chain` returns *four* in the stale branch, not one: `rebuild_nodes`
chains its roots behind `SetWanted` with `after_ok` rather than nesting them
under it, so `SetWanted` rolls up only itself. Naming it alone would have
reported the start complete while the rebuild was still running — the same
under-reporting bug one level down.
2026-08-04 19:57:32 +02:00
atlas
dfb88e2dc2 wip(#3001): catch the multi-line .insert( chains the grep missed 2026-08-04 19:57:32 +02:00
atlas
fe52037b0d wip(#3001): rename insert -> insert_job per mara's 50056 2026-08-04 19:57:32 +02:00
atlas
102ebdc03d wip(#3001): 9 of 10 test helpers off the container id; sweep stale prose
The helpers keep shrinking the same way: resolve_id goes, 'n.id != root'
goes (the container was the only non-work node), root_of goes, and the
container-parent normalisation goes because a group root now genuinely
has parent = None. pending_kinds_filtered drops from a four-clause
multi-line filter to one line.

Also removed a doc block my earlier edit had orphaned above the renamed
helper, and swept 'under `dag`' / '`submit` returns' out of the prose.

state_of stays untouched: it reads a roll-up, which is the same question
as hivectl's queued_dags.
2026-08-04 19:57:32 +02:00
atlas
be4763678b wip(#3001): test helpers off the container id
submit() -> insert() in tests, and the two shape walkers lose their dag
param: with no container there is no per-DAG root to filter on, nothing
to exclude (every node is real work now), and a group root genuinely has
parent = None, so the container-parent normalisation goes too. Each test
builds a fresh JobQueue, so "the DAG" is "the graph".

20 errors remain, all in tests.rs, and they are the point: changing the
helper's type from u64 to () made every site that consumed the container
id light up as `expected u64, found ()`. A type error is an exhaustive
grep -- ten helpers take a dag id, not the three I had measured.

state_of(q, dag_id) is not mechanical: it read the DAG's ROLLED-UP state,
which was the container node's own. That makes it the second consumer of
the container-as-roll-up-point, alongside hivectl's queued_dags poll.
Both want the same answer, so it waits on the same ruling.
2026-08-04 19:57:32 +02:00
atlas
f04a0cee92 wip(#3001): convert remaining unblocked call sites; sweep docs
21 of 28 non-test call sites now insert directly. power.rs compiles.
The only remaining errors are server.rs's 5, which are blocked: those
sites feed the returned id into HostResponse::queued -> `queued_dags`,
a wire field hivectl polls via QueueDag. Removing the container without
answering that breaks hivectl's wait/progress loop; asked on the issue.

Also swept the deleted symbol out of prose, not just code:
- docs/coordinator.md: "the submit layer (job_queue/submit.rs)" ->
  the power layer (job_queue/power.rs), and "submits" -> "inserts".
- templates.rs module doc: points at super::power for the power ops.
- lifecycle_ops.rs module doc: says which path each op takes now.
- mod.rs's insert_group comment restated the open issue verbatim
  ("a DAG is addressed by its container node, which submit inserts
  itself"). Replaced with what is actually true for that path.

Dashboard behaviour deltas worth review: insert failures are now
logged per agent instead of swallowed, and UPDATE-ALL emits one queue
snapshot after the loop rather than one per agent.
2026-08-04 19:57:32 +02:00
atlas
7c0d9d2379 wip(#3001): remove submit layer, rescue power ops into job_queue/power.rs
TREE IS RED ON PURPOSE — there is no compiling intermediate between
deleting submit and converting every caller. Checkpoint commit so the
work is durable; do not "fix" it by restoring submit.

Done:
- JobQueue::submit -> JobQueue::insert (no source/reason/container;
  returns the ids insert_job names).
- submit.rs deleted. Its 6 pure chain builders + 3 async *_many
  gatherers were NOT wrapper code and are rescued into
  job_queue/power.rs (templates.rs documents power ops as living
  outside it, because their shape needs a live is_running read).
- Converted: meta_inputs 1, topology 2, permissions 3, auto_update 2,
  actions 3, lifecycle_handlers 3.
- Dropped source/reason at every converted site: nothing ever read
  NodeKind::Dag's fields (only `{ .. }` matches exist), so they are
  write-only. Dead reason-only locals deleted; the boot sweep's summary
  became a tracing::info! rather than being lost.

Remaining: dashboard/lifecycle_ops 7, server.rs 7, and the test suite —
tests.rs has its own submit() helper whose u64 return is used as the
handle to navigate the inserted DAG, so those need a different way to
find nodes, not a mechanical port.
2026-08-04 19:57:32 +02:00
17 changed files with 791 additions and 880 deletions

View file

@ -106,7 +106,7 @@ subgraph each (independent roots, run concurrently on their own leases), not
N separate DAGs. N separate DAGs.
**These are built dynamically from each agent's live running state** (an **These are built dynamically from each agent's live running state** (an
async `lifecycle::is_running` read), so they live in `job_queue/submit.rs`, async `lifecycle::is_running` read), so they live in `job_queue/power.rs`,
not the pure/sync `templates.rs`. Per-agent shape rule: `stop`/`start` carry not the pure/sync `templates.rs`. Per-agent shape rule: `stop`/`start` carry
a head `SetWanted` (intent) — `restart` does not; the tail `Reconcile` a head `SetWanted` (intent) — `restart` does not; the tail `Reconcile`
(convergence guarantee — cheap, noops when already converged) is ALWAYS (convergence guarantee — cheap, noops when already converged) is ALWAYS
@ -161,12 +161,12 @@ Notable collapses:
Per-agent power *intent*`wanted: Up | Offline` — is durable as the Per-agent power *intent*`wanted: Up | Offline` — is durable as the
`agent_power` table in the coordinator DB (`hive-c0re/src/stores/power.rs`). `agent_power` table in the coordinator DB (`hive-c0re/src/stores/power.rs`).
`container_view` remains the observed *status*; `Reconcile` nodes converge the `container_view` remains the observed *status*; `Reconcile` nodes converge the
two. Setting `wanted` is never a queued node: the submit layer two. Setting `wanted` is never a queued node: the power layer
(`job_queue/submit.rs`) writes the row synchronously, then submits the DAG (`job_queue/power.rs`) writes the row synchronously, then inserts the DAG
whose `Reconcile` reads the fresh value — rapid toggles are last-writer-wins. whose `Reconcile` reads the fresh value — rapid toggles are last-writer-wins.
Power toggles never commit to the meta repo. Every operator power surface — Power toggles never commit to the meta repo. Every operator power surface —
dashboard buttons, the MCP tools, and `hivectl stop/start/restart/kill` dashboard buttons, the MCP tools, and `hivectl stop/start/restart/kill`
rides the queue through that submit layer, so intent, lease serialization, rides the queue through that power layer, so intent, lease serialization,
and crash-watch suppression can't drift per surface; the only direct starts and crash-watch suppression can't drift per surface; the only direct starts
left are the root-agent bootstrap and infra containers (no lease, no left are the root-agent bootstrap and infra containers (no lease, no
harness). Cancelling a still-queued power DAG reverts `wanted` to the harness). Cancelling a still-queued power DAG reverts `wanted` to the

View file

@ -55,13 +55,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// nothing). // nothing).
let inputs: Vec<String> = let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default(); serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let submitted = coord.job_queue.submit( let inserted = coord.job_queue.insert_job(|b| {
crate::job_queue::Source::Approval, crate::job_queue::templates::meta_update(b, inputs, Some(id));
format!("approval #{id} meta input update"), Vec::new()
|b| crate::job_queue::templates::meta_update(b, inputs, Some(id)), });
); if let Err(e) = inserted {
if let Err(e) = submitted { return Err(e.context("insert meta-update dag"));
return Err(e.context("submit meta-update dag"));
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
Ok(()) Ok(())
@ -75,13 +74,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
{ {
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed"); tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
} }
let submitted = coord.job_queue.submit( let inserted = coord.job_queue.insert_job(|b| {
crate::job_queue::Source::Approval, crate::job_queue::templates::spawn(b, approval.agent.as_str(), id);
format!("approval #{id} spawn"), Vec::new()
|b| crate::job_queue::templates::spawn(b, approval.agent.as_str(), id), });
); if let Err(e) = inserted {
if let Err(e) = submitted { return Err(e.context("insert spawn dag"));
return Err(e.context("submit spawn dag"));
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
Ok(()) Ok(())
@ -106,12 +104,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// `run_deploy_apply` (ff-merge, then grows the rebuild subgraph), // `run_deploy_apply` (ff-merge, then grows the rebuild subgraph),
// `run_finalize_deploy` (deploy tag + lock commit) and // `run_finalize_deploy` (deploy tag + lock commit) and
// `run_deploy_tail` (compensation + forge mirror). // `run_deploy_tail` (compensation + forge mirror).
enqueue_approval_rebuild( enqueue_approval_rebuild(&coord, approval.agent.as_str(), id);
&coord,
approval.agent.as_str(),
id,
format!("approval #{id} merge config pr"),
);
Ok(()) Ok(())
} }
} }
@ -121,19 +114,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
/// dispatch arm — the work ends in a container rebuild routed through the /// dispatch arm — the work ends in a container rebuild routed through the
/// queue. See [`crate::job_queue::templates::approval_deploy`] for the node /// queue. See [`crate::job_queue::templates::approval_deploy`] for the node
/// shape; the executor dispatches each node to the `run_deploy_*` bodies below. /// shape; the executor dispatches each node to the `run_deploy_*` bodies below.
fn enqueue_approval_rebuild( fn enqueue_approval_rebuild(coord: &Arc<Coordinator>, agent: &str, approval_id: i64) {
coord: &Arc<Coordinator>, if let Err(e) = coord.job_queue.insert_job(|b| {
agent: &str, crate::job_queue::templates::approval_deploy(b, agent, approval_id);
approval_id: i64, Vec::new()
reason: String, }) {
) { tracing::error!(%agent, approval_id, error = ?e, "insert approval deploy dag failed");
if let Err(e) = coord
.job_queue
.submit(crate::job_queue::Source::Approval, reason, |b| {
crate::job_queue::templates::approval_deploy(b, agent, approval_id);
})
{
tracing::error!(%agent, approval_id, error = ?e, "submit approval deploy dag failed");
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
} }

View file

@ -1,11 +1,12 @@
//! Container lifecycle endpoints for the dashboard. //! Container lifecycle endpoints for the dashboard.
//! //!
//! Rebuild / restart / start / stop (hard + graceful) / update-all all //! Rebuild / restart / start / stop (hard + graceful) / update-all all
//! submit DAGs to the job queue (`job_queue::submit`), so each shows a //! insert DAGs into the job queue — the power ops via
//! visible queued→running transient on the dashboard — a direct //! [`crate::job_queue::power`], the static shapes straight through
//! sub-second start/stop only flashed the badge. Start/stop also //! `JobQueue::insert` — so each shows a visible queued→running transient on
//! persist the agent's `wanted` power intent before submitting; the //! the dashboard; a direct sub-second start/stop only flashed the badge.
//! DAG's `Reconcile` converges to it. Destroy delegates to //! Start/stop also persist the agent's `wanted` power intent before
//! inserting; the DAG's `Reconcile` converges to it. Destroy delegates to
//! `actions::destroy` (optionally purging). //! `actions::destroy` (optionally purging).
use axum::{ use axum::{
@ -27,7 +28,6 @@ pub(super) struct GracefulParams {
} }
use super::{AppState, Ident, error_response, guard_agent_name, strip_container_prefix}; use super::{AppState, Ident, error_response, guard_agent_name, strip_container_prefix};
use crate::job_queue::{Source, submit};
use crate::{actions, lifecycle}; use crate::{actions, lifecycle};
/// Queue a rebuild DAG for `name`. /// Queue a rebuild DAG for `name`.
@ -50,12 +50,13 @@ pub(super) async fn post_rebuild(
if let Some(reject) = guard_agent_name(&state, &logical).await { if let Some(reject) = guard_agent_name(&state, &logical).await {
return reject; return reject;
} }
submit::rebuild( if let Err(e) = state.coord.job_queue.insert_job(|b| {
&state.coord, crate::job_queue::templates::rebuild(b, &logical, true);
&logical, Vec::new()
Source::Manual, }) {
"manual via dashboard ↻ R3BU1LD button".to_owned(), tracing::error!(agent = %logical, error = ?e, "rebuild: insert failed");
); }
state.coord.emit_rebuild_queue_snapshot();
(StatusCode::OK, "ok").into_response() (StatusCode::OK, "ok").into_response()
} }
@ -92,13 +93,12 @@ pub(super) async fn post_kill(
// timeout fallback to a hard stop). The agent's lifecycle // timeout fallback to a hard stop). The agent's lifecycle
// lease keeps it from racing an in-flight rebuild for the same // lease keeps it from racing an in-flight rebuild for the same
// agent, and per-node progress surfaces on the queue snapshot. // agent, and per-node progress surfaces on the queue snapshot.
submit::graceful_stop( if let Err(e) =
&state.coord, crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), true)
&logical, .await
Source::Manual, {
"manual via dashboard graceful stop".to_owned(), tracing::error!(agent = %logical, error = ?e, "graceful stop: insert failed");
) }
.await;
return (StatusCode::OK, "ok").into_response(); return (StatusCode::OK, "ok").into_response();
} }
// Manager is stoppable from the dashboard like any other // Manager is stoppable from the dashboard like any other
@ -111,13 +111,12 @@ pub(super) async fn post_kill(
// `socket_server.rs::Request::Kill` stays in place: a // `socket_server.rs::Request::Kill` stays in place: a
// manager calling Kill on its own container is self-suicide // manager calling Kill on its own container is self-suicide
// mid-call, not a legitimate operator action. // mid-call, not a legitimate operator action.
submit::stop( if let Err(e) =
&state.coord, crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), false)
&logical, .await
Source::Manual, {
"manual via dashboard stop".to_owned(), tracing::error!(agent = %logical, error = ?e, "stop: insert failed");
) }
.await;
(StatusCode::OK, "ok").into_response() (StatusCode::OK, "ok").into_response()
} }
@ -149,22 +148,23 @@ pub(super) async fn post_restart(
return reject; return reject;
} }
if params.graceful { if params.graceful {
submit::graceful_restart( if let Err(e) = crate::job_queue::power::restart_many(
&state.coord, &state.coord,
&logical, std::slice::from_ref(&logical),
Source::Manual, true,
"manual via dashboard graceful restart".to_owned(),
) )
.await; .await
{
tracing::error!(agent = %logical, error = ?e, "graceful restart: insert failed");
}
return (StatusCode::OK, "ok").into_response(); return (StatusCode::OK, "ok").into_response();
} }
submit::restart( if let Err(e) =
&state.coord, crate::job_queue::power::restart_many(&state.coord, std::slice::from_ref(&logical), false)
&logical, .await
Source::Manual, {
"manual via dashboard ↺ R3START button".to_owned(), tracing::error!(agent = %logical, error = ?e, "restart: insert failed");
) }
.await;
(StatusCode::OK, "ok").into_response() (StatusCode::OK, "ok").into_response()
} }
@ -226,13 +226,11 @@ pub(super) async fn post_start(
return (StatusCode::OK, "ok").into_response(); return (StatusCode::OK, "ok").into_response();
} }
} }
submit::start( if let Err(e) =
&state.coord, crate::job_queue::power::start_many(&state.coord, std::slice::from_ref(&logical)).await
&logical, {
Source::Manual, tracing::error!(agent = %logical, error = ?e, "start: insert failed");
"manual via dashboard start".to_owned(), }
)
.await;
(StatusCode::OK, "ok").into_response() (StatusCode::OK, "ok").into_response()
} }
@ -411,13 +409,14 @@ pub(super) async fn post_update_all(State(state): State<AppState>) -> Response {
else { else {
continue; continue;
}; };
submit::rebuild( if let Err(e) = state.coord.job_queue.insert_job(|b| {
&state.coord, crate::job_queue::templates::rebuild(b, &logical, true);
&logical, Vec::new()
Source::Manual, }) {
"manual via dashboard 🌀 UPDATE ALL".to_owned(), tracing::error!(agent = %logical, error = ?e, "update-all: insert failed");
); }
} }
state.coord.emit_rebuild_queue_snapshot();
(StatusCode::OK, "ok").into_response() (StatusCode::OK, "ok").into_response()
} }

View file

@ -208,16 +208,18 @@ pub(super) async fn post_meta_update(
if inputs.is_empty() { if inputs.is_empty() {
return error_response("meta-update: no inputs selected"); return error_response("meta-update: no inputs selected");
} }
let inputs_label = inputs.join(", ");
// Cascade rebuild children fan out from the MetaLock node when the // Cascade rebuild children fan out from the MetaLock node when the
// lock bump lands — appended by the scheduler so they build against // lock bump lands — appended by the scheduler so they build against
// the post-bump lock, and a failed bump simply fans out nothing. // the post-bump lock, and a failed bump simply fans out nothing.
crate::job_queue::submit::meta_update( state
&state.coord, .coord
inputs, .job_queue
crate::job_queue::Source::Manual, .insert_job(|b| {
format!("meta-update via dashboard ({inputs_label})"), crate::job_queue::templates::meta_update(b, inputs, None);
); Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
(StatusCode::OK, "ok").into_response() (StatusCode::OK, "ok").into_response()
} }

View file

@ -155,15 +155,21 @@ pub(super) async fn post_tool_groups(
// META_LOCK inside the WritePermFile node, so concurrent // META_LOCK inside the WritePermFile node, so concurrent
// batch-apply actions for different agents never race on the // batch-apply actions for different agents never race on the
// shared tool-groups.json. // shared tool-groups.json.
crate::job_queue::submit::perm_change( state
&state.coord, .coord
&logical, .job_queue
crate::job_queue::Source::Manual, .insert_job(|b| {
"tool-group change via permissions UI".to_owned(), crate::job_queue::templates::perm_change(
crate::job_queue::PermPayload::ToolGroups { b,
groups: body.groups.clone(), &logical,
}, crate::job_queue::PermPayload::ToolGroups {
); groups: body.groups.clone(),
},
);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard"); tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard");
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }
@ -264,15 +270,21 @@ pub(super) async fn post_capabilities(
// META_LOCK inside the WritePermFile node, so concurrent // META_LOCK inside the WritePermFile node, so concurrent
// batch-apply actions for different agents never race on the // batch-apply actions for different agents never race on the
// shared capabilities.json. // shared capabilities.json.
crate::job_queue::submit::perm_change( state
&state.coord, .coord
&logical, .job_queue
crate::job_queue::Source::Manual, .insert_job(|b| {
"capability change via dashboard".to_owned(), crate::job_queue::templates::perm_change(
crate::job_queue::PermPayload::Capabilities { b,
caps: body.caps.clone(), &logical,
}, crate::job_queue::PermPayload::Capabilities {
); caps: body.caps.clone(),
},
);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard"); tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard");
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }
@ -359,13 +371,19 @@ pub(super) async fn post_permissions(
} }
// Phase 2 — submit one combined PermChange DAG per affected agent. // Phase 2 — submit one combined PermChange DAG per affected agent.
for (logical, groups, caps) in staged { for (logical, groups, caps) in staged {
crate::job_queue::submit::perm_change( state
&state.coord, .coord
&logical, .job_queue
crate::job_queue::Source::Manual, .insert_job(|b| {
"batch permission change via permissions UI".to_owned(), crate::job_queue::templates::perm_change(
crate::job_queue::PermPayload::Combined { groups, caps }, b,
); &logical,
crate::job_queue::PermPayload::Combined { groups, caps },
);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
tracing::info!(agent = %logical, "operator: batch perm change via dashboard"); tracing::info!(agent = %logical, "operator: batch perm change via dashboard");
} }
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())

View file

@ -21,7 +21,6 @@ use utoipa::ToSchema;
use problem_details::ProblemDetails; use problem_details::ProblemDetails;
use super::{AppState, error_problem}; use super::{AppState, error_problem};
use crate::job_queue::{Source, submit};
/// `POST /api/topology/set-parent` body. `child` is required. /// `POST /api/topology/set-parent` body. `child` is required.
/// `new_parent` may be: /// `new_parent` may be:
@ -94,12 +93,15 @@ pub(super) async fn post_set_parent(
new_parent = ?new_parent, new_parent = ?new_parent,
"operator: set-parent via dashboard" "operator: set-parent via dashboard"
); );
submit::reparent( state
&state.coord, .coord
vec![(child, new_parent)], .job_queue
Source::Manual, .insert_job(|b| {
"manual set-parent via dashboard".to_owned(), crate::job_queue::templates::reparent(b, vec![(child, new_parent)]);
); Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }
@ -150,11 +152,14 @@ pub(super) async fn post_set_parent_bulk(
.map_err(|e| error_problem(&e))?; .map_err(|e| error_problem(&e))?;
let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect(); let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect();
tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard"); tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard");
submit::reparent( state
&state.coord, .coord
moves, .job_queue
Source::Manual, .insert_job(|b| {
"manual set-parent-bulk via dashboard".to_owned(), crate::job_queue::templates::reparent(b, moves);
); Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }

View file

@ -53,7 +53,7 @@ pub(super) async fn run_node(
kind: &NodeKind, kind: &NodeKind,
) -> (super::JobBuilder, Result<()>) { ) -> (super::JobBuilder, Result<()>) {
// The agent this node targets rides the payload — empty for the agentless // The agent this node targets rides the payload — empty for the agentless
// container kinds (`MetaLock`, `Dag`), which never read it. // kinds (`MetaLock`, `Reparent`), which never read it.
let agent = kind.agent(); let agent = kind.agent();
// Every arm is `Result<()>`; the three that grow work declare into `builder` // Every arm is `Result<()>`; the three that grow work declare into `builder`
// *synchronously*, after their own awaits have finished. Borrowing `&builder` // *synchronously*, after their own awaits have finished. Borrowing `&builder`
@ -115,26 +115,21 @@ pub(super) async fn run_node(
run_finalize_deploy(coord, *approval_id).await run_finalize_deploy(coord, *approval_id).await
} }
NodeKind::DeployTail { approval_id, .. } => { NodeKind::DeployTail { approval_id, .. } => {
run_deploy_tail(coord, coord.job_queue.dag_of(id), agent, *approval_id).await run_deploy_tail(coord, coord.job_queue.root_of(id), agent, *approval_id).await
} }
NodeKind::ResolveApproval { NodeKind::ResolveApproval {
approval_id, approval_id,
outcome, outcome,
} => run_resolve_approval(coord, coord.job_queue.dag_of(id), *approval_id, *outcome).await, } => run_resolve_approval(coord, coord.job_queue.root_of(id), *approval_id, *outcome).await,
NodeKind::EmitRebuilt { ok, .. } => { NodeKind::EmitRebuilt { ok, .. } => {
run_emit_rebuilt(coord, agent, coord.job_queue.dag_of(id), *ok).await; run_emit_rebuilt(coord, agent, coord.job_queue.root_of(id), *ok).await;
Ok(()) Ok(())
} }
NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up), NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
// The nodes that carry no work of their own; completing one lets it // Braces carry no work of their own; completing one lets it reach
// reach `Finishing` so the nodes under it start. // `Finishing` so the nodes under it start. What they declare stays held
// - `Dag`: pure grouping container. The DAG's terminal side effect, if // until their whole subtree settles.
// any, is its own tail node in the graph. NodeKind::DeployWindow { .. } | NodeKind::AgentWindow { .. } => Ok(()),
// - `DeployWindow` / `AgentWindow`: pure resource holders (braces) —
// what they declare stays held until their subtree settles.
NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } | NodeKind::AgentWindow { .. } => {
Ok(())
}
}; };
(builder, result) (builder, result)
} }

View file

@ -10,17 +10,16 @@
//! carries the agent it targets ([`NodeKind::agent`]); the two resource //! carries the agent it targets ([`NodeKind::agent`]); the two resource
//! classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease //! classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease
//! subtree-held), declared per node at its construction site; //! subtree-held), declared per node at its construction site;
//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`) //! - **a job has no container node.** A template declares its nodes and names
//! carrying the group's metadata, with the work nodes hung under it as //! the roots it wants back; `insert_job` returns those ids. Grouping is the
//! its subtree (the **parent axis** groups; `deps` order). So the container's //! parent axis (a root's rolled-up state *is* its subtree's), so membership is
//! `NodeId` is the DAG id, its rolled-up state is the DAG state, and membership //! a graph walk with no host-side side-tables. The lease is owned by a subtree
//! is a graph walk — there are no host grouping side-tables. The lease is owned //! root and borrowed by its descendants (continuity);
//! by a subtree root and borrowed by its descendants (continuity); //! - terminal work is an ordinary **tail node**
//! - per-DAG terminal work is an ordinary **tail node**
//! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder //! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
//! appends in [`templates`], edged onto the DAG's other group roots by the //! appends in [`templates`], edged onto the job's group roots by the outcome it
//! outcome it reports. Templates emit one tail per outcome and the graph runs //! reports. Templates emit one tail per outcome and the graph runs exactly one,
//! exactly one, so nothing branches at runtime. //! so nothing branches at runtime.
//! //!
//! The queue is runtime-only (no persistence): an empty graph on boot; desired //! The queue is runtime-only (no persistence): an empty graph on boot; desired
//! state is re-derived by the reconcile sweep. A single scheduler task //! state is re-derived by the reconcile sweep. A single scheduler task
@ -29,9 +28,9 @@
pub mod exec; pub mod exec;
pub mod model; pub mod model;
pub mod power;
pub mod resource; pub mod resource;
pub mod scheduler; pub mod scheduler;
pub mod submit;
pub mod templates; pub mod templates;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
@ -46,7 +45,7 @@ use hive_jobq_wire::{GraphNode, GraphWire};
use tokio::sync::Notify; use tokio::sync::Notify;
pub use hive_jobq::TerminalState; pub use hive_jobq::TerminalState;
pub use model::{NodeKind, PermPayload, Source, State}; pub use model::{NodeKind, PermPayload, State};
use resource::Resource; use resource::Resource;
/// A job under construction: `hive_jobq`'s builder over this queue's payload /// A job under construction: `hive_jobq`'s builder over this queue's payload
@ -90,12 +89,10 @@ pub struct RunningTransient {
/// The crate scheduler, specialised to this host's node + resource types. /// The crate scheduler, specialised to this host's node + resource types.
/// ///
/// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`) /// A job is **just its nodes** — no container, no grouping side-tables. A
/// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id, /// root's rolled-up state is its subtree's, so membership is a graph walk and
/// its rolled-up state is the DAG state, and there are no grouping side-tables: /// "which job is this node in" is [`JobQueue::root_of`]. One shared crate
/// membership + meta are graph queries ([`container`] + the `hive_jobq::Graph` /// [`Graph`] holds every job's nodes.
/// accessors, with the meta read straight off the container's payload). One
/// shared crate [`Graph`] holds every DAG.
/// ///
/// There is deliberately **no wrapper struct and no per-node side map**. The /// There is deliberately **no wrapper struct and no per-node side map**. The
/// last map held the `build_logs` row id; that link now lives on the log row /// last map held the `build_logs` row id; that link now lives on the log row
@ -140,35 +137,12 @@ fn outcome_of(result: Result<(), String>) -> Outcome {
} }
} }
/// Insert a declared `job` into the shared graph, returning the inserted ids. // `insert_group` lived here: a `group_parent`-taking insert whose only
/// // remaining caller was the DAG container, everything under it. Runtime growth
/// A node that declared no parent hangs under `group_parent` — the DAG // never went through it — an executor declares into the builder `hive_jobq`
/// container for a template, the emitting node for a runtime-appended // hands it, which parents the new work under the emitting node by
/// subgraph. Templates declare the parent axis + sibling ordering directly, so // construction. With no container to be the other kind of parent, the
/// there is no dep-on-root to drop and no lease to hoist: each node declares // distinction it existed to express is gone.
/// its own resources, and the crate's borrow model keeps a resource continuous
/// across a subtree (a root owns it, descendants borrow it). Independent group
/// roots carry no cross-links, so a multi-agent DAG's per-agent subgraphs run
/// concurrently, each on its own lease.
///
/// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group(
inner: &mut Sched,
declare: impl FnOnce(&JobBuilder),
group_parent: Option<NodeId>,
) -> anyhow::Result<()> {
inner
.insert_job(group_parent, |b| {
declare(b);
// c0re names no handles: a DAG is addressed by its container node,
// which `submit` inserts itself, and nothing downstream looks an
// individual step up by id.
Vec::new()
})
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
Ok(())
}
impl JobQueue { impl JobQueue {
#[must_use] #[must_use]
@ -188,40 +162,34 @@ impl JobQueue {
self.sched.lock().expect("job_queue mutex poisoned") self.sched.lock().expect("job_queue mutex poisoned")
} }
/// Submit a DAG: insert a [`NodeKind::Dag`] **container node** carrying the /// Insert a job's nodes into the shared graph, then wake the run loop.
/// group's metadata, then insert the template's nodes as its subtree (their
/// roots re-parented to the container). Returns the container's id as the
/// DAG id — its rolled-up state is the DAG state.
/// ///
/// The container is an ordinary node: it declares no resources, so the /// Deliberately named for the [`hive_jobq`] primitive it wraps, because
/// scheduler claims it on the next pass, runs its (empty) logic and parks /// that is nearly all it is. **The wrapper earns its place on the wake**:
/// it in `Finishing`, at which point its children become runnable. Nothing /// the crate is sync and runtime-free — it holds no `Notify` at all — so
/// here completes it by hand — a node with no work of its own still goes /// the channel the run loop parks on belongs to the host, and something has
/// the way every other node goes. /// to ping it. Left to call sites, an insert whose ping was forgotten would
/// leave a correct DAG sitting unscheduled until an unrelated event
/// happened along; nothing would fail, and no test in isolation would see
/// it.
/// ///
/// `source` and `reason` are the container node's own payload — they are /// Returns exactly what the primitive returns: the ids of the nodes the
/// arguments here rather than fields of a spec struct because that is all /// template named, in the order it named them.
/// they ever were. `declare` is the recipe, taken by generic and run
/// against a builder `hive_jobq` owns: it goes from the template straight
/// into this call, so there is nothing to allocate for.
/// ///
/// # Errors /// # Errors
/// Propagates a graph-insert error (dependencies that aren't /// Propagates a graph-insert error (dependencies that aren't
/// dependency-topological). /// dependency-topological).
pub fn submit( pub fn insert_job(
&self, &self,
source: Source, declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
reason: String, ) -> anyhow::Result<Vec<NodeId>> {
declare: impl FnOnce(&JobBuilder),
) -> anyhow::Result<u64> {
let mut inner = self.lock(); let mut inner = self.lock();
let container = inner let named = inner
.append(NodeKind::Dag { source, reason }, Vec::new(), None) .insert_job(None, declare)
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
insert_group(&mut inner, declare, Some(container))?;
drop(inner); drop(inner);
self.notify.notify_one(); self.notify.notify_one();
Ok(container.get()) Ok(named)
} }
/// The scheduler itself, for `hive_jobq`'s run-loop seam /// The scheduler itself, for `hive_jobq`'s run-loop seam
@ -235,11 +203,16 @@ impl JobQueue {
&self.sched &self.sched
} }
/// The DAG container id owning `node`, for log lines and the dashboard. /// The id of the **group root** `node` belongs to, for log lines and the
/// Derived from the graph rather than carried alongside the node — the /// dashboard. Derived from the graph rather than carried alongside the node
/// parent axis already knows it. /// — the parent axis already knows it.
///
/// Was `dag_of`, when a job's nodes hung under a container node that *was*
/// the group. Without it the parent chain ends at whichever root the
/// template declared, so this answers "which root owns this node", not
/// "which DAG is this in" — there is no longer such a thing.
#[must_use] #[must_use]
pub fn dag_of(&self, node: NodeId) -> Option<u64> { pub fn root_of(&self, node: NodeId) -> Option<u64> {
self.lock().graph().root_of(node).map(NodeId::get) self.lock().graph().root_of(node).map(NodeId::get)
} }
@ -424,9 +397,9 @@ fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
/// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones. /// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones.
/// ///
/// Selected *structurally* — a root is a node with no parent. The typed /// Selected *structurally* — a root is a node with no parent. The typed
/// projection this replaced keyed on `NodeKind::Dag` instead, which made the /// projection this replaced keyed on the since-removed container kind instead,
/// visible set depend on one host node kind; nothing here knows what a node /// which made the visible set depend on one host node kind; nothing here knows
/// means. /// what a node means.
/// ///
/// **This bound is load-bearing, not tidiness.** Nothing ever removes a node /// **This bound is load-bearing, not tidiness.** Nothing ever removes a node
/// from the graph (bounded pruning is a Stage-C follow-up), so serving /// from the graph (bounded pruning is a Stage-C follow-up), so serving

View file

@ -1,18 +1,18 @@
//! Data model for the generic job-DAG queue: node kinds (the primitive //! Data model for the generic job-DAG queue: the node kinds — the primitive
//! operations), dependency edges, and the runtime `Dag` / `Node` store. //! operations — and what each one carries. The `State` / `PermPayload` wire
//! The `Source` / `State` / `PermPayload` wire enums live in //! enums live in `hive_host_sock::jobs` (they travel on the host admin
//! `hive_host_sock::jobs` (they travel on the host admin socket) and are //! socket) and are re-exported here for the queue's internal use. The graph
//! re-exported here for the queue's internal use. The graph itself is //! itself is served through `hive_jobq_wire`'s generic projection — there is
//! served through `hive_jobq_wire`'s generic projection — there is no //! no second, typed view of it any more.
//! second, typed view of it any more.
//! //!
//! Two levels: the **DAG** is the unit of cancel / approval-resolution //! **One level, not two.** The node is the unit of everything: scheduling,
//! and the dashboard group; the **node** is the unit of scheduling / //! execution, build-log, cancel, and the dashboard group (a group root's
//! execution / build-log, and carries its own `agent` (a //! subtree *is* the group). A DAG used to be a second level above it, with
//! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! its own store and its own id; there is no container node any more, so a
//! full design. //! job is exactly the nodes it declared. See `docs/coordinator.md::Job queue`
//! for the full design.
pub use hive_host_sock::jobs::{PermPayload, Source, State}; pub use hive_host_sock::jobs::{PermPayload, State};
use serde::Serialize; use serde::Serialize;
use hive_jobq::TerminalState; use hive_jobq::TerminalState;
@ -280,18 +280,6 @@ pub enum NodeKind {
/// `Prebuild`, but that's a no-op there — the agent is down, so prebuild /// `Prebuild`, but that's a no-op there — the agent is down, so prebuild
/// is skipped.) /// is skipped.)
SetWanted { agent: String, up: bool }, SetWanted { agent: String, up: bool },
/// The **DAG container** node: one per submitted DAG, carrying the group's
/// domain metadata. Every node hangs *under* it (its subtree), so
/// the container's `NodeId` **is** the DAG id and its rolled-up state **is**
/// the DAG state. Pure grouping — lease- and
/// build-slot-exempt; the executor instant-completes it (`Done`) so it
/// reaches `Finishing` and its children start.
///
/// No `created_at` here: the graph stamps [`hive_jobq::Node::created_at`] on
/// every node at insert, so the container already has one. A second copy in
/// the payload would be the same instant recorded twice, with only this
/// variant's version reachable to a generic viewer.
Dag { source: Source, reason: String },
} }
/// How a hive-c0re node describes itself to a generic graph viewer. /// How a hive-c0re node describes itself to a generic graph viewer.
@ -362,14 +350,13 @@ impl NodeKind {
NodeKind::ResolveApproval { .. } => "resolve_approval", NodeKind::ResolveApproval { .. } => "resolve_approval",
NodeKind::EmitRebuilt { .. } => "emit_rebuilt", NodeKind::EmitRebuilt { .. } => "emit_rebuilt",
NodeKind::SetWanted { .. } => "set_wanted", NodeKind::SetWanted { .. } => "set_wanted",
NodeKind::Dag { .. } => "dag",
} }
} }
/// The agent this node targets, or `""` for agentless kinds /// The agent this node targets, or `""` for agentless kinds
/// ([`NodeKind::MetaLock`] on the `hyperhive` pseudo-agent, /// ([`NodeKind::MetaLock`] on the `hyperhive` pseudo-agent,
/// [`NodeKind::Reparent`] which can span multiple agents, and the /// [`NodeKind::Reparent`] which can span multiple agents, and
/// [`NodeKind::Dag`] container). /// [`NodeKind::ResolveApproval`] which acts on an approval row).
#[must_use] #[must_use]
pub fn agent(&self) -> &str { pub fn agent(&self) -> &str {
match self { match self {
@ -397,8 +384,7 @@ impl NodeKind {
| NodeKind::SetWanted { agent, .. } => agent, | NodeKind::SetWanted { agent, .. } => agent,
NodeKind::MetaLock { .. } NodeKind::MetaLock { .. }
| NodeKind::Reparent { .. } | NodeKind::Reparent { .. }
| NodeKind::ResolveApproval { .. } | NodeKind::ResolveApproval { .. } => "",
| NodeKind::Dag { .. } => "",
} }
} }

View file

@ -1,59 +1,27 @@
//! Request-level submit API — the surface the dashboard POST handlers, //! Power ops (`stop` / `start` / `restart`) — the DAG shapes whose per-agent
//! the MCP socket handlers, and `hivectl` paths call. //! form depends on **live** container state, so they cannot be static
//! [`super::templates`] entries.
//! //!
//! The **power ops** (`stop` / `start` / `restart`) are built here, not in //! The split is the purity line, not the subject matter: the `*_chain` /
//! `templates.rs`: each agent's subgraph shape depends on its *live* running //! `*_nodes` builders below are pure (they take `running` / `stale` as
//! state, which needs an async `lifecycle::is_running` read that a pure/sync //! parameters, which is what keeps them unit-testable without a container),
//! template can't do. So these fns are async — they read each agent's state, //! and only the `*_many` entry points do the async `lifecycle::is_running`
//! assemble a per-agent subgraph out of the shared pure primitives //! read that produces those parameters.
//! (`JobBuilder::node` + `templates::rebuild_nodes`), all declaring into ONE job
//! (independent per-agent roots, concurrent on their own leases).
//! //!
//! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable //! Each entry point declares its group and inserts it. There is no metadata
//! intent write) — `restart` does NOT (it bounces the container but leaves //! parameter and no container node: attribution is not something every caller
//! `wanted` untouched, so a deliberately-stopped agent isn't forced up). The //! has to invent, and a DAG is addressed by the nodes a template names.
//! tail `Reconcile` (the convergence guarantee — cheap, noops when already
//! converged) is ALWAYS present; only the *mechanical* nodes
//! (`Signal`/`Drain`/`StopForUpdate`) are state-conditional (skipped for a
//! down agent — nothing to quiesce/stop). Keeping `Reconcile` in every shape
//! closes the TOCTOU window: if an agent flips state between the `is_running`
//! read and node execution, the tail `Reconcile` still converges it in-DAG,
//! with `StopForUpdate`-noop as the backstop — no reliance on an external
//! reconcile sweep. Every helper emits a fresh queue snapshot so the
//! dashboard shows the new DAG immediately.
use std::sync::Arc; use std::sync::Arc;
use super::model::NodeKind; use hive_jobq::NodeId;
use super::resource::Resource; use super::resource::Resource;
use super::templates::rebuild_nodes; use super::templates::rebuild_nodes;
use super::{JobBuilder, Source, templates}; use super::{JobBuilder, NodeKind};
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
use crate::lifecycle; use crate::lifecycle;
fn submit_and_emit(
coord: &Arc<Coordinator>,
source: Source,
reason: String,
declare: impl FnOnce(&JobBuilder),
) -> u64 {
let id = coord
.job_queue
.submit(source, reason, declare)
.expect("template-declared shapes are acyclic");
coord.emit_rebuild_queue_snapshot();
id
}
/// Manual/approval-independent rebuild (always relocks the agent's
/// meta input — the meta-update cascade grows its own rebuild subgraphs
/// in-DAG instead of going through this surface).
pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::rebuild(builder, agent, true);
})
}
// ---- dynamic power-op DAG assembly ---------------------------------------- // ---- dynamic power-op DAG assembly ----------------------------------------
// //
// The pure per-agent chain builders below take `running` (and `stale`) // The pure per-agent chain builders below take `running` (and `stale`)
@ -67,7 +35,16 @@ pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: St
/// actually running (nothing to drain on a down container). The `Reconcile` /// actually running (nothing to drain on a down container). The `Reconcile`
/// stays even for a down agent so a race-up between the state read and exec /// stays even for a down agent so a race-up between the state read and exec
/// is still stopped in-DAG. /// is still stopped in-DAG.
fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) { /// Returns the group root's guid, which is what the caller names so
/// `insert_job` hands its id back — that id is how `hivectl` polls this agent's
/// progress. A chain that returned nothing would insert correctly and leave the
/// caller with nothing to wait on.
fn stop_chain(
builder: &JobBuilder,
agent: &str,
graceful: bool,
running: bool,
) -> hive_jobq::NodeGuid {
// `SetWanted` is the group root and owns the agent lease; the mechanical // `SetWanted` is the group root and owns the agent lease; the mechanical
// steps are its children (borrow the lease, run once it reaches `Finishing`, // steps are its children (borrow the lease, run once it reaches `Finishing`,
// dep-ordered among themselves). // dep-ordered among themselves).
@ -96,13 +73,26 @@ fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool)
.needs(Resource::Agent(a())) .needs(Resource::Agent(a()))
.part_of(wanted); .part_of(wanted);
} }
wanted.guid()
} }
/// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev /// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev
/// agent gets the rebuild subgraph (its tail `Reconcile` starts it on /// agent gets the rebuild subgraph (its tail `Reconcile` starts it on
/// current derivations), otherwise a plain `Reconcile` (which starts a down /// current derivations), otherwise a plain `Reconcile` (which starts a down
/// agent and noops an already-running one). /// agent and noops an already-running one).
fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) { ///
/// Returns **every** group root — see [`stop_chain`] for why they are named.
///
/// ⚠️ More than one in the stale branch: `rebuild_nodes` chains its roots
/// *behind* `SetWanted` with `after_ok`, it does **not** nest them under it. So
/// `SetWanted` rolls up only itself, and naming it alone would report the whole
/// start finished while the rebuild was still running.
fn start_chain(
builder: &JobBuilder,
agent: &str,
running: bool,
stale: bool,
) -> Vec<hive_jobq::NodeGuid> {
let wanted = builder let wanted = builder
.node(NodeKind::SetWanted { .node(NodeKind::SetWanted {
agent: agent.to_owned(), agent: agent.to_owned(),
@ -110,10 +100,16 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
}) })
.needs(Resource::Agent(agent.to_owned())); .needs(Resource::Agent(agent.to_owned()));
if !running && stale { if !running && stale {
// Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, the
// `Prebuild` + `Reconcile` are their own group roots (top-level, per // `AgentWindow` brace and `Reconcile` are their own group roots
// `rebuild_nodes`). // (top-level, per `rebuild_nodes`) — hence all four names.
rebuild_nodes(builder, agent, true, Some(wanted)); let roots = rebuild_nodes(builder, agent, true, Some(wanted));
vec![
wanted.guid(),
roots.meta_sync.guid(),
roots.agent_window.guid(),
roots.reconcile.guid(),
]
} else { } else {
let _ = builder let _ = builder
.node(NodeKind::Reconcile { .node(NodeKind::Reconcile {
@ -121,6 +117,7 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
}) })
.needs(Resource::Agent(agent.to_owned())) .needs(Resource::Agent(agent.to_owned()))
.part_of(wanted); .part_of(wanted);
vec![wanted.guid()]
} }
} }
@ -133,14 +130,22 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
/// before `Reconcile`; a down agent gets just `Reconcile`, which /// before `Reconcile`; a down agent gets just `Reconcile`, which
/// converges to intent — a stopped (`wanted = Off`) agent stays stopped, /// converges to intent — a stopped (`wanted = Off`) agent stays stopped,
/// a crashed (`wanted = Up`) agent comes back up. /// a crashed (`wanted = Up`) agent comes back up.
fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) { ///
/// Returns the group root's guid — see [`stop_chain`].
fn restart_chain(
builder: &JobBuilder,
agent: &str,
graceful: bool,
running: bool,
) -> hive_jobq::NodeGuid {
let a = || agent.to_owned(); let a = || agent.to_owned();
if !running { if !running {
// Nothing to bounce — a lone Reconcile converges to intent. // Nothing to bounce — a lone Reconcile converges to intent, and is
let _ = builder // itself the root.
return builder
.node(NodeKind::Reconcile { agent: a() }) .node(NodeKind::Reconcile { agent: a() })
.needs(Resource::Agent(a())); .needs(Resource::Agent(a()))
return; .guid();
} }
// Running: mechanical stop then Reconcile. The first stop node is the group // Running: mechanical stop then Reconcile. The first stop node is the group
// ROOT (no SetWanted head) and owns the agent lease; the rest are its // ROOT (no SetWanted head) and owns the agent lease; the rest are its
@ -176,6 +181,7 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo
.needs(Resource::Agent(a())) .needs(Resource::Agent(a()))
.part_of(signal) .part_of(signal)
.after_ok(stop); .after_ok(stop);
signal.guid()
} else { } else {
let stop = builder let stop = builder
.node(NodeKind::StopForUpdate { agent: a() }) .node(NodeKind::StopForUpdate { agent: a() })
@ -184,6 +190,7 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo
.node(NodeKind::Reconcile { agent: a() }) .node(NodeKind::Reconcile { agent: a() })
.needs(Resource::Agent(a())) .needs(Resource::Agent(a()))
.part_of(stop); .part_of(stop);
stop.guid()
} }
} }
@ -201,10 +208,15 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo
// subgraph's indices onto another's used to be a function. // subgraph's indices onto another's used to be a function.
/// Declare the stop DAG from explicit `(agent, running)` targets. /// Declare the stop DAG from explicit `(agent, running)` targets.
pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) { pub(crate) fn stop_nodes(
for (agent, running) in targets { builder: &JobBuilder,
stop_chain(builder, agent, graceful, *running); targets: &[(String, bool)],
} graceful: bool,
) -> Vec<hive_jobq::NodeGuid> {
targets
.iter()
.map(|(agent, running)| stop_chain(builder, agent, graceful, *running))
.collect()
} }
/// Assemble the start DAG from explicit `(agent, running, stale)` targets. /// Assemble the start DAG from explicit `(agent, running, stale)` targets.
@ -213,76 +225,68 @@ pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], grace
/// running under its lease, so a down+stale agent that grew a rebuild subgraph /// running under its lease, so a down+stale agent that grew a rebuild subgraph
/// reports `rebuilding` during its swap and `starting` at its reconcile, /// reports `rebuilding` during its swap and `starting` at its reconcile,
/// without the DAG having to guess one label covering every target. /// without the DAG having to guess one label covering every target.
pub(crate) fn start_nodes(builder: &JobBuilder, targets: &[(String, bool, bool)]) { pub(crate) fn start_nodes(
for (agent, running, stale) in targets { builder: &JobBuilder,
start_chain(builder, agent, *running, *stale); targets: &[(String, bool, bool)],
} ) -> Vec<hive_jobq::NodeGuid> {
targets
.iter()
.flat_map(|(agent, running, stale)| start_chain(builder, agent, *running, *stale))
.collect()
} }
/// Declare the restart DAG from explicit `(agent, running)` targets. /// Declare the restart DAG from explicit `(agent, running)` targets.
pub(crate) fn restart_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) { pub(crate) fn restart_nodes(
for (agent, running) in targets { builder: &JobBuilder,
restart_chain(builder, agent, graceful, *running); targets: &[(String, bool)],
} graceful: bool,
) -> Vec<hive_jobq::NodeGuid> {
targets
.iter()
.map(|(agent, running)| restart_chain(builder, agent, graceful, *running))
.collect()
} }
/// Restart a single agent. Thin wrapper over [`restart_many`]. // ---- entry points ---------------------------------------------------------
pub async fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 { //
restart_many(coord, &[agent.to_owned()], false, source, reason).await // Each reads the live state its shape depends on, then declares + inserts.
}
/// Graceful restart of a single agent (signal → drain → stop → reconcile, /// Restart `agents` in a **single** DAG — one per-agent subgraph each, built
/// when running). Thin wrapper over [`restart_many`] with `graceful = true`. /// from live running state and run concurrently on their own leases. A running
pub async fn graceful_restart( /// agent gets the stop→reconcile chain (`graceful` prepends signal→drain); a
coord: &Arc<Coordinator>, /// down agent gets a lone `Reconcile`. Restart never writes `wanted`, so the
agent: &str,
source: Source,
reason: String,
) -> u64 {
restart_many(coord, &[agent.to_owned()], true, source, reason).await
}
/// Restart `agents` (one or many) in a **single** DAG — one per-agent
/// subgraph each, built dynamically from live running state and run
/// concurrently on their own leases. A running agent gets the stop→reconcile
/// chain (`graceful` prepends signal→drain); a down agent gets just a lone
/// `Reconcile` (nothing to stop). Restart never writes `wanted`, so the
/// tail `Reconcile` converges each agent to its EXISTING intent — a /// tail `Reconcile` converges each agent to its EXISTING intent — a
/// deliberately-stopped agent stays down. The whole hive-wide /// deliberately-stopped agent stays down.
/// `hivectl restart` is one DAG. ///
/// # Errors
/// Propagates a graph-insert error.
pub async fn restart_many( pub async fn restart_many(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
agents: &[String], agents: &[String],
graceful: bool, graceful: bool,
source: Source, ) -> anyhow::Result<Vec<NodeId>> {
reason: String,
) -> u64 {
let mut targets = Vec::with_capacity(agents.len()); let mut targets = Vec::with_capacity(agents.len());
for agent in agents { for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await)); targets.push((agent.clone(), lifecycle::is_running(agent).await));
} }
submit_and_emit(coord, source, reason, |builder| { let ids = coord
restart_nodes(builder, &targets, graceful); .job_queue
}) .insert_job(|b| restart_nodes(b, &targets, graceful))?;
coord.emit_rebuild_queue_snapshot();
Ok(ids)
} }
/// Start a single agent. Thin wrapper over [`start_many`]. /// Start `agents` in a **single** DAG. A down agent gets
pub async fn start(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 { /// `SetWanted(Up) → Reconcile` (or, rev stale, a rebuild-then-start so it comes
start_many(coord, &[agent.to_owned()], source, reason).await /// up on current derivations); an already-running agent gets the same shape
} /// with the reconcile noop'ing.
///
/// Start `agents` (one or many) in a **single** DAG — one per-agent subgraph /// # Errors
/// each, built dynamically from live state and run concurrently on their own /// Propagates a graph-insert error.
/// leases. A down agent gets `SetWanted(Up) → Reconcile` (or, rev stale, a
/// rebuild-then-start so it comes up on current derivations); an already-
/// running agent gets `SetWanted(Up) → Reconcile` (the reconcile noops). The
/// whole hive-wide `hivectl start` is one DAG.
pub async fn start_many( pub async fn start_many(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
agents: &[String], agents: &[String],
source: Source, ) -> anyhow::Result<Vec<NodeId>> {
reason: String,
) -> u64 {
let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake); let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake);
let mut targets = Vec::with_capacity(agents.len()); let mut targets = Vec::with_capacity(agents.len());
for agent in agents { for agent in agents {
@ -296,88 +300,30 @@ pub async fn start_many(
} }
targets.push((agent.clone(), running, stale)); targets.push((agent.clone(), running, stale));
} }
submit_and_emit(coord, source, reason, |builder| { let ids = coord.job_queue.insert_job(|b| start_nodes(b, &targets))?;
start_nodes(builder, &targets); coord.emit_rebuild_queue_snapshot();
}) Ok(ids)
} }
/// Hard stop a single agent. Thin wrapper over [`stop_many`]. /// Stop `agents` in a **single** DAG. A running agent gets
pub async fn stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 { /// `SetWanted(Off) → [Signal → Drain →](graceful) Reconcile`; a down agent
stop_many(coord, &[agent.to_owned()], false, source, reason).await /// skips the pointless quiesce but keeps the `Reconcile` as the race-up
} /// backstop.
///
/// Graceful stop of a single agent (signal → drain → reconcile, when /// # Errors
/// running). Thin wrapper over [`stop_many`] with `graceful = true`. /// Propagates a graph-insert error.
pub async fn graceful_stop(
coord: &Arc<Coordinator>,
agent: &str,
source: Source,
reason: String,
) -> u64 {
stop_many(coord, &[agent.to_owned()], true, source, reason).await
}
/// Stop `agents` (one or many) in a **single** DAG — one per-agent subgraph
/// each, built dynamically from live state and run concurrently on their own
/// leases. A running agent gets `SetWanted(Off) → [Signal → Drain →](graceful)
/// Reconcile`; a down agent gets just `SetWanted(Off) → Reconcile` (skips the
/// pointless quiesce, keeps the Reconcile as the race-up backstop). The whole
/// hive-wide `hivectl stop` is one DAG.
pub async fn stop_many( pub async fn stop_many(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
agents: &[String], agents: &[String],
graceful: bool, graceful: bool,
source: Source, ) -> anyhow::Result<Vec<NodeId>> {
reason: String,
) -> u64 {
let mut targets = Vec::with_capacity(agents.len()); let mut targets = Vec::with_capacity(agents.len());
for agent in agents { for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await)); targets.push((agent.clone(), lifecycle::is_running(agent).await));
} }
submit_and_emit(coord, source, reason, |builder| { let ids = coord
stop_nodes(builder, &targets, graceful); .job_queue
}) .insert_job(|b| stop_nodes(b, &targets, graceful))?;
} coord.emit_rebuild_queue_snapshot();
Ok(ids)
/// Perm change: commit the JSON file(s) then rebuild.
pub fn perm_change(
coord: &Arc<Coordinator>,
agent: &str,
source: Source,
reason: String,
payload: super::PermPayload,
) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::perm_change(builder, agent, payload);
})
}
/// Meta-input lock bump; cascade rebuilds fan out on completion.
pub fn meta_update(
coord: &Arc<Coordinator>,
inputs: Vec<String>,
source: Source,
reason: String,
) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::meta_update(builder, inputs, None);
})
}
/// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs —
/// one entry for `set-parent`, N for `set-parent-bulk`. Fire-and-forget like
/// everything else in this module: submits and returns a DAG id
/// immediately, the caller learns the outcome async (dashboard job view /
/// `hivectl`'s `QueueDag` poll). Wired from `server.rs`'s `HostRequest::
/// SetParent` (hivectl) and `dashboard/topology.rs`'s `set-parent`/
/// `set-parent-bulk` handlers.
pub fn reparent(
coord: &Arc<Coordinator>,
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
source: Source,
reason: String,
) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::reparent(builder, moves);
})
} }

View file

@ -93,7 +93,7 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
let coord = node_coord; let coord = node_coord;
async move { async move {
tracing::info!( tracing::info!(
dag = coord.job_queue.dag_of(id).unwrap_or_default(), dag = coord.job_queue.root_of(id).unwrap_or_default(),
node = id.get(), node = id.get(),
kind = kind.as_str(), kind = kind.as_str(),
agent = %kind.agent(), agent = %kind.agent(),

View file

@ -17,8 +17,8 @@
//! //!
//! The hive-wide **power ops** (`stop` / `start` / `restart`) are NOT here: //! The hive-wide **power ops** (`stop` / `start` / `restart`) are NOT here:
//! their per-agent shape depends on live running state (an async //! their per-agent shape depends on live running state (an async
//! `lifecycle::is_running` read), so `submit.rs` assembles them out of the //! `lifecycle::is_running` read), so [`super::power`] assembles them out of
//! primitives this module exports ([`rebuild_nodes`]). //! the primitives this module exports ([`rebuild_nodes`]).
use hive_jobq::TerminalState; use hive_jobq::TerminalState;
@ -317,13 +317,22 @@ pub(crate) fn deploy_rebuild_nodes(builder: &JobBuilder, agent: &str, approval_i
/// children. /// children.
/// ///
/// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots /// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots
/// (`MetaSync`, `Prebuild`, `Reconcile`) — `Prebuild`'s roll-up carries the /// (`MetaSync`, the `AgentWindow` brace, `Reconcile`) — the brace's roll-up
/// whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so those three cover every /// carries the whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so
/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so /// those three cover every node. Edging `Reconcile` alone would not do: it is
/// it reaches `Done` even after a failed swap and the tail would report success. /// `AfterAny` the brace, so it reaches `Done` even after a failed swap and the
pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) { /// tail would report success.
///
/// Returns those three roots, so a caller that needs to wait on the rebuild can
/// name them.
pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec<hive_jobq::NodeGuid> {
let roots = rebuild_nodes(builder, agent, relock, None); let roots = rebuild_nodes(builder, agent, relock, None);
emit_rebuilt_tails(builder, agent, &roots.all()); emit_rebuilt_tails(builder, agent, &roots.all());
vec![
roots.meta_sync.guid(),
roots.agent_window.guid(),
roots.reconcile.guid(),
]
} }
/// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the /// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the
@ -483,10 +492,16 @@ pub fn meta_update(builder: &JobBuilder, inputs: Vec<String>, approval_id: Optio
/// checks), so a parent move needs no container rebuild to take effect. /// checks), so a parent move needs no container rebuild to take effect.
/// No transient pill either — the node is agentless (no lease to hang one /// No transient pill either — the node is agentless (no lease to hang one
/// off of) and near-instant. No tail node: the write is the whole effect. /// off of) and near-instant. No tail node: the write is the whole effect.
pub fn reparent(builder: &JobBuilder, moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>) { ///
let _reparent = builder /// Returns the single node's guid so a caller can wait on it.
pub fn reparent(
builder: &JobBuilder,
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
) -> hive_jobq::NodeGuid {
builder
.node(NodeKind::Reparent { moves }) .node(NodeKind::Reparent { moves })
.needs(Resource::MetaWindow); .needs(Resource::MetaWindow)
.guid()
} }
// The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree` // The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree`

File diff suppressed because it is too large Load diff

View file

@ -180,13 +180,19 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
// submit returns a DAG id immediately, the caller polls // submit returns a DAG id immediately, the caller polls
// `QueueDag` (`hivectl`'s wait/progress loop) for the // `QueueDag` (`hivectl`'s wait/progress loop) for the
// outcome instead of blocking here on the commit. // outcome instead of blocking here on the commit.
let id = crate::job_queue::submit::reparent( let inserted = coord.job_queue.insert_job(|b| {
&coord, vec![crate::job_queue::templates::reparent(
vec![(child.clone(), new_parent.clone())], b,
crate::job_queue::Source::Manual, vec![(child.clone(), new_parent.clone())],
"manual set-parent via hivectl".to_owned(), )]
); });
HostResponse::queued(vec![id]) match inserted {
Ok(ids) => {
coord.emit_rebuild_queue_snapshot();
HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
}
Err(e) => HostResponse::error(format!("queue reparent: {e}")),
}
} }
HostRequest::SetResourceLimits { HostRequest::SetResourceLimits {
name, name,
@ -777,39 +783,36 @@ enum Verb {
} }
async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse { async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
use crate::job_queue::{Source, submit}; use crate::job_queue::power;
let id = match verb { // A single-target op is the N-target one with N = 1 — the shapes are
// identical, so there is no separate builder to keep in sync.
let targets = [name.to_owned()];
let ids = match verb {
Verb::Kill => { Verb::Kill => {
tracing::info!(%name, "kill"); tracing::info!(%name, "kill");
submit::stop( power::stop_many(coord, &targets, false).await
coord,
name,
Source::Manual,
"manual kill via hivectl".to_owned(),
)
.await
} }
Verb::Restart => { Verb::Restart => {
tracing::info!(%name, "restart"); tracing::info!(%name, "restart");
submit::restart( power::restart_many(coord, &targets, false).await
coord,
name,
Source::Manual,
"manual restart via hivectl".to_owned(),
)
.await
} }
Verb::Rebuild => { Verb::Rebuild => {
tracing::info!(%name, "rebuild"); tracing::info!(%name, "rebuild");
submit::rebuild( // Not a power op: a rebuild's shape doesn't depend on live state,
coord, // so it is a plain template insert rather than a `*_many` gather.
name, let inserted = coord
Source::Manual, .job_queue
"manual rebuild via hivectl".to_owned(), .insert_job(|b| crate::job_queue::templates::rebuild(b, name, true));
) if inserted.is_ok() {
coord.emit_rebuild_queue_snapshot();
}
inserted
} }
}; };
HostResponse::queued(vec![id]) match ids {
Ok(ids) => HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()),
Err(e) => HostResponse::error(format!("queue insert failed: {e}")),
}
} }
/// Stop the given `agents` (resolved logical names) then `infra` containers /// Stop the given `agents` (resolved logical names) then `infra` containers
@ -837,27 +840,17 @@ async fn handle_stop(
let mut errors: Vec<String> = Vec::new(); let mut errors: Vec<String> = Vec::new();
let mut queued: Vec<u64> = Vec::new(); let mut queued: Vec<u64> = Vec::new();
// One DAG for all targeted agents — a per-agent stop subgraph each // One insert for all targeted agents — a per-agent stop subgraph each
// (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent // (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent
// roots that run concurrently on their own leases. A hive-wide // roots that run concurrently on their own leases.
// `hivectl stop` is now a single DAG, not N.
if !agents.is_empty() { if !agents.is_empty() {
let reason = if graceful { match crate::job_queue::power::stop_many(coord, agents, graceful).await {
"manual via hivectl graceful stop" Ok(ids) => {
} else { queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
"manual via hivectl stop" ok_items.extend(agents.iter().cloned());
}; }
queued.push( Err(e) => errors.push(format!("queue stop: {e}")),
crate::job_queue::submit::stop_many( }
coord,
agents,
graceful,
crate::job_queue::Source::Manual,
reason.to_owned(),
)
.await,
);
ok_items.extend(agents.iter().cloned());
} }
// Agents go down before infra so they're not mid-request against a // Agents go down before infra so they're not mid-request against a
@ -944,16 +937,13 @@ async fn handle_start(
// hivectl's wait loop. // hivectl's wait loop.
let mut queued: Vec<u64> = Vec::new(); let mut queued: Vec<u64> = Vec::new();
if !agents.is_empty() { if !agents.is_empty() {
queued.push( match crate::job_queue::power::start_many(coord, agents).await {
crate::job_queue::submit::start_many( Ok(ids) => {
coord, queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
agents, ok_items.extend(agents.iter().cloned());
crate::job_queue::Source::Manual, }
"manual via hivectl start".to_owned(), Err(e) => errors.push(format!("queue start: {e}")),
) }
.await,
);
ok_items.extend(agents.iter().cloned());
} }
let mut resp = finish_lifecycle(ok_items, &errors); let mut resp = finish_lifecycle(ok_items, &errors);
@ -982,28 +972,20 @@ async fn handle_restart_scoped(
let mut errors: Vec<String> = Vec::new(); let mut errors: Vec<String> = Vec::new();
let mut queued: Vec<u64> = Vec::new(); let mut queued: Vec<u64> = Vec::new();
// One DAG for all targeted agents — a per-agent restart subgraph each // One insert for all targeted agents — a per-agent restart subgraph each
// (`SetWanted → [Signal → Drain →] StopForUpdate → Reconcile`), // (`[Signal → Drain →] StopForUpdate → Reconcile`; no `SetWanted`, since a
// independent roots that run concurrently on their own leases. A // restart converges to the agent's *existing* intent rather than rewriting
// hive-wide `hivectl restart` is now a single DAG, not N. No // it), independent roots running concurrently on their own leases. No
// client-side stop-then-start composition — the whole restart survives // client-side stop-then-start composition — the whole restart survives a
// a dropped connection because the DAG owns it. // dropped connection because the graph owns it.
if !agents.is_empty() { if !agents.is_empty() {
queued.push( match crate::job_queue::power::restart_many(coord, &agents, graceful).await {
crate::job_queue::submit::restart_many( Ok(ids) => {
coord, queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
&agents, ok_items.extend(agents.iter().cloned());
graceful, }
crate::job_queue::Source::Manual, Err(e) => errors.push(format!("queue restart: {e}")),
if graceful { }
"manual via hivectl restart --graceful".to_owned()
} else {
"manual restart via hivectl restart".to_owned()
},
)
.await,
);
ok_items.extend(agents.iter().cloned());
} }
for &container in &infra { for &container in &infra {

View file

@ -20,13 +20,9 @@ pub(super) async fn handle_start(coord: &Arc<Coordinator>, agent: &str, name: &s
// Persist `wanted = Up` and submit the Start DAG; the submit layer // Persist `wanted = Up` and submit the Start DAG; the submit layer
// upgrades a stale-rev start to a full rebuild so the container // upgrades a stale-rev start to a full rebuild so the container
// runs current nix derivations before it starts. // runs current nix derivations before it starts.
crate::job_queue::submit::start( if let Err(e) = crate::job_queue::power::start_many(coord, &[name.to_owned()]).await {
coord, tracing::error!(%agent, %name, error = ?e, "start: insert failed");
name, }
crate::job_queue::Source::Manual,
format!("agent `{agent}` start tool"),
)
.await;
Response::Ok Response::Ok
} }
@ -48,13 +44,9 @@ pub(super) async fn handle_restart(coord: &Arc<Coordinator>, agent: &str, name:
return err; return err;
} }
tracing::info!(%agent, %name, "submit restart"); tracing::info!(%agent, %name, "submit restart");
crate::job_queue::submit::restart( if let Err(e) = crate::job_queue::power::restart_many(coord, &[name.to_owned()], false).await {
coord, tracing::error!(%agent, %name, error = ?e, "restart: insert failed");
name, }
crate::job_queue::Source::Manual,
format!("agent `{agent}` restart tool"),
)
.await;
Response::Ok Response::Ok
} }
@ -153,12 +145,13 @@ pub(super) fn handle_update(coord: &Arc<Coordinator>, agent: &str, name: &str) -
return err; return err;
} }
tracing::info!(%agent, %name, "submit rebuild"); tracing::info!(%agent, %name, "submit rebuild");
crate::job_queue::submit::rebuild( if let Err(e) = coord.job_queue.insert_job(|b| {
coord, crate::job_queue::templates::rebuild(b, name, true);
name, Vec::new()
crate::job_queue::Source::Manual, }) {
format!("agent `{agent}` update tool"), tracing::error!(%agent, %name, error = ?e, "update: insert failed");
); }
coord.emit_rebuild_queue_snapshot();
Response::Ok Response::Ok
} }

View file

@ -109,12 +109,11 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
tracing::warn!( tracing::warn!(
"manager container exists but no applied flake — forcing rebuild to migrate" "manager container exists but no applied flake — forcing rebuild to migrate"
); );
if let Err(e) = coord.job_queue.submit( if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::Source::AutoUpdate, crate::job_queue::templates::rebuild(b, MANAGER_NAME, true);
"manager migration: no applied flake".to_owned(), Vec::new()
|b| crate::job_queue::templates::rebuild(b, MANAGER_NAME, true), }) {
) { tracing::warn!(error = ?e, "manager migration rebuild insert failed");
tracing::warn!(error = ?e, "manager migration rebuild submit failed");
} }
} else { } else {
tracing::debug!("manager container already present"); tracing::debug!("manager container already present");
@ -378,18 +377,19 @@ fn submit_boot_tree(
n_deferred: usize, n_deferred: usize,
n_skipped: usize, n_skipped: usize,
) { ) {
use crate::job_queue::Source; // Fully-quiet boot (nothing stale, nothing drifted) inserts nothing.
// Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
if !any_stale && drifted.is_empty() { if !any_stale && drifted.is_empty() {
return; return;
} }
let reason = format!( // The summary the sweep used to hand the container as its `reason` is a log
"boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date", // line now: it was only ever stored on a node nobody read, and the counts
fanout.len(), // are worth having where they can actually be seen.
drifted.len(), tracing::info!(
n_deferred, rebuilds = fanout.len(),
n_skipped, reconciles = drifted.len(),
deferred = n_deferred,
up_to_date = n_skipped,
"boot: sweep"
); );
// The sweep's own rebuild subgraphs emit their `Rebuilt` events as they // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they
@ -397,10 +397,11 @@ fn submit_boot_tree(
// The subgraphs also carry their own per-agent crash-watch suppression // The subgraphs also carry their own per-agent crash-watch suppression
// during their `Swap` (applied at claim time); a reconcile-only boot needs // during their `Swap` (applied at claim time); a reconcile-only boot needs
// no transient. // no transient.
if let Err(e) = coord.job_queue.submit(Source::AutoUpdate, reason, |b| { if let Err(e) = coord.job_queue.insert_job(|b| {
boot_nodes(b, any_stale, fanout, drifted); boot_nodes(b, any_stale, fanout, drifted);
Vec::new()
}) { }) {
tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); tracing::warn!(error = ?e, "boot: sweep DAG insert failed");
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
} }

View file

@ -1,6 +1,11 @@
//! Vocabulary hive-c0re's job queue shares with its clients: where a job //! Vocabulary hive-c0re's job queue shares with its clients: what a
//! came from ([`Source`]), what a permission change carries //! permission change carries ([`PermPayload`]) and the scheduler's
//! ([`PermPayload`]), and the scheduler's lifecycle [`State`]. //! lifecycle [`State`].
//!
//! A `Source` enum lived here too — where a job came from, rendered as the
//! "why" chip. It was a field on the DAG container, and it went with it: a
//! job is its nodes now, and a node says what it does rather than who asked
//! for it.
//! //!
//! **The typed `DagView`/`NodeView` projection that used to live here is //! **The typed `DagView`/`NodeView` projection that used to live here is
//! gone.** One graph is served one way now — `hive_jobq_wire`'s generic //! gone.** One graph is served one way now — `hive_jobq_wire`'s generic
@ -12,37 +17,6 @@
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
/// Where the submit request originated — drives the "why" chip on the
/// dashboard.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Source {
/// Operator action (dashboard button, CLI, manager tool).
Manual,
/// Meta-update cascade rebuild (grown into the meta-update DAG).
MetaUpdate,
/// Boot-time submission (the boot sweep DAG + boot reconciles).
AutoUpdate,
/// Crash recovery path (future use).
CrashRecover,
/// Operator approved a pending `Approval` row; `approval_id` on
/// the DAG points back at the source row.
Approval,
}
impl Source {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Source::Manual => "manual",
Source::MetaUpdate => "meta_update",
Source::AutoUpdate => "auto_update",
Source::CrashRecover => "crash_recover",
Source::Approval => "approval",
}
}
}
pub use hive_jobq::State; pub use hive_jobq::State;
/// Kind-specific payload for `Template::PermChange` DAGs. /// Kind-specific payload for `Template::PermChange` DAGs.