diff --git a/core/continuum-core/src/airc/inbound_attach.rs b/core/continuum-core/src/airc/inbound_attach.rs index c963ee5bf4..4deeb056d6 100644 --- a/core/continuum-core/src/airc/inbound_attach.rs +++ b/core/continuum-core/src/airc/inbound_attach.rs @@ -323,6 +323,31 @@ pub async fn publish_transcript_event( "first capacity offer heard from a grid peer", ); } + } else if let Some(beacon) = residency_beacon_from_envelope(&envelope) { + // Residency beacon (grid-overflow eligibility): fold the heard beacon into the + // process-global residency ledger, keyed on the WIRE's peer id — the orthogonal + // sibling of the capacity fold above. Our own echo lands here too (the loopback + // proof that publish→hear works before a second node exists). + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let model_count = beacon.resident_models.len(); + let is_new = crate::capacity::model_residency::global_residency_ledger().hear( + event.peer_id.as_uuid(), + beacon, + now_ms, + ); + if is_new { + crate::probe!( + class = "grid.residency.heard", + from_peer = %event.peer_id.as_uuid(), + model_count = model_count, + heard_peers = + crate::capacity::model_residency::global_residency_ledger().heard_count(), + "first residency beacon heard from a grid peer", + ); + } } else if let Some((name, payload)) = chat_posted_from_envelope(&envelope, event) { crate::probe!( class = "airc.chat.projected", @@ -400,6 +425,22 @@ fn capacity_offer_from_envelope( serde_json::from_value(payload.inline.clone()?).ok() } +/// Decode a `grid_residency` envelope's inline payload into a [`ResidencyBeacon`]. +/// Returns `None` for any other envelope — the residency sibling of +/// [`capacity_offer_from_envelope`], same honest schema gate. +fn residency_beacon_from_envelope( + envelope: &AircRealtimeEnvelope, +) -> Option { + let crate::airc::realtime::AircRealtimePayload::ExistingSchema { payload } = &envelope.payload + else { + return None; + }; + if payload.schema != crate::airc::realtime::AircRealtimeSchema::GridResidency { + return None; + } + serde_json::from_value(payload.inline.clone()?).ok() +} + /// Project a plain airc chat message into the THIN `chat:posted` bus /// payload the positron chat projection consumes (`AircChatPosted` in /// `ipc/positron_source.rs`). Returns `None` for any non-message event diff --git a/core/continuum-core/src/airc/realtime.rs b/core/continuum-core/src/airc/realtime.rs index 1d412b1a79..8606c64a30 100644 --- a/core/continuum-core/src/airc/realtime.rs +++ b/core/continuum-core/src/airc/realtime.rs @@ -53,6 +53,13 @@ pub enum AircRealtimeSchema { /// presence-of-compute. EphemeralCoalesced: latest wins, never replayed /// (a stale capacity reading is a lie). GridCapacity, + /// A node's residency beacon (`capacity::model_residency::ResidencyBeacon`) — + /// which models it holds resident, the grid-overflow ELIGIBILITY signal (a peer + /// is only a fast overflow target for a model it already holds). Orthogonal to + /// GridCapacity (concurrency): the governor composes the two. EphemeralCoalesced + /// like capacity, but published on a slower cadence — residency changes on model + /// page-in/out (minute-scale), not the 10s capacity beat. + GridResidency, } /// Handle to a payload already defined by a Continuum schema. @@ -458,7 +465,9 @@ impl AircRealtimePayload { | AircRealtimeSchema::LiveKitBridgeEvent => AircRealtimeDelivery::Control, // Capacity offers are presence-of-compute: latest wins, never // replayed — a stale reading must not outlive its freshness. - AircRealtimeSchema::GridCapacity => AircRealtimeDelivery::EphemeralCoalesced, + AircRealtimeSchema::GridCapacity | AircRealtimeSchema::GridResidency => { + AircRealtimeDelivery::EphemeralCoalesced + } _ => AircRealtimeDelivery::Durable, }, Self::Presence { event } => event.delivery(), diff --git a/core/continuum-core/src/capacity/grid_overflow.rs b/core/continuum-core/src/capacity/grid_overflow.rs new file mode 100644 index 0000000000..b34283d8d8 --- /dev/null +++ b/core/continuum-core/src/capacity/grid_overflow.rs @@ -0,0 +1,237 @@ +//! Grid-overflow routing — the DECISION half of the governor consumer: given a serving plan +//! that overflowed local capacity, decide which eligible peers take the overflow lanes. The +//! EFFECT half (the actual `Commands.execute("ai/generate", {aircPeer})` hop) is thin and lives +//! at the live seam; THIS is pure, deterministic, and fully unit-tested. +//! +//! ## Why overflow placement is REMOTE-ONLY (not [`super::grid::LocalFirstFitPolicy`]) +//! +//! [`super::grid::LocalFirstFitPolicy`] fills local first and floors local at `≥1` (a resident +//! model must be able to run one prefill). That floor is correct for a FRESH placement but +//! WRONG for overflow: overflow lanes are BY DEFINITION the ones that already could not fit +//! locally (`ServingPlan.grid_overflow_lanes = demand − local_lanes`). Placing them local-first +//! would re-cram the very lanes the planner just declared didn't fit — the thrash the honest +//! "over local capacity by N" signal exists to avoid. So overflow placement never touches local: +//! it spills ONLY to eligible remote peers. +//! +//! ## The two gates, composed (never absorbed) +//! +//! 1. RESIDENCY ([`ModelResidencyView::residency_eligible`]): a peer is a fast overflow target +//! only for a model it ALREADY holds — else the hop pays a cold full-weights load, defeating +//! the point. Filters the snapshot to peers holding the model. +//! 2. CONCURRENCY (the misfit-parts fit, [`super::lanes_that_fit`]): among residency-eligible + +//! REACHABLE peers, each takes at most what its OWN free budget fits for the prefill spike — +//! the same per-node-fit rule the single-device and grid policies run. +//! +//! Reachability is applied HERE (an unreachable-but-resident peer is a memory, not an offer), +//! composing cleanly with residency without either abstraction absorbing the other. +//! +//! ## Unplaced lanes are SURFACED, never dropped +//! +//! When no eligible peer can take a lane, it lands in [`OverflowRouting::unplaced`] — the honest +//! "the grid couldn't absorb N of your overflow" signal the caller queues or degrades on. Never +//! silently swallowed ([[fallbacks-are-illegal-fail-loud]]). + +use crate::identity::PeerId; + +use super::grid::GridSnapshot; +use super::lanes_that_fit; +use super::model_residency::ModelResidencyView; +use super::LeaseRequest; + +/// The routing decision for a plan's overflow lanes: where each lane lands (remote-only) plus +/// the honest count of lanes the reachable, residency-eligible grid could NOT absorb. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OverflowRouting { + /// Overflow lanes placed on named peers, most-free-first, each capped by its own fit. + pub remote: Vec<(PeerId, u32)>, + /// Overflow lanes no eligible+reachable peer could take — queue or degrade on these, + /// never drop them silently. + pub unplaced: u32, +} + +impl OverflowRouting { + /// Total overflow lanes actually placed on peers. + pub fn placed(&self) -> u32 { + self.remote.iter().map(|(_, n)| n).sum() + } +} + +/// Decide where a plan's overflow lanes run. `lease` is the demand→capacity projection the +/// serving side already built (`ModelFootprint::grid_lease_request(served_window, overflow_lanes)`): +/// `want_concurrency` = the overflow lane count, `spike_bytes` = the per-lane prefill transient. +/// `model_id` is what the overflowing node is serving (its `ServingPlan.base_model_id`) — the +/// residency key. REMOTE-ONLY by construction (see module docs): local is already saturated. +pub fn route_grid_overflow( + model_id: &str, + lease: &LeaseRequest, + residency: &ModelResidencyView, + snapshot: &GridSnapshot, + safety_margin_bytes: u64, +) -> OverflowRouting { + let want = lease.want_concurrency; + if want == 0 { + return OverflowRouting { remote: Vec::new(), unplaced: 0 }; + } + + // Gate 1 — residency: keep only peers that hold the model resident. + let eligible = residency.residency_eligible(snapshot, model_id); + + // Gate 2 — reachability + per-node fit: reachable eligible peers, most-free-first (fewest + // peers touched), each capped by its OWN budget for the prefill spike. + let mut reachable: Vec<_> = eligible.peers.iter().filter(|p| p.reachable).collect(); + reachable.sort_by(|a, b| { + b.capacity + .gpu_free_bytes_live + .cmp(&a.capacity.gpu_free_bytes_live) + }); + + let mut remaining = want; + let mut remote = Vec::new(); + for peer in reachable { + if remaining == 0 { + break; + } + let fit = lanes_that_fit( + peer.capacity.gpu_free_bytes_live, + safety_margin_bytes, + lease.spike_bytes, + ); + let take = fit.min(remaining); + if take > 0 { + remote.push((peer.peer, take)); + remaining -= take; + } + } + + // Whatever the reachable, residency-eligible grid couldn't absorb is surfaced, not dropped. + OverflowRouting { remote, unplaced: remaining } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::capacity::grid::{GridSnapshot, PeerCapacity}; + use crate::capacity::DeviceCapacity; + use uuid::Uuid; + + const GB: u64 = 1024 * 1024 * 1024; + + fn peer_id(n: u128) -> PeerId { + PeerId::from_uuid(Uuid::from_u128(n)) + } + + fn dev(free_gb: u64) -> DeviceCapacity { + DeviceCapacity { + gpu_total_bytes: 80 * GB, + gpu_free_bytes_live: free_gb * GB, + system_ram_free_bytes: 64 * GB, + } + } + + fn peer(n: u128, free_gb: u64, reachable: bool) -> PeerCapacity { + PeerCapacity { + peer: peer_id(n), + capacity: dev(free_gb), + reachable, + } + } + + fn lease(want: u32, spike_gb: u64) -> LeaseRequest { + LeaseRequest { + consumer: "qwen-coder".into(), + want_concurrency: want, + spike_bytes: spike_gb * GB, + } + } + + fn view_holding(peers: &[(u128, &[&str])]) -> ModelResidencyView { + let mut v = ModelResidencyView::new(); + for (n, models) in peers { + v.set_resident(peer_id(*n), models.iter().map(|s| s.to_string())); + } + v + } + + // what this catches: overflow placement is REMOTE-ONLY and residency+reachability gated. + // Local is never assigned lanes (it's the saturated node that overflowed). Only a peer that + // holds the model AND is reachable takes lanes; a resident-but-unreachable peer and a + // reachable-but-not-resident peer both take nothing. + #[test] + fn overflow_routes_remote_only_to_reachable_resident_peers() { + let snap = GridSnapshot { + local: dev(1), // saturated — must never receive overflow lanes + peers: vec![ + peer(1, 40, true), // resident + reachable → takes lanes + peer(2, 40, false), // resident + UNREACHABLE → nothing + peer(3, 40, true), // reachable but NOT resident → nothing + ], + }; + let residency = view_holding(&[(1, &["qwen-coder"]), (2, &["qwen-coder"])]); + + let routing = route_grid_overflow("qwen-coder", &lease(2, 1), &residency, &snap, GB); + + assert_eq!(routing.remote.len(), 1, "only peer 1 is eligible + reachable"); + assert_eq!(routing.remote[0].0.as_uuid(), Uuid::from_u128(1)); + assert_eq!(routing.placed(), 2, "both overflow lanes fit on peer 1"); + assert_eq!(routing.unplaced, 0); + } + + // what this catches: unplaced lanes are SURFACED, never dropped. When the eligible grid + // can't fit all overflow lanes (one small peer, a big per-lane spike), the shortfall is + // reported so the caller queues/degrades — the honest "grid couldn't absorb N" signal. + #[test] + fn lanes_the_grid_cannot_absorb_are_surfaced_not_dropped() { + let snap = GridSnapshot { + local: dev(1), + peers: vec![peer(1, 10, true)], // ~10GB free, but each lane spikes 8GB + }; + let residency = view_holding(&[(1, &["qwen-coder"])]); + + // want 3 lanes, 8GB spike each: only 1 fits on the 10GB peer (net of 1GB margin). + let routing = route_grid_overflow("qwen-coder", &lease(3, 8), &residency, &snap, GB); + + assert_eq!(routing.placed(), 1, "only one lane fits the peer's budget"); + assert_eq!(routing.unplaced, 2, "the other two are surfaced, not silently dropped"); + } + + // what this catches: no overflow (want == 0) is a clean no-op — nothing placed, nothing + // unplaced. The common case (demand fit locally, grid_overflow_lanes == 0) costs nothing. + #[test] + fn zero_overflow_is_a_clean_noop() { + let snap = GridSnapshot { + local: dev(20), + peers: vec![peer(1, 40, true)], + }; + let residency = view_holding(&[(1, &["qwen-coder"])]); + + let routing = route_grid_overflow("qwen-coder", &lease(0, 1), &residency, &snap, GB); + + assert!(routing.remote.is_empty()); + assert_eq!(routing.unplaced, 0); + assert_eq!(routing.placed(), 0); + } + + // what this catches: spill spreads across peers most-free-first, each capped by its OWN fit + // (the misfit-parts rule) — 12 aggregate GB across three 4GB peers can't run a 6-lane + // placement if no single peer fits it, but distinct small lanes DO spread. Two reachable + // resident peers each take their share until demand is met. + #[test] + fn spill_spreads_across_peers_most_free_first() { + let snap = GridSnapshot { + local: dev(1), + peers: vec![ + peer(1, 20, true), // more free → sorted first + peer(2, 12, true), + ], + }; + let residency = view_holding(&[(1, &["qwen-coder"]), (2, &["qwen-coder"])]); + + // 4 lanes, 8GB spike: peer1 (20-1 margin=19 → 2 lanes), peer2 (12-1=11 → 1 lane) = 3, + // one unplaced. Peer 1 (most free) is filled before peer 2. + let routing = route_grid_overflow("qwen-coder", &lease(4, 8), &residency, &snap, GB); + + assert_eq!(routing.remote[0].0.as_uuid(), Uuid::from_u128(1), "most-free peer first"); + assert_eq!(routing.placed() + routing.unplaced, 4, "every lane accounted for"); + assert!(routing.placed() >= 3, "peers absorb what their own budgets fit"); + } +} diff --git a/core/continuum-core/src/capacity/mod.rs b/core/continuum-core/src/capacity/mod.rs index e6dcfb0e76..3ae5d3de6f 100644 --- a/core/continuum-core/src/capacity/mod.rs +++ b/core/continuum-core/src/capacity/mod.rs @@ -42,9 +42,11 @@ pub mod expert_residency; pub mod expert_tier_policy; pub mod gossip; pub mod grid; +pub mod grid_overflow; pub mod lease; pub mod market; pub mod moe_arch_profile; +pub mod model_residency; pub mod moe_serving; pub mod pager_capture; pub mod trace_tail; diff --git a/core/continuum-core/src/capacity/model_residency.rs b/core/continuum-core/src/capacity/model_residency.rs new file mode 100644 index 0000000000..96cccc0f71 --- /dev/null +++ b/core/continuum-core/src/capacity/model_residency.rs @@ -0,0 +1,382 @@ +//! Model residency across the grid — WHICH peer currently holds WHICH model resident. The +//! ELIGIBILITY half of grid-overflow routing; the CONCURRENCY half is [`super::grid`]. +//! +//! ## Why this is a separate abstraction (orthogonal, never absorbed) +//! +//! `serving_plan` reasons about MODEL RESIDENCY — can a node hold model M's weights + per-lane +//! KV at the served window. `capacity/grid` reasons about CONCURRENCY SPIKES — does a node have +//! a free lane RIGHT NOW ([`super::LeaseRequest`]). Settled with BigMama 2026-07-27: these are +//! **orthogonal and compose** — neither should absorb the other, the mapping is the only +//! crossing point ([`super::serving_plan`]'s `grid_lease_request` is that one bridge). So +//! residency does NOT belong on [`super::grid::PeerCapacity`] (that would blur the concurrency +//! abstraction with a residency fact); it lives here, and the governor **composes** the two at +//! the placement filter. +//! +//! ## Why residency gates grid overflow +//! +//! A grid-overflow hop routes a persona's generation to a peer. If that peer already holds M +//! resident, the hop is fast — it needs only a free lane (the concurrency check). If it merely +//! has free VRAM but NOT M, accepting the hop forces a cold full-weights load (seconds to +//! minutes for a large model) — which defeats the entire point of overflowing for speed. So the +//! fast overflow path is eligible only for peers that already hold M. A peer with unknown +//! residency (never beaconed) is NOT eligible for the fast path — conservative by construction, +//! same spirit as [`super::residency_detect`] never claiming a promotion is faster than it is. +//! +//! ## The compose point +//! +//! [`ModelResidencyView::residency_eligible`] takes a live [`GridSnapshot`] and returns a +//! SMALLER snapshot — local device untouched (the overflowing node holds M by definition; it's +//! the one serving it), peers filtered to those holding M. The unchanged capacity placement +//! policy ([`super::grid::LocalFirstFitPolicy`]) then runs on that smaller snapshot: it never +//! learns about residency, it just sees fewer peers. Reachability stays the policy's job — an +//! unreachable-but-resident peer survives this filter and is dropped downstream by `place()`, +//! keeping the two concerns cleanly separate. + +use std::collections::{HashMap, HashSet}; +use std::sync::OnceLock; + +use dashmap::DashMap; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use super::grid::GridSnapshot; +use crate::identity::PeerId; + +/// Per-peer set of model ids that peer currently holds resident, folded from residency beacons. +/// +/// Keyed on the peer's `Uuid` (via [`PeerId::as_uuid`]) — the same choice +/// [`super::gossip::GridCapacityLedger`] makes for capacity offers, so residency and capacity +/// index the grid identically. A peer absent from the map has UNKNOWN residency (never +/// beaconed), which [`Self::holds`] reports as `false`: not eligible for the fast overflow path. +#[derive(Debug, Clone, Default)] +pub struct ModelResidencyView { + by_peer: HashMap>, +} + +impl ModelResidencyView { + pub fn new() -> Self { + Self::default() + } + + /// Record (latest-wins) the full set of models a peer currently holds resident. Latest-wins + /// because residency is a live fact — a model paged out is no longer held, so a fresh beacon + /// REPLACES the peer's set rather than merging (a merge would resurrect evicted models). + pub fn set_resident(&mut self, peer: PeerId, models: I) + where + I: IntoIterator, + S: Into, + { + self.by_peer + .insert(peer.as_uuid(), models.into_iter().map(Into::into).collect()); + } + + /// Does this peer currently hold `model_id` resident? `false` for a peer that never beaconed + /// (unknown residency) — the conservative default that keeps the fast overflow path honest. + pub fn holds(&self, peer: &PeerId, model_id: &str) -> bool { + self.by_peer + .get(&peer.as_uuid()) + .is_some_and(|set| set.contains(model_id)) + } + + /// Number of peers with a known residency beacon — probe surface, mirrors + /// [`super::gossip::GridCapacityLedger::heard_count`]. + pub fn known_peers(&self) -> usize { + self.by_peer.len() + } + + /// Compose this residency view with a live capacity snapshot for `model_id`: keep the local + /// device (the overflowing node holds M by definition) and keep only peers that hold M + /// resident. The returned snapshot feeds the UNCHANGED capacity placement policy — concurrency + /// logic never learns about residency, it just sees a shorter peer list. Reachability is NOT + /// applied here (that stays the policy's job downstream), so an unreachable-but-resident peer + /// survives this filter and is reclaimed by `place()`. Orthogonal, composed at exactly one + /// point, neither abstraction absorbing the other. + pub fn residency_eligible(&self, snapshot: &GridSnapshot, model_id: &str) -> GridSnapshot { + let mut eligible = snapshot.clone(); + eligible.peers.retain(|peer| self.holds(&peer.peer, model_id)); + eligible + } +} + +/// One node's residency beacon — the wire payload advertising which models it currently holds +/// resident. Rides its OWN `grid_residency` `EphemeralCoalesced` envelope, separate from +/// [`super::gossip::CapacityOffer`]'s `grid_capacity`: residency changes on model page-in/out +/// (minute-scale), not the 10s capacity beat, so coupling them would either over-publish +/// residency or under-refresh capacity. Names + timestamp only; peer identity is the WIRE's +/// (the transcript event's authenticated peer id), never the payload's — same rule as +/// `CapacityOffer`, so a peer cannot beacon residency on another's behalf. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ResidencyBeacon { + /// Model ids this node holds resident (warm) RIGHT NOW — its base plus any co-resident. + pub resident_models: Vec, + /// Sender clock when the reading was taken (ms since epoch). Displayed, not trusted: + /// freshness is judged by RECEIVER clock at hear-time, exactly like `CapacityOffer.at_ms`. + pub at_ms: u64, +} + +/// A heard beacon + the receiver-clock instant it arrived (the freshness anchor). +#[derive(Debug, Clone)] +struct HeardBeacon { + models: Vec, + heard_at_ms: u64, +} + +/// Beacons silent past this drop from the projected view entirely. GENEROUS versus capacity's +/// eviction because the two abstractions stay orthogonal: residency is STICKY (a model stays +/// resident across many capacity beats), and the COMPOSED capacity snapshot already gates +/// reachability — so a residency reading never has to prove liveness itself (that would blur +/// residency into concurrency). 6× the capacity eviction window: a peer whose residency we +/// haven't reheard in that long falls back to UNKNOWN (not asserted-resident), keeping the fast +/// overflow path honest rather than routing to a model a long-silent peer may have paged out. +pub const RESIDENCY_EVICTION_WINDOW_MS: u64 = 6 * super::gossip::EVICTION_WINDOW_MS; + +/// The residency beacon heartbeat — deliberately SLOWER than capacity's 10s +/// ([`super::gossip::PUBLISH_INTERVAL_MS`]) because residency changes on model page-in/out +/// (minute-scale), not the free-VRAM beat. 12 beats fit inside +/// [`RESIDENCY_EVICTION_WINDOW_MS`] — the same publish:eviction ratio capacity gossip uses, +/// so a couple of dropped beacons never evicts a still-resident peer. +pub const RESIDENCY_PUBLISH_INTERVAL_MS: u64 = RESIDENCY_EVICTION_WINDOW_MS / 12; + +/// Process-global ledger of heard residency beacons, keyed by the WIRE's peer id — the +/// residency sibling of [`super::gossip::GridCapacityLedger`], same identity + freshness +/// discipline, different (slower) cadence and payload. +#[derive(Default)] +pub struct ResidencyLedger { + heard: DashMap, +} + +/// The one process-global residency ledger — the resource it mirrors (this node's view of who +/// holds what across the grid) is process-global, same granularity argument as +/// [`super::gossip::global_ledger`]. +pub fn global_residency_ledger() -> &'static ResidencyLedger { + static LEDGER: OnceLock = OnceLock::new(); + LEDGER.get_or_init(ResidencyLedger::default) +} + +impl ResidencyLedger { + /// Fold one heard beacon in (latest per peer wins — residency is a live fact). `from_peer` + /// is the transcript event's transport identity, never payload-declared. Returns `true` when + /// this peer is NEW to the ledger — the probe-on-join surface; steady re-beacons stay silent. + pub fn hear(&self, from_peer: Uuid, beacon: ResidencyBeacon, heard_at_ms: u64) -> bool { + self.heard + .insert( + from_peer, + HeardBeacon { models: beacon.resident_models, heard_at_ms }, + ) + .is_none() + } + + /// Project the ledger into a [`ModelResidencyView`] the governor composes with the capacity + /// snapshot, evicting beacons silent past [`RESIDENCY_EVICTION_WINDOW_MS`] as it goes. + /// `own_peer` is excluded — the local node's residency is its OWN serving truth (it holds M + /// by definition when it overflows), not a round-tripped beacon. + pub fn view(&self, own_peer: Uuid, now_ms: u64) -> ModelResidencyView { + let mut view = ModelResidencyView::new(); + self.heard.retain(|peer, heard| { + let age = now_ms.saturating_sub(heard.heard_at_ms); + if age > RESIDENCY_EVICTION_WINDOW_MS { + return false; // silent too long — residency falls back to unknown + } + if *peer != own_peer { + view.by_peer + .insert(*peer, heard.models.iter().cloned().collect()); + } + true + }); + view + } + + /// Number of peers currently on the ledger (self included if echoed) — probe surface, + /// mirrors [`super::gossip::GridCapacityLedger::heard_count`]. + pub fn heard_count(&self) -> usize { + self.heard.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::capacity::grid::{GridPlacementPolicy, GridSnapshot, LocalFirstFitPolicy, PeerCapacity}; + use crate::capacity::{DeviceCapacity, LeaseRequest}; + + const GB: u64 = 1024 * 1024 * 1024; + + fn peer_id(n: u128) -> PeerId { + PeerId::from_uuid(Uuid::from_u128(n)) + } + + fn dev(free_gb: u64) -> DeviceCapacity { + DeviceCapacity { + gpu_total_bytes: 80 * GB, + gpu_free_bytes_live: free_gb * GB, + system_ram_free_bytes: 64 * GB, + } + } + + fn peer(n: u128, free_gb: u64, reachable: bool) -> PeerCapacity { + PeerCapacity { + peer: peer_id(n), + capacity: dev(free_gb), + reachable, + } + } + + // what this catches: the residency ELIGIBILITY gate. Only peers that beaconed the model as + // resident survive the filter; a peer with plenty of free VRAM but NOT holding M is dropped + // (routing to it would force a cold full-weights load — the slow path the gate exists to + // avoid), and a peer that never beaconed at all (unknown residency) is dropped too. The local + // device is always kept — the overflowing node holds M by definition. + #[test] + fn only_peers_holding_the_model_are_eligible() { + let snap = GridSnapshot { + local: dev(2), + peers: vec![ + peer(1, 40, true), // holds qwen-coder + peer(2, 40, true), // holds something else, NOT qwen-coder + peer(3, 40, true), // never beaconed — unknown residency + ], + }; + let mut view = ModelResidencyView::new(); + view.set_resident(peer_id(1), ["qwen-coder", "embed-small"]); + view.set_resident(peer_id(2), ["llama-70b"]); + // peer 3 intentionally not recorded. + + let eligible = view.residency_eligible(&snap, "qwen-coder"); + assert_eq!(eligible.local, snap.local, "local device is always kept — it holds M"); + assert_eq!(eligible.peers.len(), 1, "only the peer holding qwen-coder survives"); + assert_eq!(eligible.peers[0].peer.as_uuid(), Uuid::from_u128(1)); + } + + // what this catches: latest-wins REPLACE, not merge. A peer that paged qwen-coder OUT (its + // fresh beacon lists only what it still holds) must stop being eligible — a merge would + // resurrect the evicted model and route a hop to a peer that no longer has it. + #[test] + fn fresh_beacon_replaces_so_evicted_models_stop_being_eligible() { + let mut view = ModelResidencyView::new(); + view.set_resident(peer_id(1), ["qwen-coder", "llama-70b"]); + assert!(view.holds(&peer_id(1), "qwen-coder")); + + // Peer paged qwen-coder out; next beacon lists only llama-70b. + view.set_resident(peer_id(1), ["llama-70b"]); + assert!(!view.holds(&peer_id(1), "qwen-coder"), "evicted model must not linger"); + assert!(view.holds(&peer_id(1), "llama-70b")); + } + + // what this catches: the COMPOSE contract end-to-end — residency filters WHO is eligible, + // then the unchanged capacity policy places lanes on the survivors and applies reachability + // itself. A resident+reachable peer gets lanes; a resident+UNREACHABLE peer survives the + // residency filter (reachability isn't residency's job) but is reclaimed by place(); a + // non-resident peer is absent from placement entirely. Two orthogonal gates, composed. + #[test] + fn residency_then_capacity_policy_compose() { + let snap = GridSnapshot { + local: dev(1), // no local room — force the spill onto peers + peers: vec![ + peer(1, 40, true), // resident + reachable → should get lanes + peer(2, 40, false), // resident + UNREACHABLE → reclaimed by place() + peer(3, 40, true), // reachable but NOT resident → filtered out before place() + ], + }; + let mut view = ModelResidencyView::new(); + view.set_resident(peer_id(1), ["qwen-coder"]); + view.set_resident(peer_id(2), ["qwen-coder"]); + // peer 3 holds nothing relevant. + + let eligible = view.residency_eligible(&snap, "qwen-coder"); + assert_eq!(eligible.peers.len(), 2, "peers 1 & 2 are resident; peer 3 filtered out"); + + // Small spike so many lanes fit per peer — we're testing WHO gets lanes, not how many. + let req = LeaseRequest { + consumer: "qwen-coder".into(), + want_concurrency: 4, + spike_bytes: GB, + }; + let placement = LocalFirstFitPolicy { safety_margin_bytes: GB }.place(&eligible, &req); + + // The reachable resident peer carries the remote lanes; the unreachable one gets none. + let peer1_lanes: u32 = placement + .remote + .iter() + .filter(|(p, _)| p.as_uuid() == Uuid::from_u128(1)) + .map(|(_, n)| *n) + .sum(); + let peer2_lanes: u32 = placement + .remote + .iter() + .filter(|(p, _)| p.as_uuid() == Uuid::from_u128(2)) + .map(|(_, n)| *n) + .sum(); + let peer3_present = placement.remote.iter().any(|(p, _)| p.as_uuid() == Uuid::from_u128(3)); + + assert!(peer1_lanes > 0, "resident + reachable peer must carry overflow lanes"); + assert_eq!(peer2_lanes, 0, "resident but unreachable peer is reclaimed by place()"); + assert!(!peer3_present, "non-resident peer never reaches placement"); + } + + fn beacon(models: &[&str], at_ms: u64) -> ResidencyBeacon { + ResidencyBeacon { + resident_models: models.iter().map(|s| s.to_string()).collect(), + at_ms, + } + } + + // what this catches: the beacon RECEIVE→PROJECT pipeline — a heard beacon projects into a + // ModelResidencyView that holds() reflects, and the node's OWN echoed beacon is excluded + // (the local node's residency is its own serving truth, arriving live, not a round-trip). + // This is the residency sibling of gossip's loopback contract. + #[test] + fn heard_beacon_projects_and_own_echo_is_excluded() { + let ledger = ResidencyLedger::default(); + let me = Uuid::from_u128(1); + let other = Uuid::from_u128(2); + ledger.hear(me, beacon(&["qwen-coder"], 1_000), 1_000); // our own beacon, round-tripped + ledger.hear(other, beacon(&["qwen-coder", "llama-70b"], 1_000), 1_000); + + let view = ledger.view(me, 2_000); + assert!(!view.holds(&peer_id(1), "qwen-coder"), "own echo excluded from the peer view"); + assert!(view.holds(&peer_id(2), "qwen-coder"), "the real peer's residency projects"); + assert!(view.holds(&peer_id(2), "llama-70b")); + assert_eq!(view.known_peers(), 1, "only the genuine peer, not the echo"); + } + + // what this catches: eviction after long silence — a peer we haven't reheard past the + // (generous) residency window falls back to UNKNOWN residency, so the fast overflow path + // won't route to a model that long-silent peer may since have paged out. A fresh beacon + // brings it right back (grow is first-class). + #[test] + fn stale_beacon_evicts_then_a_fresh_one_restores() { + let ledger = ResidencyLedger::default(); + let me = Uuid::from_u128(1); + let peer = Uuid::from_u128(2); + ledger.hear(peer, beacon(&["qwen-coder"], 0), 0); + assert!(ledger.view(me, 1_000).holds(&peer_id(2), "qwen-coder")); + + // Silent past the residency eviction window: gone from the view (unknown, not asserted). + let t_evict = RESIDENCY_EVICTION_WINDOW_MS + 1; + assert!( + !ledger.view(me, t_evict).holds(&peer_id(2), "qwen-coder"), + "long-silent peer's residency falls back to unknown" + ); + + // It beacons again: instantly back. + ledger.hear(peer, beacon(&["qwen-coder"], t_evict), t_evict); + assert!( + ledger.view(me, t_evict + 1).holds(&peer_id(2), "qwen-coder"), + "a returning peer's residency is adopted on its first fresh beacon" + ); + } + + // what this catches: the wire payload survives serde round-trip byte-for-byte (camelCase + // like every realtime inline payload). If resident_models renamed or at_ms narrowed, heard + // residency would silently drift from published — the grid would gate overflow on models + // nobody actually beaconed. + #[test] + fn beacon_round_trips_through_json() { + let b = beacon(&["qwen-coder", "embed-small"], 42); + let json = serde_json::to_string(&b).expect("serialize"); + assert!(json.contains("residentModels"), "camelCase wire field: {json}"); + let back: ResidencyBeacon = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back, b, "beacon must survive the wire byte-for-byte"); + } +} diff --git a/core/continuum-core/src/cognition/serving_plan.rs b/core/continuum-core/src/cognition/serving_plan.rs index d52aafd6ce..fc983bb1ca 100644 --- a/core/continuum-core/src/cognition/serving_plan.rs +++ b/core/continuum-core/src/cognition/serving_plan.rs @@ -315,6 +315,27 @@ impl ModelFootprint { self.resident_bytes(served_window, lanes) .saturating_add(self.prefill_compute_reserve(served_window, lanes)) } + + /// The capacity-fabric [`LeaseRequest`](crate::capacity::LeaseRequest) for serving + /// this model at `served_window` with `demand_lanes` concurrent minds — the bridge + /// from the serving plan's MODEL-RESIDENCY view (weights + per-lane KV) to the grid's + /// CONCURRENCY-SPIKE view, so the grid can place overflow lanes onto peers (#180 spill). + /// `want_concurrency` = the minds that want a lane; `spike_bytes` = ONE lane's transient + /// prefill compute buffer at this window (the term the 2026-07-14 OOM turned on), so a + /// peer is only offered a spill lane it can actually hold. NOTE: this sizes the + /// CONCURRENCY spike only; a peer must ALSO already hold this model resident — that + /// residency gate is the gossip-side half (capacity::model_residency), NOT this pure map. + pub fn grid_lease_request( + &self, + served_window: u32, + demand_lanes: u32, + ) -> crate::capacity::LeaseRequest { + crate::capacity::LeaseRequest { + consumer: self.model_id.clone(), + want_concurrency: demand_lanes.max(1), + spike_bytes: self.prefill_compute_reserve(served_window, 1), + } + } } /// The serving decision for this host. diff --git a/core/continuum-core/src/ipc/mod.rs b/core/continuum-core/src/ipc/mod.rs index f46d8b8f80..6414c6510d 100644 --- a/core/continuum-core/src/ipc/mod.rs +++ b/core/continuum-core/src/ipc/mod.rs @@ -1995,6 +1995,14 @@ pub fn start_server( default_room, ), )); + // Grid residency beacon (grid-overflow eligibility, slice 3): advertise which models + // this node holds resident on a slower cadence, folded by inbound_attach into + // capacity::model_residency::global_residency_ledger. Reads the SAME serving plan the + // daemon computes (watch snapshot, no parallel probe); orthogonal sibling of capacity. + runtime.register(Arc::new(crate::modules::grid_residency::GridResidencyModule::new( + serving_daemon.subscribe(), + default_room, + ))); let continuum_root = crate::modules::persona_instance_manager::resolve_continuum_root(); let daemon_socket_for_rag_inspect = daemon_socket.clone(); let registry = crate::persona::PersonaAircRuntimeRegistry::new(); @@ -2331,6 +2339,15 @@ pub fn start_server( // below). Subscribe BEFORE the spawn so no plan edge is missed while // the task waits on the executor-ready oneshot. let mut serving_plan_rx = serving_daemon.subscribe(); + // A DEDICATED clone of the interceptor's airc-handle cell for the grid-overflow + // effector (slice 4b): the boot-spawn `async move` below captures by move, and the + // interceptor still needs the original `airc_interceptor_cell` further down — so the + // effector rides its own Arc clone of the SAME shared cell (both see the handle once + // attach_as fills it). + let overflow_airc_cell = airc_interceptor_cell.clone(); + // Same reason for the serving daemon: the reconcile task below still needs the + // original `serving_daemon` handle, so the effector rides its own clone. + let overflow_serving = serving_daemon.clone(); rt_handle.spawn(async move { // Wait for the IPC thread to deliver the WIRED executor (this both // gates ordering AND hands us the executor the personas' hands ride). @@ -2425,7 +2442,22 @@ pub fn start_server( } else { attempt += 1; let summary = supervisor - .spawn_all(&mut provider, Some(tool_executor.clone())) + .spawn_all( + &mut provider, + Some(tool_executor.clone()), + // Grid-overflow effector (slice 4b): the LIVE closure. Reads + // the serving plan's grid_overflow_lanes, filters residency- + // eligible reachable peers (from the beacon ledger), runs + // route_grid_overflow, and re-homes the persona's adapter to an + // AircRemoteInferenceAdapter over the interceptor's airc handle. + // DEFENSIVE: None on any uncertainty → local adapter (no + // regression); can only be a safe no-op or a correct off-box + // route (self excluded via airc.peer_id()). + crate::persona::grid_overflow_effector::build_overflow_effector( + overflow_airc_cell.clone(), + overflow_serving.clone(), + ), + ) .await; if summary.hosted > 0 { tracing::info!( diff --git a/core/continuum-core/src/modules/grid_residency.rs b/core/continuum-core/src/modules/grid_residency.rs new file mode 100644 index 0000000000..3f5fb81f0d --- /dev/null +++ b/core/continuum-core/src/modules/grid_residency.rs @@ -0,0 +1,166 @@ +//! GridResidencyModule — this node's residency-beacon gossip publisher (grid-overflow slice 3). +//! +//! The residency sibling of [`super::grid_capacity::GridCapacityModule`]. Every +//! [`RESIDENCY_PUBLISH_INTERVAL_MS`] tick it reads the SAME live serving plan the daemon +//! already computes (one source, no parallel probe) and broadcasts a [`ResidencyBeacon`] over +//! airc as an `EphemeralCoalesced` `grid_residency` realtime envelope: which models this node +//! holds resident — the grid-overflow ELIGIBILITY signal. The receive half lives in +//! [`crate::airc::inbound_attach`], which folds heard beacons (our own echo included — the +//! loopback proof) into [`crate::capacity::model_residency::global_residency_ledger`], whose +//! `view()` IS the `ModelResidencyView` the governor composes with the capacity snapshot. +//! +//! ## Why a SEPARATE module + slower cadence +//! +//! Residency and capacity are orthogonal (settled with BigMama 2026-07-27): capacity is +//! free-VRAM RIGHT NOW (10s beat, [`super::grid_capacity`]); residency is which models are warm +//! (minute-scale, changes only on page-in/out). Coupling them onto one envelope would either +//! over-publish residency or under-refresh capacity. Same module shape as the style guide +//! mandates — no new tokio task, the runtime tick drives it; the plan read is a lock-free +//! `watch` snapshot; the publish is one small envelope through the existing +//! `airc/realtime-publish` command surface (the same path capacity + chat ride). +//! +//! ## Today's local residency = the base model +//! +//! `ServingPlan` carries a single `base_model_id` + a `resident_models` COUNT (not a per-id +//! set), so today this node's honest resident set is `[base_model_id]`. When multi-model +//! residency lands (a per-id resident list on the plan), only [`Self::current_beacon`] changes +//! — the wire, ledger, and governor compose are already model-set-shaped. + +use std::any::Any; +use std::sync::Mutex; +use std::time::Duration; + +use async_trait::async_trait; +use serde_json::{json, Value}; +use tokio::sync::watch; + +use crate::capacity::model_residency::{ + global_residency_ledger, ResidencyBeacon, RESIDENCY_PUBLISH_INTERVAL_MS, +}; +use crate::cognition::serving_plan::ServingPlan; +use crate::runtime::{ + CommandExecutor, CommandResult, LateBound, ModuleConfig, ModulePriority, ServiceModule, +}; +use airc_core::RoomId; +use std::sync::Arc; + +pub struct GridResidencyModule { + /// Lock-free live snapshot of the daemon's serving plan — the SAME source the prefill + /// valve and serving control derive from (no parallel probe). + plan_rx: watch::Receiver>, + /// The node's discovered default room — where the grid rendezvous happens today. + room: RoomId, + executor_slot: Arc>, + /// Last published resident set — the glass box speaks on CHANGE, not on every beat (a + /// steady residency is silence). `None` = never published, so the first real plan speaks. + last_published: Mutex>>, +} + +impl GridResidencyModule { + pub fn new(plan_rx: watch::Receiver>, room: RoomId) -> Self { + Self { + plan_rx, + room, + executor_slot: Arc::new(LateBound::new("grid-residency::executor")), + last_published: Mutex::new(None), + } + } + + /// Build this node's residency beacon from the live serving plan. `None` when no plan is + /// computed yet (nothing resident to advertise) — honest silence, never a fabricated set. + /// Today the resident set is `[base_model_id]`; a multi-model plan extends only this line. + fn current_beacon(&self) -> Option { + let plan = self.plan_rx.borrow().clone()?; + Some(ResidencyBeacon { + resident_models: vec![plan.base_model_id], + at_ms: now_ms(), + }) + } +} + +fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +#[async_trait] +impl ServiceModule for GridResidencyModule { + fn config(&self) -> ModuleConfig { + ModuleConfig { + name: "grid-residency", + priority: ModulePriority::Background, + command_prefixes: &[], + event_subscriptions: &[], + needs_dedicated_thread: false, + max_concurrency: 0, + tick_interval: Some(Duration::from_millis(RESIDENCY_PUBLISH_INTERVAL_MS)), + } + } + + async fn initialize(&self, _ctx: &crate::runtime::ModuleContext) -> Result<(), String> { + Ok(()) + } + + async fn handle_command(&self, command: &str, _params: Value) -> Result { + Err(format!( + "grid-residency has no command surface — '{command}' (beacons publish on the module \ + tick; read the grid via capacity::model_residency::global_residency_ledger)" + )) + } + + async fn tick(&self) -> Result<(), String> { + // Boot ordering: a beat or two may fire before start_server installs the executor — + // transient, skip (the next beat publishes). Same guaranteed-installed path as capacity. + let Some(executor) = self.executor_slot.cloned() else { + return Ok(()); + }; + let Some(beacon) = self.current_beacon() else { + // No serving plan yet — nothing resident to advertise. Honest silence. + return Ok(()); + }; + + let envelope = json!({ + "eventId": uuid::Uuid::new_v4().to_string(), + "roomId": self.room.as_uuid().to_string(), + "sourceId": "grid-residency", + "createdAtMs": beacon.at_ms, + "delivery": "ephemeral_coalesced", + "payload": { + "kind": "existing_schema", + "payload": { + "schema": "grid_residency", + "inline": serde_json::to_value(&beacon) + .map_err(|e| format!("residency beacon encode failed: {e}"))?, + } + }, + }); + executor + .execute_json("airc/realtime-publish", json!({ "envelope": envelope })) + .await + .map_err(|e| format!("grid-residency beacon publish failed: {e}"))?; + + // Glass box: speak on change (resident set differs), silent on a steady residency. + let mut last = self.last_published.lock().map_err(|e| format!("residency lock: {e}"))?; + if last.as_deref() != Some(beacon.resident_models.as_slice()) { + crate::probe!( + class = "grid.residency.beacon", + models = ?beacon.resident_models, + heard_peers = global_residency_ledger().heard_count(), + "residency beacon published to the grid", + ); + *last = Some(beacon.resident_models); + } + Ok(()) + } + + fn install_executor(&self, executor: Arc) { + self.executor_slot.install(executor); + } + + fn as_any(&self) -> &dyn Any { + self + } +} diff --git a/core/continuum-core/src/modules/mod.rs b/core/continuum-core/src/modules/mod.rs index fb9f016da5..890b1ed0fc 100644 --- a/core/continuum-core/src/modules/mod.rs +++ b/core/continuum-core/src/modules/mod.rs @@ -47,6 +47,7 @@ pub mod gpu; pub mod grant_issuance; pub mod grid; pub mod grid_capacity; +pub mod grid_residency; pub mod health; pub mod hippocampus; pub mod inference_coordinator_module; diff --git a/core/continuum-core/src/modules/serving_daemon.rs b/core/continuum-core/src/modules/serving_daemon.rs index 5f5bfaa378..ed363c9e73 100644 --- a/core/continuum-core/src/modules/serving_daemon.rs +++ b/core/continuum-core/src/modules/serving_daemon.rs @@ -813,7 +813,7 @@ impl ServingDaemonModule { /// one candidate, so the reconcile serves that model or (if it has dropped off /// disk) nothing. Suppress subtracts; pin intersects; the planner still owns /// the choice among whatever remains. - fn live_candidates(&self) -> Vec { + pub(crate) fn live_candidates(&self) -> Vec { let suppressed = self.suppressed.borrow(); let pinned = self.pinned.borrow(); servable_candidates(&self.catalog.snapshot(), &**suppressed, &pinned) diff --git a/core/continuum-core/src/persona/grid_overflow_effector.rs b/core/continuum-core/src/persona/grid_overflow_effector.rs new file mode 100644 index 0000000000..4acc4a7286 --- /dev/null +++ b/core/continuum-core/src/persona/grid_overflow_effector.rs @@ -0,0 +1,115 @@ +//! The grid-overflow EFFECTOR (governor consumer slice 4b-ii) — composes the tested +//! decision path into the live per-persona adapter override that +//! [`super::supervisor::materialize_adapters`] consumes. +//! +//! When the node is over local capacity (`ServingPlan.grid_overflow_lanes > 0`) and a +//! reachable peer already holds a persona's model, this routes HER BRAIN to that peer: her +//! `DeliberationModelBinding.adapter` becomes an [`AircRemoteInferenceAdapter`] over airc, so +//! her inference crosses the grid transparently — the exact re-home the binding was designed +//! for. The persona doesn't know or care that her model runs on another machine. +//! +//! ## Defensive by construction — safe to ship before the live smoke +//! +//! The closure returns `None` (→ the local factory adapter, zero behavior change) on ANY +//! uncertainty: airc not yet attached, no serving plan, no overflow, no matching footprint, or +//! no residency-eligible reachable peer. So it can only ever be a **safe no-op or a correct +//! remote route** — never a self-route (this node's own peer is excluded via `airc.peer_id()`, +//! so its own residency-beacon loopback can't select itself) and never a panic. The one thing +//! the unit path can't prove is that the remote hop SUCCEEDS — that is what the live two-node +//! smoke validates; a hop that can't warm surfaces as a loud per-slot `AdapterWarmup` failure +//! ([[fallbacks-are-illegal-fail-loud]]), never a silent local downgrade. + +use std::sync::Arc; + +use airc_lib::Airc; +use tokio::sync::OnceCell; + +use crate::ai::adapter::AIProviderAdapter; +use crate::capacity::gossip::global_ledger; +use crate::capacity::grid_overflow::route_grid_overflow; +use crate::capacity::model_residency::global_residency_ledger; +use crate::capacity::DeviceCapacity; +use crate::inference::airc_remote::adapter::AircRemoteInferenceAdapter; +use crate::inference::airc_remote::transport::AircLiveTransport; +use crate::modules::serving_daemon::ServingDaemonModule; +use crate::persona::inference_profile::PersonaInferenceProfile; + +/// Headroom kept free on a peer before it may accept an overflow lane — same 1 GiB spirit as +/// the single-device [`crate::capacity::grid::LocalFirstFitPolicy`] safety margin. +const OVERFLOW_SAFETY_MARGIN_BYTES: u64 = 1024 * 1024 * 1024; + +fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Build the effector closure `spawn_all` / `materialize_adapters` consume. Captures the +/// late-bound airc handle cell (shared with the interceptor) and the serving daemon (the live +/// plan + footprint source). See the module docs for the defensive contract. +pub fn build_overflow_effector( + airc_cell: Arc>>, + serving: Arc, +) -> impl Fn(&PersonaInferenceProfile, usize) -> Option> { + move |profile, _slot| { + // airc not attached yet → local (the boot window before attach_as fills the cell). + let airc = airc_cell.get()?.clone(); + // No plan, or demand fits locally → local. grid_overflow_lanes is the honest + // "over local capacity by N" signal; 0 means nothing to spill. + let plan = serving.compute_plan()?; + if plan.grid_overflow_lanes == 0 { + return None; + } + // The footprint for THIS persona's model — the lease's per-lane prefill spike. + // Absent (model not a live candidate) → local, never a fabricated footprint. + let footprint = serving + .live_candidates() + .into_iter() + .find(|f| f.model_id == profile.model_id)?; + let lease = + footprint.grid_lease_request(plan.served_context_window, plan.grid_overflow_lanes); + + // Own peer id from the airc handle → excludes self from the residency view + gossip + // snapshot (this node's own beacon loopback must never select itself as the target). + let own = airc.peer_id().as_uuid(); + let now = now_ms(); + let residency = global_residency_ledger().view(own, now); + // Overflow placement is REMOTE-ONLY, so local capacity is never read — a zeroed local + // is the honest input (route_grid_overflow only ever inspects the peer list). + let snapshot = global_ledger().snapshot( + own, + DeviceCapacity { + gpu_total_bytes: 0, + gpu_free_bytes_live: 0, + system_ram_free_bytes: 0, + }, + now, + ); + + let routing = route_grid_overflow( + &profile.model_id, + &lease, + &residency, + &snapshot, + OVERFLOW_SAFETY_MARGIN_BYTES, + ); + // First-cut assignment: this persona takes the first placed peer. No residency-eligible + // reachable peer with room → local (queue/degrade is the planner's, not a silent drop). + let (peer, _lanes) = routing.remote.first()?; + let peer_uuid = peer.as_uuid(); + + let transport = AircLiveTransport::new(airc, peer_uuid); + let adapter = AircRemoteInferenceAdapter::new(transport); + crate::probe!( + class = "grid.overflow.route", + persona = %profile.persona_name, + model = %profile.model_id, + peer = %peer_uuid, + overflow_lanes = plan.grid_overflow_lanes, + "routing persona brain OFF-BOX to a residency-eligible peer (grid overflow)", + ); + Some(Arc::new(adapter) as Arc) + } +} diff --git a/core/continuum-core/src/persona/host.rs b/core/continuum-core/src/persona/host.rs index d42ad0df19..c9ec318822 100644 --- a/core/continuum-core/src/persona/host.rs +++ b/core/continuum-core/src/persona/host.rs @@ -264,6 +264,15 @@ impl PersonaSpawnSupervisor { // persona's HANDS are built over it (identity-scoped), so the ACL gates // what they may do. `None` → personas spawn speak-only (no hands). tool_command_executor: Option>, + // The grid-overflow effector (governor consumer slice 4b): given a persona's + // profile + slot, `Some(remote adapter)` routes her brain to a peer that holds + // her model (the ipc bootstrap builds this from the live serving plan + residency + // ledger + airc handle); `None` → local adapter. Forwarded verbatim to + // `materialize_adapters`. `|_, _| None` is the pre-effector (all-local) behavior. + overflow_adapter_for: impl Fn( + &crate::persona::inference_profile::PersonaInferenceProfile, + usize, + ) -> Option>, ) -> BootSummary { let plans = match bootstrap_planned( &self.spawner, @@ -328,6 +337,10 @@ impl PersonaSpawnSupervisor { as Arc }) }, + // Grid-overflow effector: the ipc bootstrap supplies the live decision + // (capacity + residency + airc). `|_, _| None` from a caller that doesn't + // route keeps the pre-effector all-local behavior. + overflow_adapter_for, ) .await; diff --git a/core/continuum-core/src/persona/mod.rs b/core/continuum-core/src/persona/mod.rs index bf7a5ab61a..01b2e344b1 100644 --- a/core/continuum-core/src/persona/mod.rs +++ b/core/continuum-core/src/persona/mod.rs @@ -52,6 +52,7 @@ pub mod evaluator; pub mod focus; pub mod genome_paging; pub mod home; +pub mod grid_overflow_effector; pub mod host; pub mod hw_tier_descriptor; pub mod identity_provider; diff --git a/core/continuum-core/src/persona/supervisor.rs b/core/continuum-core/src/persona/supervisor.rs index 66471a2d4d..d55a966b1c 100644 --- a/core/continuum-core/src/persona/supervisor.rs +++ b/core/continuum-core/src/persona/supervisor.rs @@ -499,6 +499,14 @@ pub async fn materialize_adapters( uuid::Uuid, ) -> Option>, + // The grid-overflow EFFECTOR (governor consumer slice 4b). Given a persona's + // profile + slot, returns `Some(remote adapter)` when the governor routed her + // off-box — her node is over local capacity and a reachable peer holds her model + // (`capacity::grid_overflow::route_grid_overflow`) — so her brain runs on that + // peer via `AircRemoteInferenceAdapter`. `None` → build the local adapter from the + // factory (the common case; demand fit locally). Closure DI keeps the supervisor + // decoupled from the capacity fabric + airc handle, same shape as the lookups above. + overflow_adapter_for: impl Fn(&PersonaInferenceProfile, usize) -> Option>, ) -> Vec> { let mut out = Vec::with_capacity(plans.len()); for (slot_index, plan) in plans.into_iter().enumerate() { @@ -525,16 +533,22 @@ pub async fn materialize_adapters( continue; } }; - let adapter = match factory.build_adapter(&profile).await { - Ok(a) => a, - Err(message) => { - out.push(Err(SupervisorError::AdapterFactory { - slot_index, - role: plan.role, - message, - })); - continue; - } + // Grid-overflow effector: if the governor routed this persona off-box, her + // adapter is the airc-remote one (her brain runs on the peer that holds her + // model). Else build the local adapter from the factory — the common case. + let adapter = match overflow_adapter_for(&profile, slot_index) { + Some(remote) => remote, + None => match factory.build_adapter(&profile).await { + Ok(a) => a, + Err(message) => { + out.push(Err(SupervisorError::AdapterFactory { + slot_index, + role: plan.role, + message, + })); + continue; + } + }, }; // Warm the adapter's KV-cache / kernels BEFORE the persona // enters her service loop. Per [[init-once-handle-then-lease-zero-copy-refs]] @@ -1106,7 +1120,7 @@ mod tests { let factory = ScriptedPersonaAdapterFactory::heuristic(); let hosted = - materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None).await; + materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None, |_, _| None).await; assert_eq!(hosted.len(), 2); assert_eq!(factory.build_count(), 2); @@ -1153,7 +1167,7 @@ mod tests { let factory = ScriptedPersonaAdapterFactory::heuristic(); let hosted = - materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None).await; + materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None, |_, _| None).await; assert_eq!(hosted.len(), 2); // Factory called exactly once — for the Ok row only. @@ -1188,7 +1202,7 @@ mod tests { let factory = ScriptedPersonaAdapterFactory::always_fails("simulated factory rejection"); let hosted = - materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None).await; + materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None, |_, _| None).await; assert_eq!(hosted.len(), 1); match &hosted[0] { @@ -1211,7 +1225,7 @@ mod tests { #[tokio::test] async fn empty_plans_yields_empty_hosted() { let factory = ScriptedPersonaAdapterFactory::heuristic(); - let hosted = materialize_adapters(vec![], &factory, |_| None, |_| None).await; + let hosted = materialize_adapters(vec![], &factory, |_| None, |_| None, |_, _| None).await; assert!(hosted.is_empty()); assert_eq!(factory.build_count(), 0); } @@ -1240,7 +1254,7 @@ mod tests { let factory = ScriptedPersonaAdapterFactory::heuristic(); // `|_| None` here is the substrate-bug shape we're locking in: // the registry exists but doesn't contain this persona_id. - let hosted = materialize_adapters(plans, &factory, |_| None, |_| None).await; + let hosted = materialize_adapters(plans, &factory, |_| None, |_| None, |_, _| None).await; assert_eq!(hosted.len(), 1); // Factory MUST NOT be called when the runtime lookup fails — @@ -1300,7 +1314,7 @@ mod tests { as Arc) } }; - let hosted = materialize_adapters(plans, &factory, lookup, |_| None).await; + let hosted = materialize_adapters(plans, &factory, lookup, |_| None, |_, _| None).await; assert_eq!(hosted.len(), 2); // Factory ran exactly once — for Paige, not Pax. @@ -1345,7 +1359,7 @@ mod tests { let (factory, counts) = ScriptedPersonaAdapterFactory::heuristic_with_counters(); let hosted = - materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None).await; + materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None, |_, _| None).await; // Both slots materialize cleanly. assert_eq!(hosted.len(), 2); @@ -1359,6 +1373,56 @@ mod tests { ); } + // what this catches: the grid-overflow EFFECTOR seam — when the governor routes a + // persona off-box, `overflow_adapter_for` supplies her adapter (the airc-remote one, + // her brain on a peer) and the LOCAL factory is NOT called for that slot. Here slot 0 + // is overflow-routed (override returns a stand-in remote), slot 1 is local. Both host; + // the factory builds ONLY slot 1 (build_count == 1); both adapters still warm. This is + // the composition point where route_grid_overflow's decision becomes a real remote brain. + #[tokio::test] + async fn overflow_effector_supplies_remote_adapter_and_bypasses_the_local_factory() { + use crate::ai::heuristic_adapter::HeuristicInferenceAdapter; + let plans = vec![ + MaterializedPersonaPlan { + role: RoleId::Helper, + instance: fake_instance("Paige"), + profile: Ok(fake_profile("Paige", "model-a")), // slot 0 → overflow-routed + }, + MaterializedPersonaPlan { + role: RoleId::Coder, + instance: fake_instance("Pax"), + profile: Ok(fake_profile("Pax", "model-b")), // slot 1 → local + }, + ]; + + let (factory, _counts) = ScriptedPersonaAdapterFactory::heuristic_with_counters(); + let hosted = materialize_adapters( + plans, + &factory, + StubAircCitizen::fresh_lookup(), + |_| None, + // Overflow effector: slot 0 is routed off-box → supply a stand-in "remote" + // adapter; every other slot stays local (None). + |_profile, slot| { + if slot == 0 { + Some(Arc::new(HeuristicInferenceAdapter::new()) as Arc) + } else { + None + } + }, + ) + .await; + + assert_eq!(hosted.len(), 2); + assert!(hosted.iter().all(|r| r.is_ok()), "both personas host (one remote, one local)"); + assert_eq!( + factory.build_count(), + 1, + "the local factory builds ONLY the non-overflow slot — the overflow slot's \ + adapter came from the effector, her brain runs on the peer" + ); + } + /// Warmup failure surfaces as `SupervisorError::AdapterWarmup` — /// the persona does NOT reach hosted state. Per [[no-fallbacks-ever]] /// an adapter that refuses to warm gets a typed slot failure; @@ -1375,7 +1439,7 @@ mod tests { "simulated warmup failure", ); let hosted = - materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None).await; + materialize_adapters(plans, &factory, StubAircCitizen::fresh_lookup(), |_| None, |_, _| None).await; assert_eq!(hosted.len(), 1); match &hosted[0] { @@ -1411,7 +1475,7 @@ mod tests { profile: Ok(fake_profile("Paige", "model-a")), }]; let hosted_ok = - materialize_adapters(ok_plan, &factory_ok, StubAircCitizen::fresh_lookup(), |_| None) + materialize_adapters(ok_plan, &factory_ok, StubAircCitizen::fresh_lookup(), |_| None, |_, _| None) .await; assert!(hosted_ok[0].is_ok(), "ok-warmup adapter materializes"); assert_eq!(ok_counts.warmups(), 1); @@ -1429,6 +1493,7 @@ mod tests { &factory_fail, StubAircCitizen::fresh_lookup(), |_| None, + |_, _| None, ) .await; assert!( diff --git a/protocol/typescript/airc/AircRealtimeSchema.ts b/protocol/typescript/airc/AircRealtimeSchema.ts index f5040ab98c..a8efd11857 100644 --- a/protocol/typescript/airc/AircRealtimeSchema.ts +++ b/protocol/typescript/airc/AircRealtimeSchema.ts @@ -3,4 +3,4 @@ /** * Existing Continuum schema carried by an AIRC realtime envelope. */ -export type AircRealtimeSchema = "jtag_message" | "event_bridge_payload" | "grid_frame" | "live_kit_bridge_command" | "live_kit_bridge_event" | "chat_transcript" | "grid_capacity"; +export type AircRealtimeSchema = "jtag_message" | "event_bridge_payload" | "grid_frame" | "live_kit_bridge_command" | "live_kit_bridge_event" | "chat_transcript" | "grid_capacity" | "grid_residency";