diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py
index a673e059c348..b0df44401b63 100644
--- a/agent/copilot_acp_client.py
+++ b/agent/copilot_acp_client.py
@@ -11,6 +11,7 @@
import json
import os
import queue
+import re
import shlex
import subprocess
import threading
@@ -23,6 +24,14 @@
ACP_MARKER_BASE_URL = "acp://copilot"
_DEFAULT_TIMEOUT_SECONDS = 900.0
+_TOOL_CALL_BLOCK_RE = re.compile(r"\s*(\{.*?\})\s*", re.DOTALL)
+_TOOL_CALL_JSON_RE = re.compile(r"\{\s*\"id\"\s*:\s*\"[^\"]+\"\s*,\s*\"type\"\s*:\s*\"function\"\s*,\s*\"function\"\s*:\s*\{.*?\}\s*\}", re.DOTALL)
+_TERMINAL_COMMAND_HINT_RE = re.compile(
+ r"(?:ferramenta\s+terminal.*?comando|execute\s+o\s+comando|run\s+the\s+command)\s*[:\-]?\s*[`'\"]([^`'\"\n]+)[`'\"]",
+ re.IGNORECASE | re.DOTALL,
+)
+_VISION_URL_RE = re.compile(r"https?://\S+", re.IGNORECASE)
+
def _resolve_command() -> str:
return (
@@ -50,15 +59,50 @@ def _jsonrpc_error(message_id: Any, code: int, message: str) -> dict[str, Any]:
}
-def _format_messages_as_prompt(messages: list[dict[str, Any]], model: str | None = None) -> str:
+def _format_messages_as_prompt(
+ messages: list[dict[str, Any]],
+ model: str | None = None,
+ tools: list[dict[str, Any]] | None = None,
+ tool_choice: Any = None,
+) -> str:
sections: list[str] = [
"You are being used as the active ACP agent backend for Hermes.",
- "Use your own ACP capabilities and respond directly in natural language.",
- "Do not emit OpenAI tool-call JSON.",
+ "Use ACP capabilities to complete tasks.",
+ "IMPORTANT: If you take an action with a tool, you MUST output tool calls using {...} blocks with JSON exactly in OpenAI function-call shape.",
+ "If no tool is needed, answer normally.",
]
if model:
sections.append(f"Hermes requested model hint: {model}")
+ if isinstance(tools, list) and tools:
+ tool_specs: list[dict[str, Any]] = []
+ for t in tools:
+ if not isinstance(t, dict):
+ continue
+ fn = t.get("function") or {}
+ if not isinstance(fn, dict):
+ continue
+ name = fn.get("name")
+ if not isinstance(name, str) or not name.strip():
+ continue
+ tool_specs.append(
+ {
+ "name": name.strip(),
+ "description": fn.get("description", ""),
+ "parameters": fn.get("parameters", {}),
+ }
+ )
+ if tool_specs:
+ sections.append(
+ "Available tools (OpenAI function schema). "
+ "When using a tool, emit ONLY {...} with one JSON object "
+ "containing id/type/function{name,arguments}. arguments must be a JSON string.\n"
+ + json.dumps(tool_specs, ensure_ascii=False)
+ )
+
+ if tool_choice is not None:
+ sections.append(f"Tool choice hint: {json.dumps(tool_choice, ensure_ascii=False)}")
+
transcript: list[str] = []
for message in messages:
if not isinstance(message, dict):
@@ -114,6 +158,174 @@ def _render_message_content(content: Any) -> str:
return str(content).strip()
+def _extract_tool_calls_from_text(text: str) -> tuple[list[SimpleNamespace], str]:
+ if not isinstance(text, str) or not text.strip():
+ return [], ""
+
+ extracted: list[SimpleNamespace] = []
+ consumed_spans: list[tuple[int, int]] = []
+
+ def _try_add_tool_call(raw_json: str) -> None:
+ try:
+ obj = json.loads(raw_json)
+ except Exception:
+ return
+ if not isinstance(obj, dict):
+ return
+ fn = obj.get("function")
+ if not isinstance(fn, dict):
+ return
+ fn_name = fn.get("name")
+ if not isinstance(fn_name, str) or not fn_name.strip():
+ return
+ fn_args = fn.get("arguments", "{}")
+ if not isinstance(fn_args, str):
+ fn_args = json.dumps(fn_args, ensure_ascii=False)
+ call_id = obj.get("id")
+ if not isinstance(call_id, str) or not call_id.strip():
+ call_id = f"acp_call_{len(extracted)+1}"
+
+ extracted.append(
+ SimpleNamespace(
+ id=call_id,
+ call_id=call_id,
+ response_item_id=None,
+ type="function",
+ function=SimpleNamespace(name=fn_name.strip(), arguments=fn_args),
+ )
+ )
+
+ for m in _TOOL_CALL_BLOCK_RE.finditer(text):
+ raw = m.group(1)
+ _try_add_tool_call(raw)
+ consumed_spans.append((m.start(), m.end()))
+
+ for m in _TOOL_CALL_JSON_RE.finditer(text):
+ raw = m.group(0)
+ _try_add_tool_call(raw)
+ consumed_spans.append((m.start(), m.end()))
+
+ if not consumed_spans:
+ return extracted, text.strip()
+
+ consumed_spans.sort()
+ merged: list[tuple[int, int]] = []
+ for start, end in consumed_spans:
+ if not merged or start > merged[-1][1]:
+ merged.append((start, end))
+ else:
+ merged[-1] = (merged[-1][0], max(merged[-1][1], end))
+
+ parts: list[str] = []
+ cursor = 0
+ for start, end in merged:
+ if cursor < start:
+ parts.append(text[cursor:start])
+ cursor = max(cursor, end)
+ if cursor < len(text):
+ parts.append(text[cursor:])
+
+ cleaned = "\n".join(p.strip() for p in parts if p and p.strip()).strip()
+ return extracted, cleaned
+
+
+def _latest_user_text(messages: list[dict[str, Any]] | None) -> str:
+ if not isinstance(messages, list):
+ return ""
+ for msg in reversed(messages):
+ if not isinstance(msg, dict):
+ continue
+ role = str(msg.get("role") or "").strip().lower()
+ if role != "user":
+ continue
+ return _render_message_content(msg.get("content"))
+ return ""
+
+
+def _heuristic_fallback_tool_calls(
+ messages: list[dict[str, Any]] | None,
+ response_text: str,
+ tools: list[dict[str, Any]] | None = None,
+) -> list[SimpleNamespace]:
+ """Best-effort fallback when ACP returns prose instead of structured tool calls."""
+ user_text = _latest_user_text(messages)
+ if not user_text:
+ return []
+
+ allowed_tool_names: set[str] = set()
+ if isinstance(tools, list):
+ for t in tools:
+ if not isinstance(t, dict):
+ continue
+ fn = t.get("function") or {}
+ if not isinstance(fn, dict):
+ continue
+ name = fn.get("name")
+ if isinstance(name, str) and name.strip():
+ allowed_tool_names.add(name.strip())
+
+ # Additional guard: only trigger when assistant prose indicates intent/action.
+ lower_resp = (response_text or "").lower()
+ intent_like = (not lower_resp) or any(
+ token in lower_resp
+ for token in (
+ "vou executar",
+ "i will run",
+ "executar o comando",
+ "running",
+ "não tenho acesso",
+ "nao tenho acesso",
+ "não está disponível",
+ "not available",
+ )
+ )
+ if not intent_like:
+ return []
+
+ # Terminal fallback
+ m = _TERMINAL_COMMAND_HINT_RE.search(user_text)
+ if m:
+ command = (m.group(1) or "").strip()
+ if command and ((not allowed_tool_names) or ("terminal" in allowed_tool_names)):
+ args = json.dumps({"command": command}, ensure_ascii=False)
+ call_id = "acp_fallback_terminal_1"
+ return [
+ SimpleNamespace(
+ id=call_id,
+ call_id=call_id,
+ response_item_id=None,
+ type="function",
+ function=SimpleNamespace(name="terminal", arguments=args),
+ )
+ ]
+
+ # Vision fallback
+ asks_vision = (
+ "vision_analyze" in user_text.lower()
+ or "visão" in user_text.lower()
+ or "visao" in user_text.lower()
+ or "imagem" in user_text.lower()
+ or "image" in user_text.lower()
+ )
+ if asks_vision:
+ url_match = _VISION_URL_RE.search(user_text)
+ if url_match:
+ image_url = url_match.group(0).strip().rstrip('.,)')
+ args = json.dumps({"image_url": image_url, "user_prompt": "Descreva em 1 frase."}, ensure_ascii=False)
+ call_id = "acp_fallback_vision_1"
+ return [
+ SimpleNamespace(
+ id=call_id,
+ call_id=call_id,
+ response_item_id=None,
+ type="function",
+ function=SimpleNamespace(name="vision_analyze", arguments=args),
+ )
+ ]
+
+ return []
+
+
def _ensure_path_within_cwd(path_text: str, cwd: str) -> Path:
candidate = Path(path_text)
if not candidate.is_absolute():
@@ -190,14 +402,25 @@ def _create_chat_completion(
model: str | None = None,
messages: list[dict[str, Any]] | None = None,
timeout: float | None = None,
+ tools: list[dict[str, Any]] | None = None,
+ tool_choice: Any = None,
**_: Any,
) -> Any:
- prompt_text = _format_messages_as_prompt(messages or [], model=model)
+ prompt_text = _format_messages_as_prompt(
+ messages or [],
+ model=model,
+ tools=tools,
+ tool_choice=tool_choice,
+ )
response_text, reasoning_text = self._run_prompt(
prompt_text,
timeout_seconds=float(timeout or _DEFAULT_TIMEOUT_SECONDS),
)
+ tool_calls, cleaned_text = _extract_tool_calls_from_text(response_text)
+ if not tool_calls:
+ tool_calls = _heuristic_fallback_tool_calls(messages, response_text, tools=tools)
+
usage = SimpleNamespace(
prompt_tokens=0,
completion_tokens=0,
@@ -205,13 +428,14 @@ def _create_chat_completion(
prompt_tokens_details=SimpleNamespace(cached_tokens=0),
)
assistant_message = SimpleNamespace(
- content=response_text,
- tool_calls=[],
+ content=cleaned_text,
+ tool_calls=tool_calls,
reasoning=reasoning_text or None,
reasoning_content=reasoning_text or None,
reasoning_details=None,
)
- choice = SimpleNamespace(message=assistant_message, finish_reason="stop")
+ finish_reason = "tool_calls" if tool_calls else "stop"
+ choice = SimpleNamespace(message=assistant_message, finish_reason=finish_reason)
return SimpleNamespace(
choices=[choice],
usage=usage,
diff --git a/run_agent.py b/run_agent.py
index af40344df5e9..ac987bfe6400 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -694,6 +694,7 @@ def __init__(
is_native_anthropic = self.api_mode == "anthropic_messages"
self._use_prompt_caching = (is_openrouter and is_claude) or is_native_anthropic
self._cache_ttl = "5m" # Default 5-minute TTL (1.25x write cost)
+ self._acp_hint_notice_emitted = False
# Iteration budget pressure: warn the LLM as it approaches max_iterations.
# Warnings are injected into the last tool result JSON (not as separate
@@ -3853,11 +3854,24 @@ def _create_request_openai_client(self, *, reason: str) -> Any:
primary_client = self._ensure_primary_openai_client(reason=reason)
if isinstance(primary_client, Mock):
return primary_client
+ if self.provider == "copilot-acp" or str(getattr(self, "base_url", "") or "").startswith("acp://copilot"):
+ return primary_client
with self._openai_client_lock():
request_kwargs = dict(self._client_kwargs)
return self._create_openai_client(request_kwargs, reason=reason, shared=False)
def _close_request_openai_client(self, client: Any, *, reason: str) -> None:
+ if client is getattr(self, "client", None):
+ is_copilot_acp = self.provider == "copilot-acp" or str(getattr(self, "base_url", "") or "").startswith("acp://copilot")
+ if is_copilot_acp and reason in {"interrupt_abort", "stream_interrupt_abort"}:
+ self._close_openai_client(client, reason=reason, shared=True)
+ return
+ logger.debug(
+ "Skipping request-level close for shared client (%s) %s",
+ reason,
+ self._client_log_context(),
+ )
+ return
self._close_openai_client(client, reason=reason, shared=False)
def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta: callable = None):
@@ -6770,6 +6784,14 @@ def run_conversation(
self._stream_callback = stream_callback
self._persist_user_message_idx = None
self._persist_user_message_override = persist_user_message
+ is_copilot_acp = self.provider == "copilot-acp" or str(getattr(self, "base_url", "") or "").startswith("acp://copilot")
+ if is_copilot_acp and not getattr(self, "_acp_hint_notice_emitted", False):
+ self._emit_status(
+ "ℹ️ Copilot ACP model selection is currently hint-only in this runtime. "
+ "Hermes preserves session/context and sends model/reasoning intent, but backend enforcement may differ."
+ )
+ self._acp_hint_notice_emitted = True
+
# Generate unique task_id if not provided to isolate VMs between concurrent tasks
effective_task_id = task_id or str(uuid.uuid4())
@@ -7231,7 +7253,9 @@ def _stop_spinner():
self.thinking_callback("")
_use_streaming = True
- if not self._has_stream_consumers():
+ if self.provider == "copilot-acp" or str(getattr(self, "base_url", "") or "").startswith("acp://copilot"):
+ _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
# (mocks return SimpleNamespace, not stream iterators).
diff --git a/tests/test_run_agent.py b/tests/test_run_agent.py
index a407d27a9f54..0baf9bfb965d 100644
--- a/tests/test_run_agent.py
+++ b/tests/test_run_agent.py
@@ -3007,6 +3007,99 @@ def test_empty_choices_chunk_skipped(self, agent):
assert resp.model == "gpt-4"
+class TestCopilotAcpIntegration:
+ def _setup_agent(self, agent):
+ agent._cached_system_prompt = "You are helpful."
+ agent._use_prompt_caching = False
+ agent.tool_delay = 0
+ agent.compression_enabled = False
+ agent.save_trajectories = False
+
+ def test_run_conversation_bypasses_streaming_for_copilot_acp_provider(self, agent):
+ self._setup_agent(agent)
+ agent.provider = "copilot-acp"
+ agent.base_url = "acp://copilot"
+ response = _mock_response(content="ACP answer", finish_reason="stop")
+
+ with (
+ patch.object(agent, "_persist_session"),
+ patch.object(agent, "_save_trajectory"),
+ patch.object(agent, "_cleanup_task_resources"),
+ patch.object(agent, "_interruptible_streaming_api_call", side_effect=AssertionError("streaming should not be used")),
+ patch.object(agent, "_interruptible_api_call", return_value=response) as mock_nonstream,
+ ):
+ result = agent.run_conversation("hello")
+
+ assert result["final_response"] == "ACP answer"
+ mock_nonstream.assert_called_once()
+
+ def test_run_conversation_bypasses_streaming_for_acp_base_url_marker(self, agent):
+ self._setup_agent(agent)
+ agent.provider = "custom"
+ agent.base_url = "acp://copilot"
+ response = _mock_response(content="ACP base marker", finish_reason="stop")
+
+ with (
+ patch.object(agent, "_persist_session"),
+ patch.object(agent, "_save_trajectory"),
+ patch.object(agent, "_cleanup_task_resources"),
+ patch.object(agent, "_interruptible_streaming_api_call", side_effect=AssertionError("streaming should not be used")),
+ patch.object(agent, "_interruptible_api_call", return_value=response) as mock_nonstream,
+ ):
+ result = agent.run_conversation("hello")
+
+ assert result["final_response"] == "ACP base marker"
+ mock_nonstream.assert_called_once()
+
+ def test_close_request_client_interrupt_closes_shared_acp_client(self, agent):
+ agent.provider = "copilot-acp"
+ agent.base_url = "acp://copilot"
+ shared = MagicMock()
+ agent.client = shared
+
+ with patch.object(agent, "_close_openai_client") as mock_close:
+ agent._close_request_openai_client(shared, reason="interrupt_abort")
+
+ mock_close.assert_called_once_with(shared, reason="interrupt_abort", shared=True)
+
+ def test_close_request_client_stream_interrupt_closes_shared_acp_client_by_base_url(self, agent):
+ agent.provider = "custom"
+ agent.base_url = "acp://copilot"
+ shared = MagicMock()
+ agent.client = shared
+
+ with patch.object(agent, "_close_openai_client") as mock_close:
+ agent._close_request_openai_client(shared, reason="stream_interrupt_abort")
+
+ mock_close.assert_called_once_with(shared, reason="stream_interrupt_abort", shared=True)
+
+
+class TestCopilotAcpNotice:
+ def test_run_conversation_emits_hint_only_notice_once(self, agent):
+ agent._cached_system_prompt = "You are helpful."
+ agent._use_prompt_caching = False
+ agent.tool_delay = 0
+ agent.compression_enabled = False
+ agent.save_trajectories = False
+ agent.provider = "copilot-acp"
+ agent.base_url = "acp://copilot"
+ response = _mock_response(content="done", finish_reason="stop")
+ notices = []
+
+ with (
+ patch.object(agent, "_persist_session"),
+ patch.object(agent, "_save_trajectory"),
+ patch.object(agent, "_cleanup_task_resources"),
+ patch.object(agent, "_emit_status", side_effect=lambda msg: notices.append(msg)),
+ patch.object(agent, "_interruptible_api_call", return_value=response),
+ ):
+ agent.run_conversation("hello")
+ agent.run_conversation("hello again")
+
+ assert len(notices) == 1
+ assert "hint-only" in notices[0]
+
+
# ===================================================================
# Interrupt _vprint force=True verification
# ===================================================================