feat(#2446): release a DAG's per-agent lease when that agent's subgraph is terminal

A per-agent lifecycle lease gates that agent's container globally across
concurrent DAGs, so it should be held for exactly as long as the agent's
work in a DAG is in flight, no longer. settle() previously freed every
lease a DAG held only at whole-DAG terminal, so a multi-agent DAG (a
hive-wide restart) kept agent A's container locked until B and C also
finished, blocking any other DAG wanting A.

Now free each agent's lease the moment its own subgraph within the DAG is
terminal (no live node still targets it), and drop that agent's dashboard
transient pill on the same edge via a new per-agent release channel. A
single-agent DAG is unaffected: its agent's subgraph goes terminal exactly
when the whole DAG does, so behaviour is identical.

hive-c0re/src/job_queue/mod.rs

@@ -94,6 +94,17 @@ pub struct TerminalDag {
     pub error: Option<String>,
 }
 
+/// 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<Dag>,
@@ -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<TerminalDag>,
+    /// 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<AgentRelease>,
 }
 
 /// 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<AgentRelease> {
+        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<String> = 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<AgentRelease> = 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<TerminalDag> = 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);
     }
 

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

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<Coordinator>,
     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;
     }

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);