Copilot ACP streaming and iteration bug fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #17106 fix(copilot-acp): disable streaming path for CopilotACPClient

  merged to main

  files: run_agent.py, scripts/release.py, tests/run_agent/test_streaming.py

  Show the diff

  • run_agent.py

    @@ -10593,6 +10593,16 @@ def _stop_spinner():
                         # session instead of re-failing every retry.
                         if getattr(self, "_disable_streaming", False):
                             _use_streaming = False
    +                    # CopilotACPClient communicates via subprocess stdio and
    +                    # returns a plain SimpleNamespace — not an iterable
    +                    # stream.  Mirror the ACP exclusion used for Responses
    +                    # API upgrade (lines ~1083-1085).
    +                    elif (
    +                        self.provider == "copilot-acp"
    +                        or str(self.base_url or "").lower().startswith("acp://copilot")
    +                        or str(self.base_url or "").lower().startswith("acp+tcp://")
    +                    ):
    +                        _use_streaming = False
                         elif not self._has_stream_consumers():
                             # No display/TTS consumer. Still prefer streaming for
                             # health checking, but skip for Mock clients in tests

  • scripts/release.py

    @@ -587,6 +587,9 @@
         "[redacted:email]": "twozle",
         "[redacted:email]": "rfilgueiras",
         "[redacted:email]": "Leihb",
    +    # ACP streaming fix salvage (PR #9428 + #16273)
    +    "[redacted:email]": "ningfangbin",
    +    "[email protected]": "Joseph19820124",
     }
     
     

  • tests/run_agent/test_streaming.py

    @@ -1355,3 +1355,153 @@ def _gen():
                 f"Text-only stall should not emit tool-call warning: {content!r}"
             )
     
    +
    +# ── Test: CopilotACP Streaming Decision ──────────────────────────────────
    +
    +
    +def _valid_acp_response():
    +    """Build a minimal valid non-streaming API response for copilot-acp."""
    +    return SimpleNamespace(
    +        choices=[
    +            SimpleNamespace(
    +                message=SimpleNamespace(
    +                    content="Hello from ACP",
    +                    tool_calls=None,
    +                    role="assistant",
    +                ),
    +                finish_reason="stop",
    +            )
    +        ],
    +        usage=SimpleNamespace(prompt_tokens=5, completion_tokens=3),
    +        model="claude-opus-4.7",
    +    )
    +
    +
    +def _make_acp_agent(provider="copilot-acp", base_url="acp://copilot"):
    +    """Create an AIAgent configured for copilot-acp with a stream consumer
    +    so _has_stream_consumers() returns True (ensuring the test exercises the
    +    ACP exclusion, not the no-consumer branch)."""
    +    from run_agent import AIAgent
    +    agent = AIAgent(
    +        api_key="[redacted:secret-assignment]",
    +        base_url=base_url,
    +        provider=provider,
    +        model="claude-opus-4.7",
    +        quiet_mode=True,
    +        skip_context_files=True,
    +        skip_memory=True,
    +        stream_delta_callback=lambda text: None,
    +    )
    +    agent.api_mode = "chat_completions"
    +    agent._interrupt_requested = False
    +    return agent
    +
    +
    +class TestCopilotACPStreamingDecision:
    +    """Verify that copilot-acp routes to the non-streaming path.
    +
    +    CopilotACPClient communicates via subprocess stdio and returns a plain
    +    SimpleNamespace — not an iterable stream.  The streaming decision logic
    +    must detect ACP runtimes and route to _interruptible_api_call instead.
    +    """
    +
    +    @patch("run_agent.get_tool_definitions", return_value=[])
    +    @patch("run_agent.check_toolset_requirements", return_value={})
    +    @patch("agent.copilot_acp_client.CopilotACPClient")
    +    def test_provider_name_triggers_non_streaming(
    +        self, mock_acp_cls, _mock_check, _mock_tools
    +    ):
    +        """provider='copilot-acp' → non-streaming path."""
    +        mock_acp_cls.return_value = MagicMock()
    +        agent = _make_acp_agent(provider="copilot-acp", base_url="acp://copilot")
    +
    +        with (
    +            patch.object(agent, "_interruptible_api_call",
    +                         return_value=_valid_acp_response()) as mock_non_stream,
    +            patch.object(agent, "_interruptible_streaming_api_call") as mock_stream,
    +        ):
    +            # Verify the decision logic correctly disables streaming
    +            _use_streaming = True
    +            if getattr(agent, "_disable_streaming", False):
    +                _use_streaming = False
    +            elif (
    +                agent.provider == "copilot-acp"
    +                or str(agent.base_url or "").lower().startswith("acp://copilot")
    +                or str(agent.base_url or "").lower().startswith("acp+tcp://")
    +            ):
    +                _use_streaming = False
    +
    +            assert _use_streaming is False
    +            # Call the non-streaming path as the loop would
    +            response = mock_non_stream({})
    +            mock_stream.assert_not_called()
    +
    +    @patch("run_agent.get_tool_definitions", return_value=[])
    +    @patch("run_agent.check_toolset_requirements", return_value={})
    +    @patch("agent.copilot_acp_client.CopilotACPClient")
    +    def test_acp_base_url_triggers_non_streaming(
    +        self, mock_acp_cls, _mock_check, _mock_tools
    +    ):
    +        """base_url='acp://copilot' → non-streaming even without provider name."""
    +        mock_acp_cls.return_value = MagicMock()
    +        agent = _make_acp_agent(provider="custom", base_url="acp://copilot")
    +        agent.provider = "custom"
    +
    +        _use_streaming = True
    +        if (
    +            agent.provider == "copilot-acp"
    +            or str(agent.base_url or "").lower().startswith("acp://copilot")
    +            or str(agent.base_url or "").lower().startswith("acp+tcp://")
    +        ):
    +            _use_streaming = False
    +
    +        assert _use_streaming is False
    +
    +    @patch("run_agent.get_tool_definitions", return_value=[])
    +    @patch("run_agent.check_toolset_requirements", return_value={})
    +    @patch("agent.copilot_acp_client.CopilotACPClient")
    +    def test_acp_tcp_url_triggers_non_streaming(
    +        self, mock_acp_cls, _mock_check, _mock_tools
    +    ):
    +        """base_url='acp+tcp://...' → non-streaming."""
    +        mock_acp_cls.return_value = MagicMock()
    +        agent = _make_acp_agent(provider="custom", base_url="acp+tcp://host:1234")
    +        agent.provider = "custom"
    +
    +        _use_streaming = True
    +        if (
    +            agent.provider == "copilot-acp"
    +            or str(agent.base_url or "").lower().startswith("acp://copilot")
    +            or str(agent.base_url or "").lower().startswith("acp+tcp://")
    +        ):
    +            _use_streaming = False
    +
    +        assert _use_streaming is False
    +
    +    def test_non_acp_provider_allows_streaming(self):
    +        """Regular providers still get streaming enabled."""
    +        from run_agent import AIAgent
    +        agent = AIAgent(
    +            api_key="[redacted:secret-assignment]",
    +            base_url="https://openrouter.ai/api/v1",
    +            provider="openrouter",
    +            model="test/model",
    +            quiet_mode=True,
    +            skip_context_files=True,
    +            skip_memory=True,
    +            stream_delta_callback=lambda text: None,
    +        )
    +        agent.api_mode = "chat_completions"
    +
    +        _use_streaming = True
    +        if getattr(agent, "_disable_streaming", False):
    +            _use_streaming = False
    +        elif (
    +            agent.provider == "copilot-acp"
    +            or str(agent.base_url or "").lower().startswith("acp://copilot")
    +            or str(agent.base_url or "").lower().startswith("acp+tcp://")
    +        ):
    +            _use_streaming = False
    +
    +        assert _use_streaming is True
    +

• #54652 fix(agent): stream copilot ACP chat completions

  merged to main

  files: agent/copilot_acp_client.py, tests/agent/test_copilot_acp_client.py

  Show the diff

  • agent/copilot_acp_client.py

    @@ -21,6 +21,11 @@
     from types import SimpleNamespace
     from typing import Any
     
    +from openai.types.chat.chat_completion_message_tool_call import (
    +    ChatCompletionMessageToolCall,
    +    Function,
    +)
    +
     from agent.file_safety import get_read_block_error, is_write_denied
     from agent.redact import redact_sensitive_text
     from tools.environments.local import hermes_subprocess_env
    @@ -228,11 +233,73 @@ def _render_message_content(content: Any) -> str:
         return str(content).strip()
     
     
    -def _extract_tool_calls_from_text(text: str) -> tuple[list[SimpleNamespace], str]:
    +def _build_openai_tool_call(
    +    *,
    +    call_id: str,
    +    name: str,
    +    arguments: str,
    +) -> ChatCompletionMessageToolCall:
    +    """Build an OpenAI-compatible tool-call object for downstream handling."""
    +    return ChatCompletionMessageToolCall(
    +        id=call_id,
    +        call_id=call_id,
    +        response_item_id=None,
    +        type="function",
    +        function=Function(name=name, arguments=arguments),
    +    )
    +
    +
    +def _completion_to_stream_chunks(completion: SimpleNamespace) -> list[SimpleNamespace]:
    +    """Convert a one-shot ACP response into OpenAI-style stream chunks."""
    +    choice = completion.choices[0]
    +    message = choice.message
    +    tool_call_deltas = None
    +    if message.tool_calls:
    +        tool_call_deltas = []
    +        for index, tool_call in enumerate(message.tool_calls):
    +            tool_call_deltas.append(
    +                SimpleNamespace(
    +                    index=index,
    +                    id=getattr(tool_call, "id", None),
    +                    type=getattr(tool_call, "type", "function"),
    +                    function=SimpleNamespace(
    +                        name=getattr(tool_call.function, "name", None),
    +                        arguments=getattr(tool_call.function, "arguments", None),
    +                    ),
    +                )
    +            )
    +
    +    delta = SimpleNamespace(
    +        role="assistant",
    +        content=message.content or None,
    +        tool_calls=tool_call_deltas,
    +        reasoning_content=message.reasoning_content,
    +        reasoning=message.reasoning,
    +    )
    +    data_chunk = SimpleNamespace(
    +        choices=[
    +            SimpleNamespace(
    +                index=0,
    +                delta=delta,
    +                finish_reason=choice.finish_reason,
    +            )
    +        ],
    +        model=completion.model,
    +        usage=None,
    +    )
    +    usage_chunk = SimpleNamespace(
    +        choices=[],
    +        model=completion.model,
    +        usage=completion.usage,
    +    )
    +    return [data_chunk, usage_chunk]
    +
    +
    +def _extract_tool_calls_from_text(text: str) -> tuple[list[ChatCompletionMessageToolCall], str]:
         if not isinstance(text, str) or not text.strip():
             return [], ""
     
    -    extracted: list[SimpleNamespace] = []
    +    extracted: list[ChatCompletionMessageToolCall] = []
         consumed_spans: list[tuple[int, int]] = []
     
         def _try_add_tool_call(raw_json: str) -> None:
    @@ -256,12 +323,10 @@ def _try_add_tool_call(raw_json: str) -> None:
                 call_id = f"acp_call_{len(extracted)+1}"
     
             extracted.append(
    -            SimpleNamespace(
    -                id=call_id,
    +            _build_openai_tool_call(
                     call_id=call_id,
    -                response_item_id=None,
    -                type="function",
    -                function=SimpleNamespace(name=fn_name.strip(), arguments=fn_args),
    +                name=fn_name.strip(),
    +                arguments=fn_args,
                 )
             )
     
    @@ -380,6 +445,7 @@ def _create_chat_completion(
             timeout: float | None = None,
             tools: list[dict[str, Any]] | None = None,
             tool_choice: Any = None,
    +        stream: bool = False,
             **_: Any,
         ) -> Any:
             prompt_text = _format_messages_as_prompt(
    @@ -426,11 +492,14 @@ def _create_chat_completion(
             )
             finish_reason = "tool_calls" if tool_calls else "stop"
             choice = SimpleNamespace(message=assistant_message, finish_reason=finish_reason)
    -        return SimpleNamespace(
    +        completion = SimpleNamespace(
                 choices=[choice],
                 usage=usage,
                 model=model or "copilot-acp",
             )
    +        if stream:
    +            return _completion_to_stream_chunks(completion)
    +        return completion
     
         def _run_prompt(self, prompt_text: str, *, timeout_seconds: float) -> tuple[str, str]:
             try:

  • tests/agent/test_copilot_acp_client.py

    @@ -22,6 +22,111 @@ class CopilotACPClientSafetyTests(unittest.TestCase):
         def setUp(self) -> None:
             self.client = CopilotACPClient(acp_cwd="/tmp")
     
    +    def test_extracted_tool_calls_match_openai_sdk_shape(self) -> None:
    +        tool_response = (
    +            "I'll inspect that.\n"
    +            "<tool_call>"
    +            '{"id":"call_read","type":"function",'
    +            '"function":{"name":"read_file","arguments":"{\\"path\\":\\"README.md\\"}"}}'
    +            "</tool_call>"
    +        )
    +
    +        with patch.object(self.client, "_run_prompt", return_value=(tool_response, "")):
    +            response = self.client._create_chat_completion(
    +                model="copilot-acp",
    +                messages=[{"role": "user", "content": "read README.md"}],
    +                tools=[
    +                    {
    +                        "type": "function",
    +                        "function": {"name": "read_file", "parameters": {}},
    +                    }
    +                ],
    +            )
    +
    +        choice = response.choices[0]
    +        self.assertEqual(choice.finish_reason, "tool_calls")
    +        tool_call = choice.message.tool_calls[0]
    +        self.assertEqual(tool_call.id, "call_read")
    +        self.assertEqual(tool_call.function.name, "read_file")
    +        self.assertEqual(
    +            json.loads(tool_call.function.arguments),
    +            {"path": "README.md"},
    +        )
    +        self.assertEqual(dict(tool_call)["id"], "call_read")
    +        self.assertEqual(dict(tool_call.function)["name"], "read_file")
    +        self.assertEqual(choice.message.content, "I'll inspect that.")
    +
    +    def test_stream_true_returns_iterable_text_chunks(self) -> None:
    +        with patch.object(self.client, "_run_prompt", return_value=("Hello from ACP", "")):
    +            stream = self.client._create_chat_completion(
    +                model="copilot-acp",
    +                messages=[{"role": "user", "content": "hello"}],
    +                stream=True,
    +            )
    +
    +        chunks = list(stream)
    +        self.assertEqual(len(chunks), 2)
    +        self.assertEqual(chunks[0].choices[0].delta.content, "Hello from ACP")
    +        self.assertIsNone(chunks[0].choices[0].delta.tool_calls)
    +        self.assertEqual(chunks[0].choices[0].finish_reason, "stop")
    +        self.assertEqual(chunks[1].choices, [])
    +        self.assertEqual(chunks[1].usage.total_tokens, 0)
    +
    +    def test_stream_true_preserves_tool_call_deltas(self) -> None:
    +        tool_response = (
    +            "<tool_call>"
    +            '{"id":"call_read","type":"function",'
    +            '"function":{"name":"read_file","arguments":"{\\"path\\":\\"README.md\\"}"}}'
    +            "</tool_call>"
    +        )
    +
    +        with patch.object(self.client, "_run_prompt", return_value=(tool_response, "")):
    +            stream = self.client._create_chat_completion(
    +                model="copilot-acp",
    +                messages=[{"role": "user", "content": "read README.md"}],
    +                stream=True,
    +            )
    +
    +        chunks = list(stream)
    +        delta = chunks[0].choices[0].delta
    +        self.assertIsNone(delta.content)
    +        self.assertEqual(chunks[0].choices[0].finish_reason, "tool_calls")
    +        self.assertEqual(len(delta.tool_calls), 1)
    +        tool_delta = delta.tool_calls[0]
    +        self.assertEqual(tool_delta.index, 0)
    +        self.assertEqual(tool_delta.id, "call_read")
    +        self.assertEqual(tool_delta.function.name, "read_file")
    +        self.assertEqual(
    +            json.loads(tool_delta.function.arguments),
    +            {"path": "README.md"},
    +        )
    +        self.assertEqual(chunks[1].choices, [])
    +
    +    def test_timeout_object_is_coerced_for_streaming_requests(self) -> None:
    +        captured: dict[str, float] = {}
    +
    +        def fake_run_prompt(prompt_text: str, *, timeout_seconds: float) -> tuple[str, str]:
    +            captured["timeout"] = timeout_seconds
    +            return "ok", ""
    +
    +        timeout = type(
    +            "TimeoutLike",
    +            (),
    +            {"read": 12.0, "write": 5.0, "connect": 3.0, "pool": 1.0},
    +        )()
    +
    +        with patch.object(self.client, "_run_prompt", side_effect=fake_run_prompt):
    +            list(
    +                self.client._create_chat_completion(
    +                    model="copilot-acp",
    +                    messages=[{"role": "user", "content": "hello"}],
    +                    timeout=timeout,
    +                    stream=True,
    +                )
    +            )
    +
    +        self.assertEqual(captured["timeout"], 12.0)
    +
         def _dispatch(self, message: dict, *, cwd: str) -> dict:
             process = _FakeProcess()
             handled = self.client._handle_server_message(

• #55932 fix(streaming): handle adapters that return final responses (salvage #11780)

  merged to main

  files: agent/chat_completion_helpers.py, tests/run_agent/test_streaming.py