feat: Remove use of vertex builds table in workflow API background mode - #14353
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughBackground workflows now capture terminal outputs in ChangesBackground output recovery
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant WorkflowExecution
participant BackgroundRunner
participant JobResult
participant WorkflowStatusAPI
WorkflowExecution->>BackgroundRunner: internal terminal output frame
BackgroundRunner->>JobResult: store captured output
WorkflowStatusAPI->>JobResult: read completed result
JobResult-->>WorkflowStatusAPI: output events and session ID
🚥 Pre-merge checks | ✅ 9✅ Passed checks (9 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
✅ Migration Validation Passed All migrations follow the Expand-Contract pattern correctly. |
✅ Test Coverage AdvisorNo source changes detected without accompanying tests. Thanks for keeping coverage up! 🎉
|
github-actions
Bot
added
enhancement
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/backend/base/langflow/services/background_execution/runner.py (2)
235-241: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUpdate the stale comment: agui now populates captured outputs.
This comment states that the agui adapter does not emit terminal outputs and that agui-protocol runs leave
output_eventsempty. The code that follows (lines 251-261) capturesWORKFLOW_OUTPUT_CAPTURE_EVENTframes for both protocols, which is exactly the mechanism this PR adds to close that agui gap. The comment now contradicts the code it precedes and will mislead future readers into thinking agui results are still empty.Update the comment to describe the off-wire capture mechanism instead of the old langflow-only behavior.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/runner.py` around lines 235 - 241, Update the comment preceding the terminal-output capture logic in the background runner to reflect that both langflow and agui adapters capture WORKFLOW_OUTPUT_CAPTURE_EVENT frames into output_events, and remove the outdated claim that agui-protocol runs leave results empty. Describe that captured outputs are persisted for status responses while remaining available on the /events log.
228-261: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winCapture resumed terminal outputs before setting
result.outputs.
graph.resume_from_checkpoint()only restores vertices and queues the resume layer; vertices built before the pause are skipped unless they are dropped downstream. Sinceoutput_eventsresets in_drive()andJob.result["outputs"]is sent as-is, a resumed run can leave terminal outputs incomplete. Persist the prioroutput_eventswith the job/checkpoint and merge them on resume, or replay them from the durable output frames beforeset_result.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/background_execution/runner.py` around lines 228 - 261, Update _drive and the resume flow so previously captured terminal output_events are restored before processing new frames and before Job.result is set. Persist output_events with the job/checkpoint and merge prior outputs on resume, or replay the durable output frames, while preserving newly captured outputs and avoiding duplicates.
🧹 Nitpick comments (1)
src/backend/tests/unit/api/v2/test_workflow_background.py (1)
152-155: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
monkeypatchinstead of manual save/restore for settings overrides. Both new tests temporarily mutate the shared settings singleton with a hand-rolledoriginal = ...; settings.x = False; try/finally: settings.x = originalpattern.pytest'smonkeypatch.setattrfixture is the idiomatic mechanism for this and guarantees restoration automatically, including on test failure inside the body.
src/backend/tests/unit/api/v2/test_workflow_background.py#L152-L155: replace the manualoriginal/assignment forvertex_builds_storage_enabledwithmonkeypatch.setattr(settings, "vertex_builds_storage_enabled", False).src/backend/tests/unit/api/v2/test_workflow_background.py#L194-L195: drop the manualfinally: settings.vertex_builds_storage_enabled = originalrestoration oncemonkeypatchis used.src/backend/tests/unit/api/v2/test_workflow_background.py#L256-L258: replace the manualoriginal/assignment forjob_events_storage_enabledwithmonkeypatch.setattr(settings, "job_events_storage_enabled", False).src/backend/tests/unit/api/v2/test_workflow_background.py#L291-L292: drop the manualfinally: settings.job_events_storage_enabled = originalrestoration oncemonkeypatchis used.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/api/v2/test_workflow_background.py` around lines 152 - 155, In src/backend/tests/unit/api/v2/test_workflow_background.py at lines 152-155 and 256-258, replace the manual settings save-and-override logic with monkeypatch.setattr for vertex_builds_storage_enabled and job_events_storage_enabled respectively; at lines 194-195 and 291-292, remove the corresponding manual finally restorations, relying on monkeypatch cleanup for both tests.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/backend/tests/unit/api/v2/test_workflow_background.py`:
- Around line 190-193: Update the comment above the session_id assertion to
state that Job.result obtains session_id from the persisted submit request in
job.job_metadata["request"], not from terminal events, while preserving the
explanation that background GET continues the same chat thread with vertex-build
storage disabled.
---
Outside diff comments:
In `@src/backend/base/langflow/services/background_execution/runner.py`:
- Around line 235-241: Update the comment preceding the terminal-output capture
logic in the background runner to reflect that both langflow and agui adapters
capture WORKFLOW_OUTPUT_CAPTURE_EVENT frames into output_events, and remove the
outdated claim that agui-protocol runs leave results empty. Describe that
captured outputs are persisted for status responses while remaining available on
the /events log.
- Around line 228-261: Update _drive and the resume flow so previously captured
terminal output_events are restored before processing new frames and before
Job.result is set. Persist output_events with the job/checkpoint and merge prior
outputs on resume, or replay the durable output frames, while preserving newly
captured outputs and avoiding duplicates.
---
Nitpick comments:
In `@src/backend/tests/unit/api/v2/test_workflow_background.py`:
- Around line 152-155: In
src/backend/tests/unit/api/v2/test_workflow_background.py at lines 152-155 and
256-258, replace the manual settings save-and-override logic with
monkeypatch.setattr for vertex_builds_storage_enabled and
job_events_storage_enabled respectively; at lines 194-195 and 291-292, remove
the corresponding manual finally restorations, relying on monkeypatch cleanup
for both tests.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 3383102b-a487-43e5-851b-79a3236faf38
📒 Files selected for processing (7)
src/backend/base/langflow/api/v2/workflow.pysrc/backend/base/langflow/api/v2/workflow_execution.pysrc/backend/base/langflow/services/background_execution/runner.pysrc/backend/tests/unit/api/v2/test_workflow_background.pysrc/lfx/src/lfx/services/settings/groups/telemetry.pysrc/lfx/src/lfx/workflow/adapters/langflow.pysrc/lfx/src/lfx/workflow/converters.py
Frontend Unit Test Coverage ReportCoverage Summary
Unit Test Results
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## release-1.12.0 #14353 +/- ##
==================================================
+ Coverage 61.95% 63.26% +1.31%
==================================================
Files 2417 2390 -27
Lines 242524 243285 +761
Branches 36184 37295 +1111
==================================================
+ Hits 150251 153924 +3673
+ Misses 90354 87436 -2918
- Partials 1919 1925 +6
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
f3ae922 to
2a07971
Compare
github-actions
Bot
added
enhancement
github-actions
Bot
added
enhancement
erichare
left a comment
erichare
left a comment
There was a problem hiding this comment.
Hi @Jkavia , could you take a look?
Code Review Summary
Found 2 critical issues that need to be fixed:
🔴 Critical (Must Fix)
1. [P1] Preserve outputs produced before a HITL pause
FilePath: [src/backend/base/langflow/api/v2/workflow.py](
) line 667if output_events:
return workflow_response_from_output_events(...)Explanation
Any non-empty Job.result.outputs is now treated as the complete result. However, each runner pass starts a fresh output accumulator, suspension exits before saving it, and the resumed pass overwrites Job.result with only its own outputs.
For parallel branches, an output completed before a HITL pause is therefore omitted when another output completes after resume. The non-empty final result also prevents vertex-build fallback; with vertex-build storage disabled, recovery is impossible.
A two-pass real-JobService reproduction confirmed that only the post-resume output survives.
Suggested Fix
- Persist partial captures when suspending.
- Merge captures by
component_idacross resume passes. - Add a parallel-branch HITL regression test with vertex-build storage disabled.
2. [P1] Event-storage-off mode drops live HITL requests
FilePath: [src/backend/base/langflow/services/background_execution/runner.py](
) line 283seq = last_durable_seq + 1Explanation
With event persistence disabled, normal frames use a local sequence. The human-input frame is intercepted instead of published, then _suspend() writes it to the otherwise-empty database log with sequence 1.
If the connected stream has already advanced to local cursor N, replay requests only rows after N, so the lower-numbered HITL prompt is discarded before the stream closes as suspended. Resume starts another local sequence at 1, creating the same cursor collision.
A connected-tail reproduction received live sequences [1, 2] while the database contained only the unseen human-input event at sequence 1. This contradicts the setting’s “live streaming … unaffected” contract and can strand the job awaiting an invisible decision.
Suggested Fix
- Use one job-scoped monotonic cursor across normal, control, and resumed frames.
- Publish the HITL request with that cursor, or store pending-request state outside
job_events. - Add storage-disabled live pause/resume coverage.
✅ What's Good
- The off-wire capture cleanly preserves AG-UI wire behavior.
- The PR merges cleanly onto current
release-1.12.0(e73fd4bf) and passesgit diff --check. - Focused validation passed: 163 LFX tests, 31 runner/resume tests, and all 4 newly added API tests.
- Required CI is currently red from Playwright shard 53’s file-upload failure/SQLite lock and shard 58’s Microsoft apt repository 403; neither overlaps the changed files, but a rerun is still required.
github-actions
Bot
added
enhancement
|
Addressed the first "Preserve outputs produced before a HITL pause" issue and removed the events flag for now as it was only optional addition, will skip that for now as its needed so that should resolve second review comment as well. |
github-actions
Bot
added
enhancement
erichare
left a comment
erichare
left a comment
There was a problem hiding this comment.
@Jkavia looking good, can you take a look at these comments?
Code Review Summary
Found 1 critical issue that needs to be fixed:
🔴 Critical (Must Fix)
1. [P1] Make suspended state and resume metadata atomic
FilePath: [src/backend/base/langflow/services/background_execution/runner.py](
langflow/src/backend/base/langflow/services/background_execution/runner.py
Lines 149 to 163 in 76d0248
await self._jobs.update_job_status(job_id, JobStatus.SUSPENDED)
# pending_request_id and pre_pause_outputs are persisted afterwardExplanation
SUSPENDED is committed before pending_request_id and pre_pause_outputs. A client can observe that status and [claim the resume](
langflow/src/backend/base/langflow/services/background_execution/service.py
Lines 421 to 431 in 76d0248
I reproduced this with a barrier around update_job_metadata(): the continuation completed while the stash was unavailable, and Job.result.outputs contained only the post-resume output. The same window can expose a missing or previous pending_request_id.
Suggested Fix
- Add a job-service operation that persists the suspended status and metadata in one transaction.
- Add a barrier-based regression test that resumes while suspension is being persisted.
Found 3 suggestions for improvement:
🟡 Suggestions (Should Consider)
1. [P2] Continue capturing standard output frames
FilePath: [src/backend/base/langflow/services/background_execution/runner.py](
langflow/src/backend/base/langflow/services/background_execution/runner.py
Lines 255 to 296 in 76d0248
Explanation
The runner now populates results exclusively from the private WORKFLOW_OUTPUT_CAPTURE_EVENT, removing the previous standard "output" handling. An injected/exported FrameSource can emit a valid durable output followed by end; that output is persisted and published, but the job completes with outputs: []. I reproduced this against the current head.
Suggested Fix
- Retain
"output"as a fallback capture path. - Deduplicate the private and standard captures by
component_id. - Add a runner test using only normal adapter events.
2. [P2] Preserve the effective session on vertex-build fallback
FilePath: [src/backend/base/langflow/api/v2/workflow.py](
langflow/src/backend/base/langflow/api/v2/workflow.py
Lines 651 to 686 in 76d0248
Explanation
The endpoint derives effective_session_id from the persisted request, but returns the reconstructed vertex-build response unchanged. When Job.result.outputs is empty and the terminal builds are data-only, reconstruction cannot infer a session and returns null despite an explicit request session. That violates the documented response contract.
Suggested Fix
Apply effective_session_id when the reconstructed response does not already contain a session, and add a data-only fallback test.
3. [P2] Fall back when stored outputs cannot be deserialized
FilePath: [src/backend/base/langflow/api/v2/workflow.py](
langflow/src/backend/base/langflow/api/v2/workflow.py
Lines 665 to 673 in 76d0248
FilePath: [src/lfx/src/lfx/workflow/converters.py](
langflow/src/lfx/src/lfx/workflow/converters.py
Lines 509 to 522 in 76d0248
Explanation
Any truthy stored output list suppresses vertex-build reconstruction, while the converter silently skips malformed or version-skewed entries. A completed job can consequently return empty or partial outputs despite having valid vertex-build recovery data. I reproduced an invalid stored capture returning {} without invoking reconstruction.
Suggested Fix
Make conversion report rejected entries and fall back—or merge with vertex reconstruction—when the stored result cannot be reconstructed completely.
Found 1 optional nit:
🟢 Nits (Optional)
1. Refresh stale AG-UI comments and the PR description
FilePath: [runner.py](
langflow/src/backend/base/langflow/services/background_execution/runner.py
Lines 235 to 246 in 76d0248
Explanation
Several comments still say AG-UI leaves results empty, while the new off-wire capture populates them. The PR description also still advertises LANGFLOW_JOB_EVENTS_STORAGE_ENABLED, which this head removes.
Suggested Fix
Update the comments and PR description to match the final design.
✅ What's Good
- The event-storage-off failure is resolved by removing that mode.
- Pre-pause output preservation works in ordinary, non-racing resume sequences.
- The off-wire capture preserves both Langflow and AG-UI wire protocols.
- Focused validation passed: 154 LFX tests, 23 SQLite runner tests, 3 new SQLite HITL tests, and 2 API integration tests.
- The head
76d02486a0is mergeable onto currentrelease-1.12.0, andgit diff --checkpasses.
github-actions
Bot
added
enhancement
erichare
left a comment
erichare
left a comment
There was a problem hiding this comment.
Re-reviewed at 46fee33. The suspend transition and resume-critical metadata now commit atomically; standard output frames retain a Job.result fallback with deduplication; vertex-build recovery preserves the effective session; malformed or incomplete stored outputs recover without returning a partial authoritative result or a 500. The stale protocol comments and PR description are corrected. Focused and broader backend tests, the full LFX converter suite, Ruff, formatting, and merge-tree validation are green. No remaining findings.
github-actions
Bot
added
lgtm
github-actions
Bot
added
enhancement
dkaushik94
left a comment
dkaushik94
left a comment
There was a problem hiding this comment.
One point I'f fix:
src/backend/base/langflow/api/v2/workflow_execution.py:464 catches only (RuntimeError, ValueError, OSError). But set_result commits through session_scope(), and a DB-layer failure raises a sqlalchemy.exc.SQLAlchemyError (e.g. OperationalError — SQLite "database is locked", which is a documented pain point in this very repo). That's not in the caught tuple, so it escapes the helper, propagates to the broad except Exception at src/backend/base/langflow/api/v2/workflow_execution.py:637, and gets returned as create_error_response(...).
So under DB contention, a run that fully succeeded (response already built inline) reports an error to the client — the exact opposite of the docstring's promise that "a persistence failure must never fail the run." It's a separate session, so there's no transaction-state risk; the fix is just to broaden the except to except Exception (with # noqa: BLE001). None of the new tests cover this path.
Approving since everything else looks good. Pushing this fix to the branch myself.
Should be good to go @Jkavia
…h session_id Flip the v2 workflow GET-status construction so the durable Job.result blob is the primary output source and vertex_build reconstruction is the fallback. Echo session_id (from job_metadata request, falling back to flow id) on the response so completed background runs return the same shape as sync. Populate Job.result for AG-UI runs via protocol-neutral off-wire capture: a synthesized WORKFLOW_OUTPUT_CAPTURE_EVENT frame built from the raw end_vertex before adapter.translate, captured in-memory only (never appended or published, so it cannot leak to /events). This makes AG-UI runs carry full outputs even though the wire protocol emits no output event. Add LANGFLOW_JOB_EVENTS_STORAGE_ENABLED to gate job_events persistence. When off, the runner uses a local seq fallback so live subscribers still work while per-milestone DB writes and reattach/replay are skipped; Job.result and GET-status output are unaffected. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> fix(workflow-api): preserve pre-pause HITL outputs; drop job-events storage flag A background run that produced a terminal output on one branch, then paused for human input on another, lost that output: the resumed pass starts a fresh capture list and the output loop skips re-emitting checkpoint-restored vertices, so finalize only had the resumed pass's captures. Fix: PauseRequested carries the pre-pause output_events out of _drive; _suspend stashes them (deduped) in job_metadata.pre_pause_outputs; a resumed _drive pre-seeds its capture list from that stash; set_result dedups by component_id (first position, latest value) so a re-emitted vertex overwrites its stale entry and branch order is preserved. Also remove the job_events_storage_enabled setting — the durable event log is now always persisted (removes the event-storage-off path that dropped live HITL requests). Drops the field from EXPECTED_FIELDS and the now-impossible off-mode test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> fix: harden background workflow result handling feat(workflow-api): persist sync run outputs to Job.result A sync run already creates a Job row (to support HITL suspend + run_id-keyed vertex builds) and reaches COMPLETED, but never wrote Job.result — so a later GET status on its job_id could not return the outputs the request returned inline. Persist the terminal outputs in the same list-of-OutputEvent shape the background runner writes, so the status read is protocol-uniform across sync and background. Best-effort: the caller already holds the response inline, so a result-cache write failure is logged and never fails the run. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Fixed exception handling upon persistence failure but execution success. now persistence issue won't block users from getting the output if the workflow ran successfully.
00c9ce0 to
646b0f2
Compare
github-actions
Bot
added
enhancement
github-actions
Bot
added
enhancement
Both conflicts were additive: langflow-ai#14353 added emit_output_capture to _stream_event_frames and its background call site at the same spot this branch adds execution_timeout. Kept both parameters.
3 participants
Make the v2 workflow GET-status endpoint use the durable
Job.resultblob as its primary output source, withvertex_buildreconstruction as the recovery path for legacy, missing, malformed, or incomplete captures. The completed response preserves the submittedsession_id(falling back to the flow ID) across both paths.Populate
Job.resultfor both Langflow and AG-UI runs through protocol-neutral off-wire output capture. The capture is kept in memory only and never appended or published, so it cannot leak onto/events. The runner also accepts the standard durableoutputframe as a fallback for custom frame sources, with final deduplication preventing double capture.Keep HITL suspension coherent by committing the
SUSPENDEDstatus and resume-critical metadata (pending_request_idand pre-pause outputs) atomically. This prevents a racing resume from observing a suspended job before the metadata needed to continue it is visible.Store workflow API sync results in the Job table as well.
Verified with focused background-workflow, HITL suspend/resume, status-recovery, and converter tests, including the vertex-build-storage-off integration path and malformed stored-output recovery.