fix(parsing): drop TextFormatT parameterization in parse_response to fix memory leak (openai#3084) (openai#3088)

## Why

Closes openai#3084.

`AsyncResponses.parse()` leaks pydantic schema objects without bound.
The issue author captured it with a flame graph — every call allocates a
fresh Rust-backed `SchemaValidator`/`SchemaSerializer` that never gets
freed.

## Root cause

`parse_response()` called `construct_type_unchecked` with three types
parameterized by a free module-level `TypeVar`:

```python
construct_type_unchecked(type_=ParsedResponseOutputText[TextFormatT], ...)
construct_type_unchecked(type_=ParsedResponseOutputMessage[TextFormatT], ...)
construct_type_unchecked(type_=ParsedResponse[TextFormatT], ...)
```

Pydantic can't resolve a free `TypeVar`, so
`model_rebuild(raise_errors=False)` returns `False`.
`MockCoreSchema._built_memo` only caches the rebuilt schema when
`model_rebuild` succeeds — so the cache is never populated and a new
schema is allocated on every call.

## Fix

Drop the `[TextFormatT]` parameterization at the `type_=` argument. At
runtime Python's generics are erased anyway — the constructed object's
type is identical either way. Only the pydantic schema rebuild path
differs:

| | Before | After |
|---|---|---|
| `type_` arg | `ParsedResponse[TextFormatT]` | `ParsedResponse` |
| `model_rebuild` | always returns `False` | succeeds |
| `_built_memo` | never populated | populated once, reused |
| Per-call allocation | heavy Rust objects | cache hit |

This is exactly what the issue author diagnosed.

## Verification

- Ran the existing `tests/lib/responses/test_responses.py` suite
locally: **5/5 passing** — no functional regression.
- The change is purely about schema caching behavior. Callers see no
difference in returned objects or their types.

## Scope

Minimal — 3 sites in one file, inline comments pointing back to openai#3084
for maintainability. No public API change, no test changes needed
(existing tests cover the parse path).

---------

Signed-off-by: Mukunda Katta <mukunda.vjcs6@gmail.com>
Co-authored-by: Marcus Wood <marcuswood@openai.com>

Author: MukundaKatta

src/openai/lib/_parsing/_responses.py

@@ -56,6 +56,8 @@ def parse_response(
     input_tools: Iterable[ToolParam] | Omit | None,
     response: Response | ParsedResponse[object],
 ) -> ParsedResponse[TextFormatT]:
+    # Keep generic annotations in quoted casts: runtime specialization can create
+    # new model classes in each async context and retain them in the type adapter cache.
     output_list: List[ParsedResponseOutputItem[TextFormatT]] = []
 
     for output in response.output or []:
@@ -68,7 +70,7 @@ def parse_response(
 
                 content_list.append(
                     construct_type_unchecked(
-                        type_=ParsedResponseOutputText[TextFormatT],
+                        type_=cast("type[ParsedResponseOutputText[TextFormatT]]", ParsedResponseOutputText),
                         value={
                             **item.to_dict(),
                             "parsed": parse_text(item.text, text_format=text_format, phase=output.phase),
@@ -78,7 +80,7 @@ def parse_response(
 
             output_list.append(
                 construct_type_unchecked(
-                    type_=ParsedResponseOutputMessage[TextFormatT],
+                    type_=cast("type[ParsedResponseOutputMessage[TextFormatT]]", ParsedResponseOutputMessage),
                     value={
                         **output.to_dict(),
                         "content": content_list,
@@ -133,7 +135,7 @@ def parse_response(
             output_list.append(output)
 
     return construct_type_unchecked(
-        type_=ParsedResponse[TextFormatT],
+        type_=cast("type[ParsedResponse[TextFormatT]]", ParsedResponse),
         value={
             **response.to_dict(),
             "output": output_list,

src/openai/lib/streaming/responses/_responses.py

@@ -283,7 +283,7 @@ def handle_event(self, event: RawResponseStreamEvent) -> List[ResponseStreamEven
 
             events.append(
                 build(
-                    ResponseTextDoneEvent[TextFormatT],
+                    cast("type[ResponseTextDoneEvent[TextFormatT]]", ResponseTextDoneEvent),
                     content_index=event.content_index,
                     item_id=event.item_id,
                     output_index=event.output_index,

src/openai/resources/responses/responses.py

@@ -1422,7 +1422,7 @@ def parser(raw_response: Response) -> ParsedResponse[TextFormatT]:
             ),
             # we turn the `Response` instance into a `ParsedResponse`
             # in the `parser` function above
-            cast_to=cast(Type[ParsedResponse[TextFormatT]], Response),
+            cast_to=cast("Type[ParsedResponse[TextFormatT]]", Response),
         )
 
     @overload
@@ -3305,7 +3305,7 @@ def parser(raw_response: Response) -> ParsedResponse[TextFormatT]:
             ),
             # we turn the `Response` instance into a `ParsedResponse`
             # in the `parser` function above
-            cast_to=cast(Type[ParsedResponse[TextFormatT]], Response),
+            cast_to=cast("Type[ParsedResponse[TextFormatT]]", Response),
         )
 
     @overload

tests/lib/responses/test_commentary_parsing.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
 import json
+import asyncio
+import contextvars
 from typing import Any
 
 import httpx2
 import pytest
 from pydantic import BaseModel, ValidationError
 
-from openai import OpenAI, AsyncOpenAI, omit
+from openai import OpenAI, AsyncOpenAI, omit, _models
 from openai._types import Omit
 from openai.types.responses import ParsedResponse
 from openai.lib.streaming.responses import ResponseStreamEvent
@@ -309,3 +311,41 @@ async def test_phase_selection_preserves_content(sync: bool, streaming: bool, ph
                 assert content.parsed == (Result(answer="final") if actual.phase == "final_answer" else None)
             else:
                 assert content.refusal == raw["refusal"]
+
+
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+@pytest.mark.parametrize("streaming", [False, True], ids=["parse", "stream"])
+async def test_parsing_reuses_types_across_contexts(sync: bool, streaming: bool) -> None:
+    response_types: set[tuple[type, type, type]] = set()
+    event_types: set[type] = set()
+
+    async def request() -> None:
+        if streaming:
+            emitted, response = await _stream(sync, _events("Preparing.", '{"answer":"final"}'))
+            done = [event for event in emitted if event.type == "response.output_text.done"]
+            assert [event.parsed for event in done] == [None, Result(answer="final")]
+            event_types.update(type(event) for event in done)
+        else:
+            response = await _parse(sync, [_message('{"answer":"final"}')])
+        assert response.output_parsed == Result(answer="final")
+        message = response.output[-1]
+        assert message.type == "message"
+        part = message.content[0]
+        assert part.type == "output_text"
+        assert part.parsed == Result(answer="final")
+        response_types.add((type(response), type(message), type(part)))
+
+    # Inherited contexts can share Pydantic's generic cache and hide the leak.
+    for _ in range(3):
+        await contextvars.Context().run(asyncio.create_task, request())
+
+    # Pydantic v1 has no TypeAdapter cache, but still exercises type reuse below.
+    cache_info = getattr(getattr(_models, "_CachedTypeAdapter", None), "cache_info", None)
+    before = cache_info().currsize if cache_info is not None else None
+    for _ in range(10):
+        await contextvars.Context().run(asyncio.create_task, request())
+    if cache_info is not None:
+        assert cache_info().currsize == before
+    assert len(response_types) == 1
+    if streaming:
+        assert len(event_types) == 1