Tracking/integration branch for the A2A protocol epic (F1-F12). Feature
PRs merge into this branch; this branch lands on release once the
underlying v2/durable/HITL stack does. See the epic PR for the breakdown.
* fix(api/v2): address review findings on the v2 workflows endpoint
- recover session_id for completed background jobs from the persisted
terminal message instead of always returning null, so GET status can
continue the same chat/memory thread
- replay a user-cancel as a CUSTOM cancel marker + RUN_FINISHED (agui) and
a `cancelled` terminal (langflow) instead of RUN_ERROR, so a re-attaching
client no longer reads a deliberate stop as a failure
- cancel the evicted still-running buffer writer when the background-run
registry is full, so it stops appending into a run no reader can find
- derive per-component status from the error artifact / valid flag instead
of hardcoding COMPLETED, and stop the langflow adapter dropping `valid`
- throttle the unauthenticated public endpoint per IP and bound its
input_value/session_id length
- document the sync-only scope of request-body globals
- document that live event re-attach is intentionally owner-only
* test(lfx): register public_flow_rate_limit_per_minute in settings composition
* refactor(v2 workflows): split workflow.py and address review blockers
Splits the ~1.5k-line workflow.py into focused modules and folds in the
execution-timeout and error-sanitization fixes from Cristhianzl's review of #13307.
- B1: workflow.py now holds only the four route handlers. Validation guards move
to workflow_validation, the sync/stream run loop to workflow_execution, and the
durable background machinery to workflow_background (layered, acyclic).
- I1: add workflow_execution_timeout (default 300) and apply a single wall-clock
ceiling across sync, stream, background, and public via _stream_event_frames. A
timeout becomes a sanitized terminal error and marks a background job failed.
- I3: the route error handlers no longer echo raw exception text. They return a
generic, code-tagged message and log the full exception server-side.
- R1: remove the "commented out / future scope" comments that sat over live
dataframe-extraction code in converters.py.
- R4: drop the worker-routing internals from the reattach 409 message.
Tests cover the timeout terminal-error path and the error-body sanitization, and
the settings field-count guard is updated for the new setting.
* refactor(lfx): extract v2 workflow contract layer into lfx.workflow
Moves the protocol-agnostic pieces of the v2 workflows API out of the langflow
backend into lfx so both the backend and `lfx serve` can share one contract.
First step toward giving lfx (the production runtime) the v2 workflows API.
- Move api/v2/adapters/, agui_translator.py, and converters.py to lfx/workflow/.
They depend only on lfx.schema.workflow and ag_ui (already an lfx dep), so lfx
carries the contract with zero langflow imports.
- Decouple the one langflow reference: converters typed run_response against
langflow.api.v1.schemas.RunResponse (TYPE_CHECKING only). Replaced with a local
RunResponseLike Protocol (outputs + session_id), the only attributes used.
- Repoint the six backend v2 workflow modules to import from lfx.workflow.
- Move the five protocol-agnostic contract tests into src/lfx/tests/unit/workflow/
(run in the lfx-only env). test_output_event_parity and test_workflow_agui stay
in langflow (they need langflow.api.build) with repointed imports.
Coverage unchanged: 201 contract tests pass in the lfx-only env, 191 backend v2
tests pass; 392 total, same as before the move.
* [autofix.ci] apply automated fixes
---------
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Adopt the lfx.workflow shared contract layer: rewire the background
execution service and the v2 routes to import adapters/converters/
agui_translator from lfx.workflow, drop the duplicated backend copies and
the superseded inline workflow_background/execution/validation split.
Reconcile the adapter/converter divergence: keep the shared layer's
cancel_events (distinct cancelled terminal, not error) and valid-derived
component status, and port the background path's is_durable + durable-event
sets and workflow_response_from_output_events.
Moves the protocol-agnostic pieces of the v2 workflows API out of the langflow
backend into lfx so both the backend and `lfx serve` can share one contract.
First step toward giving lfx (the production runtime) the v2 workflows API.
- Move api/v2/adapters/, agui_translator.py, and converters.py to lfx/workflow/.
They depend only on lfx.schema.workflow and ag_ui (already an lfx dep), so lfx
carries the contract with zero langflow imports.
- Decouple the one langflow reference: converters typed run_response against
langflow.api.v1.schemas.RunResponse (TYPE_CHECKING only). Replaced with a local
RunResponseLike Protocol (outputs + session_id), the only attributes used.
- Repoint the six backend v2 workflow modules to import from lfx.workflow.
- Move the five protocol-agnostic contract tests into src/lfx/tests/unit/workflow/
(run in the lfx-only env). test_output_event_parity and test_workflow_agui stay
in langflow (they need langflow.api.build) with repointed imports.
Coverage unchanged: 201 contract tests pass in the lfx-only env, 191 backend v2
tests pass; 392 total, same as before the move.
Splits the ~1.5k-line workflow.py into focused modules and folds in the
execution-timeout and error-sanitization fixes from Cristhianzl's review of #13307.
- B1: workflow.py now holds only the four route handlers. Validation guards move
to workflow_validation, the sync/stream run loop to workflow_execution, and the
durable background machinery to workflow_background (layered, acyclic).
- I1: add workflow_execution_timeout (default 300) and apply a single wall-clock
ceiling across sync, stream, background, and public via _stream_event_frames. A
timeout becomes a sanitized terminal error and marks a background job failed.
- I3: the route error handlers no longer echo raw exception text. They return a
generic, code-tagged message and log the full exception server-side.
- R1: remove the "commented out / future scope" comments that sat over live
dataframe-extraction code in converters.py.
- R4: drop the worker-routing internals from the reattach 409 message.
Tests cover the timeout terminal-error path and the error-body sanitization, and
the settings field-count guard is updated for the new setting.
- recover session_id for completed background jobs from the persisted
terminal message instead of always returning null, so GET status can
continue the same chat/memory thread
- replay a user-cancel as a CUSTOM cancel marker + RUN_FINISHED (agui) and
a `cancelled` terminal (langflow) instead of RUN_ERROR, so a re-attaching
client no longer reads a deliberate stop as a failure
- cancel the evicted still-running buffer writer when the background-run
registry is full, so it stops appending into a run no reader can find
- derive per-component status from the error artifact / valid flag instead
of hardcoding COMPLETED, and stop the langflow adapter dropping `valid`
- throttle the unauthenticated public endpoint per IP and bound its
input_value/session_id length
- document the sync-only scope of request-body globals
- document that live event re-attach is intentionally owner-only
A completed background run's GET status returned a bare COMPLETED with an
empty `outputs` and a null `output`. The COMPLETED branch reconstructs
from `vertex_builds` keyed by job_id, which the durable path does not
write, so reconstruction raised ValueError and fell through to an empty
response; the result was only retrievable via a /events re-attach.
The runner now captures the terminal `output` events (the langflow
adapter's normalized ComponentOutput payloads) into `Job.result`, and the
status COMPLETED branch rebuilds the `outputs` map and resolved `output`
from them via `workflow_response_from_output_events`, matching the sync
response. agui-protocol runs emit no `output` events, so their status
stays result-less (the result remains on the /events log).
The v2 workflows endpoint applied `tweaks` only on mode=sync. The stream
and background paths build the graph via the v1 build-vertex loop
(`generate_flow_events`), which never received the tweaks, so they were
silently dropped. The confusing symptom: a model passed via tweaks
surfaced as "A model selection is required", and any per-component
override was ignored on non-sync runs.
Thread `parsed.tweaks` into `generate_flow_events` and apply them to the
built graph via `vertex.update_raw_params`. We do not use the lfx
`process_tweaks_on_graph` helper because it only sets `vertex.params`,
which does not persist to runtime (the same bug
`lfx.base.tools.run_flow._process_tweaks_on_graph` works around). No-tweaks
runs are unchanged (guarded by `if tweaks`). Adds a streaming regression
test that overrides ChatInput via tweaks and asserts the value drives the run.
The durable background-execution rewrite of workflow.py reverted
`side_channel_events` to its pre-"end" form, dropping the "end" event
from the AG-UI side-channel. That event carries `build_duration` to the
playground chat-view, and the message metadata badge only renders when
`hasDuration || hasTokens`. With build_duration gone the badge vanished,
failing the token-usage and shareable-playground "Finished In" regression
tests. Re-add "end" so the streaming playground path delivers it again.
The durable background-execution work added six Settings fields
(background_max_concurrency, background_job_timeout, background_lease_ttl_s,
background_heartbeat_interval_s, background_watchdog_interval_s, test_redis_url)
without updating EXPECTED_FIELDS, so test_field_count_unchanged failed 152 != 146.
These are intentional bg-exec config; add them to the gate.
As a module-level runtime value, FrameSourceFactory = Callable[..., Any] is a
GenericAlias that passes isinstance(obj, type) but makes issubclass(obj, Service)
raise on Python 3.10/3.14. The service factory scans this module for Service
subclasses (services/factory.py:90), so the runtime alias crashed service
initialization on those interpreters with 'issubclass() arg 1 must be a class'
-> 'Could not initialize services', erroring out dozens of unrelated tests at
setup (3.13 was unaffected, which is why local runs passed).
The alias is only referenced in a lazy annotation (from __future__ import
annotations), so moving it + the Callable import under TYPE_CHECKING removes it
from the runtime namespace with no behavior change.
Verified: alias absent from runtime module namespace, factory scan finds
BackgroundExecutionService cleanly, durable service tests 26/26 pass.
These 3 tests targeted the removed queue-service stop helper
(_cancel_workflow_queue_job / get_queue_service), inherited via the
agui->bg-default merge and failing with AttributeError across the stack:
- test_stop_workflow_success: adapted to the durable stop path
(revoke_task -> stop_job -> update_job_status(CANCELLED)).
- test_stop_workflow_allowed_for_legacy_job_with_no_user_id (IDOR): adapted to
the durable mechanism; still asserts the ownership check does not block a
legacy user_id=None row.
- test_stop_workflow_returns_503_when_queue_cancel_cannot_be_confirmed: dropped
— the durable stop writes a best-effort STOP signal and always finalizes
CANCELLED; the queue-service 'cannot confirm -> 503' path no longer exists.
test_workflow.py now passes 24/24.
A v2 background run created TWO JobType.WORKFLOW rows for one flow execution:
the durable row (submit()'s job_id, owned by JobRunner) plus an orphan keyed by
the run_id generate_flow_events mints, because the build pipeline's
track_job_status defaults True and the durable frame source never passed False.
The flow ran once (double bookkeeping), but every background run left a phantom
WORKFLOW row + job_events and double-fired the memory-base hook, skewing metrics.
Thread track_job_status through _stream_event_frames; pass False only from the
background frame source (the durable runner already owns the row + fires the
hook with the durable job_id). Stream/public paths keep default True. Also gate
build.py's memory-base hook fire behind track_job_status so background doesn't
double-fire. Adds a regression test (RED before fix: found 2 rows).
Resolves the workflow.py / test_workflow_agui.py conflict in favor of the
durable background-execution design: the in-memory _BackgroundRun buffer,
_BACKGROUND_RUNS registry, and their tests are dropped (replaced by the durable
job_events store + InMemoryLiveBus + JobRunner, which already port the
load-bearing guarantees: replay-then-tail, Last-Event-ID resume, exactly-once
non-resurrecting cancellation, FAILED-vs-COMPLETED, cross-worker stop,
privacy-404). The base's AG-UI translator fixes (inactive-vertex deltas +
dedupe) and the jest downlevelIteration fix merge in cleanly.
Verified: 209 backend tests pass (services/background_execution, v2
test_workflow_agui, test_agui_translator, test_converters).
build.py keeps reporting a conditionally-excluded vertex in
inactivated_vertices on every subsequent end_vertex (the excluded set
persists until the ConditionalRouter clears it), so the translator was
putting the same inactive STATE_DELTA on the wire once per remaining
vertex. Track emitted inactive nodes and skip re-emitting; drop a node
from the set when it actually runs again (build_start/end_vertex) so a
loop re-activation can still re-emit inactive later.
A branch component (If-Else, Conditional Router) reports its not-taken
vertices in build_data.inactivated_vertices, but the AG-UI translator only
emitted the branch node's own success/error status and dropped that list. The
canvas seeds every planned node as pending from vertices_sorted; skipped
vertices then get no build_start/end_vertex, so they stayed stuck on pending
instead of rendering as inactive (the v1 build path marked them INACTIVE).
The translator now appends an inactive STATE_DELTA op per inactivated vertex,
and the frontend bridge maps the new inactive status to BuildStatus.INACTIVE
and tears its edges down like a completed node. Fixes the If-Else regression
in general-bugs-reset-flow-run.spec.ts.
