Python: replay an empty local_shell_call with its matching shell output

python/packages/openai/agent_framework_openai/_chat_client.py

@@ -2825,15 +2825,63 @@ def _shell_item_to_contents(self, item: Any, local_shell_tool_name: str | None)
                     )
                 )
             else:
+                provider_status = _provider_status(getattr(item, "status", None))
+                empty_local_call = bool(
+                    local_shell_tool_name
+                    and local_call_id
+                    and local_call_item_id
+                    and isinstance(raw_local_command_parts, list)
+                    and not raw_local_command_parts
+                )
                 contents.append(
                     Content.from_shell_tool_call(
                         call_id=local_call_id,
                         commands=[local_command] if local_command else [],
                         timeout_ms=getattr(getattr(item, "action", None), "timeout_ms", None),
-                        status=_provider_status(getattr(item, "status", None)),
-                        raw_representation=item,
+                        status=provider_status,
+                        # An empty ``local_shell_call`` has no replayable local form: its provider
+                        # payload only validates as ``local_shell_call``, which this client never
+                        # emits on replay. A call that ran no commands is equivalent to a
+                        # ``shell_call`` with an empty action, so record that shape and pair it with
+                        # the terminal output below instead of leaving an orphaned shell call.
+                        # The item id also goes into the serialized metadata, because
+                        # ``raw_representation`` does not survive ``Content.to_dict()``.
+                        additional_properties=(
+                            _openai_additional_properties(
+                                {
+                                    _OPENAI_SHELL_OUTPUT_TYPE_NAMESPACE_KEY: OPENAI_SHELL_OUTPUT_TYPE_SHELL_CALL,
+                                    _OPENAI_LOCAL_SHELL_CALL_ITEM_ID_NAMESPACE_KEY: local_call_item_id,
+                                },
+                                legacy={
+                                    OPENAI_SHELL_OUTPUT_TYPE_KEY: OPENAI_SHELL_OUTPUT_TYPE_SHELL_CALL,
+                                    OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY: local_call_item_id,
+                                },
+                            )
+                            if empty_local_call
+                            else None
+                        ),
+                        raw_representation=(
+                            {
+                                "id": local_call_item_id,
+                                "type": "shell_call",
+                                "call_id": local_call_id,
+                                "status": provider_status or "completed",
+                                "environment": {"type": "local"},
+                                "action": {"commands": []},
+                            }
+                            if empty_local_call
+                            else item
+                        ),
                     )
                 )
+                if empty_local_call:
+                    contents.append(
+                        Content.from_shell_tool_result(
+                            call_id=local_call_id,
+                            outputs=[],
+                            additional_properties={_LOCAL_TOOL_RESULT_KEY: True},
+                        )
+                    )
         elif item_type == "shell_call_output":
             shell_output_call_id = getattr(item, "call_id", None) or ""
             shell_outputs: list[Content] = []
@@ -3072,6 +3120,8 @@ def _prepare_shell_transcript_item_for_openai(
                 output["max_output_length"] = content.max_output_length
             return output
         payload = cls._serialize_provider_payload(content.raw_representation)
+        if payload is None:
+            payload = cls._rebuild_shell_call_payload(content, expected_type)
         if isinstance(payload, Mapping):
             typed_payload = cast("Mapping[str, Any]", payload)
             raw_call_id = typed_payload.get("call_id")
@@ -3092,6 +3142,32 @@ def _prepare_shell_transcript_item_for_openai(
             "Use service-side continuation or preserve the original provider response item."
         )
 
+    @classmethod
+    def _rebuild_shell_call_payload(cls, content: Content, expected_type: str) -> dict[str, Any] | None:
+        """Rebuild a shell call payload from the fields ``Content.to_dict()`` keeps.
+
+        ``raw_representation`` is dropped on serialization, so a call restored from a checkpoint
+        only carries its typed fields. An empty local shell call has everything needed to rebuild
+        the wire payload, which is what keeps a restored conversation replayable.
+        """
+        if expected_type != "shell_call" or not content.call_id:
+            return None
+        item_id = _get_openai_metadata(
+            content.additional_properties,
+            _OPENAI_LOCAL_SHELL_CALL_ITEM_ID_NAMESPACE_KEY,
+            OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY,
+        )
+        payload: dict[str, Any] = {
+            "type": "shell_call",
+            "call_id": content.call_id,
+            "status": content.status or "completed",
+            "action": {"commands": list(content.commands or [])},
+        }
+        if isinstance(item_id, str) and item_id:
+            payload["id"] = item_id
+            payload["environment"] = {"type": "local"}
+        return payload
+
     @staticmethod
     def _serialize_provider_payload(value: Any) -> Any:
         """Convert OpenAI SDK objects into JSON-serializable Python values."""

python/packages/openai/tests/openai/test_openai_chat_client.py

@@ -68,6 +68,7 @@
     ResponseOutputTextAnnotationAddedEvent,
     ResponseStreamEvent,
 )
+from openai.types.responses.response_output_item import LocalShellCall, LocalShellCallAction
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem, Summary
 from openai.types.responses.response_reasoning_summary_text_delta_event import (
     ResponseReasoningSummaryTextDeltaEvent,
@@ -2979,6 +2980,42 @@ def test_empty_local_shell_call_replays_with_empty_output(
     assert shell_outputs == [expected_output]
 
 
+@pytest.mark.parametrize("round_trip", [False, True])
+@pytest.mark.parametrize("service_storage", [False, True])
+def test_empty_local_shell_call_item_replays_with_empty_output(round_trip: bool, service_storage: bool) -> None:
+    """An empty provider ``local_shell_call`` must replay as a matched empty shell pair."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    item = LocalShellCall(
+        id="sh_local_empty",
+        call_id="call_local_empty",
+        status="completed",
+        type="local_shell_call",
+        action=LocalShellCallAction(command=[], env={}, type="exec"),
+    )
+
+    contents = client._shell_item_to_contents(item, "run_shell")
+
+    assert [content.type for content in contents] == ["shell_tool_call", "shell_tool_result"]
+    message = Message("assistant", contents)
+    if round_trip:
+        # Round-trip the whole message, not only the result: ``Content.to_dict()`` drops
+        # ``raw_representation``, so this is the shape a persisted conversation takes.
+        message = Message.from_json(message.to_json())
+    replay = client._prepare_messages_for_openai(
+        [message, Message("user", ["Continue"])],
+        request_uses_service_side_storage=service_storage,
+    )
+
+    shell_calls = [entry for entry in replay if entry["type"] == "shell_call"]
+    assert len(shell_calls) == (0 if service_storage else 1)
+    if not service_storage:
+        assert shell_calls[0]["call_id"] == "call_local_empty"
+        assert shell_calls[0]["action"]["commands"] == []
+    assert [entry for entry in replay if entry["type"] == "shell_call_output"] == [
+        {"type": "shell_call_output", "call_id": "call_local_empty", "output": []}
+    ]
+
+
 @pytest.mark.parametrize("streaming", [False, True])
 def test_empty_local_shell_call_parsing_emits_pair_without_execution(streaming: bool) -> None:
     client = OpenAIChatClient(model="test-model", api_key="test-key")