wehub-resource-sync
1146 lines
51 KiB
Python
1146 lines
51 KiB
Python
"""AcpExecutor: drive *any* agent that speaks the Agent Client Protocol (ACP).
|
|
|
|
ACP (agentclientprotocol.com) is an open, editor-agnostic protocol: a JSON-RPC
|
|
2.0 conversation over newline-delimited JSON on a subprocess's stdin/stdout. Its
|
|
whole premise is that the *client* need not know which agent it drives — Goose
|
|
(``goose acp``), Qwen Code (``qwen --acp``), Gemini CLI
|
|
(``gemini --experimental-acp``), Zed's Claude Code bridge
|
|
(``@zed-industries/claude-code-acp``) and any in-house agent all speak the same
|
|
wire.
|
|
|
|
This executor is the **generic** counterpart to the vendor-specific
|
|
:class:`~omnigent.inner.goose_executor.GooseExecutor` /
|
|
:class:`~omnigent.inner.qwen_executor.QwenExecutor`: it spawns whatever command a
|
|
user configured (:class:`AcpAgentConfig.command`) and speaks ACP against it. The
|
|
handful of things those two hardcode become config knobs here:
|
|
|
|
* ``command`` — the argv to launch (``shlex``-split; never a shell).
|
|
* ``session_id_mode`` — ``"server"`` (agent assigns the id, Goose-style) or
|
|
``"client"`` (we generate it, Qwen-style).
|
|
* ``send_model_in_session_new`` / ``model`` — send a non-standard ``model`` field
|
|
in ``session/new`` (Qwen accepts it; most agents take
|
|
the model from their own config / the command's flags).
|
|
|
|
Protocol flow (identical for every ACP agent):
|
|
1. ``initialize`` — handshake; learn ``agentCapabilities`` (image support).
|
|
2. ``session/new`` — create/adopt a session id + ``cwd`` + ``mcpServers``.
|
|
3. ``session/prompt`` — send a user turn; consume streaming ``session/update``
|
|
notifications (``agent_message_chunk``, ``agent_thought_chunk``,
|
|
``tool_call`` / ``tool_call_update``), answer any server-initiated
|
|
``session/request_permission`` / ``fs/*`` requests, then read the final
|
|
response (``stopReason`` + optional ``usage``).
|
|
4. Re-use the same session id for later turns (the agent retains context).
|
|
|
|
The agent runs its own agent loop, tool execution, context window and compaction
|
|
internally. This executor translates the ACP event stream into Omnigent
|
|
:class:`ExecutorEvent`s and routes the agent's permission requests through
|
|
Omnigent's TOOL_CALL policy + human-consent elicitation.
|
|
|
|
Vs. the Goose executor this generalizes, it additionally: renders the agent's
|
|
tool calls as Omnigent tool cards (``tool_call`` → ``ToolCallRequest``,
|
|
``tool_call_update`` → ``ToolCallComplete``), forwards reasoning
|
|
(``agent_thought_chunk`` → ``ReasoningChunk``), and honors interrupts via the ACP
|
|
``session/cancel`` notification.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import secrets
|
|
import shlex
|
|
from collections.abc import AsyncIterator
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from omnigent.inner._acp_omnigent_mcp import OmnigentAcpMcp
|
|
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
|
|
from omnigent.inner.executor import (
|
|
Executor,
|
|
ExecutorConfig,
|
|
ExecutorError,
|
|
ExecutorEvent,
|
|
Message,
|
|
ReasoningChunk,
|
|
TextChunk,
|
|
ToolCallComplete,
|
|
ToolCallRequest,
|
|
ToolCallStatus,
|
|
TurnComplete,
|
|
)
|
|
from omnigent.inner.os_env import OSEnvironment, create_os_environment
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ACP error code an agent maps to a filesystem "not found" (ENOENT) when a
|
|
# delegated ``fs/read_text_file`` misses — the reference ACP client lib special-
|
|
# cases exactly this code. Any other code surfaces raw.
|
|
_ACP_RESOURCE_NOT_FOUND_CODE = -32002
|
|
|
|
# ACP protocol constants (JSON-RPC 2.0 method names).
|
|
_AGENT_METHOD_INITIALIZE = "initialize"
|
|
_AGENT_METHOD_SESSION_NEW = "session/new"
|
|
_AGENT_METHOD_SESSION_PROMPT = "session/prompt"
|
|
|
|
# Notification sent *from* the agent to the client (streaming progress).
|
|
_CLIENT_NOTIFICATION_SESSION_UPDATE = "session/update"
|
|
# Notification sent *from* the client to the agent to abort the current turn.
|
|
_CLIENT_NOTIFICATION_SESSION_CANCEL = "session/cancel"
|
|
|
|
# Server-initiated request methods (agent → client).
|
|
_AGENT_REQUEST_REQUEST_PERMISSION = "session/request_permission"
|
|
|
|
# session/update.update.sessionUpdate discriminator values we map.
|
|
_UPDATE_AGENT_MESSAGE_CHUNK = "agent_message_chunk"
|
|
_UPDATE_AGENT_THOUGHT_CHUNK = "agent_thought_chunk"
|
|
_UPDATE_TOOL_CALL = "tool_call"
|
|
_UPDATE_TOOL_CALL_UPDATE = "tool_call_update"
|
|
_UPDATE_USAGE = "usage_update"
|
|
|
|
# ACP tool-call lifecycle statuses (the terminal ones close a tool card).
|
|
_TOOL_STATUS_COMPLETED = "completed"
|
|
_TOOL_STATUS_FAILED = "failed"
|
|
|
|
# Idle (time-without-progress) timeouts in seconds.
|
|
_PROMPT_TIMEOUT_SECONDS = 300.0
|
|
_INIT_TIMEOUT_SECONDS = 30.0
|
|
|
|
# ACP protocol version this executor targets (matches Goose 1.38 / Qwen).
|
|
_PROTOCOL_VERSION = 1
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AcpAgentConfig:
|
|
"""Identity of the ACP agent this executor drives.
|
|
|
|
:param command: The command to launch, e.g. ``"gemini --experimental-acp"``.
|
|
Split with :func:`shlex.split` into an argv and exec'd directly (never
|
|
via a shell), so quoting works but ``$VAR`` / pipes / redirects do not.
|
|
:param name: Human label for logs / elicitation cards (e.g. ``"Gemini CLI"``).
|
|
:param model: Optional model id. Only sent to the agent when
|
|
:attr:`send_model_in_session_new` is set; otherwise inert (the agent
|
|
takes its model from its own config or from flags in ``command``).
|
|
:param session_id_mode: ``"server"`` — the agent assigns the session id and
|
|
we adopt it (Goose); ``"client"`` — we generate the id and send it
|
|
(Qwen). Defaults to ``"server"``, the ACP-idiomatic shape.
|
|
:param send_model_in_session_new: Send a non-standard ``model`` field in
|
|
``session/new``. Off by default because a strict agent may reject unknown
|
|
params; enable per-agent for Qwen-shaped agents that honor it.
|
|
:param omnigent_mcp: Expose Omnigent's builtin tools to the agent via
|
|
``session/new.mcpServers`` (the shared ``serve-mcp`` relay). On by
|
|
default; the global ``OMNIGENT_ACP_MCP=0`` kill switch also disables it.
|
|
"""
|
|
|
|
command: str
|
|
name: str = "ACP agent"
|
|
model: str | None = None
|
|
session_id_mode: str = "server"
|
|
send_model_in_session_new: bool = False
|
|
omnigent_mcp: bool = True
|
|
|
|
|
|
class _AcpRequestError(Exception):
|
|
"""A handler failure to return as a JSON-RPC error on a server request.
|
|
|
|
Carries the JSON-RPC ``code`` / ``message`` so the dispatch in
|
|
:meth:`AcpExecutor._respond_to_agent_request` can build the error reply
|
|
without each handler assembling the wire envelope itself.
|
|
"""
|
|
|
|
def __init__(self, code: int, message: str) -> None:
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
|
|
|
|
def _looks_like_missing_file(message: str) -> bool:
|
|
"""Heuristic: does an os_env error message indicate a missing path?
|
|
|
|
The os_env helper returns failures as ``{"error": "<str>"}`` rather than
|
|
typed exceptions, so the message text is the only signal that a read missed
|
|
because the file is absent. Used to map onto the ENOENT code so the model
|
|
sees "file not found".
|
|
"""
|
|
lowered = message.lower()
|
|
return (
|
|
"no such file" in lowered
|
|
or "errno 2" in lowered
|
|
or "not found" in lowered
|
|
or "does not exist" in lowered
|
|
)
|
|
|
|
|
|
def _inline_text_file_data(file_data: Any) -> str: # type: ignore[explicit-any]
|
|
"""Decode a text ``input_file`` ``file_data`` data URI into inline text.
|
|
|
|
``input_file`` blocks may carry a ``data:<mime>;base64,<payload>`` URI. Text
|
|
files are decoded so the model sees their content; binary files (PDF, images)
|
|
can't be inlined and return ``""``. A bare, non-data-URI string is treated as
|
|
already-inline text.
|
|
"""
|
|
if not isinstance(file_data, str) or not file_data:
|
|
return ""
|
|
if not file_data.startswith("data:"):
|
|
return file_data
|
|
try:
|
|
import base64
|
|
|
|
meta, b64 = file_data.split(",", 1)
|
|
mime = meta.split(";")[0].replace("data:", "")
|
|
if not mime.startswith("text/"):
|
|
return ""
|
|
return base64.b64decode(b64).decode("utf-8", errors="replace")
|
|
except Exception: # noqa: BLE001 — best-effort; never break a turn on a bad URI
|
|
return ""
|
|
|
|
|
|
def _parse_image_data_uri(data_uri: Any) -> tuple[str, str] | None: # type: ignore[explicit-any]
|
|
"""Split an ``image/*`` ``data:`` URI into ``(mime_type, base64_payload)``.
|
|
|
|
Returns ``None`` for anything that isn't an inline ``image/*`` data URI
|
|
(external URLs are never fetched — SSRF).
|
|
"""
|
|
if not isinstance(data_uri, str) or not data_uri.startswith("data:"):
|
|
return None
|
|
try:
|
|
meta, payload = data_uri.split(",", 1)
|
|
except ValueError:
|
|
return None
|
|
mime = meta.split(";")[0].replace("data:", "")
|
|
if not mime.startswith("image/") or not payload:
|
|
return None
|
|
return mime, payload
|
|
|
|
|
|
class AcpExecutor(Executor):
|
|
"""Executor that drives any ACP agent over JSON-RPC 2.0 on stdio."""
|
|
|
|
def __init__(
|
|
self,
|
|
config: AcpAgentConfig,
|
|
cwd: str | None = None,
|
|
os_env: OSEnvSpec | None = None,
|
|
) -> None:
|
|
"""Initialize the generic ACP executor.
|
|
|
|
:param config: The agent to drive (command + protocol knobs).
|
|
:param cwd: Working directory for the agent subprocess. ``None`` inherits
|
|
the caller's cwd.
|
|
:param os_env: Environment / sandbox spec. When its ``sandbox`` is not
|
|
``"none"``, the whole agent process tree is wrapped in the platform
|
|
sandbox (bwrap/seatbelt) at spawn — see :meth:`_sandbox_launch_path`.
|
|
"""
|
|
self._config = config
|
|
self._cwd = cwd or os.getcwd()
|
|
self._os_env = os_env
|
|
# Advertise ``clientCapabilities.fs`` so the agent delegates file
|
|
# reads/writes back to us (executed through the Omnigent OSEnvironment,
|
|
# which enforces the spec's sandbox read/write roots). Enabled only when
|
|
# an os_env is configured and it isn't a ``fork`` env — a forked env
|
|
# operates on a *copied* tree whose path diverges from the agent's cwd.
|
|
self._fs_delegation: bool = os_env is not None and not bool(getattr(os_env, "fork", False))
|
|
self._os_environment: OSEnvironment | None = None
|
|
|
|
# Parsed argv; the first token is the binary we resolve / sandbox.
|
|
self._argv: list[str] = shlex.split(config.command)
|
|
if not self._argv:
|
|
raise ValueError("AcpAgentConfig.command is empty")
|
|
|
|
self._proc: asyncio.subprocess.Process | None = None # type: ignore[name-defined]
|
|
self._queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() # type: ignore[explicit-any]
|
|
self._reader_task: asyncio.Task[None] | None = None
|
|
self._stderr_task: asyncio.Task[None] | None = None
|
|
# Serializes stdin writes: run_turn (prompt / request replies) and the
|
|
# adapter's interrupt_session() write from different tasks.
|
|
self._write_lock = asyncio.Lock()
|
|
|
|
self._rpc_id: int = 0
|
|
self._pending: dict[int, asyncio.Future[dict[str, Any]]] = {} # type: ignore[explicit-any]
|
|
|
|
self._session_id: str | None = None
|
|
self._initialized: bool = False
|
|
self._image_supported: bool = False
|
|
self._system_prompt_sent: bool = False
|
|
|
|
# ACP toolCallId → tool name, so a later tool_call_update can close the
|
|
# right tool card with the name from the originating tool_call.
|
|
self._tool_names: dict[str, str] = {}
|
|
|
|
# Context-window size (tokens) reported via ``usage_update``; surfaced by
|
|
# :meth:`max_context_tokens` so the UI context meter fills.
|
|
self._context_window: int | None = None
|
|
|
|
# Bridges the ExecutorAdapter installs (by attribute) so the agent's
|
|
# mid-turn ``session/request_permission`` routes through Omnigent's
|
|
# TOOL_CALL policy + human-consent elicitation. ``None`` → no bridge
|
|
# wired (standalone / unit tests) → permission falls back to allow.
|
|
self._policy_evaluator: Any | None = None # type: ignore[explicit-any]
|
|
self._elicitation_handler: Any | None = None # type: ignore[explicit-any]
|
|
# Adapter-injected tool-execution bridge (the same ``_tool_executor``
|
|
# attribute the SDK harnesses use); backs the Omnigent MCP relay.
|
|
self._tool_executor: Any | None = None # type: ignore[explicit-any]
|
|
|
|
# Omnigent-tool MCP bridge — exposes builtin tools to the agent via
|
|
# session/new.mcpServers (lazily started at first session; torn down in
|
|
# :meth:`close`). ``_omnigent_tools`` is captured each turn for the relay.
|
|
self._mcp = OmnigentAcpMcp(label=config.name)
|
|
self._omnigent_tools: list[Any] = [] # type: ignore[explicit-any]
|
|
|
|
# ------------------------------------------------------------------
|
|
# Low-level ACP transport
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _start_process(self) -> None:
|
|
"""Start the configured ACP agent as an asyncio subprocess.
|
|
|
|
The StreamReader limit is raised to 16 MiB so a large ``session/new``
|
|
response or tool-output line can't hit the default 64 KiB per-line cap.
|
|
"""
|
|
# Reset handshake state: this may be a restart after the previous
|
|
# subprocess died. ``_initialized`` is a one-way latch.
|
|
self._initialized = False
|
|
self._image_supported = False
|
|
env = os.environ.copy()
|
|
launch_path, argv = self._sandbox_launch(tuple(env.keys()))
|
|
_STREAM_LIMIT = 16 * 1024 * 1024
|
|
self._proc = await asyncio.create_subprocess_exec(
|
|
launch_path,
|
|
*argv,
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env=env,
|
|
cwd=self._cwd,
|
|
limit=_STREAM_LIMIT,
|
|
)
|
|
self._reader_task = asyncio.create_task(self._read_stdout())
|
|
self._stderr_task = asyncio.create_task(self._read_stderr())
|
|
|
|
def _sandbox_launch(self, spawn_env_names: tuple[str, ...]) -> tuple[str, list[str]]:
|
|
"""Return ``(launch_path, argv)`` — sandbox launcher or the bare binary.
|
|
|
|
When ``os_env.sandbox`` requests confinement, wraps the agent binary in
|
|
the platform sandbox so its whole process tree runs confined to the
|
|
spec's read/write roots. Falls back to the bare binary (never blocks
|
|
startup) when no sandbox is requested or the backend is unavailable.
|
|
|
|
ponytail: a sandboxed *generic* agent gets only its binary dir (read),
|
|
the cwd, and ``/tmp`` (write) — we can't know an arbitrary agent's config
|
|
dir. An agent that must write its own config under a sandbox needs
|
|
``sandbox: none`` (the default) for now; per-agent write roots is a
|
|
documented follow-up.
|
|
"""
|
|
binary = self._argv[0]
|
|
rest = self._argv[1:]
|
|
os_env = self._os_env
|
|
if os_env is None:
|
|
return binary, rest
|
|
sandbox_spec = os_env.sandbox or OSEnvSandboxSpec()
|
|
if sandbox_spec.type == "none":
|
|
return binary, rest
|
|
try:
|
|
from .sandbox import (
|
|
create_exec_launcher,
|
|
resolve_sandbox,
|
|
with_additional_read_roots,
|
|
with_additional_write_roots,
|
|
with_spawn_env_allowlist,
|
|
)
|
|
|
|
cwd = Path(self._cwd or os.getcwd()).resolve(strict=False)
|
|
sandbox = resolve_sandbox(os_env, cwd)
|
|
if not sandbox.active:
|
|
return binary, rest
|
|
resolved_bin = Path(binary)
|
|
if resolved_bin.parent != Path(".") and resolved_bin.exists():
|
|
sandbox = with_additional_read_roots(sandbox, [resolved_bin.resolve().parent])
|
|
sandbox = with_additional_write_roots(sandbox, [Path("/tmp")])
|
|
sandbox = with_spawn_env_allowlist(sandbox, spawn_env_names)
|
|
return create_exec_launcher(binary, sandbox), rest
|
|
except (OSError, ImportError, NotImplementedError) as exc:
|
|
logger.warning(
|
|
"Could not apply sandbox for ACP agent %s; running unsandboxed: %s",
|
|
self._config.name,
|
|
exc,
|
|
)
|
|
return binary, rest
|
|
|
|
async def _read_stderr(self) -> None:
|
|
"""Continuously drain the agent's stderr, logging each line at debug.
|
|
|
|
Prevents a chatty CLI from filling the OS pipe buffer (~64 KiB) and
|
|
stalling the turn.
|
|
"""
|
|
assert self._proc and self._proc.stderr
|
|
try:
|
|
while True:
|
|
raw_line = await self._proc.stderr.readline()
|
|
if not raw_line:
|
|
break
|
|
line = raw_line.decode("utf-8", errors="replace").rstrip()
|
|
if line:
|
|
logger.debug("acp[%s] stderr: %s", self._config.name, line)
|
|
except asyncio.CancelledError:
|
|
# Expected: close() cancels this reader task on teardown.
|
|
pass
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.debug("acp[%s] stderr reader stopped: %s", self._config.name, exc)
|
|
|
|
async def _read_stdout(self) -> None:
|
|
"""Continuously read NDJSON lines from the agent's stdout.
|
|
|
|
Responses (``id`` + no ``method``) resolve the matching ``_pending``
|
|
future; notifications and server-initiated requests go on ``_queue`` for
|
|
``run_turn`` to consume.
|
|
"""
|
|
assert self._proc and self._proc.stdout
|
|
try:
|
|
while True:
|
|
raw_line = await self._proc.stdout.readline()
|
|
if not raw_line:
|
|
# EOF — the subprocess exited. Wake in-flight futures so
|
|
# run_turn fails fast instead of blocking until idle timeout.
|
|
for fut in self._pending.values():
|
|
if not fut.done():
|
|
fut.set_exception(EOFError("ACP subprocess closed stdout"))
|
|
break
|
|
line = raw_line.decode("utf-8", errors="replace").strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
msg: dict[str, Any] = json.loads(line) # type: ignore[explicit-any]
|
|
except json.JSONDecodeError:
|
|
logger.debug(
|
|
"acp[%s]: non-JSON stdout line: %r", self._config.name, line[:200]
|
|
)
|
|
continue
|
|
|
|
msg_id = msg.get("id")
|
|
# Match a response by "id + no method": the agent's own requests
|
|
# (session/request_permission) also carry an id, so the method
|
|
# check prevents a colliding request from mis-resolving our future.
|
|
if msg_id is not None and "method" not in msg and msg_id in self._pending:
|
|
fut = self._pending.pop(msg_id)
|
|
if not fut.done():
|
|
fut.set_result(msg)
|
|
else:
|
|
await self._queue.put(msg)
|
|
except (asyncio.CancelledError, EOFError):
|
|
# Expected during shutdown / after the subprocess closes stdout.
|
|
pass
|
|
except Exception as exc:
|
|
logger.exception("acp[%s] stdout reader error: %s", self._config.name, exc)
|
|
for fut in self._pending.values():
|
|
if not fut.done():
|
|
fut.set_exception(exc)
|
|
await self._queue.put({"type": "error", "message": str(exc)})
|
|
|
|
async def _send(self, msg: dict[str, Any]) -> None: # type: ignore[explicit-any]
|
|
"""Write one newline-terminated JSON message to the agent's stdin."""
|
|
assert self._proc and self._proc.stdin
|
|
encoded = (json.dumps(msg) + "\n").encode("utf-8")
|
|
async with self._write_lock:
|
|
self._proc.stdin.write(encoded)
|
|
await self._proc.stdin.drain()
|
|
|
|
async def _rpc(
|
|
self,
|
|
method: str,
|
|
params: dict[str, Any], # type: ignore[explicit-any]
|
|
timeout: float = _INIT_TIMEOUT_SECONDS,
|
|
) -> dict[str, Any]: # type: ignore[explicit-any]
|
|
"""Send a JSON-RPC 2.0 request and await its response."""
|
|
self._rpc_id += 1
|
|
req_id = self._rpc_id
|
|
loop = asyncio.get_event_loop()
|
|
fut: asyncio.Future[dict[str, Any]] = loop.create_future() # type: ignore[explicit-any]
|
|
self._pending[req_id] = fut
|
|
|
|
await self._send({"jsonrpc": "2.0", "id": req_id, "method": method, "params": params})
|
|
try:
|
|
return await asyncio.wait_for(fut, timeout=timeout)
|
|
except asyncio.TimeoutError:
|
|
self._pending.pop(req_id, None)
|
|
raise
|
|
|
|
# ------------------------------------------------------------------
|
|
# ACP handshake
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _ensure_initialized(self) -> None:
|
|
"""Perform the ``initialize`` handshake if not already done."""
|
|
if self._initialized:
|
|
return
|
|
resp = await self._rpc(
|
|
_AGENT_METHOD_INITIALIZE,
|
|
{
|
|
"protocolVersion": _PROTOCOL_VERSION,
|
|
"clientInfo": {"name": "omnigent", "version": "1.0"},
|
|
"clientCapabilities": {
|
|
"fs": {
|
|
"readTextFile": self._fs_delegation,
|
|
"writeTextFile": self._fs_delegation,
|
|
},
|
|
"terminal": False,
|
|
},
|
|
},
|
|
timeout=_INIT_TIMEOUT_SECONDS,
|
|
)
|
|
if "error" in resp:
|
|
raise RuntimeError(
|
|
f"ACP initialize failed: {resp['error'].get('message', resp['error'])}"
|
|
)
|
|
prompt_caps = (
|
|
(resp.get("result") or {}).get("agentCapabilities", {}).get("promptCapabilities", {})
|
|
)
|
|
self._image_supported = bool(prompt_caps.get("image"))
|
|
self._initialized = True
|
|
|
|
async def _ensure_session(self) -> str:
|
|
"""Create (or reuse) an ACP session, returning the session id.
|
|
|
|
In ``server`` mode we send only ``cwd`` + ``mcpServers`` and adopt the id
|
|
the agent returns. In ``client`` mode we generate the id and send it.
|
|
``mcpServers`` carries Omnigent's builtin tools (via the shared serve-mcp
|
|
relay) unless disabled — see :class:`OmnigentAcpMcp`.
|
|
"""
|
|
if self._session_id is not None:
|
|
return self._session_id
|
|
|
|
mcp_servers = self._mcp.session_new_servers(
|
|
tools=self._omnigent_tools,
|
|
tool_executor=getattr(self, "_tool_executor", None),
|
|
loop=asyncio.get_event_loop(),
|
|
enabled=self._config.omnigent_mcp,
|
|
)
|
|
params: dict[str, Any] = {"cwd": self._cwd, "mcpServers": mcp_servers} # type: ignore[explicit-any]
|
|
client_id: str | None = None
|
|
if self._config.session_id_mode == "client":
|
|
client_id = secrets.token_urlsafe(16)
|
|
params["sessionId"] = client_id
|
|
if self._config.send_model_in_session_new and self._config.model:
|
|
params["model"] = self._config.model
|
|
|
|
resp = await self._rpc(_AGENT_METHOD_SESSION_NEW, params, timeout=_INIT_TIMEOUT_SECONDS)
|
|
if "error" in resp:
|
|
raise RuntimeError(
|
|
f"ACP session/new failed: {resp['error'].get('message', resp['error'])}"
|
|
)
|
|
result = resp.get("result", {})
|
|
server_session_id = result.get("sessionId") if isinstance(result, dict) else None
|
|
session_id = server_session_id or client_id
|
|
if not session_id:
|
|
raise RuntimeError(
|
|
"ACP session/new response missing sessionId: " + json.dumps(resp)[:200]
|
|
)
|
|
self._session_id = session_id
|
|
return self._session_id
|
|
|
|
# ------------------------------------------------------------------
|
|
# Server-initiated requests (agent → client)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _respond_to_agent_request(self, request: dict[str, Any]) -> None: # type: ignore[explicit-any]
|
|
"""Answer a server-initiated ACP request from the agent.
|
|
|
|
- ``session/request_permission`` — decide via Omnigent's TOOL_CALL policy
|
|
+ human-consent elicitation (:meth:`_decide_permission`), then select
|
|
the matching allow/reject option. NOT a blind approve.
|
|
- ``fs/read_text_file`` / ``fs/write_text_file`` — when fs delegation is
|
|
advertised, execute through the Omnigent OSEnvironment so the spec's
|
|
sandbox read/write roots are enforced. Off → never arrive.
|
|
- anything else — reply with JSON-RPC ``method not found`` so the agent
|
|
fails loudly rather than acting on empty data.
|
|
"""
|
|
req_id = request.get("id")
|
|
method = request.get("method", "")
|
|
params = request.get("params", {}) or {}
|
|
logger.debug("acp[%s] agent request: method=%s id=%s", self._config.name, method, req_id)
|
|
|
|
result: dict[str, Any] | None = None # type: ignore[explicit-any]
|
|
error: dict[str, Any] | None = None # type: ignore[explicit-any]
|
|
try:
|
|
if method == _AGENT_REQUEST_REQUEST_PERMISSION:
|
|
allow = await self._decide_permission(params)
|
|
result = self._permission_outcome(params, allow=allow)
|
|
elif method == "fs/read_text_file" and self._fs_delegation:
|
|
result = await self._handle_fs_read(params)
|
|
elif method == "fs/write_text_file" and self._fs_delegation:
|
|
result = await self._handle_fs_write(params)
|
|
else:
|
|
error = {
|
|
"code": -32601,
|
|
"message": f"omnigent: unsupported ACP request method {method!r}",
|
|
}
|
|
except _AcpRequestError as exc:
|
|
error = {"code": exc.code, "message": exc.message}
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.debug("acp[%s] agent request %s failed: %s", self._config.name, method, exc)
|
|
error = {"code": -32603, "message": f"{method} failed: {exc}"}
|
|
|
|
reply: dict[str, Any] = {"jsonrpc": "2.0", "id": req_id} # type: ignore[explicit-any]
|
|
if error is not None:
|
|
reply["error"] = error
|
|
else:
|
|
reply["result"] = result
|
|
await self._send(reply)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Filesystem delegation (agent → client, when fs capability advertised)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _ensure_os_environment(self) -> OSEnvironment:
|
|
"""Lazily create the OSEnvironment backing fs delegation."""
|
|
if self._os_environment is None:
|
|
env = create_os_environment(self._os_env)
|
|
if env is None:
|
|
raise _AcpRequestError(-32603, "omnigent: no os_env for fs delegation")
|
|
self._os_environment = env
|
|
return self._os_environment
|
|
|
|
async def _handle_fs_read(self, params: dict[str, Any]) -> dict[str, Any]: # type: ignore[explicit-any]
|
|
"""Serve an ACP ``fs/read_text_file`` by reading through the OSEnvironment.
|
|
|
|
ACP params ``{path, line?, limit?}`` (1-based start line, max line count;
|
|
both optional → whole file) map onto :meth:`OSEnvironment.read`.
|
|
"""
|
|
path = params.get("path")
|
|
if not isinstance(path, str) or not path:
|
|
raise _AcpRequestError(-32602, "fs/read_text_file requires a string 'path'")
|
|
line = params.get("line")
|
|
limit = params.get("limit")
|
|
offset = line if isinstance(line, int) and line >= 1 else 1
|
|
read_limit = limit if isinstance(limit, int) and limit >= 1 else None
|
|
|
|
env = await self._ensure_os_environment()
|
|
result = await env.read(path, offset=offset, limit=read_limit)
|
|
if "error" in result:
|
|
message = str(result["error"])
|
|
code = _ACP_RESOURCE_NOT_FOUND_CODE if _looks_like_missing_file(message) else -32603
|
|
raise _AcpRequestError(code, message)
|
|
if result.get("encoding") != "utf-8":
|
|
raise _AcpRequestError(-32603, f"{path}: not a UTF-8 text file")
|
|
return {"content": result.get("content", "")}
|
|
|
|
async def _handle_fs_write(self, params: dict[str, Any]) -> dict[str, Any]: # type: ignore[explicit-any]
|
|
"""Serve an ACP ``fs/write_text_file`` by writing through the OSEnvironment.
|
|
|
|
ACP params ``{path, content}``; the write goes through the helper so the
|
|
spec's sandbox write roots are enforced at the Python layer.
|
|
"""
|
|
path = params.get("path")
|
|
content = params.get("content")
|
|
if not isinstance(path, str) or not path:
|
|
raise _AcpRequestError(-32602, "fs/write_text_file requires a string 'path'")
|
|
if not isinstance(content, str):
|
|
raise _AcpRequestError(-32602, "fs/write_text_file requires string 'content'")
|
|
|
|
env = await self._ensure_os_environment()
|
|
result = await env.write(path, content)
|
|
if "error" in result:
|
|
raise _AcpRequestError(-32603, str(result["error"]))
|
|
return {}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Permission (session/request_permission) → policy + elicitation
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _extract_tool_call(params: dict[str, Any]) -> tuple[str, dict[str, Any]]: # type: ignore[explicit-any]
|
|
"""Pull ``(tool_name, tool_input)`` from a ``session/request_permission``.
|
|
|
|
ACP's ``toolCall`` carries a human ``title`` (e.g. ``"shell"``), a
|
|
``kind`` (e.g. ``"execute"``), and a ``rawInput`` dict. We prefer the
|
|
title, else the kind. (Vendor-specific ``_meta`` tool names — e.g.
|
|
Goose's ``_meta.goose.toolCall.toolName`` — are not read here; ``title``
|
|
is the portable name every ACP agent supplies.)
|
|
"""
|
|
tool_call = params.get("toolCall") or {}
|
|
name = tool_call.get("title") or tool_call.get("kind") or "tool"
|
|
args = tool_call.get("rawInput")
|
|
if not isinstance(args, dict):
|
|
args = {}
|
|
return str(name), args
|
|
|
|
async def _decide_permission(self, params: dict[str, Any]) -> bool: # type: ignore[explicit-any]
|
|
"""Decide allow/deny for a permission request — policy then elicitation.
|
|
|
|
1. **TOOL_CALL policy** (:attr:`_policy_evaluator`): a hard
|
|
``POLICY_ACTION_DENY`` denies; ``POLICY_ACTION_ASK`` defers to
|
|
elicitation (and **fails closed** when no handler is wired);
|
|
``ALLOW`` / unspecified falls through.
|
|
2. **Human-consent elicitation** (:attr:`_elicitation_handler`): routes
|
|
to the user via a web approval card and returns their accept/deny.
|
|
|
|
When neither bridge is wired (standalone / unit tests), falls back to
|
|
allow so direct use of the executor isn't blocked. In normal runner
|
|
operation the adapter installs both, so destructive actions are gated.
|
|
"""
|
|
tool_name, tool_input = self._extract_tool_call(params)
|
|
handler = getattr(self, "_elicitation_handler", None)
|
|
policy_eval = getattr(self, "_policy_evaluator", None)
|
|
|
|
if policy_eval is not None:
|
|
action: str | None
|
|
try:
|
|
verdict = await policy_eval(
|
|
"PHASE_TOOL_CALL", {"name": tool_name, "arguments": tool_input}
|
|
)
|
|
action = getattr(verdict, "action", None)
|
|
except Exception as exc: # noqa: BLE001 — fail open to elicitation
|
|
logger.warning("acp TOOL_CALL policy eval failed for %s: %s", tool_name, exc)
|
|
action = None
|
|
if action == "POLICY_ACTION_DENY":
|
|
logger.info("acp permission denied by policy: tool=%s", tool_name)
|
|
return False
|
|
if action == "POLICY_ACTION_ASK":
|
|
if handler is None:
|
|
logger.warning(
|
|
"acp TOOL_CALL policy ASK with no elicitation handler; denying tool=%s",
|
|
tool_name,
|
|
)
|
|
return False
|
|
allowed = bool(await handler(tool_name, tool_input))
|
|
logger.info(
|
|
"acp permission %s by user (policy ASK): tool=%s",
|
|
"allowed" if allowed else "denied",
|
|
tool_name,
|
|
)
|
|
return allowed
|
|
# ALLOW / UNSPECIFIED / unknown → fall through to elicitation.
|
|
|
|
if handler is not None:
|
|
allowed = bool(await handler(tool_name, tool_input))
|
|
logger.info(
|
|
"acp permission %s by user: tool=%s",
|
|
"allowed" if allowed else "denied",
|
|
tool_name,
|
|
)
|
|
return allowed
|
|
|
|
logger.debug("acp permission allowed (no policy/elicitation wired): tool=%s", tool_name)
|
|
return True
|
|
|
|
@staticmethod
|
|
def _permission_outcome( # type: ignore[explicit-any]
|
|
params: dict[str, Any], *, allow: bool
|
|
) -> dict[str, Any]:
|
|
"""Map an allow/deny decision to an ACP permission ``outcome``.
|
|
|
|
On allow, prefer a once-scoped grant (``allow_once``) over
|
|
``allow_always`` so we never persist a blanket "always allow". On deny,
|
|
pick a ``reject_*`` option, or ``cancelled`` when none is offered. The
|
|
agent's options carry both ``optionId`` and ``kind`` (e.g. ``allow_once``).
|
|
"""
|
|
options = [o for o in (params.get("options") or []) if isinstance(o, dict)]
|
|
|
|
def _pick(*kinds: str) -> dict[str, Any] | None: # type: ignore[explicit-any]
|
|
for kind in kinds:
|
|
for opt in options:
|
|
if opt.get("kind") == kind:
|
|
return opt
|
|
return None
|
|
|
|
if allow:
|
|
chosen = _pick("allow_once", "allow_always") or next(
|
|
(o for o in options if "allow" in str(o.get("kind", ""))), None
|
|
)
|
|
else:
|
|
chosen = _pick("reject_once", "reject_always") or next(
|
|
(o for o in options if "reject" in str(o.get("kind", ""))), None
|
|
)
|
|
if chosen is None:
|
|
return {"outcome": {"outcome": "cancelled"}}
|
|
return {"outcome": {"outcome": "selected", "optionId": chosen.get("optionId")}}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Prompt building
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _image_blocks_from_content(content: Any) -> list[dict[str, Any]]: # type: ignore[explicit-any]
|
|
"""Build ACP ``image`` prompt blocks from a message's ``input_image`` blocks."""
|
|
out: list[dict[str, Any]] = [] # type: ignore[explicit-any]
|
|
if not isinstance(content, list):
|
|
return out
|
|
for block in content:
|
|
if not isinstance(block, dict) or block.get("type") != "input_image":
|
|
continue
|
|
parsed = _parse_image_data_uri(block.get("image_url") or block.get("file_data"))
|
|
if parsed:
|
|
mime, data = parsed
|
|
out.append({"type": "image", "mimeType": mime, "data": data})
|
|
return out
|
|
|
|
@staticmethod
|
|
def _text_from_blocks(
|
|
blocks: list[Any],
|
|
*,
|
|
emit_image_marker: bool = False, # type: ignore[explicit-any]
|
|
) -> str:
|
|
"""Extract prompt text from a Responses-API content-block list.
|
|
|
|
ACP's ``session/prompt`` text part is plain text, so each block is folded:
|
|
``input_text``/``output_text``/``text`` verbatim; ``input_file`` inlined
|
|
(fenced) when the runner resolved it to a text data URI, else a marker;
|
|
``input_image`` as a marker only when *emit_image_marker* is set.
|
|
"""
|
|
parts: list[str] = []
|
|
for block in blocks:
|
|
if not isinstance(block, dict):
|
|
continue
|
|
btype = block.get("type")
|
|
if btype in ("input_text", "output_text", "text"):
|
|
text = block.get("text")
|
|
if isinstance(text, str) and text:
|
|
parts.append(text)
|
|
elif btype == "input_file":
|
|
name = block.get("filename") or block.get("file_id") or "file"
|
|
inlined = _inline_text_file_data(block.get("file_data"))
|
|
if inlined:
|
|
parts.append(
|
|
f"--- attached file: {name} ---\n{inlined}\n--- end of {name} ---"
|
|
)
|
|
else:
|
|
parts.append(f"[attached file: {name}]")
|
|
elif btype == "input_image" and emit_image_marker:
|
|
name = block.get("filename") or block.get("file_id")
|
|
parts.append(f"[attached image: {name}]" if name else "[attached image]")
|
|
return "\n".join(parts)
|
|
|
|
@classmethod
|
|
def _history_prefix(cls, prior: list[Any]) -> str: # type: ignore[explicit-any]
|
|
"""Serialize prior conversation turns into a text prefix.
|
|
|
|
On a *fresh* ACP session (the first turn of a newly spawned/respawned
|
|
process, or after a session reset) the agent holds none of the earlier
|
|
conversation. Since :meth:`run_turn` normally sends only the latest user
|
|
turn, we'd lose everything before the switch. Replaying the transcript as
|
|
a labeled ``role: content`` block restores that context.
|
|
"""
|
|
lines = ["Conversation so far:"]
|
|
for msg in prior:
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
role = str(msg.get("role", "user")).replace("_", " ")
|
|
raw = msg.get("content")
|
|
if raw is None:
|
|
content = ""
|
|
elif isinstance(raw, str):
|
|
content = raw
|
|
elif isinstance(raw, list):
|
|
content = cls._text_from_blocks(raw, emit_image_marker=True)
|
|
else:
|
|
content = json.dumps(raw, ensure_ascii=True)
|
|
lines.append(f"{role}: {content}")
|
|
lines.append("")
|
|
lines.append(
|
|
"Respond to the latest user message, using the conversation above as context."
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Executor interface
|
|
# ------------------------------------------------------------------
|
|
|
|
def handles_tools_internally(self) -> bool:
|
|
"""True — the ACP agent runs its own tool loop.
|
|
|
|
The Session must NOT re-execute the ``ToolCallRequest`` /
|
|
``ToolCallComplete`` events we emit from ``tool_call`` updates; they are
|
|
informational (they render tool cards showing what the agent did).
|
|
"""
|
|
return True
|
|
|
|
def supports_streaming(self) -> bool:
|
|
return True
|
|
|
|
def max_context_tokens(self) -> int | None:
|
|
"""Return the agent's reported context-window size, if observed yet."""
|
|
return self._context_window
|
|
|
|
@staticmethod
|
|
def _usage_from_result(result: dict[str, Any]) -> dict[str, Any] | None: # type: ignore[explicit-any]
|
|
"""Map an agent's final ``result.usage`` to Omnigent's usage keys.
|
|
|
|
ACP does not standardize usage, but agents that report it (Goose) use
|
|
``{totalTokens, inputTokens, outputTokens}``; Omnigent's
|
|
``TurnComplete.usage`` uses ``{input_tokens, output_tokens, total_tokens}``.
|
|
Absent → ``None`` (usage simply isn't shown for agents that don't report).
|
|
"""
|
|
usage = result.get("usage")
|
|
if not isinstance(usage, dict):
|
|
return None
|
|
out: dict[str, Any] = {}
|
|
if isinstance(usage.get("inputTokens"), int):
|
|
out["input_tokens"] = usage["inputTokens"]
|
|
if isinstance(usage.get("outputTokens"), int):
|
|
out["output_tokens"] = usage["outputTokens"]
|
|
if isinstance(usage.get("totalTokens"), int):
|
|
out["total_tokens"] = usage["totalTokens"]
|
|
return out or None
|
|
|
|
def _handle_session_update(self, update: dict[str, Any]) -> list[ExecutorEvent]: # type: ignore[explicit-any]
|
|
"""Translate one ``session/update`` payload into ExecutorEvents.
|
|
|
|
Returns the events to yield (usually 0 or 1). Side effects: records the
|
|
context window from ``usage_update`` and tracks tool-call names so a
|
|
later ``tool_call_update`` can close the right card.
|
|
"""
|
|
update_type = update.get("sessionUpdate", "")
|
|
events: list[ExecutorEvent] = []
|
|
|
|
if update_type == _UPDATE_AGENT_MESSAGE_CHUNK:
|
|
content = update.get("content", {})
|
|
text = content.get("text", "") if isinstance(content, dict) else ""
|
|
if text:
|
|
events.append(TextChunk(text=text))
|
|
elif update_type == _UPDATE_AGENT_THOUGHT_CHUNK:
|
|
content = update.get("content", {})
|
|
text = content.get("text", "") if isinstance(content, dict) else ""
|
|
if text:
|
|
events.append(ReasoningChunk(delta=text, event_type="reasoning_text"))
|
|
elif update_type == _UPDATE_TOOL_CALL:
|
|
call_id = update.get("toolCallId")
|
|
name = update.get("title") or update.get("kind") or "tool"
|
|
raw_input = update.get("rawInput")
|
|
args = raw_input if isinstance(raw_input, dict) else {}
|
|
if isinstance(call_id, str) and call_id:
|
|
self._tool_names[call_id] = str(name)
|
|
events.append(
|
|
ToolCallRequest(name=str(name), args=args, metadata={"call_id": call_id})
|
|
)
|
|
elif update_type == _UPDATE_TOOL_CALL_UPDATE:
|
|
call_id = update.get("toolCallId")
|
|
status = update.get("status")
|
|
if isinstance(call_id, str) and status in (
|
|
_TOOL_STATUS_COMPLETED,
|
|
_TOOL_STATUS_FAILED,
|
|
):
|
|
name = self._tool_names.pop(call_id, "tool")
|
|
events.append(
|
|
ToolCallComplete(
|
|
name=name,
|
|
status=(
|
|
ToolCallStatus.SUCCESS
|
|
if status == _TOOL_STATUS_COMPLETED
|
|
else ToolCallStatus.ERROR
|
|
),
|
|
result=update.get("content") or update.get("rawOutput"),
|
|
metadata={"call_id": call_id},
|
|
)
|
|
)
|
|
elif update_type == _UPDATE_USAGE:
|
|
size = update.get("size")
|
|
if isinstance(size, int) and size > 0:
|
|
self._context_window = size
|
|
|
|
return events
|
|
|
|
async def run_turn(
|
|
self,
|
|
messages: list[Message],
|
|
tools: list[Any], # type: ignore[explicit-any]
|
|
system_prompt: str,
|
|
config: ExecutorConfig | None = None, # noqa: ARG002 — unused; required by the interface
|
|
) -> AsyncIterator[ExecutorEvent]:
|
|
"""Run one turn of the agent loop via ACP.
|
|
|
|
Sends ``session/prompt`` and yields streaming events (text, reasoning,
|
|
tool-call cards) as the agent works, answering any
|
|
``session/request_permission`` mid-turn, until the final response
|
|
(``stopReason``) arrives — then yields ``TurnComplete`` with usage.
|
|
|
|
``tools`` (Omnigent's builtin tool schemas) are captured for the Omnigent
|
|
MCP relay set up at ``session/new`` — the agent still runs its OWN tools.
|
|
"""
|
|
# Captured before the (lazy) session so the MCP relay can advertise them.
|
|
self._omnigent_tools = tools or []
|
|
try:
|
|
if self._proc is None or self._proc.returncode is not None:
|
|
await self._start_process()
|
|
await self._ensure_initialized()
|
|
session_id = await self._ensure_session()
|
|
except Exception as exc: # noqa: BLE001
|
|
yield ExecutorError(message=str(exc), retryable=False)
|
|
return
|
|
|
|
# A fresh ACP session holds no prior context. Captured before the latch
|
|
# flips so we know whether to replay history into this turn.
|
|
fresh_session = not self._system_prompt_sent
|
|
|
|
user_text = ""
|
|
image_blocks: list[dict[str, Any]] = [] # type: ignore[explicit-any]
|
|
latest_user_idx: int | None = None
|
|
for idx in range(len(messages) - 1, -1, -1):
|
|
msg = messages[idx]
|
|
role = msg.get("role", "") if isinstance(msg, dict) else ""
|
|
if role == "user":
|
|
latest_user_idx = idx
|
|
content = msg.get("content", "") if isinstance(msg, dict) else ""
|
|
if isinstance(content, str):
|
|
user_text = content
|
|
elif isinstance(content, list):
|
|
if self._image_supported:
|
|
image_blocks = self._image_blocks_from_content(content)
|
|
user_text = self._text_from_blocks(
|
|
content, emit_image_marker=not self._image_supported
|
|
)
|
|
break
|
|
|
|
# On a fresh session, replay prior conversation so a model switch (which
|
|
# respawns the subprocess) or a session reset doesn't drop the thread.
|
|
if fresh_session and latest_user_idx is not None and latest_user_idx > 0:
|
|
history_prefix = self._history_prefix(messages[:latest_user_idx])
|
|
user_text = f"{history_prefix}\n\nuser: {user_text}" if user_text else history_prefix
|
|
|
|
# ACP has no system-prompt field, so fold it into the first turn.
|
|
if fresh_session:
|
|
if system_prompt:
|
|
user_text = f"{system_prompt}\n\n{user_text}" if user_text else system_prompt
|
|
self._system_prompt_sent = True
|
|
|
|
prompt_blocks: list[dict[str, Any]] = [] # type: ignore[explicit-any]
|
|
if user_text or not image_blocks:
|
|
prompt_blocks.append({"type": "text", "text": user_text})
|
|
prompt_blocks.extend(image_blocks)
|
|
|
|
# Drain stale items from a prior turn; answer any leftover server request.
|
|
while not self._queue.empty():
|
|
try:
|
|
stale = self._queue.get_nowait()
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
if isinstance(stale, dict) and stale.get("id") is not None and stale.get("method"):
|
|
await self._respond_to_agent_request(stale)
|
|
|
|
self._rpc_id += 1
|
|
req_id = self._rpc_id
|
|
loop = asyncio.get_event_loop()
|
|
fut: asyncio.Future[dict[str, Any]] = loop.create_future() # type: ignore[explicit-any]
|
|
self._pending[req_id] = fut
|
|
|
|
await self._send(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"method": _AGENT_METHOD_SESSION_PROMPT,
|
|
"params": {"sessionId": session_id, "prompt": prompt_blocks},
|
|
}
|
|
)
|
|
|
|
deadline = loop.time() + _PROMPT_TIMEOUT_SECONDS
|
|
accumulated_text: list[str] = []
|
|
|
|
while True:
|
|
remaining = deadline - loop.time()
|
|
if remaining <= 0:
|
|
yield ExecutorError(message="Timeout waiting for ACP response", retryable=True)
|
|
return
|
|
|
|
# Complete only once the future is resolved AND the queue is drained,
|
|
# so trailing chunks aren't truncated.
|
|
if fut.done() and self._queue.empty():
|
|
try:
|
|
response = fut.result()
|
|
except Exception as exc: # noqa: BLE001
|
|
self._session_id = None
|
|
self._system_prompt_sent = False
|
|
yield ExecutorError(message=f"ACP process error: {exc}", retryable=True)
|
|
return
|
|
if "error" in response:
|
|
error_msg = response["error"].get("message", "Unknown ACP error")
|
|
if "Session not found" in error_msg:
|
|
self._session_id = None
|
|
self._system_prompt_sent = False
|
|
yield ExecutorError(message=error_msg, retryable=True)
|
|
return
|
|
result = response.get("result", {}) if isinstance(response, dict) else {}
|
|
usage = self._usage_from_result(result) if isinstance(result, dict) else None
|
|
yield TurnComplete(response="".join(accumulated_text), usage=usage)
|
|
return
|
|
|
|
try:
|
|
notification = await asyncio.wait_for(
|
|
self._queue.get(), timeout=min(remaining, 2.0)
|
|
)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
|
|
method = notification.get("method", "")
|
|
params = notification.get("params", {})
|
|
|
|
if method == _CLIENT_NOTIFICATION_SESSION_UPDATE:
|
|
update = params.get("update", {})
|
|
for event in self._handle_session_update(update):
|
|
if isinstance(event, TextChunk):
|
|
accumulated_text.append(event.text)
|
|
yield event
|
|
elif notification.get("id") is not None and notification.get("method"):
|
|
# Server-initiated request (session/request_permission / fs/*):
|
|
# routes through policy + elicitation. Blocks while the human decides.
|
|
await self._respond_to_agent_request(notification)
|
|
|
|
# Inbound message = progress; reset the idle deadline.
|
|
deadline = loop.time() + _PROMPT_TIMEOUT_SECONDS
|
|
|
|
async def interrupt_session(self, session_key: str) -> bool: # noqa: ARG002 — one ACP session per process
|
|
"""Abort the running turn via the ACP ``session/cancel`` notification.
|
|
|
|
The agent responds by ending the in-flight ``session/prompt`` with a
|
|
``cancelled`` stop reason, which the ``run_turn`` loop then surfaces.
|
|
Best-effort: a no-op (returns ``False``) if there's no live session.
|
|
"""
|
|
if self._proc is None or self._proc.returncode is not None or self._session_id is None:
|
|
return False
|
|
try:
|
|
await self._send(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"method": _CLIENT_NOTIFICATION_SESSION_CANCEL,
|
|
"params": {"sessionId": self._session_id},
|
|
}
|
|
)
|
|
return True
|
|
except Exception as exc: # noqa: BLE001 — interrupt is best-effort
|
|
logger.debug("acp[%s] session/cancel failed: %s", self._config.name, exc)
|
|
return False
|
|
|
|
async def close_session(self, session_key: str) -> None:
|
|
"""Close a named session (no-op; the ACP session is per-process)."""
|
|
|
|
async def close(self) -> None:
|
|
"""Terminate the agent subprocess and clean up."""
|
|
# Tear down the Omnigent MCP relay HTTP server + its bridge dir first.
|
|
with contextlib.suppress(Exception):
|
|
self._mcp.close()
|
|
if self._reader_task:
|
|
self._reader_task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._reader_task
|
|
self._reader_task = None
|
|
if self._stderr_task:
|
|
self._stderr_task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._stderr_task
|
|
self._stderr_task = None
|
|
if self._os_environment is not None:
|
|
with contextlib.suppress(Exception):
|
|
self._os_environment.close()
|
|
self._os_environment = None
|
|
if self._proc:
|
|
with contextlib.suppress(Exception):
|
|
self._proc.stdin.close() # type: ignore[union-attr]
|
|
try:
|
|
self._proc.terminate()
|
|
await asyncio.wait_for(self._proc.wait(), timeout=5)
|
|
except Exception: # noqa: BLE001
|
|
with contextlib.suppress(Exception):
|
|
self._proc.kill()
|
|
finally:
|
|
self._proc = None
|