fix: async vision streaming + stateful SSE parser (#6284)

[beastoin]

Fixes #6284 — image chat messages fail with "server error" despite 200 OK.

Two confirmed bugs, both reproduced end-to-end with local dev backend + dev app + TCP fragmenting proxy.

Bug 1 — Backend: sync call blocks event loop (5-8s per image chat)

_ask_vision_stream() used sync openai.chat.completions.create(stream=True) inside an async handler, blocking the event loop for 5-8s per request. Under load, this starves all other requests on the pod.

Fix: Convert to AsyncOpenAI client. _ask_vision_stream and process_chat_with_file_stream are now async. _execute_file_chat_stream in graph.py runs the producer as asyncio.create_task so chunks stream progressively while the queue drains concurrently. All failure paths (files.content, streaming, _ensure_thread_and_assistant, _fill_question) guarantee the callback sentinel is sent via try/finally.

Bug 2 — App: SSE parser fails on TCP-fragmented responses

shared.dart:326 used a 1024-byte heuristic to decide if a chunk is "complete." When TCP segments split an SSE line at arbitrary byte boundaries, fragments < 1024 bytes pass through unbuffered → parseMessageChunk returns null → failedMessage() → "Looks like we are having issues with the server."

Fix: Replace the 1024-byte heuristic with a stateful remainder buffer. Accumulate data across TCP reads and only emit complete events delimited by \n\n. Applied to both makeStreamingApiCall and makeMultipartStreamingApiCall.

Deploy

Backend (Bug 1 fix):

1. gh workflow run "Deploy Backend to Cloud RUN" --repo BasedHardware/omi --ref main -f environment=prod
2. Verify: new revision serving traffic, 0 5xx at T+15m (baseline ~4/hr)
3. Pusher (GKE) deploy NOT required — chat_file.py and graph.py are not in pusher's import chain

App (Bug 2 fix):

• shared.dart SSE parser fix ships with the next mobile release (no separate deploy needed)
• Both fixes are complementary — backend async fix alone improves things, full fix needs both deployed

No new dependencies, no env vars, no secrets.

Post-deploy: watch Cloud Run 5xx for 30min, especially /v2/messages endpoint which hits the vision path.

Testing

Backend (11 tests):

• Async producer/consumer streams chunks progressively (not burst)
• Empty stream terminates cleanly
• Mid-stream producer error sends sentinel via try/finally
• Early producer error (before any data) sends sentinel via try/finally
• Concurrent "health check" completes during slow vision stream
• Sync callback: _fill_question failure → end_nowait via try/finally
• Sync callback: mid-stream failure → end_nowait + partial data
• Setup failure: _ensure_thread_and_assistant → end_nowait before re-raise
• Producer/consumer: success populates callback_data (answer, memories, nps)
• Producer/consumer: error populates callback_data['error']
• Producer/consumer: None callback_data doesn't crash

App (16 tests):

• Complete events: single, multiple, empty lines between
• TCP fragmentation: prefix split (da + ta:), payload split, delimiter split, many small fragments, large events (>1024 bytes)
• Base64 events: done: payload intact and split across fragments
• Edge cases: trailing data, empty stream, mixed event types

Live testing (CP9A + CP9B)

• Backend L1: Synced branch to venv, started uvicorn on port 10221 with async code confirmed
• App L1: Built flutter build apk --flavor dev --debug, installed on emulator
• TCP fragment proxy: Port 10220 → 10221, 10-80 byte random fragments with 1ms delays
• L2 integrated: App → fragment proxy → backend → OpenAI → streamed response
  • Text chat: response streamed through fragments, no "server issues" error
  • Image chat: async vision path (_ask_vision_stream) processed 2 image rounds with 4 async OpenAI calls
  • Multi-turn: follow-up question loaded 3 files via _async_openai.files.content, all streamed correctly

Review cycle

6 review rounds via CODEx → PR_APPROVED_LGTM. Fixes across rounds:

• R1: Add try/finally to _ask_vision_stream for error-path cleanup
• R2: Expand try/finally to cover the entire method body (files.content too)
• R3: Fix ask_stream (non-vision path) with try/finally
• R4: Move _fill_question inside try/finally, guard _ensure_thread_and_assistant

Tester cycle: 2 rounds → TESTS_APPROVED. Added 6 error-path/callback_data tests.

Files changed

File	Change
backend/utils/other/chat_file.py	Async _ask_vision_stream + process_chat_with_file_stream, try/finally on all paths
backend/utils/retrieval/graph.py	asyncio.create_task producer + concurrent queue drain
app/lib/backend/http/shared.dart	Stateful SSE remainder buffer in both streaming helpers
backend/tests/unit/test_vision_stream_async.py	11 async producer/consumer + error-path tests
app/test/unit/sse_parser_test.dart	16 SSE fragmentation tests
backend/test.sh	Added new test file

Risks

• ask_stream (non-vision file chat via Assistants API) remains sync — separate issue, different code path
• AsyncOpenAI() at module level requires OPENAI_API_KEY env var (always set in production)

---

by AI for @beastoin

[greptile-apps]

Greptile Summary

This PR fixes two confirmed bugs: (1) _ask_vision_stream was calling the sync OpenAI client inside an async handler, blocking the event loop for 5-8 s per image chat, and (2) the Dart SSE parser used a 1024-byte heuristic that broke on TCP-fragmented responses. Both fixes are well-motivated and the approach is correct.

• P1 — consumer deadlock on producer exception (graph.py lines 87-97): if _ask_vision_stream raises mid-stream (OpenAI error, network timeout), callback.end() is never called. Because asyncio.create_task stores exceptions in the task rather than propagating them immediately, the consumer loop blocks on await callback.queue.get() indefinitely — the except block is never reached and the streaming response hangs. The accompanying test (test_producer_error_terminates_queue) does not reproduce this path: it manually reads one item then awaits the task, bypassing the while True consumer loop entirely.

Confidence Score: 4/5

Safe to merge after fixing the producer-exception deadlock in graph.py; all other changes are correct.

One P1 issue remains: if the OpenAI stream raises mid-response the consumer loop hangs indefinitely, which is worse than the original sync-blocking behavior it replaces. The Dart SSE fix and the async conversion itself are correct and well-tested.

backend/utils/retrieval/graph.py — producer exception path needs a try/finally or sentinel guarantee before merge.

Important Files Changed

Filename	Overview
backend/utils/retrieval/graph.py	Converts file chat producer to asyncio.create_task, but the consumer loop can hang indefinitely if the producer raises without signaling callback.end().
backend/utils/other/chat_file.py	Correctly converts _ask_vision_stream and process_chat_with_file_stream to async with AsyncOpenAI; minor memory management gap for image byte buffers.
app/lib/backend/http/shared.dart	Replaces 1024-byte heuristic with correct stateful SSE remainder buffer in both makeStreamingApiCall and makeMultipartStreamingApiCall; logic is sound.
app/test/unit/sse_parser_test.dart	Comprehensive 16-test suite covering TCP fragmentation, base64 payloads, and edge cases; correctly mirrors production parser logic.
backend/tests/unit/test_vision_stream_async.py	4 async producer/consumer tests; test_producer_error_terminates_queue does not reproduce the production hang scenario (uses manual single-item drain instead of the while-True consumer loop).
backend/test.sh	Adds the new vision stream test file to the CI test runner as required.

Sequence Diagram

sequenceDiagram
    participant Router as FastAPI Router
    participant Graph as graph.py
    participant Callback as AsyncQueue
    participant Producer as _ask_vision_stream
    participant OAI as AsyncOpenAI

    Router->>Graph: async for chunk in _execute_file_chat_stream()
    Graph->>Callback: create AsyncStreamingCallback
    Graph->>Producer: asyncio.create_task(_produce())
    activate Producer
    Producer->>OAI: await chat.completions.create(stream=True)
    OAI-->>Producer: async stream
    loop each chunk
        Producer->>Callback: await callback.put_data(delta)
        Callback-->>Graph: queue.get() → yield chunk
        Graph-->>Router: yield "data: ..."
    end
    alt Normal completion
        Producer->>Callback: await callback.end() → put(None)
        Callback-->>Graph: queue.get() → None → break
        Graph->>Producer: await task → answer
    else Exception raised (P1 bug)
        Producer--xCallback: end() never called
        Note over Graph,Callback: Consumer blocks on queue.get() forever
    end

Loading

_{Reviews (1): Last reviewed commit: "fix: async vision streaming + stateful S..." | Re-trigger Greptile}

[greptile-apps]

Consumer hangs indefinitely when producer raises without signaling end

If _produce() raises an exception (e.g., OpenAI API error mid-stream, network timeout), _ask_vision_stream propagates the exception before reaching await callback.end(). Because asyncio.create_task stores the exception in the task object — not propagating it to the creator — the consumer loop is stuck at await callback.queue.get() forever: no None sentinel is ever enqueued, the except block is never reached, and the generator hangs indefinitely.

The test test_producer_error_terminates_queue does not cover this case; it manually reads one item then awaits the task, skipping the production while True drain loop entirely.

Fix: guard _produce with a try/finally that guarantees the sentinel:

async def _produce():
    try:
        return await fc_tool.process_chat_with_file_stream(question, file_ids, callback=callback)
    except Exception:
        await callback.end()  # ensure consumer is always unblocked
        raise

[greptile-apps]

Large image bytes not freed after base64 encoding

file_content.read() loads the full image into memory as a bytes object. After base64.b64encode() produces b64 (which is ~33% larger), the raw bytes remain live until file_content is garbage-collected. For multi-image chats the loop keeps all raw buffers in memory simultaneously. Per the project memory-management guidelines, large byte arrays should be freed immediately after use.

Suggested change

	file_content = await _async_openai.files.content(file.openai_file_id)
	b64 = base64.b64encode(file_content.read()).decode('utf-8')
	raw_bytes = file_content.read()
	b64 = base64.b64encode(raw_bytes).decode('utf-8')
	del raw_bytes

Context Used: Memory management - free large objects immediately... (source)

[beastoin]

CP8 Test Cycle — TESTS_APPROVED

Test Detail Table

Sequence ID	Path ID	Scenario ID	Changed path	Exact test command	Test name(s)	Assertion intent	Result	Evidence
N/A	P1	S1	chat_file.py:_ask_vision_stream happy	pytest tests/unit/test_vision_stream_async.py::TestAsyncVisionStreamPattern::test_chunks_arrive_progressively -v	test_chunks_arrive_progressively	Async producer yields chunks progressively via callback queue	PASS	11 passed below
N/A	P2	S2	chat_file.py:_ask_vision_stream early error	pytest tests/unit/test_vision_stream_async.py::TestAsyncVisionStreamPattern::test_producer_error_before_stream_terminates_queue -v	test_producer_error_before_stream_terminates_queue	files.content failure → end sentinel delivered	PASS	11 passed below
N/A	P3	S3	chat_file.py:_ask_vision_stream mid-stream error	pytest tests/unit/test_vision_stream_async.py::TestAsyncVisionStreamPattern::test_producer_error_mid_stream_terminates_queue -v	test_producer_error_mid_stream_terminates_queue	Mid-stream error → partial data + end sentinel	PASS	11 passed below
N/A	P4	S4	chat_file.py:process_chat_with_file_stream image routing	pytest tests/unit/test_vision_stream_async.py::TestAsyncVisionStreamPattern::test_empty_stream_terminates -v	test_empty_stream_terminates	Empty producer sends end signal	PASS	11 passed below
N/A	P5	S5	chat_file.py:process_chat_with_file_stream setup failure	pytest tests/unit/test_vision_stream_async.py::TestSyncCallbackErrorPaths::test_ensure_thread_failure_ends_callback_before_ask_stream -v	test_ensure_thread_failure_ends_callback_before_ask_stream	_ensure_thread_and_assistant failure → end_nowait before re-raise	PASS	11 passed below
N/A	P6a	S6a	chat_file.py:ask_stream _fill_question failure	pytest tests/unit/test_vision_stream_async.py::TestSyncCallbackErrorPaths::test_ask_stream_fill_question_failure_ends_callback -v	test_ask_stream_fill_question_failure_ends_callback	_fill_question failure → end_nowait via try/finally	PASS	11 passed below
N/A	P6b	S6b	chat_file.py:ask_stream mid-stream failure	pytest tests/unit/test_vision_stream_async.py::TestSyncCallbackErrorPaths::test_ask_stream_mid_stream_failure_ends_callback -v	test_ask_stream_mid_stream_failure_ends_callback	Mid-stream sync error → end_nowait + partial data	PASS	11 passed below
N/A	P7a	S7a	graph.py:_execute_file_chat_stream success	pytest tests/unit/test_vision_stream_async.py::TestProducerConsumerCallbackData::test_success_populates_callback_data -v	test_success_populates_callback_data	callback_data gets answer, memories_found, ask_for_nps	PASS	11 passed below
N/A	P7b	S7b	graph.py:_execute_file_chat_stream error	pytest tests/unit/test_vision_stream_async.py::TestProducerConsumerCallbackData::test_error_populates_callback_data_error -v	test_error_populates_callback_data_error	callback_data gets error string	PASS	11 passed below
N/A	P7c	S7c	graph.py:_execute_file_chat_stream None callback_data	pytest tests/unit/test_vision_stream_async.py::TestProducerConsumerCallbackData::test_none_callback_data_no_crash -v	test_none_callback_data_no_crash	None callback_data doesn't crash	PASS	11 passed below
N/A	P7d	S7d	async concurrency	pytest tests/unit/test_vision_stream_async.py::TestAsyncVisionStreamPattern::test_concurrent_health_not_blocked -v	test_concurrent_health_not_blocked	Slow producer doesn't block other async work	PASS	11 passed below
N/A	P8	S8	shared.dart:makeStreamingApiCall SSE parser	cd app && flutter test test/unit/sse_parser_test.dart	16 tests (fragmentation, delimiters, edge cases)	Remainder buffer correctly handles TCP-fragmented SSE	PASS	16 passed below
N/A	P9	S9	shared.dart:makeMultipartStreamingApiCall SSE parser	cd app && flutter test test/unit/sse_parser_test.dart	Same 16 tests (identical algorithm)	Same remainder parser in multipart path	PASS	Covered by same algorithm tests

Test Results

Backend (11 tests):

pytest tests/unit/test_vision_stream_async.py -v
11 passed in 0.28s

App (16 tests):

cd app && flutter test test/unit/sse_parser_test.dart
All 16 tests passed

Tester Assessment

• Round 1: identified coverage gaps in error paths and callback_data population
• Added 6 new tests (SyncCallbackErrorPaths + ProducerConsumerCallbackData)
• Round 2: TESTS_APPROVED — all unit-scope gaps closed
• Residual risk (real OpenAI/GCP wiring) deferred to CP9 live testing

---

by AI for @beastoin

[beastoin]

CP9A — Level 1 Live Test (Build & Run Changed Components Standalone)

Backend L1 — Async Vision Streaming

• Build: Synced branch code to full-clone venv, started uvicorn on port 10221
• Startup log: Application startup complete. Uvicorn running on http://0.0.0.0:10221
• Code verification: async def _ask_vision_stream at line 128 confirmed in running backend

App L1 — SSE Parser Fix

• Build: flutter build apk --flavor dev --debug → ✓ Built build/app/outputs/flutter-apk/app-dev-debug.apk
• Install: adb install -r → Performing Streamed Install, Success

TCP Fragment Proxy (chaos engineering)

• Fragment proxy on port 10220 → backend 10221, splitting responses into 10-80 byte random fragments with 1ms delays
• App pointed at http://100.125.36.102:10220/ (all traffic goes through fragmenter)

Changed-Path Coverage Checklist

Path ID	Changed path	Happy-path test	Non-happy-path test	L1 result + evidence
P1	chat_file.py:_ask_vision_stream async streaming	Send image + question through fragment proxy	N/A (error paths covered by unit tests)	PASS — vision response streamed through 10-80 byte fragments
P7	graph.py:_execute_file_chat_stream producer/consumer	Image chat routed to file chat, producer/consumer drained queue	N/A	PASS — log shows [FileChat] All 1 files are images, using Chat Completions vision API
P8	shared.dart:makeStreamingApiCall SSE remainder parser	Text chat through fragment proxy	N/A	PASS — text response rendered correctly
P9	shared.dart:makeMultipartStreamingApiCall SSE remainder parser	Same algorithm as P8	N/A	PASS (same parser logic)
P4-P6	Error paths (setup failure, fill_question, mid-stream)	Covered by unit tests (11 backend)	Covered by unit tests	PASS (unit scope)

Evidence

Text chat through fragment proxy (P8):

Image chat — async vision path (P1, P7):

Backend log trace:

INFO:routers.chat:send_message Describe this image None R2IxlZVs8sRU20j9jLNTBiiFAoO2
INFO:utils.retrieval.graph:execute_chat_stream app: <none>
INFO:utils.retrieval.graph:Processing file chat with 1 files
INFO:utils.other.chat_file:[FileChat] All 1 files are images, using Chat Completions vision API

No errors in backend logs from our code paths. Pre-existing errors (LangSmith not configured, encryption) are unrelated.

L1 Synthesis

All changed paths proven at L1. P1 (async vision) and P7 (producer/consumer) confirmed via image chat that routed through the async _ask_vision_stream path, streamed through the TCP fragment proxy, and rendered correctly in the app. P8/P9 (SSE parser) confirmed via text chat through the same fragment proxy without "server issues" error. Error paths (P4-P6) are proven at unit-test scope (11 backend tests). No paths remain UNTESTED.

---

by AI for @beastoin

[beastoin]

CP9B — Level 2 Live Test (Service + App Integrated)

Infrastructure

• Backend: uvicorn on port 10221 (PID 3260080) with branch code
• TCP Fragment Proxy: port 10220 → 10221, 10-80 byte random fragments
• App: dev APK built from branch, installed on emulator-5554
• Wiring: API_BASE_URL=http://100.125.36.102:10220/ (all traffic through fragmenter)

Test Scenarios

1. Text chat through fragment proxy (P8/P9):

• Sent "What color is the sky?" → received full streaming response
• Response rendered correctly despite TCP fragmentation
• No "Looks like we are having issues with the server" error

2. Image chat — async vision path (P1/P4/P7):

• Attached chaos-test image, sent "Describe this image"
• Backend log: [FileChat] All 1 files are images, using Chat Completions vision API
• Response: detailed description of all 11 objects, text overlays, gradient background
• Streamed through fragment proxy without error

3. Multi-turn image follow-up (P1/P7 repeat):

• Sent "How many shapes are in the image" as follow-up
• Backend log: Processing file chat with 3 files → All 3 files are images, using Chat Completions vision API
• 3x async files.content GET + 1x async chat.completions POST — all via _async_openai
• Response: "43 shapes" with breakdown (16 rectangles/squares, 14 ellipses/circles, 13 lines)

Changed-Path Coverage Checklist (L2 update)

Path ID	Changed path	L2 result + evidence
P1	chat_file.py:_ask_vision_stream async streaming	PASS — 2 image chats, 4 async OpenAI calls, all streamed correctly
P4	chat_file.py:process_chat_with_file_stream image routing	PASS — routed to vision path for all-image input
P7	graph.py:_execute_file_chat_stream producer/consumer	PASS — queue drained, response rendered in app
P8	shared.dart:makeStreamingApiCall SSE parser	PASS — text chat through 10-80 byte fragments
P9	shared.dart:makeMultipartStreamingApiCall SSE parser	PASS (same algorithm as P8)
P5-P6	Error paths	PASS (unit scope — 11 backend tests)

Evidence

Follow-up image response (multi-turn, 3 files, async):

Backend log trace (L2):

INFO:utils.retrieval.graph:Processing file chat with 3 files
INFO:utils.other.chat_file:[FileChat] All 3 files are images, using Chat Completions vision API
INFO:httpx:HTTP Request: GET https://api.openai.com/v1/files/file-H2B3Y4jgJ6PXSBjmV9VqCF/content "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET https://api.openai.com/v1/files/file-FJQ1Gh6vubVYtkTsRBhgZi/content "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET https://api.openai.com/v1/files/file-PPK1MS6kUUH3sst5HMUiLU/content "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"

L2 Synthesis

All changed paths proven at L2 integrated scope. P1/P4/P7 (async vision + producer/consumer) confirmed via 2 image chat rounds with a total of 4 async OpenAI API calls, all streamed through the TCP fragment proxy and rendered correctly in the app. P8/P9 (SSE parser) confirmed via text chat through the same fragment proxy. Error paths (P5/P6) proven at unit scope. No paths remain UNTESTED. No "server issues" errors observed across any test.

---

by AI for @beastoin

[beastoin]

lgtm

[beastoin]

@kodjima33 — both fixes for #6284 are live:

• Backend: async vision streaming deployed to Cloud Run (0 5xx at T+15m)
• App: SSE parser fix in TestFlight build 803 (VALID, ready for internal testing)

Image chat should no longer show "Looks like we are having issues with the server."

---

by AI for @beastoin