feat: Handle error events in OpenAI response streaming#10844
feat: Handle error events in OpenAI response streaming#10844edwinjosechittilappilly merged 26 commits intomainfrom
Conversation
Adds error event handling to the OpenAI response stream, yielding an OpenAI-compatible error response and stopping the stream on error. Includes a test to verify error propagation during streaming.
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. WalkthroughThe PR adds error event handling to the OpenAI streaming response handler. When an error event is detected, the code extracts the error message, logs it, constructs an OpenAI-compatible error response, yields it to the client, and terminates the stream. A comprehensive test validates this error propagation behavior. Changes
Estimated code review effort🎯 2 (Simple) | ⏱️ ~10 minutes
Possibly related PRs
Suggested labels
Suggested reviewers
Pre-merge checks and finishing touches❌ Failed checks (3 warnings)
✅ Passed checks (4 passed)
✨ Finishing touches
🧪 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 |
github-actions
bot
added
enhancement
github-actions
bot
added
enhancement
github-actions
bot
added
enhancement
github-actions
bot
added
enhancement
github-actions
bot
added
enhancement
github-actions
bot
added
enhancement
github-actions
bot
added
enhancement
|
Sample Output Structure:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py (1)
83-87: Shorten and moveevent: errorassertion comment to fix E501 and clarify intentThe inline comment makes this line exceed 120 characters and is a bit hard to scan. Moving it above the assertion both fixes E501 and clarifies why we’re checking for the absence of
event: error:- # Verify we got the error event in the stream - assert ( - "event: error" not in content - ) # OpenAI format doesn't use event: error for the data payload itself usually, but let's check the data + # Verify we got the error event in the stream via the data payload, not an `event: error` line. + assert "event: error" not in contentThis matches the intended OpenAI-compatible behavior and satisfies the linter.
🧹 Nitpick comments (2)
src/backend/base/langflow/api/v1/openai_responses.py (1)
168-177: Streaming error-event behavior is correct; align generic exception path JSON formattingThe new
event_type == "error"handling does the right thing: extracts a message, logs it, builds an OpenAI-style error viacreate_openai_error, yields it as JSON in adata:line, and terminates the stream. One small consistency improvement is to JSON-serialize the error response in the genericexceptpath as well so clients never see a Python dict repr in the stream.You can mirror the new behavior with a minimal change:
except Exception as e: # noqa: BLE001 logger.error(f"Error in stream generator: {e}") error_response = create_openai_error( message=str(e), type_="processing_error", ) - yield f"data: {error_response}\n\n" + yield f"data: {json.dumps(error_response)}\n\n"This keeps all streaming errors consistently JSON-encoded for clients.
Also applies to: 390-396
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py (1)
15-19: Tighten test docstring to satisfy Ruff D205Ruff treats the current multi-line docstring as summary + description without a blank line. Collapsing it into a single-line summary is simplest and keeps the intent clear:
-@pytest.mark.asyncio -async def test_openai_response_stream_error_handling(client): - """Test that errors during streaming are correctly propagated to the client - as OpenAI-compatible error responses. - """ +@pytest.mark.asyncio +async def test_openai_response_stream_error_handling(client): + """Test that errors during streaming are propagated as OpenAI-compatible error responses."""This removes the D205 warning without changing behavior.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/backend/base/langflow/api/v1/openai_responses.py(1 hunks)src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (3)
src/backend/**/*.py
📄 CodeRabbit inference engine (.cursor/rules/backend_development.mdc)
src/backend/**/*.py: Use FastAPI async patterns withawaitfor async operations in component execution methods
Useasyncio.create_task()for background tasks and implement proper cleanup with try/except forasyncio.CancelledError
Usequeue.put_nowait()for non-blocking queue operations andasyncio.wait_for()with timeouts for controlled get operations
Files:
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.pysrc/backend/base/langflow/api/v1/openai_responses.py
**/test_*.py
📄 CodeRabbit inference engine (Custom checks)
**/test_*.py: Review test files for excessive use of mocks that may indicate poor test design - check if tests have too many mock objects that obscure what's actually being tested
Warn when mocks are used instead of testing real behavior and interactions, and suggest using real objects or test doubles when mocks become excessive
Ensure mocks are used appropriately for external dependencies only, not for core logic
Backend test files should follow the naming convention test_*.py with proper pytest structure
Test files should have descriptive test function names that explain what is being tested
Tests should be organized logically with proper setup and teardown
Consider including edge cases and error conditions for comprehensive test coverage
Verify tests cover both positive and negative scenarios where appropriate
For async functions in backend tests, ensure proper async testing patterns are used with pytest
For API endpoints, verify both success and error response testing
Files:
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
src/backend/base/langflow/api/**/*.py
📄 CodeRabbit inference engine (.cursor/rules/backend_development.mdc)
Backend API endpoints should be organized by version (v1/, v2/) under
src/backend/base/langflow/api/with specific modules for features (chat.py, flows.py, users.py, etc.)
Files:
src/backend/base/langflow/api/v1/openai_responses.py
🧠 Learnings (5)
📚 Learning: 2025-11-24T19:47:28.997Z
Learnt from: CR
Repo: langflow-ai/langflow PR: 0
File: .cursor/rules/testing.mdc:0-0
Timestamp: 2025-11-24T19:47:28.997Z
Learning: Applies to src/backend/tests/**/*.py : Test Langflow REST API endpoints using the `client` fixture with appropriate HTTP methods (GET, POST, etc.), headers (logged_in_headers), and payload validation
Applied to files:
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
📚 Learning: 2025-11-24T19:47:28.997Z
Learnt from: CR
Repo: langflow-ai/langflow PR: 0
File: .cursor/rules/testing.mdc:0-0
Timestamp: 2025-11-24T19:47:28.997Z
Learning: Applies to src/backend/tests/**/*.py : Use `monkeypatch` fixture to mock internal functions for testing error handling scenarios; validate error status codes and error message content in responses
Applied to files:
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
📚 Learning: 2025-11-24T19:47:28.997Z
Learnt from: CR
Repo: langflow-ai/langflow PR: 0
File: .cursor/rules/testing.mdc:0-0
Timestamp: 2025-11-24T19:47:28.997Z
Learning: Applies to src/backend/tests/**/*.py : Test real-time event streaming endpoints by consuming NDJSON lines using `response.aiter_lines()`, parsing JSON, and validating event structure and job_id consistency
Applied to files:
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
📚 Learning: 2025-11-24T19:47:28.997Z
Learnt from: CR
Repo: langflow-ai/langflow PR: 0
File: .cursor/rules/testing.mdc:0-0
Timestamp: 2025-11-24T19:47:28.997Z
Learning: Applies to src/backend/tests/**/*.py : Test both sync and async code paths, mock external dependencies appropriately, test error handling and edge cases, validate input/output behavior, and test component initialization and configuration
Applied to files:
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
📚 Learning: 2025-11-24T19:46:09.104Z
Learnt from: CR
Repo: langflow-ai/langflow PR: 0
File: .cursor/rules/backend_development.mdc:0-0
Timestamp: 2025-11-24T19:46:09.104Z
Learning: Applies to src/backend/base/langflow/api/**/*.py : Backend API endpoints should be organized by version (v1/, v2/) under `src/backend/base/langflow/api/` with specific modules for features (chat.py, flows.py, users.py, etc.)
Applied to files:
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
🪛 GitHub Actions: Ruff Style Check
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
[error] 1-1: INP001 File is part of an implicit namespace package. Add an init.py.
🪛 GitHub Check: Ruff Style Check (3.13)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py
[failure] 86-86: Ruff (E501)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:86:121: E501 Line too long (121 > 120)
[failure] 66-66: Ruff (ARG001)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:66:52: ARG001 Unused function argument: kwargs
[failure] 66-66: Ruff (ARG001)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:66:44: ARG001 Unused function argument: args
[failure] 51-51: Ruff (F841)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:51:78: F841 Local variable mock_run_flow is assigned to but never used
[failure] 51-59: Ruff (SIM117)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:51:9: SIM117 Use a single with statement with multiple contexts instead of nested with statements
[failure] 17-19: Ruff (D205)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:17:5: D205 1 blank line required between summary line and description
[failure] 1-1: Ruff (INP001)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:1:1: INP001 File src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py is part of an implicit namespace package. Add an __init__.py.
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (18)
- GitHub Check: Run Backend Tests / Unit Tests - Python 3.10 - Group 4
- GitHub Check: Run Backend Tests / Unit Tests - Python 3.10 - Group 5
- GitHub Check: Run Backend Tests / Unit Tests - Python 3.10 - Group 2
- GitHub Check: Run Backend Tests / Unit Tests - Python 3.10 - Group 1
- GitHub Check: Run Backend Tests / Integration Tests - Python 3.10
- GitHub Check: Run Backend Tests / Unit Tests - Python 3.10 - Group 3
- GitHub Check: Test Starter Templates
- GitHub Check: Lint Backend / Run Mypy (3.12)
- GitHub Check: Lint Backend / Run Mypy (3.11)
- GitHub Check: Lint Backend / Run Mypy (3.13)
- GitHub Check: Run Frontend Unit Tests / Frontend Jest Unit Tests
- GitHub Check: Lint Backend / Run Mypy (3.10)
- GitHub Check: Run Backend Tests / LFX Tests - Python 3.10
- GitHub Check: Run Frontend Tests / Determine Test Suites and Shard Distribution
- GitHub Check: Test Docker Images / Test docker images
- GitHub Check: Optimize new Python code in this PR
- GitHub Check: Update Component Index
- GitHub Check: Update Starter Projects
🔇 Additional comments (1)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py (1)
9-12: Client fixture setup is straightforwardCreating the app via
create_app()and wrapping it in aTestClientper test is a clean, idiomatic pattern here; no changes needed.
| @@ -0,0 +1,96 @@ | |||
| import json | |||
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Add __init__.py to make this tests package explicit (Ruff INP001)
Ruff reports this file as part of an implicit namespace package. To satisfy INP001 and keep tools happy, add an (optionally empty) __init__.py alongside this module, e.g.:
# src/backend/base/langflow/tests/api/v1/__init__.py
"""
Make the tests.api.v1 package explicit for tooling.
"""No changes are needed in test_openai_responses_error.py itself, but this extra file will unblock the Ruff check.
🧰 Tools
🪛 GitHub Actions: Ruff Style Check
[error] 1-1: INP001 File is part of an implicit namespace package. Add an init.py.
🪛 GitHub Check: Ruff Style Check (3.13)
[failure] 1-1: Ruff (INP001)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:1:1: INP001 File src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py is part of an implicit namespace package. Add an __init__.py.
🤖 Prompt for AI Agents
In src/backend/base/langflow/tests/api/v1 around lines 1-1, Ruff reports this
module lives in an implicit namespace package; create
src/backend/base/langflow/tests/api/v1/__init__.py (can be empty or contain a
short module docstring) to make the package explicit so the INP001 check passes
and no changes to test_openai_responses_error.py are required.
| with patch("langflow.api.v1.openai_responses.get_flow_by_id_or_endpoint_name") as mock_get_flow: | ||
| # Setup mock flow | ||
| mock_flow = MagicMock() | ||
| mock_flow.data = {"nodes": [{"data": {"type": "ChatInput"}}, {"data": {"type": "ChatOutput"}}]} | ||
| mock_get_flow.return_value = mock_flow | ||
|
|
||
| # Mock run_flow_generator to emit an error event | ||
| with patch("langflow.api.v1.openai_responses.run_flow_generator") as mock_run_flow: | ||
| # We need to simulate the event manager queue behavior | ||
| # The run_flow_generator in the actual code puts events into the event_manager | ||
| # which puts them into the queue. | ||
|
|
||
| # Instead of mocking the complex event manager interaction, we can mock | ||
| # consume_and_yield to yield our simulated error event | ||
|
|
||
| with patch("langflow.api.v1.openai_responses.consume_and_yield") as mock_consume: | ||
| # Simulate an error event from the queue | ||
| error_event = json.dumps({"event": "error", "data": {"error": "Simulated streaming error"}}).encode( | ||
| "utf-8" | ||
| ) | ||
|
|
||
| # Yield error event then None to end stream | ||
| async def event_generator(*args, **kwargs): | ||
| yield error_event | ||
| yield None | ||
|
|
||
| mock_consume.side_effect = event_generator | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Resolve Ruff F841/ARG001/SIM117 by simplifying patches and marking unused args
You can eliminate the unused mock_run_flow, unused args/kwargs, and nested with blocks in one go by combining contexts and using underscore-prefixed parameters:
- # Mock the flow execution to simulate an error during streaming
- with patch("langflow.api.v1.openai_responses.get_flow_by_id_or_endpoint_name") as mock_get_flow:
- # Setup mock flow
- mock_flow = MagicMock()
- mock_flow.data = {"nodes": [{"data": {"type": "ChatInput"}}, {"data": {"type": "ChatOutput"}}]}
- mock_get_flow.return_value = mock_flow
-
- # Mock run_flow_generator to emit an error event
- with patch("langflow.api.v1.openai_responses.run_flow_generator") as mock_run_flow:
- # We need to simulate the event manager queue behavior
- # The run_flow_generator in the actual code puts events into the event_manager
- # which puts them into the queue.
-
- # Instead of mocking the complex event manager interaction, we can mock
- # consume_and_yield to yield our simulated error event
-
- with patch("langflow.api.v1.openai_responses.consume_and_yield") as mock_consume:
- # Simulate an error event from the queue
- error_event = json.dumps({"event": "error", "data": {"error": "Simulated streaming error"}}).encode(
- "utf-8"
- )
-
- # Yield error event then None to end stream
- async def event_generator(*args, **kwargs):
- yield error_event
- yield None
-
- mock_consume.side_effect = event_generator
+ # Mock the flow execution to simulate an error during streaming
+ with (
+ patch("langflow.api.v1.openai_responses.get_flow_by_id_or_endpoint_name") as mock_get_flow,
+ patch("langflow.api.v1.openai_responses.run_flow_generator"),
+ patch("langflow.api.v1.openai_responses.consume_and_yield") as mock_consume,
+ ):
+ # Setup mock flow
+ mock_flow = MagicMock()
+ mock_flow.data = {"nodes": [{"data": {"type": "ChatInput"}}, {"data": {"type": "ChatOutput"}}]}
+ mock_get_flow.return_value = mock_flow
+
+ # Simulate an error event from the queue
+ error_event = json.dumps({"event": "error", "data": {"error": "Simulated streaming error"}}).encode("utf-8")
+
+ # Yield error event then None to end stream
+ async def event_generator(*_args, **_kwargs):
+ yield error_event
+ yield None
+
+ mock_consume.side_effect = event_generatorThis addresses Ruff F841 (unused mock_run_flow), ARG001 (unused args/kwargs), and SIM117 (nested with), while keeping the test logic the same.
🧰 Tools
🪛 GitHub Check: Ruff Style Check (3.13)
[failure] 66-66: Ruff (ARG001)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:66:52: ARG001 Unused function argument: kwargs
[failure] 66-66: Ruff (ARG001)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:66:44: ARG001 Unused function argument: args
[failure] 51-51: Ruff (F841)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:51:78: F841 Local variable mock_run_flow is assigned to but never used
[failure] 51-59: Ruff (SIM117)
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py:51:9: SIM117 Use a single with statement with multiple contexts instead of nested with statements
🤖 Prompt for AI Agents
src/backend/base/langflow/tests/api/v1/test_openai_responses_error.py around
lines 44 to 71: simplify the test patches by combining the patch contexts into a
single with and remove the unused mock_run_flow variable and nested with blocks;
replace the async event_generator signature to accept underscore-prefixed unused
params (e.g., async def event_generator(*_args, **_kwargs):) or no params if
none are used, yield the error_event then None, and set mock_consume.side_effect
= event_generator, thus eliminating Ruff F841/ARG001/SIM117 warnings while
keeping the same simulated error stream behavior.
ogabrielluiz
changed the title
github-actions
bot
added
enhancement
Adds error event handling to the OpenAI response stream, yielding an OpenAI-compatible error response and stopping the stream on error. Includes a test to verify error propagation during streaming.
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.