feat(workflows): classify adapter frames as durable vs ephemeral

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]
 

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)

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)

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