diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 25d6df7d..f9d6cdc0 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -94,6 +94,17 @@ pub struct TerminalDag { pub error: Option, } +/// A single per-agent lease release: agent `agent`'s subgraph within DAG +/// `dag_id` just reached terminal, so the scheduler drops that agent's +/// `(dag_id, agent)` transient guard — ahead of (or coinciding with) the +/// whole-DAG [`TerminalDag`]. The lease itself is freed inside `settle`; +/// this only carries the transient-drop signal out to the scheduler. +#[derive(Debug, Clone)] +pub struct AgentRelease { + pub dag_id: u64, + pub agent: String, +} + #[derive(Debug, Default)] struct Inner { dags: VecDeque, @@ -108,6 +119,12 @@ struct Inner { /// terminal hooks (approval resolution, intent revert, transient /// release) fire exactly once per DAG no matter how it ended. pending_terminal: Vec, + /// Per-agent lease releases not yet consumed by the scheduler + /// ([`JobQueue::drain_agent_releases`]). Fed by `settle` the moment + /// an agent's subgraph within a DAG goes terminal — earlier than the + /// whole-DAG `pending_terminal` for a multi-agent DAG. Drives the + /// per-agent transient-guard drop. + pending_agent_release: Vec, } /// The queue. Lives on `Coordinator` (one per hive-c0re process); a @@ -419,45 +436,70 @@ impl JobQueue { std::mem::take(&mut inner.pending_terminal) } - /// Propagate cancellations, release the leases of newly-terminal - /// DAGs, buffer each terminal roll-up exactly once (the - /// `terminal_reported` flag) for [`Self::drain_terminal`], and trim - /// history. + /// Take the per-agent lease releases accumulated since the last + /// drain. The scheduler calls this every wakeup and drops the + /// matching `(dag_id, agent)` transient guard for each — freeing an + /// agent's dashboard pill the moment its subgraph settles, ahead of + /// the whole-DAG terminal for a multi-agent DAG. + pub fn drain_agent_releases(&self) -> Vec { + let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); + std::mem::take(&mut inner.pending_agent_release) + } + + /// Propagate cancellations, release each agent's lease the moment its + /// own subgraph settles (not at whole-DAG terminal), buffer each + /// terminal roll-up exactly once (the `terminal_reported` flag) for + /// [`Self::drain_terminal`], and trim history. fn settle(inner: &mut Inner) { Self::propagate_cancellations(inner); - let mut freed: Vec = Vec::new(); + + // Per-agent early lease release. A lease gates an agent's + // container globally, so it must be held for exactly as long as + // that agent's work in the DAG is in flight — no longer. The + // moment no live node of a DAG still targets an agent, free that + // agent's lease so a concurrent DAG wanting the same agent can + // proceed, even while the rest of this DAG runs on. For a + // single-agent DAG the agent's subgraph goes terminal exactly + // when the whole DAG does, so this reduces to the old behaviour. + // Runs over ALL dags, gated on this dag actually holding the + // lease, so it fires exactly once per (dag, agent). + let mut released: Vec = Vec::new(); + for dag in &inner.dags { + for agent in dag.agents() { + if inner.leases.get(agent.as_str()) == Some(&dag.id) + && dag.agent_subgraph_terminal(&agent) + { + released.push(AgentRelease { + dag_id: dag.id, + agent, + }); + } + } + } + for rel in &released { + inner.leases.remove(&rel.agent); + } + inner.pending_agent_release.extend(released); + + // Whole-DAG terminal roll-up (approval resolution, intent revert, + // `Rebuilt` events) — still fires once per DAG. Leases are already + // freed by the per-agent pass above by the time we get here. let mut reports: Vec = Vec::new(); for dag in &mut inner.dags { if !dag.is_terminal() || dag.terminal_reported { continue; } dag.terminal_reported = true; - // Free every agent-lease this DAG holds (one per distinct - // agent it touched). Single-agent DAGs release their one lease; - // a future multi-agent DAG releases all of them at terminal. - // (Per-agent early release — freeing an agent's lease the moment - // that agent's subgraph is terminal rather than at whole-DAG - // terminal — is a refinement for the multi-agent-emission - // follow-up, where it actually matters.) - let agents = dag.agents(); - for agent in &agents { - if inner.leases.get(agent.as_str()) == Some(&dag.id) { - freed.push(agent.clone()); - } - } reports.push(TerminalDag { dag_id: dag.id, template: dag.template, - agents, + agents: dag.agents(), approval_id: dag.approval_id, state: dag.rollup(), error: dag.first_error().map(str::to_owned), }); } inner.pending_terminal.append(&mut reports); - for agent in freed { - inner.leases.remove(&agent); - } Self::trim_history(inner, now_unix() - HISTORY_GRACE_SECS); } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index fed808a4..7ffe4a04 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -302,6 +302,19 @@ impl Dag { self.nodes.iter().all(|n| n.state.is_terminal()) } + /// True when no live (non-terminal) node of this DAG still targets + /// `agent` — i.e. that agent's subgraph within the DAG has settled. + /// Used to release an agent's lifecycle lease the moment its own + /// work is done, rather than waiting for the whole DAG to terminate. + /// Vacuously true for an agent the DAG has no node for; callers gate + /// on actually holding that agent's lease first. + pub fn agent_subgraph_terminal(&self, agent: &str) -> bool { + self.nodes + .iter() + .filter(|n| n.agent == agent) + .all(|n| n.state.is_terminal()) + } + /// First failed node's error, for the roll-up `error` field. pub fn first_error(&self) -> Option<&str> { self.nodes diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index c0f5e59c..3e9959d5 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -139,15 +139,25 @@ async fn handle_completion( coord.emit_rebuild_queue_snapshot(); } -/// Drain buffered terminal roll-ups: drop each DAG's lease-window -/// transient guard, then run the terminal hook (approval resolution, +/// Drain per-agent lease releases and buffered terminal roll-ups. +/// +/// Per-agent first: an agent's subgraph within a DAG went terminal (its +/// lease was freed in `settle`), so drop that agent's `(dag, agent)` +/// transient pill now — ahead of whole-DAG terminal for a multi-agent +/// DAG. Then the whole-DAG terminals: drop any remaining transient the +/// DAG still held and run the terminal hook (approval resolution, /// `Rebuilt` events, cancelled-power-op intent revert). async fn process_terminals( coord: &Arc, transients: &mut HashMap<(u64, String), crate::coordinator::TransientGuard>, ) { + for rel in coord.job_queue.drain_agent_releases() { + transients.remove(&(rel.dag_id, rel.agent)); + } for terminal in coord.job_queue.drain_terminal() { - // Drop every per-agent transient guard this DAG held. + // Drop any per-agent transient guard the DAG still held (the + // per-agent pass above already dropped the ones whose subgraphs + // settled early). transients.retain(|(dag_id, _), _| *dag_id != terminal.dag_id); exec::on_dag_terminal(coord, &terminal).await; } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index ef981f4d..2a40034f 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -328,6 +328,52 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { ); } +/// A multi-agent DAG frees an agent's lease the moment THAT agent's +/// subgraph is terminal — not when the whole DAG finishes. So a +/// concurrent DAG wanting the finished agent can proceed while the rest +/// of the first DAG runs on. +#[test] +fn multi_agent_lease_frees_per_subgraph_not_whole_dag() { + let q = JobQueue::new(4); + let id = submit(&q, restart_online(&["agent-a", "agent-b"], false, "r")); + + // Drive agent-a's ENTIRE subgraph to Done while leaving agent-b's + // head running (so agent-b keeps holding its lease). + let mut b_in_flight = false; + loop { + let mut progressed = false; + for c in q.claim_ready() { + if c.agent == "agent-a" { + q.complete_node(id, c.node_id, Ok(())); + progressed = true; + } else { + b_in_flight = true; // leave agent-b's node running + } + } + if !progressed { + break; + } + } + assert!(b_in_flight, "agent-b subgraph should still be in flight"); + // The DAG as a whole is NOT terminal — agent-b runs on. + assert_eq!(state_of(&q, id), State::Running); + + // agent-a's lease is freed early → a concurrent agent-a DAG runs; + // an agent-b DAG still blocks on the lease agent-b's subgraph holds. + submit(&q, restart_online(&["agent-a"], false, "concurrent-a")); + submit(&q, restart_online(&["agent-b"], false, "concurrent-b")); + let claims = q.claim_ready(); + let agents: Vec<&str> = claims.iter().map(|c| c.agent.as_str()).collect(); + assert!( + agents.contains(&"agent-a"), + "agent-a lease freed the moment its subgraph settled" + ); + assert!( + !agents.contains(&"agent-b"), + "agent-b lease still held — its subgraph is still in flight" + ); +} + #[test] fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() { let q = JobQueue::new(4);