Files
ruvnet--RuView/v2/crates/wifi-densepose-bfld/src/pipeline.rs
T
ruv 38676aa2bd feat(adr-118/p6.4): spawn_with_oracle for Soul Signature deployments (227/227 GREEN)
Iter 34. Closes the gap where BfldPipelineHandle had no path for an
operator-supplied SoulMatchOracle to reach the worker thread. The
emit_with_oracle surface added in iter 14 was unreachable through the
handle API — Soul Signature deployments (ADR-118 §1.4) had to either
drop down to BfldEmitter directly or accept Recalibrate gate-drops on
known-enrolled matches.

Added (in src/pipeline.rs):
- BfldPipeline::process_with_oracle<O: SoulMatchOracle>(
      inputs, embedding, oracle,
  ) -> Option<BfldEvent>
  Wraps emitter.emit_with_oracle then applies the same privacy_mode
  post-processing as process(). Privacy_mode and oracle are independent
  — class-3 demote still happens AFTER any oracle Recalibrate exemption.

Added (in src/pipeline_handle.rs):
- BfldPipelineHandle::spawn_with_oracle<P, O>(pipeline, publisher, oracle) -> Self
  where O: SoulMatchOracle + Send + Sync + 'static
  The worker thread owns the oracle and consults it on every recv().
  Worker loop now calls pipeline.process_with_oracle(...) instead of
  pipeline.process(...).

tests/handle_soul_oracle.rs (3 named tests, all green):
  spawn_with_oracle_null_is_equivalent_to_spawn
    Parity: 3 identical low-risk inputs through spawn() and
    spawn_with_oracle(NullOracle) produce the same publish count
    and the same motion-topic count.
  spawn_with_always_match_oracle_lets_events_publish_under_high_risk
    *** Headline test ***
    3 high-risk inputs spaced > DEBOUNCE_NS apart. With AlwaysMatch
    oracle, all 3 produce motion topics — the gate never reaches
    Recalibrate because the oracle reports an enrolled-person match.
  spawn_with_null_oracle_drops_events_under_sustained_recalibrate_score
    Negative control for the above: same 3 inputs through NullOracle,
    only 1 motion topic survives (the first input lands at Accept;
    the second and third hit Recalibrate after debounce and are
    dropped per ADR-121 §2.4).

ADR-124 status (iter step 0 sibling check):
- docs/adr/ADR-124-rvagent-mcp-ruvector-npm-integration.md unchanged
  at 431 lines. SENSE-BRIDGE scope remains orthogonal to BFLD core;
  no overlap with this iter.

ACs progressed:
- ADR-118 §1.4 Soul Signature companion contract end-to-end through
  the public handle API. Operators wiring Soul Signature into a
  RuView deployment now use:
      BfldPipelineHandle::spawn_with_oracle(pipeline, publisher, my_oracle)
  …and the rest of the per-frame flow stays identical to spawn().
- ADR-121 §2.6 Recalibrate exemption proven over the worker-thread
  boundary, not just at the unit level (iter 12 covered the gate-only
  case).

Test config:
- cargo test --no-default-features → 72 passed
- cargo test                       → 227 passed (224 + 3)

Out of scope (next iter target):
- GitHub Actions workflow with mosquitto Docker (lifts iters 24+29
  live-broker e2e from skip-mode). Remaining unmet ACs require
  either external resources (KIT BFId, Pi5/Nexmon) or CI infra.

Co-Authored-By: claude-flow <ruv@ruv.net>
2026-05-24 18:45:54 -04:00

201 lines
7.4 KiB
Rust

//! `BfldPipeline` — public entry point. ADR-118 §2.1.
//!
//! Thin facade over [`crate::BfldEmitter`] that adds:
//!
//! - A configuration struct ([`BfldConfig`]) for ergonomic construction.
//! - A `privacy_mode` toggle that flips the active class to
//! [`PrivacyClass::Restricted`] (and back to the configured baseline)
//! without rebuilding the underlying emitter state.
//! - A single named consumer call ([`Self::process`]) so callers don't have
//! to navigate the lower-level emitter API.
//!
//! Future iters add `process_to_frame()` (BfldFrame production) and a `tokio`
//! MQTT loop wrapper on top of this same facade.
#![cfg(feature = "std")]
use crate::coherence_gate::SoulMatchOracle;
use crate::emitter::{BfldEmitter, SensingInputs};
use crate::identity_risk::GateAction;
use crate::signature_hasher::SignatureHasher;
use crate::{BfldEvent, BfldFrame, BfldFrameHeader, BfldPayload, IdentityEmbedding, PrivacyClass};
/// Construction parameters for [`BfldPipeline`]. Matches the ADR-118 default-
/// secure posture: `class = Anonymous`, no zone, no signature hasher.
#[derive(Debug, Clone)]
pub struct BfldConfig {
/// Node identifier published in every `BfldEvent.node_id`.
pub node_id: String,
/// Optional default zone; passed through to every event.
pub default_zone_id: Option<String>,
/// Baseline privacy class. `privacy_mode = true` overrides to Restricted.
pub privacy_class: PrivacyClass,
/// Optional signature hasher; when present, the pipeline derives
/// `rf_signature_hash` via [`crate::IdentityFeatures`].
pub signature_hasher: Option<SignatureHasher>,
}
impl BfldConfig {
/// Build a minimal config: node_id only, class defaulted to Anonymous.
#[must_use]
pub fn new(node_id: impl Into<String>) -> Self {
Self {
node_id: node_id.into(),
default_zone_id: None,
privacy_class: PrivacyClass::Anonymous,
signature_hasher: None,
}
}
/// Set the default zone.
#[must_use]
pub fn with_zone(mut self, zone_id: impl Into<String>) -> Self {
self.default_zone_id = Some(zone_id.into());
self
}
/// Override the baseline privacy class.
#[must_use]
pub const fn with_privacy_class(mut self, class: PrivacyClass) -> Self {
self.privacy_class = class;
self
}
/// Install a signature hasher.
#[must_use]
pub fn with_signature_hasher(mut self, hasher: SignatureHasher) -> Self {
self.signature_hasher = Some(hasher);
self
}
}
/// Public BFLD entry point. Owns the configured emitter and the
/// `privacy_mode` toggle.
pub struct BfldPipeline {
/// Baseline class — the class to which `disable_privacy_mode()` returns.
baseline_class: PrivacyClass,
privacy_mode: bool,
emitter: BfldEmitter,
}
impl BfldPipeline {
/// Build a pipeline from `config`. The underlying emitter is initialized
/// with the configured class; `privacy_mode` is initially `false`.
#[must_use]
pub fn new(config: BfldConfig) -> Self {
let mut emitter = BfldEmitter::new(config.node_id);
if let Some(zone) = config.default_zone_id {
emitter = emitter.with_zone(zone);
}
emitter = emitter.with_privacy_class(config.privacy_class);
if let Some(hasher) = config.signature_hasher {
emitter = emitter.with_signature_hasher(hasher);
}
Self {
baseline_class: config.privacy_class,
privacy_mode: false,
emitter,
}
}
/// Process a single sensing frame. Delegates to the underlying emitter,
/// then post-processes the resulting event to honor `privacy_mode`. When
/// privacy mode is engaged the published event is demoted to Restricted
/// (identity-derived fields stripped) regardless of the configured baseline.
pub fn process(
&mut self,
inputs: SensingInputs,
embedding: Option<IdentityEmbedding>,
) -> Option<BfldEvent> {
let mut event = self.emitter.emit(inputs, embedding)?;
if self.privacy_mode {
event.privacy_class = PrivacyClass::Restricted;
event.apply_privacy_gating();
}
Some(event)
}
/// Variant of [`Self::process`] that consults a [`SoulMatchOracle`] before
/// the coherence gate fires `Recalibrate`. See ADR-121 §2.6 and ADR-118
/// §1.4. The privacy_mode post-processing still applies; the oracle only
/// affects whether the gate transitions to Recalibrate at all.
pub fn process_with_oracle<O: SoulMatchOracle>(
&mut self,
inputs: SensingInputs,
embedding: Option<IdentityEmbedding>,
oracle: &O,
) -> Option<BfldEvent> {
let mut event = self.emitter.emit_with_oracle(inputs, embedding, oracle)?;
if self.privacy_mode {
event.privacy_class = PrivacyClass::Restricted;
event.apply_privacy_gating();
}
Some(event)
}
/// Wire-bytes variant of [`Self::process`]: returns a [`BfldFrame`] ready
/// to serialize via `BfldFrame::to_bytes()`. Caller supplies a
/// `header_template` carrying AP / STA / session identity fields and a
/// `payload` typed via [`BfldPayload`]. The pipeline overrides the
/// template's `timestamp_ns` and `privacy_class` from its own state, then
/// builds the frame via [`BfldFrame::from_payload`] so the CRC covers the
/// section-prefixed bytes.
///
/// Returns `None` whenever the gate drops the underlying event (Reject or
/// Recalibrate), so `process_to_frame` is a strict subset of `process`.
pub fn process_to_frame(
&mut self,
inputs: SensingInputs,
header_template: BfldFrameHeader,
payload: BfldPayload,
embedding: Option<IdentityEmbedding>,
) -> Option<BfldFrame> {
let timestamp_ns = inputs.timestamp_ns;
let _gate_signal = self.process(inputs, embedding)?;
let mut header = header_template;
header.timestamp_ns = timestamp_ns;
header.privacy_class = self.current_privacy_class().as_u8();
Some(BfldFrame::from_payload(header, &payload))
}
/// `true` if `enable_privacy_mode()` has been called more recently than
/// `disable_privacy_mode()`.
#[must_use]
pub const fn is_privacy_mode_enabled(&self) -> bool {
self.privacy_mode
}
/// Read the currently active class. Returns Restricted if privacy mode is
/// engaged, otherwise the baseline.
#[must_use]
pub const fn current_privacy_class(&self) -> PrivacyClass {
if self.privacy_mode {
PrivacyClass::Restricted
} else {
self.baseline_class
}
}
/// Read-only access to the current gate action — for diagnostics.
#[must_use]
pub const fn current_gate_action(&self) -> GateAction {
self.emitter.current_action()
}
/// Engage privacy mode: future `process()` calls return events demoted
/// to Restricted (identity_risk_score + rf_signature_hash stripped)
/// regardless of the configured baseline.
///
/// The override is applied post-emission so the underlying gate / ring /
/// hasher state remains unchanged and recoverable when privacy mode is
/// later disabled.
pub fn enable_privacy_mode(&mut self) {
self.privacy_mode = true;
}
/// Disengage privacy mode: future events return to the configured baseline.
pub fn disable_privacy_mode(&mut self) {
self.privacy_mode = false;
}
}