0fc1ebca93
tool.completed carries is_error, but the ACP bridge dropped it and re-derived status from the result text alone, so a tool cancelled by a user interrupt (plain-text result) or one returning an error dict closed as completed. Pass the flag through close_tool_call and OR it into build_tool_complete's failed predicate; the text heuristic stays as the fallback for the step-closer path.
306 lines
13 KiB
Python
306 lines
13 KiB
Python
"""Callback factories for bridging AIAgent events to ACP notifications.
|
|
|
|
Each factory returns a callable with the signature AIAgent expects for its
|
|
callbacks. AIAgent runs in a worker thread while the event loop lives on the
|
|
main thread, so updates are pushed via ``conn.session_update()`` scheduled
|
|
thread-safely onto the loop.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from collections import deque
|
|
from typing import Any, Callable, Deque, Dict
|
|
|
|
import acp
|
|
from acp.schema import AgentPlanUpdate, PlanEntry
|
|
|
|
from .tools import (
|
|
_json_loads_maybe, build_tool_abandoned, build_tool_complete, build_tool_start, coerce_tool_args,
|
|
make_tool_call_id,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ACP plans only support pending/in_progress/completed. Cancelled tasks are kept
|
|
# as terminal entries so the client's full-list replacement doesn't drop them.
|
|
_PLAN_STATUS = {"pending": "pending", "in_progress": "in_progress", "completed": "completed", "cancelled": "completed"}
|
|
|
|
|
|
def _build_plan_update_from_todo_result(result: Any) -> AgentPlanUpdate | None:
|
|
"""Translate Hermes' todo tool result into ACP's native plan update.
|
|
|
|
Zed renders ``sessionUpdate: plan`` as its first-class task panel, so the
|
|
todo state is exposed natively rather than only as a tool-call transcript."""
|
|
if not isinstance(result, str) or not result.strip():
|
|
return None
|
|
data = _json_loads_maybe(result)
|
|
if not isinstance(data, dict) or not isinstance(data.get("todos"), list):
|
|
return None
|
|
|
|
entries: list[PlanEntry] = []
|
|
for item in data["todos"]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
content = str(item.get("content") or item.get("id") or "").strip()
|
|
if not content:
|
|
continue
|
|
raw_status = str(item.get("status") or "pending").strip()
|
|
if raw_status == "cancelled":
|
|
content = f"[cancelled] {content}"
|
|
entries.append(PlanEntry(content=content, priority="medium", status=_PLAN_STATUS.get(raw_status, "pending")))
|
|
return AgentPlanUpdate(session_update="plan", entries=entries)
|
|
|
|
|
|
def _send_update(conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop, update: Any) -> None:
|
|
"""Fire-and-forget an ACP session update from a worker thread."""
|
|
from agent.async_utils import safe_schedule_threadsafe
|
|
|
|
future = safe_schedule_threadsafe(
|
|
conn.session_update(session_id, update), loop, logger=logger, log_message="Failed to send ACP update",
|
|
)
|
|
if future is None:
|
|
return
|
|
try:
|
|
future.result(timeout=5)
|
|
except Exception:
|
|
logger.debug("Failed to send ACP update", exc_info=True)
|
|
|
|
|
|
def _upgrade_queue(tool_call_ids: Dict[str, Deque[str]], name: str) -> Deque[str] | None:
|
|
"""Fetch the per-tool FIFO of pending call IDs, upgrading a legacy bare-string entry in place."""
|
|
queue = tool_call_ids.get(name)
|
|
if isinstance(queue, str):
|
|
queue = tool_call_ids[name] = deque([queue])
|
|
return queue
|
|
|
|
|
|
def close_tool_call(
|
|
conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop, tool_call_ids: Dict[str, Deque[str]],
|
|
tool_call_meta: Dict[str, Dict[str, Any]], name: str, result: Any = None, is_error: bool = False,
|
|
) -> str | None:
|
|
"""Close the oldest open ACP tool call for ``name``; returns its id, or None when none is open."""
|
|
queue = _upgrade_queue(tool_call_ids, name)
|
|
if not queue:
|
|
return None
|
|
tc_id = queue.popleft()
|
|
meta = tool_call_meta.pop(tc_id, {})
|
|
_send_update(conn, session_id, loop, build_tool_complete(
|
|
tc_id, name, result=str(result) if result is not None else None,
|
|
function_args=meta.get("args"), snapshot=meta.get("snapshot"), is_error=is_error,
|
|
))
|
|
if not queue:
|
|
tool_call_ids.pop(name, None)
|
|
return tc_id
|
|
|
|
|
|
def flush_open_tool_calls(
|
|
conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop, tool_call_ids: Dict[str, Deque[str]],
|
|
tool_call_meta: Dict[str, Dict[str, Any]],
|
|
) -> int:
|
|
"""Close every tool call still open at the end of a turn, and report how many there were.
|
|
|
|
A tool blocked by scope, guardrail or an editor permission prompt never
|
|
projects ``tool.completed``, so without this its bubble stays ``in_progress``
|
|
forever and clients read the turn as one that never ran a tool."""
|
|
open_calls = [(name, list(queue)) for name, queue in list(tool_call_ids.items()) if queue]
|
|
flushed = 0
|
|
for name, ids in open_calls:
|
|
for tc_id in ids:
|
|
tool_call_meta.pop(tc_id, None)
|
|
_send_update(conn, session_id, loop, build_tool_abandoned(tc_id, name))
|
|
flushed += 1
|
|
tool_call_ids.pop(name, None)
|
|
if flushed:
|
|
logger.debug("Flushed %d ACP tool call(s) left open at turn end", flushed)
|
|
return flushed
|
|
|
|
|
|
def make_tool_progress_cb(
|
|
conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop, tool_call_ids: Dict[str, Deque[str]],
|
|
tool_call_meta: Dict[str, Dict[str, Any]],
|
|
edit_approval_policy_getter: Callable[[], tuple[str, str | None]] | None = None,
|
|
turn_state: Dict[str, Any] | None = None,
|
|
) -> Callable:
|
|
"""Create a ``tool_progress_callback`` for AIAgent.
|
|
|
|
Signature: ``tool_progress_callback(event_type, name, preview, args, **kwargs)``.
|
|
Emits ``ToolCallStart`` for ``tool.started`` and tracks IDs in a FIFO per tool
|
|
name so parallel same-name calls complete against the right ACP tool call.
|
|
``tool.completed`` closes that call with its own result — the step callback
|
|
only fires on the *next* step, which leaves a turn's last tools open."""
|
|
|
|
def _tool_progress(event_type: str, name: str = None, preview: str = None, args: Any = None, **kwargs) -> None:
|
|
if event_type == "tool.completed" and name:
|
|
if turn_state is not None:
|
|
turn_state["saw_completion"] = True
|
|
# The executor's verdict: a cancelled/errored tool may return plain text the heuristic misses.
|
|
close_tool_call(
|
|
conn, session_id, loop, tool_call_ids, tool_call_meta, name, kwargs.get("result"),
|
|
is_error=bool(kwargs.get("is_error")),
|
|
)
|
|
return
|
|
if event_type != "tool.started":
|
|
return
|
|
args = coerce_tool_args(args)
|
|
tc_id = make_tool_call_id()
|
|
queue = _upgrade_queue(tool_call_ids, name)
|
|
if queue is None:
|
|
queue = tool_call_ids[name] = deque()
|
|
queue.append(tc_id)
|
|
|
|
snapshot = None
|
|
if name in {"write_file", "patch", "skill_manage"}:
|
|
try:
|
|
from agent.display import capture_local_edit_snapshot
|
|
|
|
snapshot = capture_local_edit_snapshot(name, args)
|
|
except Exception:
|
|
logger.debug("Failed to capture ACP edit snapshot for %s", name, exc_info=True)
|
|
tool_call_meta[tc_id] = {"args": args, "snapshot": snapshot}
|
|
|
|
edit_diff = None
|
|
if name in {"write_file", "patch"} and edit_approval_policy_getter is not None:
|
|
try:
|
|
from acp_adapter.edit_approval import build_edit_proposal, should_auto_approve_edit
|
|
|
|
proposal = build_edit_proposal(name, args)
|
|
if proposal is not None:
|
|
policy, cwd = edit_approval_policy_getter()
|
|
if should_auto_approve_edit(proposal, policy, cwd):
|
|
edit_diff = proposal
|
|
except Exception:
|
|
logger.debug("Failed to prepare auto-approved ACP edit diff for %s", name, exc_info=True)
|
|
|
|
_send_update(conn, session_id, loop, build_tool_start(tc_id, name, args, edit_diff=edit_diff))
|
|
|
|
return _tool_progress
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Assistant message identity
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
class AssistantMessageIdAllocator:
|
|
"""Allocates stable per-message ids for streamed assistant chunks.
|
|
|
|
ACP clients group streamed ``agent_message_chunk`` / ``agent_thought_chunk``
|
|
deltas into one assistant reply by ``messageId`` and use a NEW id to start
|
|
the next reply (root-reply replacement semantics). Without ids, a client
|
|
that replaces "the current assistant message" on each chunk collapses
|
|
separate autonomous turns into one bubble.
|
|
|
|
One allocator lives per ACP session; a contiguous run of deltas shares
|
|
``current()`` and ``close()`` marks the message finished so the next delta
|
|
allocates a fresh id. Ids are UUID4 strings because the ACP schema requires
|
|
UUID-format message ids, and a fresh UUID can never collide with an earlier
|
|
turn's id.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._active: str | None = None
|
|
self._last: str | None = None
|
|
|
|
def current(self) -> str:
|
|
"""Return the active message id, allocating one if none is open."""
|
|
if self._active is None:
|
|
self._active = self._last = str(uuid.uuid4())
|
|
return self._active
|
|
|
|
def last(self) -> str | None:
|
|
"""Return the most recently allocated id (open or closed)."""
|
|
return self._last
|
|
|
|
def close(self) -> None:
|
|
"""End the active message; the next chunk starts a new id."""
|
|
self._active = None
|
|
|
|
|
|
def _make_text_cb(
|
|
conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop, wrap: Callable[[str], Any],
|
|
message_ids: AssistantMessageIdAllocator | None = None,
|
|
) -> Callable:
|
|
# ``None`` is the flush sentinel Hermes core sends between assistant messages
|
|
# (before tool execution / at end of stream): it closes the active messageId so
|
|
# the next delta opens a new bubble instead of merging into the previous one.
|
|
def _cb(text: str | None) -> None:
|
|
if text:
|
|
update = wrap(text)
|
|
if message_ids is not None:
|
|
update.message_id = message_ids.current()
|
|
_send_update(conn, session_id, loop, update)
|
|
elif text is None and message_ids is not None:
|
|
message_ids.close()
|
|
|
|
return _cb
|
|
|
|
|
|
def make_thinking_cb(
|
|
conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop,
|
|
message_ids: AssistantMessageIdAllocator | None = None,
|
|
) -> Callable:
|
|
"""Create a ``thinking_callback`` for AIAgent."""
|
|
return _make_text_cb(conn, session_id, loop, acp.update_agent_thought_text, message_ids)
|
|
|
|
|
|
def make_message_cb(
|
|
conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop,
|
|
message_ids: AssistantMessageIdAllocator | None = None,
|
|
) -> Callable:
|
|
"""Create a callback that streams agent response text to the editor."""
|
|
return _make_text_cb(conn, session_id, loop, acp.update_agent_message_text, message_ids)
|
|
|
|
|
|
def make_step_cb(
|
|
conn: acp.Client, session_id: str, loop: asyncio.AbstractEventLoop, tool_call_ids: Dict[str, Deque[str]],
|
|
tool_call_meta: Dict[str, Dict[str, Any]], turn_state: Dict[str, Any] | None = None,
|
|
) -> Callable:
|
|
"""Create a ``step_callback(api_call_count: int, prev_tools: list)`` for AIAgent."""
|
|
|
|
def _step(api_call_count: int, prev_tools: Any = None) -> None:
|
|
if not isinstance(prev_tools, list):
|
|
return
|
|
for tool_info in prev_tools:
|
|
tool_name = result = function_args = None
|
|
if isinstance(tool_info, dict):
|
|
tool_name = tool_info.get("name") or tool_info.get("function_name")
|
|
# Key presence, not truthiness: "", 0 and False are real results (#10845).
|
|
result = tool_info.get("result") if "result" in tool_info else tool_info.get("output")
|
|
function_args = tool_info.get("arguments") or tool_info.get("args")
|
|
elif isinstance(tool_info, str):
|
|
tool_name = tool_info
|
|
|
|
if not tool_name:
|
|
continue
|
|
# ``tool.completed`` already closed this call with its own result;
|
|
# this callback is the fallback for runtimes that never project one.
|
|
if not (turn_state or {}).get("saw_completion"):
|
|
queue = _upgrade_queue(tool_call_ids, tool_name)
|
|
if not queue:
|
|
continue
|
|
tc_id = queue.popleft()
|
|
meta = tool_call_meta.pop(tc_id, {})
|
|
# ``prev_tools`` carries the wire ``arguments`` JSON *string*; the content
|
|
# builders index it as a dict, so an uncoerced string raised inside this
|
|
# (swallowed) callback and the bubble never closed.
|
|
_send_update(conn, session_id, loop, build_tool_complete(
|
|
tc_id, tool_name, result=str(result) if result is not None else None,
|
|
function_args=coerce_tool_args(function_args) if function_args else meta.get("args"),
|
|
snapshot=meta.get("snapshot"),
|
|
))
|
|
if not queue:
|
|
tool_call_ids.pop(tool_name, None)
|
|
if tool_name == "todo" and (plan_update := _build_plan_update_from_todo_result(result)) is not None:
|
|
_send_update(conn, session_id, loop, plan_update)
|
|
|
|
return _step
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
import json # noqa: F401,E402
|
|
# ---- END PLUGIN-COMPAT ----
|