From 5c24ef0cd1cf3037774de7ddf5ac6df18e752e33 Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Thu, 4 Jun 2026 00:02:02 -0300 Subject: [PATCH] feat(workflows): classify adapter frames as durable vs ephemeral --- .../base/langflow/api/v2/adapters/__init__.py | 10 ++++ .../base/langflow/api/v2/adapters/agui.py | 24 +++++++++ .../base/langflow/api/v2/adapters/langflow.py | 20 ++++++++ .../api/v2/adapters/test_event_durability.py | 51 +++++++++++++++++++ 4 files changed, 105 insertions(+) create mode 100644 src/backend/tests/unit/api/v2/adapters/test_event_durability.py diff --git a/src/backend/base/langflow/api/v2/adapters/__init__.py b/src/backend/base/langflow/api/v2/adapters/__init__.py index 3ffe9e18e2..1f2dc879b9 100644 --- a/src/backend/base/langflow/api/v2/adapters/__init__.py +++ b/src/backend/base/langflow/api/v2/adapters/__init__.py @@ -76,6 +76,16 @@ class StreamAdapter(Protocol): via event type; the buffer task falls back to other signals. """ + def is_durable(self, event_type: str) -> bool: + """True when a frame of ``event_type`` must be persisted to the durable log. + + Durable frames (milestones) are appended to ``job_events`` so a + reattaching client can rebuild state after the live bus is gone. + Ephemeral frames (token deltas) are published to the live bus only. + Unknown types default to ephemeral — a new milestone must opt in + explicitly rather than silently bloat the durable log. + """ + StreamAdapterFactory = Callable[[StreamAdapterContext], StreamAdapter] diff --git a/src/backend/base/langflow/api/v2/adapters/agui.py b/src/backend/base/langflow/api/v2/adapters/agui.py index 495e46cfdd..ffe63379f0 100644 --- a/src/backend/base/langflow/api/v2/adapters/agui.py +++ b/src/backend/base/langflow/api/v2/adapters/agui.py @@ -19,6 +19,27 @@ from langflow.api.v2.adapters import ( ) from langflow.api.v2.agui_translator import AGUITranslator +# Durable AG-UI milestones. ``TEXT_MESSAGE_CONTENT`` is the per-token delta and +# is ephemeral; the START/END lifecycle frames around it are durable so a +# reattaching client knows a message happened even without the token stream. +_AGUI_DURABLE_EVENTS: frozenset[str] = frozenset( + { + "RUN_STARTED", + "RUN_FINISHED", + "RUN_ERROR", + "STEP_STARTED", + "STEP_FINISHED", + "TEXT_MESSAGE_START", + "TEXT_MESSAGE_END", + "TOOL_CALL_START", + "TOOL_CALL_ARGS", + "TOOL_CALL_END", + "STATE_SNAPSHOT", + "STATE_DELTA", + "CUSTOM", + } +) + def _to_stream_event(event: BaseEvent) -> StreamEvent: """Frame one AG-UI event for SSE consumption.""" @@ -62,5 +83,8 @@ class AGUIAdapter: def terminal_error_type(self) -> str | None: return "RUN_ERROR" + def is_durable(self, event_type: str) -> bool: + return event_type in _AGUI_DURABLE_EVENTS + register_stream_adapter("agui", AGUIAdapter) diff --git a/src/backend/base/langflow/api/v2/adapters/langflow.py b/src/backend/base/langflow/api/v2/adapters/langflow.py index cdc3ec250d..4a0417eb04 100644 --- a/src/backend/base/langflow/api/v2/adapters/langflow.py +++ b/src/backend/base/langflow/api/v2/adapters/langflow.py @@ -20,6 +20,23 @@ from langflow.api.v2.adapters import ( ) from langflow.api.v2.converters import build_component_output, resolve_output_type +# Durable milestones for the langflow wire protocol. ``token`` is the only +# high-volume ephemeral type; everything else the build loop emits is a +# milestone worth persisting for reattach. +_LANGFLOW_DURABLE_EVENTS: frozenset[str] = frozenset( + { + "build_start", + "build_end", + "vertices_sorted", + "end_vertex", + "output", + "add_message", + "remove_message", + "error", + "end", + } +) + class LangflowAdapter: """Passthrough adapter: each EventManager event becomes one wire event. @@ -91,5 +108,8 @@ class LangflowAdapter: def terminal_error_type(self) -> str | None: return "error" + def is_durable(self, event_type: str) -> bool: + return event_type in _LANGFLOW_DURABLE_EVENTS + register_stream_adapter("langflow", LangflowAdapter) diff --git a/src/backend/tests/unit/api/v2/adapters/test_event_durability.py b/src/backend/tests/unit/api/v2/adapters/test_event_durability.py new file mode 100644 index 0000000000..39f82cbfa1 --- /dev/null +++ b/src/backend/tests/unit/api/v2/adapters/test_event_durability.py @@ -0,0 +1,51 @@ +"""Each adapter classifies frames as durable (persisted) vs ephemeral (live-only). + +Milestones (run/vertex start+end, outputs, tool_call, error, STATE, run_finished) +are durable so a reattaching client can rebuild state from the durable log. +Token deltas (TEXT_MESSAGE_CONTENT / langflow ``token``) are ephemeral: high +volume, only useful live. +""" + +from __future__ import annotations + +from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter + + +def _ctx() -> StreamAdapterContext: + return StreamAdapterContext(run_id="run-1", thread_id="thread-1") + + +class TestLangflowDurability: + def test_milestones_are_durable(self): + adapter = get_stream_adapter("langflow", _ctx()) + for event_type in ("build_start", "end_vertex", "output", "error", "end"): + assert adapter.is_durable(event_type) is True, event_type + + def test_token_is_ephemeral(self): + adapter = get_stream_adapter("langflow", _ctx()) + assert adapter.is_durable("token") is False + + def test_unknown_event_defaults_to_ephemeral(self): + adapter = get_stream_adapter("langflow", _ctx()) + assert adapter.is_durable("some_unknown_event") is False + + +class TestAguiDurability: + def test_milestones_are_durable(self): + adapter = get_stream_adapter("agui", _ctx()) + for event_type in ( + "RUN_STARTED", + "RUN_FINISHED", + "RUN_ERROR", + "TEXT_MESSAGE_START", + "TEXT_MESSAGE_END", + "TOOL_CALL_START", + "TOOL_CALL_END", + "STATE_SNAPSHOT", + "CUSTOM", + ): + assert adapter.is_durable(event_type) is True, event_type + + def test_token_delta_is_ephemeral(self): + adapter = get_stream_adapter("agui", _ctx()) + assert adapter.is_durable("TEXT_MESSAGE_CONTENT") is False