diff --git a/scripts/test b/scripts/test
index 68546c7c4d..01935e6a10 100755
--- a/scripts/test
+++ b/scripts/test
@@ -53,7 +53,7 @@ else
   echo
 fi
 
-export DEFER_PYDANTIC_BUILD=false
+export OPENAI_PYDANTIC_DEFER_BUILD=false
 
 if [ "${OPENAI_TEST_HTTP_CLIENT:-httpx}" = "httpx2" ]; then
   echo "==> Using HTTPX2 for sync and async API-resource clients (including RESPX-backed cases)"
diff --git a/src/openai/_models.py b/src/openai/_models.py
index ed4c1f82d6..6614264728 100644
--- a/src/openai/_models.py
+++ b/src/openai/_models.py
@@ -126,7 +126,8 @@ def __repr_args__(self) -> ReprArgs:
             return [arg for arg in super().__repr_args__() if arg[0] not in {"_request_id", "__exclude_fields__"}]
     else:
         model_config: ClassVar[ConfigDict] = ConfigDict(
-            extra="allow", defer_build=coerce_boolean(os.environ.get("DEFER_PYDANTIC_BUILD", "true"))
+            extra="allow",
+            defer_build=coerce_boolean(os.environ.get("OPENAI_PYDANTIC_DEFER_BUILD", "true")),
         )
 
     if TYPE_CHECKING:
diff --git a/src/openai/_utils/_utils.py b/src/openai/_utils/_utils.py
index 02046a352c..bf47d350c7 100644
--- a/src/openai/_utils/_utils.py
+++ b/src/openai/_utils/_utils.py
@@ -337,7 +337,8 @@ def coerce_float(val: str) -> float:
     return float(val)
 
 
-def coerce_boolean(val: str) -> bool:
+def coerce_boolean(obj: object) -> bool:
+    val = str(obj).lower()
     return val == "true" or val == "1" or val == "on"
 
 
diff --git a/src/openai/lib/streaming/responses/_responses.py b/src/openai/lib/streaming/responses/_responses.py
index 6975a9260d..e859b3c73c 100644
--- a/src/openai/lib/streaming/responses/_responses.py
+++ b/src/openai/lib/streaming/responses/_responses.py
@@ -86,6 +86,15 @@ def get_final_response(self) -> ParsedResponse[TextFormatT]:
 
         return response
 
+    def get_abort_reconciliation_items(self) -> List[Dict[str, Any]]:
+        """Returns a list of synthetic function_call_output items for any pending tool calls.
+        
+        This should be used if the stream is closed before it has been read to completion
+        to ensure that the conversation state remains consistent.
+        """
+        return self._state.get_abort_reconciliation_items()
+
+
     def until_done(self) -> Self:
         """Blocks until the stream has been consumed."""
         consume_sync_iterator(self)
@@ -188,6 +197,15 @@ async def get_final_response(self) -> ParsedResponse[TextFormatT]:
 
         return response
 
+    def get_abort_reconciliation_items(self) -> List[Dict[str, Any]]:
+        """Returns a list of synthetic function_call_output items for any pending tool calls.
+        
+        This should be used if the stream is closed before it has been read to completion
+        to ensure that the conversation state remains consistent.
+        """
+        return self._state.get_abort_reconciliation_items()
+
+
     async def until_done(self) -> Self:
         """Blocks until the stream has been consumed."""
         await consume_async_iterator(self)
@@ -340,6 +358,11 @@ def accumulate_event(self, event: RawResponseStreamEvent) -> ParsedResponseSnaps
                 )
             else:
                 snapshot.output.append(event.item)
+        elif event.type == "response.output_item.done":
+            # The JS SDK tracks this to know which tool calls are "complete" from the model's perspective.
+            # In the Python SDK, ParsedResponseFunctionToolCall is updated via deltas,
+            # but we can use this to mark it as done if needed.
+            pass
         elif event.type == "response.content_part.added":
             output = snapshot.output[event.output_index]
             if output.type == "message":
@@ -369,4 +392,35 @@ def _create_initial_response(self, event: RawResponseStreamEvent) -> ParsedRespo
         if event.type != "response.created":
             raise RuntimeError(f"Expected to have received `response.created` before `{event.type}`")
 
-        return construct_type_unchecked(type_=ParsedResponseSnapshot, value=event.response.to_dict())
+        snapshot = construct_type_unchecked(type_=ParsedResponseSnapshot, value=event.response.to_dict())
+        if snapshot.output is None:
+            snapshot.output = []
+        return snapshot
+
+
+    def get_abort_reconciliation_items(self) -> List[Dict[str, Any]]:
+        """Returns a list of synthetic function_call_output items for any pending tool calls.
+        
+        This is used to reconcile the conversation state if the stream is aborted
+        after the model has generated a tool call but before it was completed.
+        """
+        snapshot = self.__current_snapshot
+        if not snapshot:
+            return []
+
+        items: List[Dict[str, Any]] = []
+        for output in snapshot.output:
+            if output.type == "function_call":
+                # We consider it "pending" if we haven't received a corresponding result yet.
+                # In the Responses API, results are submitted in a separate turn or turn-part.
+                # If we are in the middle of a stream that generated a function_call,
+                # and the stream is aborted, we should mark it as incomplete.
+                items.append({
+                    "type": "function_call_output",
+                    "call_id": output.call_id,
+                    "name": output.name,
+                    "status": "incomplete",
+                    "output": "aborted"
+                })
+        return items
+