bb-ops 0.3.3

Canonical concrete components for the bytesandbrains framework — syscalls, wire transport, backends, protocols, role implementations.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

//! `PeerHealthGateTx` syscall op - gate inserted by the compiler
//! pass `bb-compiler/src/insert_peer_health_gate_tx.rs` upstream of
//! every `wire::Send`. Consults [`bb_runtime::framework::PeerGovernor`]
//! and emits a `TriggerValue` on Allow; the downstream `wire::Send`
//! fires only on Allow. On Deny the op returns an `OpError` whose
//! `detail` carries a stable label (`blocklisted` / `not_allowlisted`
//! / `cooldown`).
//!
//! The peer identity rides as `ATTR_PEER` on the gate NodeProto.
//! The canonical form is `attribute.s: bytes` (multihash); the
//! legacy i64 form is accepted during the wire-IR migration.

use bb_ir::proto::onnx::NodeProto;
use bb_ir::wire_shape;
use bb_runtime::atomic::DispatchResult;
use bb_runtime::bus::OpError;
use bb_runtime::framework::{BlockReason, Decision};
use bb_runtime::ids::PeerId;
use bb_runtime::runtime::RuntimeResourceRef;
use bb_runtime::slot_value::SlotValue;
use bb_runtime::syscall::values::TriggerValue;

/// Marker struct for `register_syscall::<PeerHealthGateTxOp>`.
pub struct PeerHealthGateTxOp;

/// `(domain, op_type)` registration key.
pub const DOMAIN: &str = "ai.bytesandbrains.syscall";
/// Op type name.
pub const OP_TYPE: &str = "PeerHealthGateTx";
/// Re-export the canonical ATTR_PEER constant from `bb_ir::wire_shape`.
pub const ATTR_PEER: &str = wire_shape::ATTR_PEER;

/// Invoke fn - consults `PeerGovernor::check_outbound` and emits a
/// `TriggerValue` on Allow, fails with a stable `reason` label on
/// any Deny variant.
pub fn invoke(
    node: &NodeProto,
    _inputs: &[(&str, &dyn SlotValue)],
    ctx: &mut RuntimeResourceRef<'_>,
) -> Result<DispatchResult, OpError> {
    let peer = read_peer_attr(node).ok_or_else(|| OpError {
        detail: format!("PeerHealthGateTx missing required `{ATTR_PEER}` attribute"),
        ..Default::default()
    })?;
    let now_ns = ctx.time.scheduler.now_ns();

    match ctx
        .peers
        .governor
        .check_outbound(peer, ctx.peers.backoff, now_ns)
    {
        Decision::Allow => Ok(DispatchResult::Immediate(vec![(
            "trigger".to_string(),
            Box::new(TriggerValue),
        )])),
        Decision::Deny(reason) => Err(OpError {
            detail: format!(
                "PeerHealthGateTx denied send to peer {peer:?}: reason={}",
                reason_label(&reason),
            ),
            ..Default::default()
        }),
    }
}

/// Reduce a `BlockReason` to a stable diagnostic string used in
/// `OpError.detail`. Consumers match on the label to react.
pub fn reason_label(reason: &BlockReason) -> &'static str {
    match reason {
        BlockReason::Blocklisted => "blocklisted",
        BlockReason::NotAllowlisted => "not_allowlisted",
        BlockReason::Cooldown { .. } => "cooldown",
    }
}

/// Read the gate's destination peer. Tries the canonical bytes
/// form first (`attribute.s` carrying multihash bytes), falls
/// back to the legacy `attribute.i` i64 form for nodes emitted
/// before the wire-IR migration.
fn read_peer_attr(node: &NodeProto) -> Option<PeerId> {
    wire_shape::read_peer_bytes(node).and_then(|bytes| PeerId::from_bytes(bytes).ok())
}


use bb_runtime::registry::OpRegistration as _BbOpsSyscallReg;

inventory::submit! {
    _BbOpsSyscallReg {
        domain: DOMAIN,
        op_type: OP_TYPE,
        invoke,
        kind: bb_runtime::registry::RegistrationKind::Syscall,
    }
}