Doc role: Implementation plan for the signal routing and SignalBus infrastructure Status: Implemented Date: 2026-03-08 Parent: 2026-03-08_registry_development_plan.md Registries / infrastructure covered: SignalRoutingLayer → SignalBus Specs: system/signal_bus_spec.md (primary), SYSTEM_REGISTER.md (routing policy) Lanes: lane:runtime-followon (#91) — SR2/SR3 signal routing

Sector H completes the signal infrastructure that all other sectors depend on for decoupled cross-registry coordination. The SignalRoutingLayer exists as a functional skeleton (SR2/SR3 done gates met) but is narrow: three signal topics, no async observers, no dead-letter policy, no input or registry-event topics.

The SR3 → SR4 target is a SignalBus-class abstraction that replaces remaining direct inter-registry wiring with typed publish/subscribe. Sector H does not need to complete before other sectors begin — the existing skeleton is sufficient for basic cross-registry signals — but the SignalBus upgrade must complete before cross-registry wiring in Sectors D and E can be considered clean.

Current state:     SignalRoutingLayer (direct fanout, sync observers, lifecycle now includes `SemanticIndexUpdated`)
SR4 target:        SignalBus (typed, async observers, backpressure, dead-letter, full topic set)

SignalRoutingLayer in shell/desktop/runtime/registries/signal_routing.rs now provides the concrete runtime implementation behind an explicit SignalBus trait:

• 5 topics: Navigation, Lifecycle, Sync, RegistryEvent, InputEvent.
• Typed topic sub-enums plus SignalEnvelope source/causality metadata.
• Sync observers plus async tokio::broadcast fanout with all-topic subscription support.
• Dead-letter retention, zero-observer warnings, observer-failure logging, and lag diagnostics.
• RegistryRuntime now owns Arc<dyn SignalBus> instead of a concrete routing-layer field.

Implementation note (2026-03-10):

• The async API returns an AsyncSignalSubscription wrapper rather than a raw broadcast::Receiver. This is intentional: Tokio reports lag on receive, so the wrapper is where lag diagnostics and warning policy live.
• Agent consumption remains a Sector G follow-on. Sector H is no longer blocked on AgentRegistry; the subscription surface it needs now exists.

Unlocks: Agent signal subscription (Sector G G3.2); registry state change notifications.

Registry state changes (lens update, theme switch, workflow activation, identity rotation, mod load/unload) need to propagate to observers without direct inter-registry calls. This is the SignalBus's primary use case.

pub enum SignalKind {
    // Existing
    Navigation(NavigationSignal),
    Lifecycle(LifecycleSignal),
    Sync(SyncSignal),

    // New
    RegistryEvent(RegistryEventSignal),
    InputEvent(InputEventSignal),
}

pub enum RegistryEventSignal {
    ThemeChanged { new_theme_id: ThemeId },
    LensChanged { new_lens_id: LensId },
    WorkflowChanged { new_workflow_id: WorkflowId },
    PhysicsProfileChanged { new_profile_id: PhysicsProfileId },
    SemanticIndexUpdated,
    ModLoaded { mod_id: ModId },
    ModUnloaded { mod_id: ModId },
    AgentSpawned { agent_id: AgentId },
    IdentityRotated { identity_id: IdentityId },
}

pub enum InputEventSignal {
    ContextChanged { new_context: InputContext },
    BindingRemapped { action_id: ActionId },
}

Done gates:

• RegistryEvent and InputEvent topic variants added.
• RegistryEventSignal variants cover the currently implemented state-changing runtime surfaces.
• Current registry operations in implemented Sectors A–F that change observable state emit the appropriate signal.
• Unit tests cover navigation/lifecycle/registry/input signal publication and observation.

Extend the existing Lifecycle topic with signals needed by Sector F:

pub enum LifecycleSignal {
    // Existing
    NodeActivated { node_key: NodeKey },
    NodeDeactivated { node_key: NodeKey },
    WorkspaceRestored,

    // New
    SemanticIndexUpdated,   // from KnowledgeRegistry::reconcile_semantics()
    MimeResolved { node_key: NodeKey, mime: String },  // from Sector A probe
    WorkflowActivated { workflow_id: WorkflowId },
}

Done gates:

• SemanticIndexUpdated added (with GUI/runtime observer consumption for registry-backed lens refresh).
• MimeResolved variant added.
• KnowledgeRegistry::reconcile_semantics() emits SemanticIndexUpdated.
• Sector A MIME probe emits MimeResolved.

Implements SYSTEM_REGISTER misroute-visibility policy and signal bus spec backpressure requirements.

A signal published with no registered observers is currently silently counted as "unrouted". This masks misroutes during development.

pub fn publish(&self, envelope: SignalEnvelope) {
    let observer_count = self.observer_count_for(envelope.kind.topic());
    if observer_count == 0 {
        log::warn!(
            "signal_routing: signal {:?} has no observers (source: {:?})",
            envelope.kind,
            envelope.source
        );
        self.diagnostics.unrouted += 1;
        return;
    }
    // ...
}

Done gates:

• log::warn! on zero-observer publish.
• unrouted counter increments on zero-observer.
• Tests cover unrouted publish accounting and dead-letter capture.

Observer call failures (panics, lock poisoning) currently increment failed counter silently. Surfaces as log::error! with observer identity:

if let Err(e) = observer(envelope.clone()) {
    log::error!("signal_routing: observer {:?} failed on {:?}: {:?}", observer_id, kind, e);
    self.diagnostics.failed += 1;
}

Done gates:

• Observer errors logged with observer identity.
• DIAG_SIGNAL_ROUTING failure channel emits at Error severity on observer failure.

Register DIAG_SIGNAL_ROUTING_* channels with versioned payload schema (Sector F F1.2 dependency):

register.signal_routing.published   — Info
register.signal_routing.unrouted    — Warn
register.signal_routing.failed      — Error
register.signal_routing.queue_depth — Info (future: async queue)

Done gates:

• Core signal-routing diagnostic channels are registered (published, unrouted, failed, queue_depth) with an additional lagged warn channel for async backpressure.
• SignalRoutingDiagnostics fields emit through the diagnostics layer on publish/receive boundaries.
• SR3 done gate (diagnostics channels report signal routing health) confirmed complete.

Unlocks: Agent signal subscription (Sector G G3.2); ControlPanel worker signal reception.

The current observer model uses synchronous Box<dyn Fn> callbacks. Agents and workers need to receive signals without blocking the frame loop. The solution is a tokio::broadcast channel per topic, alongside the existing sync observer map.

pub struct SignalRoutingLayer {
    // Sync path (existing — for in-frame-loop observers)
    sync_observers: HashMap<SignalTopic, Vec<SyncObserver>>,

    // Async path (new — for workers and agents)
    broadcast_tx: HashMap<SignalTopic, broadcast::Sender<SignalEnvelope>>,
}

impl SignalRoutingLayer {
    /// Subscribe to async signals for a topic. Returns a Receiver clone.
    pub fn subscribe_async(&self, topic: SignalTopic)
        -> broadcast::Receiver<SignalEnvelope>;

    /// Subscribe to all topics (for agents that need full signal stream).
    pub fn subscribe_all(&self) -> broadcast::Receiver<SignalEnvelope>;
}

broadcast::Sender has a fixed capacity; lagging receivers are detected on receive and surfaced through AsyncSignalSubscription::recv() with a warn diagnostic emission. That is the implemented backpressure policy.

Done gates:

• broadcast_tx map added to SignalRoutingLayer with one channel per topic plus an all-topics channel.
• subscribe_async() and subscribe_all() implemented.
• publish() sends to both sync observers and broadcast channels.
• Lagging receiver detection is implemented on receive with a Warn diagnostic emission.
• Sector G is unblocked: AgentContext::signal_rx can be wired from subscribe_all() when AgentRegistry lands.
• Tests cover topic-scoped, all-topic, and lagged async subscriber behavior.

Unlocks: SR4 done gates; complete replacement of remaining direct inter-registry wiring.

The signal_bus_spec.md defines the SignalBus as the Register-owned publish/subscribe fabric. The SignalRoutingLayer is the SR2/SR3 transitional implementation; SignalBus is the stabilised SR4 API.

The key distinction: SignalRoutingLayer is a concrete struct with internal direct fanout. SignalBus is a typed API facade — callers interact with it through trait methods, allowing the internal implementation to evolve (e.g. move to an async message broker) without changing callsites.

pub trait SignalBus: Send + Sync {
    fn publish(&self, envelope: SignalEnvelope);
    fn subscribe_sync(&self, topic: SignalTopic, observer: SyncObserver) -> ObserverId;
    fn unsubscribe(&self, id: ObserverId);
    fn subscribe_async(&self, topic: SignalTopic) -> broadcast::Receiver<SignalEnvelope>;
    fn subscribe_all(&self) -> broadcast::Receiver<SignalEnvelope>;
    fn diagnostics(&self) -> SignalRoutingDiagnostics;
}

SignalRoutingLayer implements SignalBus. RegistryRuntime holds Arc<dyn SignalBus>.

Done gates:

• SignalBus trait defined.
• SignalRoutingLayer implements SignalBus.
• RegistryRuntime field changed from SignalRoutingLayer to Arc<dyn SignalBus>.
• Runtime callsites now route through SignalBus trait methods.

Scan for direct registry-to-registry calls that should route through the SignalBus. Each one that bypasses the bus is an SR4 violation. Replace with signal publication + observer subscription.

Known candidates after Sectors A–G:

• LensRegistry reactivity to KnowledgeRegistry updates (Sector A A4.3 / Sector F F2.4).
• PresentationDomainRegistry reactivity to ThemeRegistry changes (Sector D D4.2).
• WorkflowRegistry cross-profile application (Sector E E2.2).

Done gates:

• Audit complete: current cross-registry coordination routes through SignalBus/runtime publication seams.
• No direct Arc<OtherRegistry> field references exist in registry structs outside the explicit composition/coordinator roles.
• SR4 done gate: legacy dispatch callsites are removed or wrapped behind Register APIs.

• RegistryEvent and InputEvent topics exist with the implemented runtime variant sets.
• Implemented registry state changes emit the appropriate RegistryEventSignal.
• Zero-observer publish emits log::warn!; observer failures emit log::error!.
• DIAG_SIGNAL_ROUTING_* channels are registered and emit with the correct severity.
• Async subscriber path exists through AsyncSignalSubscription; Sector G can consume it without new signal-layer work.
• SignalBus trait is defined; SignalRoutingLayer implements it; RegistryRuntime uses Arc<dyn SignalBus>.
• No direct inter-registry wiring outside explicit composition/coordinator roles remains in the registry layer.
• SR2/SR3/SR4 done gates are confirmed complete.

• system/signal_bus_spec.md — canonical SignalBus component spec
• SYSTEM_REGISTER.md — SR1/SR2/SR3/SR4 roadmap and done gates
• 2026-03-08_sector_g_mod_agent_plan.md — AgentContext signal subscription
• 2026-03-08_sector_f_knowledge_index_plan.md — SemanticIndexUpdated signal
• 2026-03-08_registry_development_plan.md — master index