P3 FLIP-LLM: the llm plane served through its door on the plane driver (dev-only llm-on-driver switch)

CHANGELOG.md

@@ -661,13 +661,14 @@ silently lost every counter. Reaching a recorder is no longer representable from
 host accepts only the current one. A sink that wants the observability back-channel implements one
 defaulted SDK method, `drain_observations`.
 
-**A new `plugins.logs` block is accepted and checked; plugin log lines have not moved.** The block
+**A new `plugins.logs` block is accepted and checked, and plugin log lines move to per-plugin log files.** The block
 names the directory for per-plugin log files (`dir`, default `logs/plugins`), the default level
 (`level`, default `info`), per-instance levels (`levels`), and rotation (`rotate_mb` and `keep`).
-A level word that names no level, or a `rotate_mb` of `0`, refuses the boot and names the key. This
-build writes no per-plugin log file: the plugins it loads do not run through the path that writes
-them, so a plugin's log lines reach the Busbar log exactly as they did in 1.5.5, and setting
-`plugins.logs` changes nothing an operator can observe beyond that check.
+A level word that names no level, or a `rotate_mb` of `0`, refuses the boot and names the key.
+
+- 1.6.0 Changed: a plugin's log lines, the LLM plane's included (it is a plugin in 1.6.0), are written to that plugin's own log file under `plugins.logs.dir` (for example `logs/plugins/llm.log`), not to the Busbar log.
+  In 1.5.5 the plane's warnings (for example `dropping image_url.detail: ...`) were lines of the
+  Busbar log. **Migration:** point log collection at `plugins.logs.dir` as well as the Busbar log.
 
 ### Spec fidelity
 

crates/busbar-contract/include/busbar_plugin.h

@@ -417,6 +417,7 @@ extern "C" {
 #define BB_PLANE_AUDIT_NONE UINT32_C(0) /* [`ServeOut::audit`]: no audit row. */
 #define BB_PLANE_AUDIT_APPLIED UINT32_C(1) /* [`ServeOut::audit`]: the request applied; the row's outcome is `applied`. */
 #define BB_PLANE_AUDIT_REJECTED UINT32_C(2) /* [`ServeOut::audit`]: the request was rejected after it was judged; the row's outcome is */
+#define BB_PLANE_AUDIT_DEGRADED UINT32_C(3) /* A [`RECORD_AUDIT`] write's outcome only (never [`ServeOut::audit`]): the unit was served with */
 #define BB_PLANE_CHAIN_LENGTH_PREFIXED UINT32_C(1) /* [`RecordChain::framing`]: each field of the record's digest is length-prefixed. */
 #define BB_PLANE_CHAIN_PIPE_SEPARATED UINT32_C(2) /* [`RecordChain::framing`]: the fields of the record's digest are joined by `|`. */
 #define BB_PLANE_CHAIN_DIGESTS_SCOPE UINT32_C(1) /* [`RecordChain::flags`]: the record's scope enters its digest. */

crates/busbar-contract/src/abi/plane/check.rs

@@ -151,7 +151,7 @@ fn records(buf: &[RecordWrite], n: u64, arena: u64, b: &Bounds) -> Result<(), Fa
                 code(
                     u64::from(r.kind),
                     u64::from(super::AUDIT_APPLIED),
-                    u64::from(super::AUDIT_REJECTED),
+                    u64::from(super::AUDIT_DEGRADED),
                     "record.audit_outcome",
                 )?;
                 if r.key.len == 0 {

crates/busbar-contract/src/abi/plane/mod.rs

@@ -591,6 +591,10 @@ pub const AUDIT_APPLIED: u32 = 1;
 /// [`ServeOut::audit`]: the request was rejected after it was judged; the row's outcome is
 /// `rejected`.
 pub const AUDIT_REJECTED: u32 = 2;
+/// A [`RECORD_AUDIT`] write's outcome only (never [`ServeOut::audit`]): the unit was served with
+/// something the caller asked for left out (a control the far end's form cannot carry); the row's
+/// outcome is `degraded`, the 1.5.5 row's.
+pub const AUDIT_DEGRADED: u32 = 3;
 
 /// [`RecordChain::framing`]: each field of the record's digest is length-prefixed.
 pub const CHAIN_LENGTH_PREFIXED: u32 = 1;

crates/busbar-contract/src/abi/tests/plane_check_tests.rs

@@ -700,15 +700,15 @@ fn an_audit_record_names_its_outcome_and_action() {
         record_kinds: 0,
         ..bounds()
     };
-    for outcome in [AUDIT_APPLIED, AUDIT_REJECTED] {
+    for outcome in [AUDIT_APPLIED, AUDIT_REJECTED, AUDIT_DEGRADED] {
         r.kind = outcome;
         assert_eq!(
             check_on_piece(Ready, &o, (&[], &[r], &[]), &caps(), &none),
             Ok(()),
             "outcome {outcome}"
         );
     }
-    for outcome in [AUDIT_NONE, AUDIT_REJECTED + 1, u32::MAX] {
+    for outcome in [AUDIT_NONE, AUDIT_DEGRADED + 1, u32::MAX] {
         r.kind = outcome;
         assert_eq!(
             piece(&o, &[], &[r], &[]),

crates/busbar-contract/src/plane_calls.rs

@@ -270,6 +270,9 @@ pub struct PlaneRegistration {
     pub trust_keys: Vec<crate::plane::TrustKeyDecl>,
     /// The tail's sentence refusing a forwarded caller credential; `None` = it states none.
     pub caller_credential_refusal: Option<&'static str>,
+    /// Whether the tail states `TAIL_FALLBACK`: the plane is the catch-all every unclaimed path
+    /// falls through to, and its card is the flat one (at most one registered plane).
+    pub fallback: bool,
     /// The door's own `validate` over a whole section (its settings, JSON): `Ok`, or the door's
     /// words. The kernel runs it where the section is parsed and where an admin write lands.
     pub validate: SectionJudge,

crates/busbar-kernel-egress/src/lib.rs

@@ -74,7 +74,7 @@ pub use pool::{
     Failover, Member, OnExhausted, Pool, DEFAULT_FAILOVER_CAP, DEFAULT_FAILOVER_DEADLINE_SECS,
 };
 pub use select::{RequestCtx, WeightedFloor};
-pub use walk::{Step, Taken, Walk, WalkPorts};
+pub use walk::{Shape, Step, Taken, Walk, WalkPorts};
 pub use wire::Shed;
 
 #[cfg(test)]

crates/busbar-kernel-egress/src/ports.rs

@@ -383,6 +383,14 @@ pub trait Clock: Send + Sync {
     /// seconds, and a budget close to a second boundary would collapse to zero.
     fn now_millis(&self) -> u128;
 
+    /// Microseconds since the same fixed point, for a latency sample: 1.5.5 measured an
+    /// attempt's latency in fractional milliseconds, so a sub-millisecond answer still records
+    /// (an integer `0` is the EWMA's "no sample"). A clock without the precision answers its
+    /// milliseconds scaled.
+    fn now_micros(&self) -> u128 {
+        self.now_millis().saturating_mul(1000)
+    }
+
     /// A future that completes no earlier than `ms` milliseconds from now.
     fn sleep(&self, ms: u64) -> BoxFut<'_, ()>;
 }
@@ -395,6 +403,10 @@ pub trait Telemetry: Send + Sync {
     /// One upstream attempt was started against this member.
     fn upstream_attempt(&self, pool: &str, destination: DestinationId);
 
+    /// A served attempt's time to the upstream's response head, in milliseconds (the destination's
+    /// latency signal, a bounded proxy that never waits out a streamed body).
+    fn upstream_latency(&self, _destination: DestinationId, _ms: f64) {}
+
     /// One attempt failed, with the disposition label the previous release used.
     fn upstream_failure(&self, pool: &str, destination: DestinationId, disposition: &'static str);
 

crates/busbar-kernel-egress/src/walk.rs

@@ -54,6 +54,21 @@ pub struct WalkPorts<'p> {
     pub pools: &'p HashMap<String, Pool>,
 }
 
+/// THE SHAPE A UNIT'S HOOKS PUT ON ITS PICKS (the design's route pump, "Hooks": the routing-stage policy
+/// keeps and orders the routed pool's members; a required restrict holds a pool the walk spills
+/// into): each pick's admissible members, the ones kept, and the order the walk prefers among them
+/// (empty = the weighted floor's own). `None` = a required restriction no member satisfies: the
+/// walk answers [`Shed::restrict_no_lane`] from there.
+pub trait Shape {
+    /// `members` of a pick, as the unit's hooks leave them: the walk's own pool (`primary`) or a
+    /// pool it spills into.
+    fn shape(
+        &self,
+        members: Vec<Member>,
+        primary: bool,
+    ) -> Option<(Vec<Member>, Vec<DestinationId>)>;
+}
+
 /// The walk's whole budget over `pool`, seconds: the pool's request timeout, `0` for a pool that is
 /// not configured.
 #[must_use]
@@ -184,6 +199,17 @@ impl Walk {
         &self.ctx
     }
 
+    /// The pool the walk is in and whether it is the walk's own (`true`) rather than a spill;
+    /// `None` once every path is spent.
+    #[must_use]
+    pub fn walking(&self) -> Option<(&str, bool)> {
+        match &self.phase {
+            Phase::Primary(name) => Some((name, true)),
+            Phase::Spill(name) | Phase::Terminal(name) => Some((name, false)),
+            Phase::Shed(_) => None,
+        }
+    }
+
     /// The walk ends here, before its pool's terminal: every step from now answers `shed`.
     pub fn refuse(&mut self, shed: Shed) {
         self.phase = Phase::Shed(shed);
@@ -203,6 +229,17 @@ impl Walk {
         ports: &WalkPorts<'_>,
         affinity: Option<u64>,
         token: &Pass<Route>,
+    ) -> Step {
+        self.next_shaped(ports, affinity, None, token)
+    }
+
+    /// [`Self::next`], each pick's members as the unit's hooks `shape` them.
+    pub fn next_shaped(
+        &mut self,
+        ports: &WalkPorts<'_>,
+        affinity: Option<u64>,
+        shape: Option<&dyn Shape>,
+        token: &Pass<Route>,
     ) -> Step {
         loop {
             if self.ctx.expired(ports.clock.now_secs()) {
@@ -218,9 +255,13 @@ impl Walk {
                         return Step::Shed(Shed::empty_pool());
                     }
                     if self.hops <= pool.failover.max_hops {
-                        if let Some(taken) = self.pick(ports, pool, affinity, token, true) {
-                            self.hops += 1;
-                            return Step::Take(taken);
+                        match self.pick(ports, pool, affinity, shape, token, true) {
+                            Ok(Some(taken)) => {
+                                self.hops += 1;
+                                return Step::Take(taken);
+                            }
+                            Ok(None) => {}
+                            Err(shed) => return Step::Shed(shed),
                         }
                     }
                     self.phase = Phase::Terminal(name);
@@ -229,8 +270,10 @@ impl Walk {
                     let Some(pool) = ports.pools.get(&name) else {
                         return Step::Shed(pool_shed(ports, &name, token));
                     };
-                    if let Some(taken) = self.pick(ports, pool, None, token, false) {
-                        return Step::Take(taken);
+                    match self.pick(ports, pool, None, shape, token, false) {
+                        Ok(Some(taken)) => return Step::Take(taken),
+                        Ok(None) => {}
+                        Err(shed) => return Step::Shed(shed),
                     }
                     self.phase = Phase::Terminal(name);
                 }
@@ -285,36 +328,64 @@ impl Walk {
 
     /// THE ONE PICK, over `pool`'s members this unit has not tried: the walk's own pool
     /// (`primary`, with the unit's affinity) and every pool it spills into (degraded).
+    ///
+    /// # Errors
+    ///
+    /// The unit's hooks hold the pool to a restriction no member satisfies: that shed, which every
+    /// step from here answers.
     fn pick(
         &mut self,
         ports: &WalkPorts<'_>,
         pool: &Pool,
         affinity: Option<u64>,
+        shape: Option<&dyn Shape>,
         token: &Pass<Route>,
         primary: bool,
-    ) -> Option<Taken> {
-        let members = pool.admissible_members();
+    ) -> Result<Option<Taken>, Shed> {
+        let admissible = pool.admissible_members().into_owned();
+        let (members, order) = match shape {
+            Some(shape) => match shape.shape(admissible, primary) {
+                Some(shaped) => shaped,
+                None => {
+                    let shed = Shed::restrict_no_lane();
+                    self.phase = Phase::Shed(shed.clone());
+                    return Err(shed);
+                }
+            },
+            None => (admissible, Vec::new()),
+        };
         let now = ports.clock.now_secs();
-        let mut picked = pick_among(
+        let Some(mut picked) = pick_among(
             &PickInput {
                 breaker: ports.breaker,
                 capacity: ports.capacity,
                 floor: ports.floor,
                 pool: &pool.name,
                 members: &members,
                 affinity,
-                preference: None,
+                preference: (!order.is_empty()).then_some(order.as_slice()),
                 now,
                 token,
             },
             &mut self.ctx,
-        )?;
-        let member = members
+        ) else {
+            return Ok(None);
+        };
+        let Some(member) = members
             .iter()
-            .find(|m| m.destination == picked.destination)?
-            .clone();
+            .find(|m| m.destination == picked.destination)
+            .cloned()
+        else {
+            return Ok(None);
+        };
         let probe = picked.take_probe_epoch();
-        Some(self.take(&pool.name, member, picked.permit, probe, !primary))
+        Ok(Some(self.take(
+            &pool.name,
+            member,
+            picked.permit,
+            probe,
+            !primary,
+        )))
     }
 
     /// Take `member` of `pool` for the next attempt: excluded from every pick of the unit after it.

crates/busbar-kernel/src/appbuild.rs

@@ -540,6 +540,19 @@ impl InstalledLimits {
     }
 }
 
+/// WHAT THE COMPOSITION ROOT KEEPS OF A RESOLVED CONFIGURATION beside the generation built from it
+/// (spec Part 1 line 569: a door plane is handed one validated `{section: value}` object per
+/// `open`/`refresh`): the root projects the sections, providers and pool bounds its door planes are
+/// sealed over, and reads the projection back when the generation is swapped in.
+pub type ConfigProjection = fn(&config::RootCfg) -> std::sync::Arc<dyn std::any::Any + Send + Sync>;
+
+static CONFIG_PROJECTION: std::sync::OnceLock<ConfigProjection> = std::sync::OnceLock::new();
+
+/// Bind the composition root's [`ConfigProjection`], once, before the first build.
+pub fn set_config_projection(project: ConfigProjection) {
+    let _ = CONFIG_PROJECTION.set(project);
+}
+
 #[cold] // boot/admin-only — keeps hot text dense (never inlined into a warm path)
 #[inline(never)]
 pub fn build_app_from_config(
@@ -551,6 +564,8 @@ pub fn build_app_from_config(
     config_paths: (Option<std::path::PathBuf>, Option<std::path::PathBuf>),
     prior: Option<&state::App>,
 ) -> Result<(state::App, Option<GovCredentialRotation>, InstalledLimits), String> {
+    // The root's projection of this configuration, taken before any part of it is consumed.
+    let config_projection = CONFIG_PROJECTION.get().map(|project| project(&cfg));
     // Install the resolved operational limits process-wide BEFORE any subsystem reads them —
     // running here (not in main) so a config APPLY/RELOAD refreshes them too. The values threaded
     // explicitly (client/store/router/TLS) read `cfg.limits` directly; the deep call-stack sites
@@ -1644,6 +1659,14 @@ pub fn build_app_from_config(
 
     // Populate the NEUTRAL carrier field-by-field from the already-resolved config (pre-resolved secret
     // plaintexts + rate-card-derived costs + resolved context/tokens are in `lane_inputs`/`pool_inputs`).
+    // THE KERNEL'S OWN TABLES over the same resolved sections, for the read seam when no plane
+    // contributes a runtime of its own.
+    let config_tables = Arc::new(busbar_kernel::plane_host::ConfigTables::of(
+        &lane_inputs,
+        &pool_inputs,
+        &by_model,
+        cfg.upstream_credentials,
+    ));
     let fallback_build_input = PlaneBuildInput {
         lanes: lane_inputs,
         pools: pool_inputs,
@@ -1757,6 +1780,7 @@ pub fn build_app_from_config(
     // read on the hook seam with a single relaxed load.
 
     let app = App {
+        config_projection,
         plane_sections: Some(cfg.plane_sections.clone()),
         // Telemetry-bank slot table for this generation (built above as a local, BEFORE the
         // config-derived collections moved into the fallback plane's runtime bundle so its
@@ -1768,6 +1792,7 @@ pub fn build_app_from_config(
         // seam; the snapshot names only the `&'static str` key, and `App::llm_runtime` downcasts the
         // slot on the money path. Absent slot (featureless build) reads the empty default.
         fallback_runtime_key,
+        config_tables,
         store,
         // The container planes' breaker cells: PROCESS-LIFETIME, reused across an apply/reload the
         // way the HTTP client pool and governance state are — a config swap must not un-trip a

crates/busbar-kernel/src/core_routes.rs

@@ -191,6 +191,15 @@ impl CoreRouter {
         self
     }
 
+    /// The methods a core route is already mounted under at exactly `path`.
+    pub(crate) fn methods_at(&self, path: &str) -> Vec<RouteMethod> {
+        self.table
+            .by_path
+            .get(path)
+            .map(|rows| rows.iter().map(|(m, _)| *m).collect())
+            .unwrap_or_default()
+    }
+
     /// Split into the mounted router and the table describing it.
     pub(crate) fn into_parts(self) -> (Router<Arc<AppHandle>>, CoreRouteTable) {
         (self.router, self.table)