fix(responses): sync conversation before yielding terminal events in streaming (#3888)

Move conversation sync logic before yield to ensure it executes even
when
streaming consumers break early after receiving response.completed
event.

## Test Plan

```
OLLAMA_URL=http://localhost:11434 \
  pytest -sv tests/integration/responses/ \
  --stack-config server:ci-tests \
  --text-model ollama/llama3.2:3b-instruct-fp16 \
  --inference-mode live \
  -k conversation_multi
```

This test now passes.

Author: ashwinb

llama_stack/providers/inline/agents/meta_reference/responses/openai_responses.py

@@ -372,14 +372,13 @@ async def _create_streaming_response(
                 final_response = stream_chunk.response
             elif stream_chunk.type == "response.failed":
                 failed_response = stream_chunk.response
-            yield stream_chunk
 
             if stream_chunk.type == "response.output_item.done":
                 item = stream_chunk.item
                 output_items.append(item)
 
-            # Store and sync immediately after yielding terminal events
-            # This ensures the storage/syncing happens even if the consumer breaks early
+            # Store and sync before yielding terminal events
+            # This ensures the storage/syncing happens even if the consumer breaks after receiving the event
             if (
                 stream_chunk.type in {"response.completed", "response.incomplete"}
                 and final_response
@@ -400,6 +399,8 @@ async def _create_streaming_response(
                     await self._sync_response_to_conversation(conversation, input, output_items)
                     await self.responses_store.store_conversation_messages(conversation, messages_to_store)
 
+            yield stream_chunk
+
     async def delete_openai_response(self, response_id: str) -> OpenAIDeleteResponseObject:
         return await self.responses_store.delete_response_object(response_id)
 

tests/integration/conftest.py

@@ -43,6 +43,7 @@ def pytest_sessionstart(session):
 
     if "SQLITE_STORE_DIR" not in os.environ:
         os.environ["SQLITE_STORE_DIR"] = tempfile.mkdtemp()
+        logger.info(f"Setting SQLITE_STORE_DIR: {os.environ['SQLITE_STORE_DIR']}")
 
     # Set test stack config type for api_recorder test isolation
     stack_config = session.config.getoption("--stack-config", default=None)

tests/integration/fixtures/common.py

@@ -40,7 +40,12 @@ def is_port_available(port: int, host: str = "localhost") -> bool:
 
 def start_llama_stack_server(config_name: str) -> subprocess.Popen:
     """Start a llama stack server with the given config."""
-    cmd = f"uv run llama stack run {config_name}"
+
+    # remove server.log if it exists
+    if os.path.exists("server.log"):
+        os.remove("server.log")
+
+    cmd = f"llama stack run {config_name}"
     devnull = open(os.devnull, "w")
     process = subprocess.Popen(
         shlex.split(cmd),