From cd47c7525432480662360c205cc46c058d2b0407 Mon Sep 17 00:00:00 2001 From: xiduzo Date: Sat, 18 Jul 2026 23:14:12 +0200 Subject: [PATCH 1/2] feat(runtime): surface node health as a diagnostics channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a per-node runtime status channel (`Effects.node_diagnostics`), mirroring the ADR-0008/0009 pattern: a new `EffectsSink::report_diagnostic` hook, compile- forced in both hosts. Nodes raise/clear via `RuntimeContext::report_diagnostic`. First use: I2C reads that never ACK. The board streams an empty I2C_REPLY plus "I2C: Too few bytes received" when a device doesn't respond; the codec used to drop the empty reply (`len < 8`), hiding the fault. Now it parses zero-data replies (`len >= 7`) so the target node sees a short read, raises an error diagnostic (attributed by address, keeps its last good value), and clears on recovery — raised only on a transition so a per-poll NACK doesn't spam. Co-Authored-By: Claude Opus 4.8 --- apps/web/src-tauri/src/runtime/host.rs | 6 +- crates/microflow-core/src/firmata/mod.rs | 17 +++-- crates/microflow-core/src/runtime/context.rs | 71 ++++++++++++++++++- .../src/runtime/input/i2c_device.rs | 34 ++++++++- crates/microflow-core/src/runtime/mod.rs | 69 +++++++++++++++++- 5 files changed, 183 insertions(+), 14 deletions(-) diff --git a/apps/web/src-tauri/src/runtime/host.rs b/apps/web/src-tauri/src/runtime/host.rs index cfc2800..c561b15 100644 --- a/apps/web/src-tauri/src/runtime/host.rs +++ b/apps/web/src-tauri/src/runtime/host.rs @@ -23,7 +23,7 @@ use microflow_core::runtime::cloud; use microflow_core::flow::FlowUpdate; use microflow_core::runtime::{ CloudRequest, CloudRequestKind, ComponentBase, ComponentEvent, ComponentValue, Effects, - EffectsSink, FlowRuntime, SubscriberWiring, Wakeup, WakeupId, + EffectsSink, FlowRuntime, NodeDiagnostic, SubscriberWiring, Wakeup, WakeupId, }; use std::collections::HashMap; use std::io::{ErrorKind, Read, Write}; @@ -433,6 +433,10 @@ impl EffectsSink for Actor { fn dispatch_event(&mut self, event: &ComponentEvent) { let _ = self.app.emit("component-event", event); } + + fn report_diagnostic(&mut self, diagnostic: &NodeDiagnostic) { + let _ = self.app.emit("node-diagnostic", diagnostic); + } } /// Performs cloud `Effects` for the desktop host (ADR-0009): the network I/O that diff --git a/crates/microflow-core/src/firmata/mod.rs b/crates/microflow-core/src/firmata/mod.rs index b9c901f..b6937c9 100644 --- a/crates/microflow-core/src/firmata/mod.rs +++ b/crates/microflow-core/src/firmata/mod.rs @@ -493,17 +493,24 @@ impl FirmataClient { } I2C_REPLY => { let len = buf.len(); - if len < 8 { + // Minimum is `F0 77 addrLSB addrMSB regLSB regMSB END_SYSEX` (7 + // bytes, zero data): a device that never ACKs a read makes the + // board stream exactly this empty reply alongside its "I2C: Too + // few bytes received" string. Parse it (`data = []`) instead of + // dropping it, so the target node sees a short read and can + // surface the fault. The old `len < 8` guard silently discarded + // every empty reply, hiding the NACK entirely. + if len < 7 { return Step::Skipped; } let mut reply = I2cReply { address: i32::from(buf[2]) | (i32::from(buf[3]) << 7), register: i32::from(buf[4]) | (i32::from(buf[5]) << 7), - data: vec![buf[6] | (buf[7] << 7)], + data: Vec::new(), }; - let mut i = 8; - while i < len - 1 { - if buf[i] == END_SYSEX || i + 2 > len { + let mut i = 6; + while i + 1 < len { + if buf[i] == END_SYSEX { break; } reply.data.push(buf[i] | (buf[i + 1] << 7)); diff --git a/crates/microflow-core/src/runtime/context.rs b/crates/microflow-core/src/runtime/context.rs index 73453d0..7bda962 100644 --- a/crates/microflow-core/src/runtime/context.rs +++ b/crates/microflow-core/src/runtime/context.rs @@ -72,9 +72,35 @@ pub enum CloudRequestKind { MidiSend { device_name: String, bytes: Vec }, } +/// Severity of a [`NodeDiagnostic`], mapped 1:1 onto the UI's existing per-node +/// `error` (red) / `warning` (amber) badge on `NodeContainer`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum DiagnosticLevel { + Warning, + Error, +} + +/// A runtime health signal a node raises about *itself* — surfaced on the node +/// in the UI (not routed across edges like a [`ComponentEvent`]). A hardware +/// node uses this to report a fault it can only see at runtime, e.g. an I2C +/// device whose reads never ACK ("too few bytes"). `message: None` clears any +/// prior diagnostic on that node (recovery). Nodes should raise these only on a +/// state *transition* so a per-poll failure doesn't spam the channel. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct NodeDiagnostic { + /// The node the diagnostic is about (and displayed on). + pub node: String, + pub level: DiagnosticLevel, + /// The message to show; `None` clears the node's diagnostic. + pub message: Option, +} + /// Everything the host must do after one runtime turn. Bytes go to the serial /// port, events to the UI stores, wakeups to host timers, cancellations clear -/// timers that are no longer wanted, cloud requests go to the network. +/// timers that are no longer wanted, cloud requests go to the network, node +/// diagnostics go to the node's badge in the UI. #[derive(Debug, Default, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct Effects { @@ -83,6 +109,7 @@ pub struct Effects { pub wakeups: Vec, pub cancellations: Vec, pub cloud_requests: Vec, + pub node_diagnostics: Vec, } /// The per-field hook surface a **Runtime Host** implements to apply one turn's @@ -111,12 +138,15 @@ pub trait EffectsSink { /// Deliver a component event to the UI (desktop Tauri `emit`, browser store /// ingest). These leave the runtime and do not feed back this turn. fn dispatch_event(&mut self, event: &ComponentEvent); + /// Surface a node's runtime health on its UI badge (desktop Tauri `emit`, + /// browser store write). `message: None` clears the node's diagnostic. + fn report_diagnostic(&mut self, diagnostic: &NodeDiagnostic); } impl Effects { /// Apply this turn's effects to `sink` in the **canonical order** (ADR-0008, /// extended by ADR-0009): `outbound_bytes → cancellations → wakeups → - /// cloud_requests → component_events`. + /// cloud_requests → component_events → node_diagnostics`. /// /// - Bytes first: lowest wire latency for the turn's hardware writes. /// - Cancel before arm: a cancel + re-arm of the same logical timer in one @@ -145,6 +175,9 @@ impl Effects { for event in &self.component_events { sink.dispatch_event(event); } + for diagnostic in &self.node_diagnostics { + sink.report_diagnostic(diagnostic); + } } } @@ -161,6 +194,9 @@ pub struct ScheduleRequests { /// this turn (ADR-0009). Resolved into [`Effects::cloud_requests`] when the /// turn drains. pub cloud_requests: Vec<(String, CloudRequestKind)>, + /// Runtime health signals nodes raised about themselves this turn. Resolved + /// into [`Effects::node_diagnostics`] when the turn drains. + pub diagnostics: Vec, } /// Capabilities handed to a component for the duration of one dispatch call: @@ -228,6 +264,27 @@ impl<'a> RuntimeContext<'a> { .cloud_requests .push((self.node_id.to_string(), kind)); } + + /// Raise a runtime diagnostic on this node (shown on its UI badge). Use for a + /// fault the node can only detect at runtime — e.g. an I2C read that never + /// ACKs. Raise on a state *transition* only, so a per-poll failure doesn't + /// spam. Clear with [`clear_diagnostic`](Self::clear_diagnostic) on recovery. + pub fn report_diagnostic(&mut self, level: DiagnosticLevel, message: impl Into) { + self.requests.diagnostics.push(NodeDiagnostic { + node: self.node_id.to_string(), + level, + message: Some(message.into()), + }); + } + + /// Clear any diagnostic previously shown on this node (recovery). + pub fn clear_diagnostic(&mut self) { + self.requests.diagnostics.push(NodeDiagnostic { + node: self.node_id.to_string(), + level: DiagnosticLevel::Error, + message: None, + }); + } } #[cfg(test)] @@ -246,6 +303,7 @@ mod apply_tests { Arm(WakeupId), Cloud(String), Event(String), + Diagnostic(String), } #[derive(Default)] @@ -269,6 +327,9 @@ mod apply_tests { fn dispatch_event(&mut self, event: &ComponentEvent) { self.calls.push(Call::Event(event.source_handle.to_string())); } + fn report_diagnostic(&mut self, diagnostic: &NodeDiagnostic) { + self.calls.push(Call::Diagnostic(diagnostic.node.clone())); + } } fn event(handle: &str) -> ComponentEvent { @@ -306,6 +367,11 @@ mod apply_tests { prompt: "hi".to_string(), }, }], + node_diagnostics: vec![NodeDiagnostic { + node: "i2c".to_string(), + level: DiagnosticLevel::Error, + message: Some("no ACK".to_string()), + }], }; let mut rec = Recorder::default(); @@ -319,6 +385,7 @@ mod apply_tests { Call::Arm(9), Call::Cloud("llm".to_string()), Call::Event("value".to_string()), + Call::Diagnostic("i2c".to_string()), ], "effects must apply in the canonical order with no double-fire" ); diff --git a/crates/microflow-core/src/runtime/input/i2c_device.rs b/crates/microflow-core/src/runtime/input/i2c_device.rs index 412b91d..0856b3a 100644 --- a/crates/microflow-core/src/runtime/input/i2c_device.rs +++ b/crates/microflow-core/src/runtime/input/i2c_device.rs @@ -11,8 +11,8 @@ //! § Hardware Callback). use crate::runtime::{ - BoardWiring, Component, ComponentBase, ComponentBuilder, ComponentValue, HardwareComponent, - I2cContinuousRead, ListenerWiring, RuntimeContext, RuntimeError, + BoardWiring, Component, ComponentBase, ComponentBuilder, ComponentValue, DiagnosticLevel, + HardwareComponent, I2cContinuousRead, ListenerWiring, RuntimeContext, RuntimeError, }; use crate::config::i2c_device::{fold_bytes, ByteDecode, I2cDeviceConfig, OutputFormat}; @@ -20,6 +20,9 @@ pub struct I2cDevice { base: ComponentBase, config: I2cDeviceConfig, initialized: bool, + /// Whether the node is currently showing a short-read fault, so a diagnostic + /// is raised/cleared only on a transition (not every 100ms poll). + faulted: bool, } impl I2cDevice { @@ -29,6 +32,7 @@ impl I2cDevice { base: ComponentBase::new(id, ComponentValue::Number(0.0)), config, initialized: false, + faulted: false, } } @@ -223,7 +227,31 @@ impl HardwareComponent for I2cDevice { Ok(()) } - fn on_i2c_reply(&mut self, bytes: &[u8], _ctx: &mut RuntimeContext) -> Result<(), RuntimeError> { + fn on_i2c_reply(&mut self, bytes: &[u8], ctx: &mut RuntimeContext) -> Result<(), RuntimeError> { + // A reply shorter than the requested read length means the device didn't + // return its data — an unACKed read (the board's "I2C: Too few bytes + // received"). Surface it on the node and keep the last good value rather + // than overwriting it with a truncated/empty decode. Raise/clear only on + // a transition so a per-poll fault doesn't spam the diagnostic channel. + let expected = usize::from(self.config.read_length); + if bytes.len() < expected { + if !self.faulted { + self.faulted = true; + ctx.report_diagnostic( + DiagnosticLevel::Error, + format!( + "No response from 0x{:02X}: got {} of {expected} bytes. Check wiring (SDA/SCL/power), the address, and pull-ups.", + self.config.address, + bytes.len(), + ), + ); + } + return Ok(()); + } + if self.faulted { + self.faulted = false; + ctx.clear_diagnostic(); + } // The runtime unmarshals the I2C-reply Hardware Callback to raw bytes // once at the dispatch site (`ComponentValue::as_byte_vec`), so decode // straight from the slice — no per-node re-unwrap. diff --git a/crates/microflow-core/src/runtime/mod.rs b/crates/microflow-core/src/runtime/mod.rs index 65d102a..57db157 100644 --- a/crates/microflow-core/src/runtime/mod.rs +++ b/crates/microflow-core/src/runtime/mod.rs @@ -48,8 +48,8 @@ pub use component::{ }; pub use reconcile::{plan_board, BoardPlan, DesiredBoard}; pub use context::{ - CloudRequest, CloudRequestKind, Effects, EffectsSink, RuntimeContext, ScheduleRequests, Wakeup, - WakeupId, + CloudRequest, CloudRequestKind, DiagnosticLevel, Effects, EffectsSink, NodeDiagnostic, + RuntimeContext, ScheduleRequests, Wakeup, WakeupId, }; pub use error::{HardwareError, RuntimeError}; pub use registry::ComponentRegistry; @@ -749,6 +749,7 @@ impl FlowRuntime { .drain(..) .map(|(source, kind)| CloudRequest { source: Arc::from(source.as_str()), kind }) .collect(); + let node_diagnostics = std::mem::take(&mut reqs.diagnostics); let (wakeups, cancellations) = self.resolve_schedule(reqs); // One wide event per turn that *did something* — drained an event, wrote // bytes, (re)armed a timer, issued a cloud call, or hit a dispatch error. @@ -762,6 +763,7 @@ impl FlowRuntime { || !wakeups.is_empty() || !cancellations.is_empty() || !cloud_requests.is_empty() + || !node_diagnostics.is_empty() || errors > 0; if produced { tracing::debug!( @@ -776,7 +778,14 @@ impl FlowRuntime { "flow tick", ); } - Effects { outbound_bytes: out, component_events: events, wakeups, cancellations, cloud_requests } + Effects { + outbound_bytes: out, + component_events: events, + wakeups, + cancellations, + cloud_requests, + node_diagnostics, + } } /// Gate stale events, branch internal/hardware callbacks, echo `set_value` @@ -1302,6 +1311,60 @@ mod tests { ); } + #[test] + fn short_i2c_reply_raises_then_clears_node_diagnostic() { + use crate::firmata::{END_SYSEX, I2C_REPLY, START_SYSEX}; + use serde_json::json; + let mut rt = FlowRuntime::new(); + rt.seed_digital_pins(20); + + // A TCS-shaped node: reads 8 bytes from reg 0xB4 at 0x29. + let dev = node( + "tcs", + "I2cDevice", + json!({ "device": "tcs34725", "address": 0x29, "register": 0xB4, "readLength": 8, "output": "raw" }), + ); + rt.update_flow(FlowUpdate { nodes: vec![dev], edges: vec![] }); + + let reply = |data: &[u8]| { + // reg 0xB4 = 0x34 | (0x01 << 7) + let mut f = vec![START_SYSEX, I2C_REPLY, 0x29, 0, 0x34, 0x01]; + for &b in data { + f.push(b & 0x7F); + f.push(b >> 7); + } + f.push(END_SYSEX); + f + }; + + // A full 8-byte reply: value updates, no diagnostic. + let ok = rt.feed_bytes(&reply(&[1, 2, 3, 4, 5, 6, 7, 8])); + assert!(ok.node_diagnostics.is_empty(), "healthy read must not diagnose"); + assert!(ok.component_events.iter().any(|e| &*e.source == "tcs")); + + // An EMPTY reply (the board's NACK: `F0 77 29 00 34 01 F7`) must raise an + // error diagnostic on the node and NOT emit a (garbage) value event. + let nack = rt.feed_bytes(&reply(&[])); + assert_eq!(nack.node_diagnostics.len(), 1, "NACK must diagnose once"); + let d = &nack.node_diagnostics[0]; + assert_eq!(&d.node, "tcs"); + assert!(matches!(d.level, DiagnosticLevel::Error)); + assert!(d.message.is_some(), "raise carries a message"); + assert!( + !nack.component_events.iter().any(|e| &*e.source == "tcs"), + "a short read must keep the last good value, not emit" + ); + + // A second NACK must NOT re-diagnose (transition-only, no spam). + let nack2 = rt.feed_bytes(&reply(&[])); + assert!(nack2.node_diagnostics.is_empty(), "repeat NACK must not spam"); + + // Recovery: a full reply clears the diagnostic (message None). + let recovered = rt.feed_bytes(&reply(&[8, 7, 6, 5, 4, 3, 2, 1])); + assert_eq!(recovered.node_diagnostics.len(), 1, "recovery clears once"); + assert!(recovered.node_diagnostics[0].message.is_none(), "clear has no message"); + } + /// The full hardware-faithful analog loop on an Uno-shaped board: seed the /// exact pin-table JSON the desktop detection / web session hand over /// (`analogChannel >= 0` marks A0..A5 = pins 14..19), apply a flow holding a From e666c7775f70bc5a0127b5fa9cf08650ac5f1a69 Mon Sep 17 00:00:00 2001 From: xiduzo Date: Sat, 18 Jul 2026 23:14:26 +0200 Subject: [PATCH 2/2] feat(web): show runtime node diagnostics on the node badge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consumes the new `node_diagnostics` effect on both hosts (browser reactor + desktop `node-diagnostic` Tauri event) into a `useNodeDiagnosticsStore`, and renders it through the existing `NodeContainer` error/warning badge — no new UI. The I2C device node now shows a red error when its reads don't ACK ("No response from 0x29: got 0 of 8 bytes. Check wiring…"), outranking the advisory shared- address warning. The exhaustive `keyof Effects` guard in effects-sink forced the new field to be handled; conformance test updated. Co-Authored-By: Claude Opus 4.8 --- .../flow/nodes/i2c-device/i2c-device.tsx | 10 ++++- apps/web/src/hooks/use-node-diagnostics.ts | 21 +++++++++ .../firmata/__tests__/effects-sink.test.ts | 15 ++++++- apps/web/src/lib/firmata/effects-sink.ts | 15 +++++-- apps/web/src/lib/firmata/flow-reactor.ts | 6 +++ apps/web/src/lib/runtime/wasm.ts | 9 ++++ apps/web/src/routes/flow/$flowId.tsx | 2 + apps/web/src/stores/node-diagnostics.ts | 44 +++++++++++++++++++ 8 files changed, 115 insertions(+), 7 deletions(-) create mode 100644 apps/web/src/hooks/use-node-diagnostics.ts create mode 100644 apps/web/src/stores/node-diagnostics.ts diff --git a/apps/web/src/components/flow/nodes/i2c-device/i2c-device.tsx b/apps/web/src/components/flow/nodes/i2c-device/i2c-device.tsx index 59c54f7..726cc66 100644 --- a/apps/web/src/components/flow/nodes/i2c-device/i2c-device.tsx +++ b/apps/web/src/components/flow/nodes/i2c-device/i2c-device.tsx @@ -12,6 +12,7 @@ import { type BaseNode, } from "../_base/_base"; import { useNodeValue } from "@/stores/node-data"; +import { useNodeDiagnostic } from "@/stores/node-diagnostics"; import { useFlowNodes, useFlowSession } from "@/session"; import { dataSchema, type Data, type Value } from "./i2c-device.schema"; import { @@ -48,10 +49,15 @@ function useSharedAddressWarning(id: string, address: number): string | undefine } export function I2cDevice(props: Props) { - const warning = useSharedAddressWarning(props.id, props.data.address); + const addressWarning = useSharedAddressWarning(props.id, props.data.address); + // A runtime fault (e.g. the device never ACKs a read) outranks the advisory + // shared-address warning: show the red error badge when the runtime raises one. + const diagnostic = useNodeDiagnostic(); + const error = diagnostic?.level === "error" ? diagnostic.message : undefined; + const warning = diagnostic?.level === "warning" ? diagnostic.message : addressWarning; return ( - + state.apply); + + useListen({ + type: "node-diagnostic", + handler: ({ payload }) => { + apply(payload); + }, + }); +} diff --git a/apps/web/src/lib/firmata/__tests__/effects-sink.test.ts b/apps/web/src/lib/firmata/__tests__/effects-sink.test.ts index b5b16ea..5167ed8 100644 --- a/apps/web/src/lib/firmata/__tests__/effects-sink.test.ts +++ b/apps/web/src/lib/firmata/__tests__/effects-sink.test.ts @@ -4,6 +4,7 @@ import { type CloudRequest, type ComponentEvent, type EffectsSink, + type NodeDiagnostic, type Wakeup, } from "../effects-sink"; import type { Effects } from "@/lib/runtime/wasm"; @@ -31,6 +32,9 @@ class Recorder implements EffectsSink { dispatchEvent(event: ComponentEvent): void { this.calls.push(`event:${event.sourceHandle}`); } + reportDiagnostic(diagnostic: NodeDiagnostic): void { + this.calls.push(`diag:${diagnostic.node}`); + } } function event(sourceHandle: string): ComponentEvent { @@ -54,12 +58,20 @@ describe("applyEffects (ADR-0008 canonical order)", () => { prompt: "hi", }, ], + nodeDiagnostics: [{ node: "i2c", level: "error", message: "no ACK" }], }; const rec = new Recorder(); applyEffects(fx, rec); - expect(rec.calls).toEqual(["write:3", "cancel:7", "arm:9", "cloud:llm", "event:value"]); + expect(rec.calls).toEqual([ + "write:3", + "cancel:7", + "arm:9", + "cloud:llm", + "event:value", + "diag:i2c", + ]); }); test("skips writeBytes when there are no outbound bytes", () => { @@ -69,6 +81,7 @@ describe("applyEffects (ADR-0008 canonical order)", () => { wakeups: [], cancellations: [], cloudRequests: [], + nodeDiagnostics: [], }; const rec = new Recorder(); diff --git a/apps/web/src/lib/firmata/effects-sink.ts b/apps/web/src/lib/firmata/effects-sink.ts index 0d1050e..87323cd 100644 --- a/apps/web/src/lib/firmata/effects-sink.ts +++ b/apps/web/src/lib/firmata/effects-sink.ts @@ -17,6 +17,8 @@ export type { CloudRequest }; export type Wakeup = Effects["wakeups"][number]; /** One emitted component event, as carried in the `Effects` serde shape. */ export type ComponentEvent = Effects["componentEvents"][number]; +/** One node health signal, as carried in the `Effects` serde shape. */ +export type NodeDiagnostic = Effects["nodeDiagnostics"][number]; /** * The platform primitives an effects application drives — the TypeScript shape @@ -32,6 +34,7 @@ export interface EffectsSink { armWakeup(wakeup: Wakeup): void; performCloud(request: CloudRequest): void; dispatchEvent(event: ComponentEvent): void; + reportDiagnostic(diagnostic: NodeDiagnostic): void; } /** @@ -58,14 +61,17 @@ const EFFECT_HANDLERS: { [K in keyof Effects]: (fx: Effects, sink: EffectsSink) componentEvents: (fx, sink) => { for (const event of fx.componentEvents) sink.dispatchEvent(event); }, + nodeDiagnostics: (fx, sink) => { + for (const diagnostic of fx.nodeDiagnostics) sink.reportDiagnostic(diagnostic); + }, }; /** * The **canonical order** (ADR-0008, extended by ADR-0009) the fields apply in: - * `outboundBytes → cancellations → wakeups → cloudRequests → componentEvents`. - * Bytes first (wire latency), cancel-before-arm (so a cancel + re-arm of one - * logical timer in a turn is safe), cloud launched before UI events leave, UI - * events last (they leave the runtime and do not feed back this turn). + * `outboundBytes → cancellations → wakeups → cloudRequests → componentEvents → + * nodeDiagnostics`. Bytes first (wire latency), cancel-before-arm (so a cancel + + * re-arm of one logical timer in a turn is safe), cloud launched before UI events + * leave, UI events last (they leave the runtime and do not feed back this turn). * * `satisfies` pins every entry to a real field; {@link AssertOrderIsExhaustive} * below pins the *reverse* — a new field absent from this tuple fails to compile, @@ -77,6 +83,7 @@ const APPLY_ORDER = [ "wakeups", "cloudRequests", "componentEvents", + "nodeDiagnostics", ] as const satisfies readonly (keyof Effects)[]; /** Errors unless {@link APPLY_ORDER} lists every key of `Effects` (the wrap in a diff --git a/apps/web/src/lib/firmata/flow-reactor.ts b/apps/web/src/lib/firmata/flow-reactor.ts index 1e76a80..75f85a1 100644 --- a/apps/web/src/lib/firmata/flow-reactor.ts +++ b/apps/web/src/lib/firmata/flow-reactor.ts @@ -33,8 +33,10 @@ import { type CloudRequest, type ComponentEvent, type EffectsSink, + type NodeDiagnostic, type Wakeup, } from "./effects-sink"; +import { useNodeDiagnosticsStore } from "@/stores/node-diagnostics"; import type { BoardConnection } from "./web-serial"; // Re-exported so the board controller keeps importing `CloudDeps` from here; the @@ -217,4 +219,8 @@ export class FlowReactor implements EffectsSink { dispatchEvent(event: ComponentEvent): void { applyComponentEvent(event, this.edges); } + + reportDiagnostic(diagnostic: NodeDiagnostic): void { + useNodeDiagnosticsStore.getState().apply(diagnostic); + } } diff --git a/apps/web/src/lib/runtime/wasm.ts b/apps/web/src/lib/runtime/wasm.ts index 1dfd594..bfcb046 100644 --- a/apps/web/src/lib/runtime/wasm.ts +++ b/apps/web/src/lib/runtime/wasm.ts @@ -97,6 +97,14 @@ export type MidiListener = { deviceName: string; }; +/** A node's runtime health signal, shown on its UI badge (matches the Rust + * `NodeDiagnostic` serde shape). `message: null` clears the node's diagnostic. */ +export type NodeDiagnostic = { + node: string; + level: "warning" | "error"; + message: string | null; +}; + /** The side effects of one runtime turn (matches the Rust `Effects` serde shape). */ export type Effects = { outboundBytes: number[]; @@ -104,4 +112,5 @@ export type Effects = { wakeups: Wakeup[]; cancellations: number[]; cloudRequests: CloudRequest[]; + nodeDiagnostics: NodeDiagnostic[]; }; diff --git a/apps/web/src/routes/flow/$flowId.tsx b/apps/web/src/routes/flow/$flowId.tsx index 98fc7b8..bb87a7a 100644 --- a/apps/web/src/routes/flow/$flowId.tsx +++ b/apps/web/src/routes/flow/$flowId.tsx @@ -12,6 +12,7 @@ import { } from "@/session"; import { usePins, type Pin } from "@/stores/board"; import { useComponentEvents } from "@/hooks/use-component-events"; +import { useNodeDiagnostics } from "@/hooks/use-node-diagnostics"; import { useHotkeyEvents } from "@/hooks/use-hotkey-events"; import { useDebouncer } from "@tanstack/react-pacer"; import { trpc } from "@/lib/trpc"; @@ -33,6 +34,7 @@ import type { Node } from "@xyflow/react"; function FlowEventListeners() { const session = useFlowSession(); useComponentEvents(); + useNodeDiagnostics(); useHotkeyEvents(); // Dispatch the live flow to the runtime — Tauri IPC on desktop, the in-browser // wasm runtime on web; the dispatcher picks the sender by platform. diff --git a/apps/web/src/stores/node-diagnostics.ts b/apps/web/src/stores/node-diagnostics.ts new file mode 100644 index 0000000..7c37463 --- /dev/null +++ b/apps/web/src/stores/node-diagnostics.ts @@ -0,0 +1,44 @@ +import { useNodeId } from "@/components/flow/nodes/_base/_base"; +import type { NodeDiagnostic } from "@/lib/firmata/effects-sink"; +import { create } from "zustand"; + +/** A node's live runtime health, fed by the runtime `node_diagnostics` effect + * (browser reactor + desktop `node-diagnostic` Tauri event). Distinct from + * {@link useNodeDataStore} — that carries edge *values*; this carries a node's + * self-reported fault, rendered on the existing `NodeContainer` error/warning + * badge. A `null` message clears the node (recovery). */ +type Diagnostic = { level: "warning" | "error"; message: string }; + +type NodeDiagnosticsState = { + diagnostics: Record; + apply: (diagnostic: NodeDiagnostic) => void; + clear: () => void; +}; + +export const useNodeDiagnosticsStore = create((set) => ({ + diagnostics: {}, + clear: () => { + set({ diagnostics: {} }); + }, + apply: ({ node, level, message }) => { + set((state) => { + const next = { ...state.diagnostics }; + if (message === null) { + delete next[node]; + } else { + next[node] = { level, message }; + } + return { diagnostics: next }; + }); + }, +})); + +/** The current diagnostic for the node in context, or `undefined` if healthy. */ +export function useNodeDiagnostic(): Diagnostic | undefined { + const id = useNodeId(); + return useNodeDiagnosticsStore((state) => state.diagnostics[id]); +} + +export function useClearNodeDiagnostics() { + return useNodeDiagnosticsStore((state) => state.clear); +}