From d9a1eeb3b6e789d7d8448d73ad7ecc15bd42d6d9 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 21 Aug 2026 14:33:41 +0200 Subject: [PATCH] Let a node's failure outlive the run that followed it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A node's error cleared the moment it ran again, so a failure that genuinely fired an alert could leave no trace on the canvas by the time anyone looked. The engine records it now — on the node's status, so it survives a reload and every client agrees — and reading the traceback is what clears it. The seam is the event bus, which is where every failing path already meets: a queued live run, an explicit run, a preview, and a single triggered node all publish `node_error`, while the controller's own observer would have seen only one of them. That was half the confusion. The other half: clicking a failed neuron on Home often landed on a flow where everything looked fine. Nodes merge into one neuron by instance key — every InfluxDB node pointing at the same bucket is one neuron — and the click went to whichever flow contributed a member first, not the one that failed. It now goes to the failing member and selects it, and the canvas marks a failing node rather than leaving it to the dot alone. The inject node emitted one payload to every port it declared, whatever their types, so an inject on a bool port carrying the text "true" raised at publish time. Each port gets its own field now, typed and parsed by that port's dtype, and remembers what it last sent. A port that is renamed carries its value with it; one that is removed takes its value with it. An inject written before this keeps emitting exactly what it did. The derived-cron chip also appeared on the delay node, where `interval` is a rate limit and a schedule derived from it means nothing. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Uq8mtNb97A7praJLyeEYgs --- backend/app/api/routes/flows.py | 36 ++++- backend/app/flow/controller.py | 64 ++++++++ backend/app/flow/nodes/inject.py | 29 +++- backend/tests/flow/test_node_types.py | 25 ++- frontend/src/client/schemas.gen.ts | 36 +++++ frontend/src/client/sdk.gen.ts | 32 +++- frontend/src/client/types.gen.ts | 13 ++ frontend/src/components/Flow/BrainNode.tsx | 30 ++-- frontend/src/components/Flow/BrainView.tsx | 22 ++- frontend/src/components/Flow/FlowBoundary.tsx | 101 +++++++++---- frontend/src/components/Flow/FlowEditor.tsx | 26 +++- frontend/src/components/Flow/FlowNode.tsx | 54 ++++++- frontend/src/components/Flow/NodePanel.tsx | 142 +++++++++++++++++- frontend/src/components/Flow/flow.css | 49 ++++++ frontend/src/components/Flow/liveStore.ts | 96 +++++++++++- frontend/src/components/Flow/useFlowSocket.ts | 14 ++ .../src/routes/_canvas/flows/$flowName.tsx | 11 +- frontend/src/routes/_layout/index.tsx | 8 +- frontend/tests/runtime.spec.ts | 36 ++++- 19 files changed, 738 insertions(+), 86 deletions(-) diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index 6ea4fa7..0109b9f 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -278,6 +278,7 @@ def read_flows(controller: FlowControllerDep) -> Any: enabled=controller.is_enabled(name), paused=controller.is_paused(name), quarantined=controller.is_quarantined(name), + version=definition.version, ) ) return FlowsPublic(data=summaries, count=len(summaries)) @@ -720,6 +721,25 @@ def cancel_node(name: str, node_id: str, controller: FlowControllerDep) -> Any: return Message(message=f"'{node_id}' was not running") +@router.post("/{name}/nodes/{node_id}/acknowledge", response_model=Message) +def acknowledge_node_error( + name: str, node_id: str, controller: FlowControllerDep +) -> Any: + """Dismiss what a node last failed with, so the canvas stops marking it. + + A failure outlives the next good run on purpose — otherwise one that fired + an alert leaves no trace by the time anyone looks. Reading the traceback is + what says it has been seen. + """ + try: + controller.acknowledge_error(f"{name}.{node_id}") + except KeyError: + raise HTTPException( + status_code=404, detail=f"No node named '{node_id}' in flow '{name}'" + ) from None + return Message(message=f"Cleared the failure on '{node_id}'") + + @router.get("/{name}/state", response_model=FlowStatePublic) def read_flow_state(name: str, controller: FlowControllerDep) -> Any: """The last value seen on every message of this flow.""" @@ -815,8 +835,12 @@ def event_for_panel(event: dict[str, Any], only: set[str]) -> bool: """Whether a panel's socket should carry this event. The same bound as the snapshot above, applied to the stream that follows - it: a value the panel draws, and nothing else on the bus. + it: a value the panel draws, and nothing else on the bus — save for a + dashboard being published, which is how a screen hears that the document + it is drawing, or the set of them it was given, has moved. """ + if event.get("type") == "dashboard_changed": + return True return event.get("type") == "message_value" and str(event.get("name") or "") in only @@ -855,6 +879,16 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None: sender.cancel() break event = sender.result() + if only is not None and event.get("type") == "dashboard_changed": + # The scope was resolved once, at the handshake. A panel + # pointed at another dashboard would otherwise fetch the + # new document and then draw tiles nothing ever updates. + # ``or set()`` because a panel that was deleted resolves to + # None, the same as a person's token — and that would widen + # this socket to everything on the bus. + only = panel_scope(token, websocket.app) or set() + if controller is not None: + await websocket.send_json(snapshot_payload(controller, only)) if only is not None and not event_for_panel(event, only): continue await websocket.send_json(event) diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 45ecd38..d427f3d 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -10,8 +10,10 @@ from __future__ import annotations import asyncio import logging +import time import traceback from collections.abc import Callable +from contextlib import suppress from dataclasses import dataclass, field from enum import Enum from typing import Any, cast @@ -91,6 +93,11 @@ class LoadedNode: error: str | None = None health: Health = "ok" health_detail: str | None = None + #: The last time this node raised while running, and what it said. Kept + #: after it has run again — a failure nobody saw is the one worth keeping — + #: so only an acknowledgement clears it, not a good run and not a rebuild. + last_error: str = "" + last_error_ts: float | None = None @dataclass @@ -307,12 +314,18 @@ class FlowController: self.supervisor = Supervisor(events) self.history_limits: dict[str, int] = {} self._lock = asyncio.Lock() + self._failures: asyncio.Task[None] | None = None # ------------------------------------------------------------------------- # Lifecycle # ------------------------------------------------------------------------- async def start(self) -> None: + # Subscribed before anything runs, so no failure falls between the + # first build and someone watching for one. + self._failures = asyncio.create_task( + self._watch_failures(), name="node-failures" + ) # Build first: the consumer must have a pipeline to execute against # before it claims anything, or work waiting from the last run would be # taken and dropped — which is the very case the queue exists for. @@ -321,10 +334,50 @@ class FlowController: self.execution.start() async def stop(self) -> None: + watcher, self._failures = self._failures, None + if watcher is not None: + watcher.cancel() + with suppress(asyncio.CancelledError): + await watcher await self._teardown() if self.execution is not None: await run_in_threadpool(self.execution.stop) + async def _watch_failures(self) -> None: + """Record every node failure the engine reports, from the event bus. + + A node can be run by the live pipeline, by a run's own, or by hand, and + each builds its own graph — but they all report through + ``Pipeline.publish_error``, so the bus is the one place they meet. + Which is the point: what is recorded here is the node's, not any + particular pipeline's, and it outlives both the good run that follows + it and the rebuild after that. + """ + if self.events is None: + return + async with self.events.subscribe() as queue: + while True: + event = await queue.get() + if event.get("type") != "node_error": + continue + entry = self.loaded.get(str(event.get("node") or "")) + if entry is None: + continue + entry.last_error = str(event.get("error") or "") + entry.last_error_ts = float(event.get("ts") or time.time()) + + def acknowledge_error(self, node_id: str) -> None: + """Forget what a node's last failure was. The only thing that clears it. + + Raises ``KeyError`` for a node the engine does not have, so a route can + answer that the way every other node call does. + """ + entry = self.loaded.get(node_id) + if entry is None: + raise KeyError(node_id) + entry.last_error = "" + entry.last_error_ts = None + async def set_enabled(self, flow: str, enabled: bool) -> None: """Stop or start one flow. Rebuilding is what applies it.""" await run_in_threadpool(self.store.write_enabled, flow, enabled) @@ -360,6 +413,15 @@ class FlowController: self._build_flows, [(flow, False) for flow in published] ) + # A rebuild is a fresh set of nodes, but not a fresh history: every + # publish rebuilds every flow, so dropping the failures here would + # wipe them constantly. They are the operator's to dismiss. + for node_id, entry in loaded.items(): + previous = self.loaded.get(node_id) + if previous is not None and previous.last_error: + entry.last_error = previous.last_error + entry.last_error_ts = previous.last_error_ts + self.loaded = loaded self.pipeline = Pipeline( nodes=nodes, @@ -600,6 +662,8 @@ class FlowController: error=entry.error, health=entry.health, health_detail=entry.health_detail, + last_error=entry.last_error, + last_error_ts=entry.last_error_ts, ) for entry in self.loaded.values() if flow is None or entry.flow == flow diff --git a/backend/app/flow/nodes/inject.py b/backend/app/flow/nodes/inject.py index 4f7552b..4e2d6c4 100644 --- a/backend/app/flow/nodes/inject.py +++ b/backend/app/flow/nodes/inject.py @@ -28,7 +28,8 @@ class InjectNode(Node): """Emit a value: on request, every n seconds, on a schedule, or at startup. The value is whatever ``payload`` says, or the current time when it says - nothing — a timestamp is what most schedules actually want. + nothing — a timestamp is what most schedules actually want. ``payloads`` + overrides that per output port, for a node that starts more than one thing. """ class Params(BaseModel): @@ -38,6 +39,13 @@ class InjectNode(Node): default=None, description="What to emit. Empty emits the current time.", ) + payloads: dict[str, Any] = Field( + default_factory=dict, + description=( + "What to emit on each output port, keyed by port name. A port not " + "named here falls back to `payload`." + ), + ) interval: float = Field( default=0, ge=0, @@ -76,13 +84,19 @@ class InjectNode(Node): name=name or "inject", ) - def _payload(self) -> Any: - return time.time() if self.cfg.payload is None else self.cfg.payload + def _values(self) -> dict[str, Any]: + """One value per output port: its own, or the node-wide payload.""" + # Read once, so an emission that falls back carries a single timestamp + # across every port rather than one per port. + fallback = time.time() if self.cfg.payload is None else self.cfg.payload + return { + spec.port: self.cfg.payloads.get(spec.port, fallback) + for spec in self.output_ports + } def _emit(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: - """Every output carries the same value; that is what injecting means.""" - value = self._payload() - return {spec.port: value for spec in self.output_ports} or None + """Each output carries what its port says to emit.""" + return self._values() or None # ------------------------------------------------------------------------- # Its own schedule @@ -149,8 +163,7 @@ class InjectNode(Node): pass async def _fire(self) -> None: - value = self._payload() - outputs = {spec.port: value for spec in self.output_ports} + outputs = self._values() if outputs: # inject runs the graph, which is blocking work. await asyncio.to_thread(self.inject, outputs) diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index c04c81d..789ffc5 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -43,7 +43,7 @@ FIXTURES: dict[str, dict] = { "provides": [MessageSpec(name="score", dtype=DType.FLOAT)], }, "inject": { - "params": {"payload": 1.0, "interval": 60}, + "params": {"payload": 1.0, "payloads": {"tick": 2.0}, "interval": 60}, "requires": [], "provides": [MessageSpec(name="tick", dtype=DType.FLOAT)], }, @@ -124,6 +124,29 @@ def test_mqtt_routes_topics_by_port(): assert node._topic_to_ports == {"house/temp": ["temp"], "house/hum": ["humidity"]} +def test_inject_emits_per_port_and_falls_back_to_one_payload(): + """A port named in `payloads` gets its own value; the rest share `payload`.""" + # Each port is published as its own declared type, which `check` enforces. + typed = NODE_TYPES["inject"].cls( + provides=[ + MessageSpec(name="flag", dtype=DType.BOOL), + MessageSpec(name="count", dtype=DType.INT), + ], + params={"payloads": {"flag": True, "count": 3}}, + ) + assert typed.execute({}) == {"flag": True, "count": 3} + + # No `payloads` at all is what every flow written so far carries. + shared = NODE_TYPES["inject"].cls( + provides=[ + MessageSpec(name="left", dtype=DType.FLOAT), + MessageSpec(name="right", dtype=DType.FLOAT), + ], + params={"payload": 1.0}, + ) + assert shared.execute({}) == {"left": 1.0, "right": 1.0} + + def test_every_offered_type_has_a_fixture(): # A new built-in without a fixture here would ship untested. Connectors are # separate packages and carry their own tests, so they are not this suite's. diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index f5e687b..ece1f8c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -423,6 +423,11 @@ export const DashboardDef_InputSchema = { title: 'Canvas Height', default: 1080 }, + icon: { + type: 'string', + title: 'Icon', + default: '' + }, pages: { items: { '$ref': '#/components/schemas/PageDef-Input' @@ -479,6 +484,11 @@ export const DashboardDef_OutputSchema = { title: 'Canvas Height', default: 1080 }, + icon: { + type: 'string', + title: 'Icon', + default: '' + }, pages: { items: { '$ref': '#/components/schemas/PageDef-Output' @@ -528,6 +538,11 @@ export const DashboardSummarySchema = { type: 'boolean', title: 'Has Draft', default: false + }, + version: { + type: 'integer', + title: 'Version', + default: 1 } }, type: 'object', @@ -1003,6 +1018,11 @@ export const FlowSummarySchema = { type: 'boolean', title: 'Quarantined', default: false + }, + version: { + type: 'integer', + title: 'Version', + default: 1 } }, type: 'object', @@ -1679,6 +1699,22 @@ export const NodeStatusPublicSchema = { } ], title: 'Health Detail' + }, + last_error: { + type: 'string', + title: 'Last Error', + default: '' + }, + last_error_ts: { + anyOf: [ + { + type: 'number' + }, + { + type: 'null' + } + ], + title: 'Last Error Ts' } }, type: 'object', diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 53d828f..fe1db27 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsReadRunsData, RunsReadRunsResponse, RunsReadRunData, RunsReadRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsReadRunsData, RunsReadRunsResponse, RunsReadRunData, RunsReadRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AlertsService { /** @@ -235,6 +235,9 @@ export class DashboardsService { /** * Create Dashboard * Start a dashboard: one page, one section, nothing on it yet. + * + * A draft, like every edit that follows it — a dashboard reaches a panel + * only once someone publishes it, so an empty one never does. * @param data The data for the request. * @param data.name * @returns DashboardDef_Output Successful Response @@ -899,6 +902,33 @@ export class FlowsService { }); } + /** + * Acknowledge Node Error + * Dismiss what a node last failed with, so the canvas stops marking it. + * + * A failure outlives the next good run on purpose — otherwise one that fired + * an alert leaves no trace by the time anyone looks. Reading the traceback is + * what says it has been seen. + * @param data The data for the request. + * @param data.name + * @param data.nodeId + * @returns Message Successful Response + * @throws ApiError + */ + public static acknowledgeNodeError(data: FlowsAcknowledgeNodeErrorData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/nodes/{node_id}/acknowledge', + path: { + name: data.name, + node_id: data.nodeId + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Read Flow State * The last value seen on every message of this flow. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index fc45a51..4f6a255 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -189,6 +189,7 @@ export type DashboardDef_Input = { columns?: number; canvas_width?: number; canvas_height?: number; + icon?: string; pages?: Array; version?: number; has_draft?: boolean; @@ -203,6 +204,7 @@ export type DashboardDef_Output = { columns?: number; canvas_width?: number; canvas_height?: number; + icon?: string; pages?: Array; version?: number; has_draft?: boolean; @@ -222,6 +224,7 @@ export type DashboardSummary = { page_count?: number; widget_count?: number; has_draft?: boolean; + version?: number; }; export type DeadLetter = { @@ -391,6 +394,7 @@ export type FlowSummary = { enabled?: boolean; paused?: boolean; quarantined?: boolean; + version?: number; }; export type HealthSummary = { @@ -633,6 +637,8 @@ export type NodeStatusPublic = { error?: (string | null); health?: 'ok' | 'degraded' | 'down'; health_detail?: (string | null); + last_error?: string; + last_error_ts?: (number | null); }; export type health = 'ok' | 'degraded' | 'down'; @@ -1294,6 +1300,13 @@ export type FlowsCancelNodeData = { export type FlowsCancelNodeResponse = (Message); +export type FlowsAcknowledgeNodeErrorData = { + name: string; + nodeId: string; +}; + +export type FlowsAcknowledgeNodeErrorResponse = (Message); + export type FlowsReadFlowStateData = { name: string; }; diff --git a/frontend/src/components/Flow/BrainNode.tsx b/frontend/src/components/Flow/BrainNode.tsx index 1441228..7645435 100644 --- a/frontend/src/components/Flow/BrainNode.tsx +++ b/frontend/src/components/Flow/BrainNode.tsx @@ -3,7 +3,7 @@ import { memo, useEffect, useRef, useState } from "react" import { duration } from "@/lib/motion" import { cn } from "@/lib/utils" -import { useGroupActive, useGroupEmits, useGroupError } from "./liveStore" +import { useGroupActive, useGroupEmits, useGroupFailure } from "./liveStore" export type BrainNodeData = { label: string @@ -47,7 +47,10 @@ function BrainNodeComponent({ data }: NodeProps) { // Whether it has published at all, which the snapshot answers for what // happened before this page connected; `emits` only counts what we saw. const active = useGroupActive(members) - const failed = useGroupError(members) + // The failure rather than the live status: it has to still be here after the + // node has run again, or Home and the canvas disagree about the same node. + const failure = useGroupFailure(members) + const failed = Boolean(failure) // Two faults, told apart the way the mark's two parts are: the ring is the // wiring around the node, so a flow that cannot run as written colours the // ring; the disc is the node itself, so a run that broke colours the disc. @@ -96,7 +99,7 @@ function BrainNodeComponent({ data }: NodeProps) { borderWidth: Math.round(size * RING), boxShadow: `inset 0 0 0 ${Math.round(size * GAP)}px var(--brain-gap)`, }} - title={`${label} · ${kind} · ${members.join(", ")}${issue ? ` · ${issue}` : ""}`} + title={`${label} · ${kind} · ${members.join(", ")}${issue ? ` · ${issue}` : ""}${failure ? ` · ${failure}` : ""}`} > {/* Both ends sit at the centre; the edge trims itself back to the rim. */} {/* The label box is only as wide as the circle it hangs under, so two words break onto two lines unless told not to. */} - {problem ? ( + {issue ? ( - {[failed && "failed", issue && "cannot run"] - .filter(Boolean) - .join(" · ")} + cannot run ) : null} diff --git a/frontend/src/components/Flow/BrainView.tsx b/frontend/src/components/Flow/BrainView.tsx index 2d236b1..82fed7e 100644 --- a/frontend/src/components/Flow/BrainView.tsx +++ b/frontend/src/components/Flow/BrainView.tsx @@ -28,6 +28,7 @@ import { scaleIn } from "@/lib/motion" import { BrainEdge, type BrainEdgeData } from "./BrainEdge" import { BrainNode, type BrainNodeData } from "./BrainNode" import "./flow.css" +import { liveStore } from "./liveStore" import { graphQueryOptions } from "./queries" import { useFlowSocket } from "./useFlowSocket" @@ -228,9 +229,24 @@ function BrainCanvas() { setRevealed(node.id) return } - const [flow] = (node.data as BrainNodeData).flows - if (flow) - navigate({ to: "/flows/$flowName", params: { flowName: flow } }) + // A neuron merges every node talking to the same thing, and its + // flows are in the order they contributed one — which is rarely the + // one that failed. So the failing member picks the flow, and takes + // its node with it; `flow.node_id`, and neither half can hold a dot. + const { members, flows } = node.data as BrainNodeData + const failing = members.find( + (member) => + liveStore.getFailure(member) || + liveStore.getStatus(member)?.status === "error", + ) + const [memberFlow, nodeId] = (failing ?? "").split(".") + const flowName = memberFlow || flows[0] + if (flowName) + navigate({ + to: "/flows/$flowName", + params: { flowName }, + search: nodeId ? { node: nodeId } : {}, + }) }} onPaneClick={() => setRevealed(null)} className="brain-flat h-full w-full" diff --git a/frontend/src/components/Flow/FlowBoundary.tsx b/frontend/src/components/Flow/FlowBoundary.tsx index 671e4c1..c3916d7 100644 --- a/frontend/src/components/Flow/FlowBoundary.tsx +++ b/frontend/src/components/Flow/FlowBoundary.tsx @@ -18,6 +18,7 @@ import { TooltipContent, TooltipTrigger, } from "@/components/ui/tooltip" +import { cn } from "@/lib/utils" import { qualify } from "./deriveEdges" import { useLiveValue } from "./liveStore" import { asText, DTYPES } from "./NodePanel" @@ -46,6 +47,66 @@ export function parseByDtype(dtype: DType | undefined, raw: string): unknown { } } +/** + * One literal, entered the way its type is entered. + * + * A flag has two values and gets a choice of them; everything else is typed + * and read back with {@link parseByDtype}. Deliberately not `type="number"` + * for the numbers — see that function on why half-typed input must survive. + * + * `className` carries no height: the two controls need different ones. + */ +export function DtypeValue({ + dtype, + value, + label, + id, + placeholder, + className, + onChange, +}: { + dtype: DType | undefined + value: unknown + /** What the field is called, for anyone not looking at it. */ + label: string + id?: string + placeholder?: string + className?: string + onChange: (next: unknown) => void +}) { + if (dtype === "bool") { + return ( + + ) + } + + return ( + onChange(parseByDtype(dtype, event.target.value))} + /> + ) +} + /** Putting a declared value into the running graph, credited to the input. */ function usePublishInput() { return useMutation({ @@ -116,38 +177,14 @@ function InputRow({ ))} - {spec.dtype === "bool" ? ( - - ) : ( - - onChange({ - ...declared, - initial: parseByDtype(spec.dtype, event.target.value), - }) - } - /> - )} + onChange({ ...declared, initial })} + /> - Show the traceback + + {failedEarlier + ? `Failed at ${failedAt} — show the traceback` + : "Show the traceback"} + ) : null} diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index 36566c9..77e53fa 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -38,6 +38,7 @@ import { Switch } from "@/components/ui/switch" import useCustomToast from "@/hooks/useCustomToast" import { inCodeEditor, useShortcuts } from "@/lib/shortcuts" import { cn } from "@/lib/utils" +import { DtypeValue } from "./FlowBoundary" import { MessageSparkline } from "./MessageSparkline" import { flowKeys, @@ -219,6 +220,7 @@ function PortList({ suggestions, onChange, onRenamed, + onRemoved, streamable = false, }: { title: string @@ -228,6 +230,8 @@ function PortList({ suggestions: string[] onChange: (next: MessageSpec[]) => void onRenamed?: (previous: string, next: string) => void + /** The port about to be dropped, reported just before the shorter list. */ + onRemoved?: (spec: MessageSpec) => void /** Outputs only: a port a node publishes on repeatedly while it runs. */ streamable?: boolean }) { @@ -355,7 +359,10 @@ function PortList({ size="icon-sm" className="text-muted-foreground" aria-label="Remove port" - onClick={() => onChange(specs.filter((_, i) => i !== index))} + onClick={() => { + onRemoved?.(spec) + onChange(specs.filter((_, i) => i !== index)) + }} > @@ -565,9 +572,15 @@ function cronFromInterval(seconds: unknown): string | null { /** What the five fields mean, and the schedule the interval beside them asks for. */ function CronHelp({ params, + derivable, onPick, }: { params: Record + /** + * Whether `interval` beside it is a schedule at all. On a delay it is a rate + * limit, and a cron built from it would say something the node never does. + */ + derivable: boolean onPick: (expression: string) => void }) { const derived = cronFromInterval(params.interval) @@ -583,7 +596,7 @@ function CronHelp({ */5 every fifth,{" "} 1-5 a range.

- {derived && params.cron !== derived ? ( + {derivable && derived && params.cron !== derived ? (