Stop the generation memory leak behind the 512 MiB OOMs

server/app/services/learning_path_service.py

@@ -18,6 +18,7 @@
     PermissionDeniedError,
     RateLimitError,
 )
+from openai.lib._parsing._responses import type_to_text_format_param
 from pydantic import BaseModel
 
 SYSTEM_PROMPT = """Generate a learning path as a directed graph for a given topic.
@@ -99,6 +100,26 @@ class LearningPathGraphOutput(BaseModel):
     edges: list[GraphEdge]
 
 
+# responses.parse(text_format=...) leaks memory under uvicorn in openai<3.16.2: each call creates
+# new ParsedResponse[...] classes and the SDK keeps a TypeAdapter for every one of them
+# (openai-python #3084). So generation sends the same strict JSON schema format through
+# responses.create and parses the reply with parse_graph_output. Once openai>=3.16.2 is locked,
+# responses.parse can return; test_generation_does_not_grow_sdk_type_cache guards the leak.
+LEARNING_PATH_TEXT_FORMAT = type_to_text_format_param(LearningPathGraphOutput)
+
+
+def parse_graph_output(response: Any) -> LearningPathGraphOutput | None:
+    """Validate every output_text part and return the first, as responses.parse does."""
+    graphs = [
+        LearningPathGraphOutput.model_validate_json(content.text)
+        for item in response.output
+        if item.type == "message"
+        for content in item.content
+        if content.type == "output_text"
+    ]
+    return graphs[0] if graphs else None
+
+
 class LearningPathOutput(BaseModel):
     """Legacy schema kept for backward compatibility in tests."""
 
@@ -308,10 +329,10 @@ def _build_usage_dict(self, usage: Any) -> dict[str, int]:
             "total_tokens": total_tokens,
         }
 
-    async def _request_completion(self, topic: str) -> Any:
-        """Call OpenAI structured-output generation, mapping SDK errors."""
+    async def _request_completion(self, topic: str) -> tuple[Any, LearningPathGraphOutput | None]:
+        """Call OpenAI structured-output generation and parse it, mapping SDK and parse errors."""
         try:
-            return await self._client.responses.parse(
+            response = await self._client.responses.create(
                 model=self._model,
                 # Luna defaults to medium reasoning. Preserve the previous
                 # non-reasoning workload; other model overrides keep their behavior.
@@ -320,9 +341,10 @@ async def _request_completion(self, topic: str) -> Any:
                     {"role": "system", "content": SYSTEM_PROMPT},
                     {"role": "user", "content": f'Generate a learning path for "{topic}".'},
                 ],
-                text_format=LearningPathGraphOutput,
+                text={"format": LEARNING_PATH_TEXT_FORMAT},
                 store=True,
             )
+            return response, parse_graph_output(response)
         except Exception as e:
             raise self._map_upstream_error(e) from e
 
@@ -350,14 +372,14 @@ async def generate_learning_path(self, topic: str) -> dict[str, Any]:
                 await generation_task
             raise
 
-        response = await generation_task
+        response, parsed = await generation_task
 
-        if response.output_parsed is None:
-            msg = "Error while reading OpenAI's response.output_parsed for learning path."
+        if parsed is None:
+            msg = "Error while reading OpenAI's parsed output for learning path."
             logger.warning(f"Structured output parsing failed: {msg}")
             raise LearningPathError(f"Structured output parsing failed: {msg}", status_code=500)
 
-        graph = response.output_parsed.model_dump()
+        graph = parsed.model_dump()
         nodes = graph["nodes"]
         edges = break_cycles(nodes, graph["edges"])
 

server/tests/test_dependency_regressions.py

@@ -11,7 +11,11 @@
 import pytest
 from openai import AsyncOpenAI
 
-from app.services.learning_path_service import LearningPathGraphOutput
+from app.services.learning_path_service import (
+    LEARNING_PATH_TEXT_FORMAT,
+    LearningPathGraphOutput,
+    parse_graph_output,
+)
 
 
 @pytest.mark.parametrize(
@@ -128,10 +132,12 @@ def respond(request):
         max_retries=0,
         http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)),
     ) as sdk:
-        response = await sdk.responses.parse(
-            model="synthetic-model", input="Synthetic test", text_format=LearningPathGraphOutput
+        response = await sdk.responses.create(
+            model="synthetic-model",
+            input="Synthetic test",
+            text={"format": LEARNING_PATH_TEXT_FORMAT},
         )
-    assert response.output_parsed == LearningPathGraphOutput(nodes=[], edges=[])
+    assert parse_graph_output(response) == LearningPathGraphOutput(nodes=[], edges=[])
     assert len(requests) == 1
     # HTTPX normalizes IDNs before AnyIO receives them, matching the advisory workaround.
     assert httpx.URL("https://faß.example").raw_host == b"xn--fa-hia.example"

server/tests/test_model_migration.py

@@ -1,11 +1,13 @@
 """Model migration checks using the real SDK over synthetic HTTP transports."""
 
 import asyncio
+import contextvars
 import json
 
 import httpx
 import pytest
 from openai import AsyncOpenAI
+from openai import _models as openai_models
 
 from app.core.config import Settings
 from app.core.dependencies import get_learning_path_service
@@ -119,6 +121,8 @@ def respond(request):
             with pytest.raises(LearningPathError) as error:
                 await service.generate_learning_path("react")
             assert error.value.status_code == (429 if outcome == "rate-limit" else 500)
+            if outcome == "refusal":
+                assert "parsing failed" in str(error.value)
         else:
             result = await service.generate_learning_path("react")
             assert result["model"] == expected_model
@@ -132,6 +136,40 @@ def respond(request):
     assert set(seen) == {"/v1/moderations", "/v1/responses"}
 
 
+@pytest.mark.asyncio
+async def test_generation_does_not_grow_sdk_type_cache():
+    # uvicorn runs each request in a fresh contextvars.Context, which empties pydantic's
+    # generic-class cache. responses.parse then built new ParsedResponse[...] classes on every
+    # request and kept a TypeAdapter for each in the SDK's unbounded lru_cache, leaking about
+    # 1 MiB per generation until Cloud Run killed the instance at 512 MiB.
+    def respond(request):
+        if request.url.path == "/v1/moderations":
+            return httpx.Response(
+                200, json={"id": "mod_synthetic", "results": [{"flagged": False}]}
+            )
+        return httpx.Response(200, json=response_payload("gpt-6-luna"))
+
+    async with AsyncOpenAI(
+        api_key="synthetic-key",
+        base_url="https://api.openai.com/v1",
+        max_retries=0,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)),
+    ) as sdk:
+        service = LearningPathService(sdk)
+        loop = asyncio.get_running_loop()
+
+        async def generate_like_uvicorn():
+            await contextvars.Context().run(
+                loop.create_task, service.generate_learning_path("react")
+            )
+
+        await generate_like_uvicorn()  # Let one-time caches fill first.
+        before = openai_models._CachedTypeAdapter.cache_info().currsize
+        for _ in range(20):
+            await generate_like_uvicorn()
+        assert openai_models._CachedTypeAdapter.cache_info().currsize == before
+
+
 @pytest.mark.asyncio
 async def test_flagged_moderation_cancels_luna_generation():
     started = asyncio.Event()

server/tests/test_services_learning_path.py

@@ -2,13 +2,15 @@
 
 import pytest
 from openai import AsyncOpenAI
+from pydantic import ValidationError
 
 from app.services.learning_path_service import (
     LearningPathError,
     LearningPathGraphOutput,
     LearningPathService,
     break_cycles,
     graph_to_levels,
+    parse_graph_output,
 )
 
 
@@ -22,6 +24,12 @@ def service(mock_client) -> LearningPathService:
     return LearningPathService(client=mock_client, model="gpt-5-mini", max_topic_length=120)
 
 
+def message_output(mocker, *texts):
+    """A Responses API output list with one message holding these output_text parts."""
+    parts = [mocker.Mock(type="output_text", text=text) for text in texts]
+    return [mocker.Mock(type="message", content=parts)]
+
+
 SAMPLE_GRAPH_DATA = {
     "nodes": [
         {
@@ -220,9 +228,9 @@ async def test_success_returns_graph_structure(
         mock_client.moderations.create = mocker.AsyncMock(
             return_value=mocker.Mock(results=[mocker.Mock(flagged=False)])
         )
-        mock_client.responses.parse = mocker.AsyncMock(
+        mock_client.responses.create = mocker.AsyncMock(
             return_value=mocker.Mock(
-                output_parsed=LearningPathGraphOutput(**graph),
+                output=message_output(mocker, LearningPathGraphOutput(**graph).model_dump_json()),
                 usage=mocker.Mock(
                     model_dump=lambda: {
                         "input_tokens": 10,
@@ -254,10 +262,9 @@ async def test_refusal_raises(self, service: LearningPathService, mock_client, m
             return_value=mocker.Mock(results=[mocker.Mock(flagged=False)])
         )
         refusal_block = mocker.Mock(type="refusal", refusal="Content refused")
-        output_item = mocker.Mock(content=[refusal_block])
-        mock_client.responses.parse = mocker.AsyncMock(
+        output_item = mocker.Mock(type="message", content=[refusal_block])
+        mock_client.responses.create = mocker.AsyncMock(
             return_value=mocker.Mock(
-                output_parsed=None,
                 output=[output_item],
                 usage=mocker.Mock(model_dump=lambda: {}),
                 model="gpt-5-mini",
@@ -268,15 +275,12 @@ async def test_refusal_raises(self, service: LearningPathService, mock_client, m
             await service.generate_learning_path("react")
         assert exc_info.value.status_code == 500
 
-    async def test_none_output_parsed_raises(
-        self, service: LearningPathService, mock_client, mocker
-    ):
+    async def test_no_output_raises(self, service: LearningPathService, mock_client, mocker):
         mock_client.moderations.create = mocker.AsyncMock(
             return_value=mocker.Mock(results=[mocker.Mock(flagged=False)])
         )
-        mock_client.responses.parse = mocker.AsyncMock(
+        mock_client.responses.create = mocker.AsyncMock(
             return_value=mocker.Mock(
-                output_parsed=None,
                 output=[],
                 usage=mocker.Mock(model_dump=lambda: {}),
                 model="gpt-5-mini",
@@ -286,3 +290,47 @@ async def test_none_output_parsed_raises(
         with pytest.raises(LearningPathError, match="parsing") as exc_info:
             await service.generate_learning_path("react")
         assert exc_info.value.status_code == 500
+
+    @pytest.mark.parametrize(
+        "text", ["", '{"nodes": [', '{"nodes": []}'], ids=["empty", "truncated", "missing-edges"]
+    )
+    async def test_invalid_output_text_raises(
+        self, service: LearningPathService, mock_client, mocker, text
+    ):
+        mock_client.moderations.create = mocker.AsyncMock(
+            return_value=mocker.Mock(results=[mocker.Mock(flagged=False)])
+        )
+        mock_client.responses.create = mocker.AsyncMock(
+            return_value=mocker.Mock(output=message_output(mocker, text))
+        )
+
+        with pytest.raises(LearningPathError, match="Unexpected upstream") as exc_info:
+            await service.generate_learning_path("react")
+        assert exc_info.value.status_code == 500
+
+
+class TestParseGraphOutput:
+    def test_first_of_several_parts_wins(self, mocker):
+        first = LearningPathGraphOutput(**SAMPLE_GRAPH_DATA)
+        second = LearningPathGraphOutput(nodes=[], edges=[])
+        output = message_output(mocker, first.model_dump_json()) + message_output(
+            mocker, second.model_dump_json()
+        )
+        assert parse_graph_output(mocker.Mock(output=output)) == first
+
+    def test_any_invalid_part_raises(self, mocker):
+        valid = LearningPathGraphOutput(**SAMPLE_GRAPH_DATA).model_dump_json()
+        with pytest.raises(ValidationError):
+            parse_graph_output(mocker.Mock(output=message_output(mocker, valid, "")))
+
+    def test_skips_other_items_and_refusals(self, mocker):
+        graph = LearningPathGraphOutput(**SAMPLE_GRAPH_DATA)
+        message = mocker.Mock(
+            type="message",
+            content=[
+                mocker.Mock(type="refusal", refusal="Content refused"),
+                mocker.Mock(type="output_text", text=graph.model_dump_json()),
+            ],
+        )
+        output = [mocker.Mock(type="reasoning"), message]
+        assert parse_graph_output(mocker.Mock(output=output)) == graph