wehub-resource-sync
983 lines
44 KiB
Python
983 lines
44 KiB
Python
"""CursorExecutor: run agents through the Cursor Python SDK (``cursor-sdk``).
|
|
|
|
Drives Cursor via :mod:`cursor_sdk` over a local bridge — one persistent
|
|
``AsyncAgent`` per Omnigent conversation, created on a
|
|
:meth:`cursor_sdk.AsyncClient.launch_bridge` client and reused turn to turn.
|
|
Each ``run_turn`` issues one ``agent.send`` and translates the streamed
|
|
``run.events()`` (``RunStreamEvent`` objects) into ExecutorEvents:
|
|
assistant text → :class:`TextChunk`, thinking → :class:`ReasoningChunk`,
|
|
tool calls → :class:`ToolCallRequest` / :class:`ToolCallComplete`, completing
|
|
on the run's terminal :class:`cursor_sdk.RunResult`.
|
|
|
|
Policy enforcement covers three phases: PHASE_LLM_REQUEST (pre-send),
|
|
PHASE_LLM_RESPONSE (post-response), and PHASE_TOOL_CALL (native tools).
|
|
Cursor's native tools execute inside the Cursor process so they cannot be
|
|
pre-blocked, but when a non-bridged tool call is observed the executor
|
|
evaluates PHASE_TOOL_CALL and cancels the run on DENY. Bridged Omnigent
|
|
tools (MCP-wrapped) are already gated server-side via the dispatch bridge
|
|
and are skipped to avoid double evaluation.
|
|
|
|
Crucially, Omnigent's spec-declared tools (``sys_session_send`` et al.) are
|
|
bridged into Cursor **in-process** via the SDK's ``custom_tools``: each
|
|
:class:`~omnigent.inner.executor.ToolSpec` becomes a ``cursor_sdk.CustomTool``
|
|
whose ``execute`` callback routes back to the executor's ``_tool_executor`` —
|
|
the same pattern the claude-sdk harness uses with its in-process MCP tools. So
|
|
a Cursor agent can call ``sys_*``, orchestrate sub-agents, and respect policies,
|
|
i.e. full first-party parity. (This replaces the earlier ``cursor-agent acp``
|
|
transport, whose ACP mode exposed MCP servers only as read-only *resources*,
|
|
never callable tools.)
|
|
|
|
The SDK's tool-callback server runs on a daemon thread, so each ``execute``
|
|
hops back to the main event loop with :func:`asyncio.run_coroutine_threadsafe`.
|
|
|
|
Auth: a Cursor **API key** (``CURSOR_API_KEY`` or a spec ``api_key``). Unlike
|
|
``cursor-agent login``, the SDK requires an API key.
|
|
|
|
Requirements:
|
|
The ``cursor-sdk`` package must be installed (it bundles / locates the
|
|
local bridge it drives).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import concurrent.futures
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
from collections.abc import AsyncIterator, Awaitable, Callable
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, TypeAlias
|
|
|
|
from omnigent.llms._usage_observer import notify_from_dict as _notify_usage_from_dict
|
|
|
|
from .datamodel import OSEnvSpec
|
|
from .executor import (
|
|
Executor,
|
|
ExecutorConfig,
|
|
ExecutorError,
|
|
ExecutorEvent,
|
|
Message,
|
|
ReasoningChunk,
|
|
TextChunk,
|
|
ToolCallComplete,
|
|
ToolCallRequest,
|
|
ToolCallStatus,
|
|
ToolSpec,
|
|
TurnCancelled,
|
|
TurnComplete,
|
|
classify_tool_result,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Omnigent's bridged-tool callback: (tool_name, args) -> awaitable result.
|
|
# Installed by the runtime adapter (see ``_executor_adapter``); mirrors the
|
|
# claude-sdk executor's ``ToolExecutor``.
|
|
ToolExecutor: TypeAlias = Callable[[str, dict[str, Any]], Awaitable[Any]] # type: ignore[explicit-any]
|
|
|
|
# Cursor's auto model-select, used when a spec pins no cursor model (the SDK
|
|
# requires a model for local agents, so unlike the old ACP path we can't pass
|
|
# ``None``).
|
|
_DEFAULT_CURSOR_MODEL = "auto"
|
|
|
|
# Upper bound (seconds) on one bridged-tool call: generous (sub-agent dispatches
|
|
# can run for minutes) but finite, so a wedged tool surfaces a timeout error
|
|
# instead of blocking the SDK's daemon callback thread forever.
|
|
_TOOL_CALL_TIMEOUT_S = 1800.0
|
|
# Maximum time (seconds) Cursor will wait for the preToolUse hook subprocess
|
|
# to return. Held at one day so the hook stays alive while the human responds
|
|
# to the web-UI approval card — mirrors the server-side ``ask_timeout`` default
|
|
# and the ``read_timeout`` used by ``cursor_policy_hook.py``.
|
|
_HOOK_APPROVAL_TIMEOUT_S = 86400
|
|
|
|
|
|
def _resolve_model(model: str | None) -> str:
|
|
"""Resolve the cursor model id, dropping ids cursor can't honor.
|
|
|
|
cursor-sdk accepts only Cursor model ids (``auto``, ``gpt-5``,
|
|
``composer-2.5``, ...), so a gateway-routed model id (carried by a spec
|
|
authored for another harness) falls back to cursor's auto-select. ``None``
|
|
likewise resolves to ``auto`` (the SDK requires a model).
|
|
"""
|
|
if not model or model.startswith(("databricks-", "databricks/")):
|
|
if model:
|
|
# Warn, not debug: the requested model is silently NOT honored, and
|
|
# a debug line is invisible in the harness subprocess — so a user who
|
|
# pinned a non-Cursor model would otherwise have no idea it was dropped.
|
|
logger.warning(
|
|
"CursorExecutor: requested model %r is not a Cursor model id; "
|
|
"falling back to %r auto-select.",
|
|
model,
|
|
_DEFAULT_CURSOR_MODEL,
|
|
)
|
|
return _DEFAULT_CURSOR_MODEL
|
|
return model
|
|
|
|
|
|
def _first_of(d: dict[str, Any], *keys: str, default: int = 0) -> int:
|
|
"""Return the value of the first key present (and not None) in *d*."""
|
|
for k in keys:
|
|
v = d.get(k)
|
|
if v is not None:
|
|
return int(v)
|
|
return default
|
|
|
|
|
|
def _normalize_cursor_usage(raw: dict[str, Any], model: str) -> dict[str, Any]:
|
|
"""Map Cursor SDK usage fields to the standard Omnigent usage dict."""
|
|
in_tok = _first_of(raw, "inputTokens", "input_tokens")
|
|
out_tok = _first_of(raw, "outputTokens", "output_tokens")
|
|
total = _first_of(raw, "totalTokens", "total_tokens", default=in_tok + out_tok)
|
|
usage: dict[str, Any] = {
|
|
"input_tokens": in_tok,
|
|
"output_tokens": out_tok,
|
|
"total_tokens": total,
|
|
"model": model,
|
|
}
|
|
# Carry cache breakdown if the backend reports it.
|
|
# The Cursor backend sends cacheReadTokens / cacheWriteTokens;
|
|
# map to the Omnigent-standard cache_read_input_tokens /
|
|
# cache_creation_input_tokens names.
|
|
for dst, *sources in (
|
|
(
|
|
"cache_read_input_tokens",
|
|
"cacheReadTokens",
|
|
"cacheReadInputTokens",
|
|
"cache_read_input_tokens",
|
|
),
|
|
(
|
|
"cache_creation_input_tokens",
|
|
"cacheWriteTokens",
|
|
"cacheCreationInputTokens",
|
|
"cache_creation_input_tokens",
|
|
),
|
|
):
|
|
for src in sources:
|
|
val = raw.get(src)
|
|
if val is not None:
|
|
usage[dst] = val
|
|
break
|
|
# cursor's inputTokens is INCLUSIVE of cache read + write. compute_llm_cost
|
|
# expects input_tokens to be the NON-cached portion and prices the cache
|
|
# buckets additively, so subtract the cached tokens here — otherwise they
|
|
# are billed twice (once at the full input rate, once at their cache rate).
|
|
# Mirrors the qwen / antigravity executors. Clamp so a malformed cached >
|
|
# input never goes negative. total_tokens keeps the reported inclusive total.
|
|
cached = (usage.get("cache_read_input_tokens") or 0) + (
|
|
usage.get("cache_creation_input_tokens") or 0
|
|
)
|
|
usage["input_tokens"] = max(0, in_tok - cached)
|
|
return usage
|
|
|
|
|
|
def _tools_fingerprint(tools: list[ToolSpec]) -> str:
|
|
"""A stable fingerprint of the tool set (names + parameter schemas).
|
|
|
|
``custom_tools`` are fixed at agent creation, so a changed tool set must
|
|
invalidate the persistent agent — otherwise removed tools stay callable and
|
|
newly-added tools are missing for the rest of the conversation.
|
|
"""
|
|
entries = sorted(
|
|
(str(t.get("name", "")), json.dumps(t.get("parameters"), sort_keys=True, default=str))
|
|
for t in tools
|
|
)
|
|
return json.dumps(entries)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Prompt building (unchanged contract from the ACP harness)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _extract_text(msg: Message) -> str:
|
|
"""Extract plain text content from a message dict."""
|
|
content = msg.get("content")
|
|
if content is None:
|
|
return ""
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts: list[str] = []
|
|
for part in content:
|
|
if isinstance(part, dict):
|
|
text = part.get("text")
|
|
if isinstance(text, str):
|
|
parts.append(text)
|
|
return " ".join(parts)
|
|
return str(content)
|
|
|
|
|
|
def _latest_user_text(messages: list[Message]) -> str:
|
|
"""Return the text of the latest user message (multimodal parts joined)."""
|
|
for msg in reversed(messages):
|
|
if msg.get("role") == "user":
|
|
return _extract_text(msg)
|
|
return ""
|
|
|
|
|
|
def _build_cursor_prompt(
|
|
messages: list[Message],
|
|
*,
|
|
is_first_turn: bool,
|
|
system_prompt: str,
|
|
) -> str:
|
|
"""Build the prompt text for an ``agent.send``.
|
|
|
|
The SDK agent persists conversation history across ``send`` calls, so on the
|
|
first turn the Omnigent system prompt is prepended (the SDK has no separate
|
|
system-prompt field), and any prior history (a sub-agent with
|
|
``pass_history=True``) is serialized for context. On subsequent turns the
|
|
agent already holds the history, so only the latest user message is sent.
|
|
|
|
:returns: The prompt string (empty when there is nothing to send).
|
|
"""
|
|
# Serialize prior history on the first turn whenever there is any (e.g. a
|
|
# ``pass_history=True`` sub-agent handed a single user message plus assistant
|
|
# / tool context) — not only when multiple *user* messages are present, which
|
|
# would drop that context.
|
|
if is_first_turn and len(messages) > 1:
|
|
lines = ["Conversation so far:"]
|
|
for msg in messages:
|
|
role = str(msg.get("role") or "user").replace("_", " ")
|
|
lines.append(f"{role}: {_extract_text(msg)}")
|
|
lines.append("")
|
|
lines.append(
|
|
"Respond to the latest user message, using the conversation above as context."
|
|
)
|
|
body = "\n".join(lines)
|
|
else:
|
|
body = _latest_user_text(messages)
|
|
|
|
if is_first_turn and system_prompt:
|
|
return f"{system_prompt}\n\n{body}" if body else system_prompt
|
|
return body
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SDKMessage → ExecutorEvent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _sdk_message_to_events(message: Any) -> list[ExecutorEvent]: # type: ignore[explicit-any]
|
|
"""Map one ``cursor_sdk`` ``SDKMessage`` to zero or more ExecutorEvents.
|
|
|
|
Handles the message types the harness surfaces; everything else (status,
|
|
system, task, user echoes) yields nothing.
|
|
"""
|
|
mtype = getattr(message, "type", None)
|
|
events: list[ExecutorEvent] = []
|
|
|
|
if mtype == "assistant":
|
|
content = getattr(getattr(message, "message", None), "content", ()) or ()
|
|
for block in content:
|
|
if getattr(block, "type", None) == "text":
|
|
text = getattr(block, "text", "") or ""
|
|
if text:
|
|
events.append(TextChunk(text=text))
|
|
return events
|
|
|
|
if mtype == "thinking":
|
|
text = getattr(message, "text", "") or ""
|
|
if text:
|
|
events.append(ReasoningChunk(delta=text, event_type="reasoning_text"))
|
|
return events
|
|
|
|
if mtype == "tool_call":
|
|
status = getattr(message, "status", "")
|
|
name = str(getattr(message, "name", "") or "tool")
|
|
raw_args = getattr(message, "args", None)
|
|
args: dict[str, Any] = raw_args if isinstance(raw_args, dict) else {}
|
|
# Cursor surfaces host custom tools under an envelope: name == "mcp",
|
|
# args == {providerIdentifier, toolName, args}. Unwrap to the real
|
|
# Omnigent tool name + args so the observed events (and any name-keyed
|
|
# policy / UI) see the actual tool, not "mcp".
|
|
if "toolName" in args:
|
|
name = str(args.get("toolName") or name)
|
|
inner = args.get("args")
|
|
args = inner if isinstance(inner, dict) else {}
|
|
is_bridged = True
|
|
else:
|
|
is_bridged = False
|
|
call_id = getattr(message, "call_id", None)
|
|
if status == "running":
|
|
events.append(
|
|
ToolCallRequest(
|
|
name=name, args=args, metadata={"call_id": call_id, "is_bridged": is_bridged}
|
|
)
|
|
)
|
|
elif status in ("completed", "error"):
|
|
result = getattr(message, "result", None)
|
|
classification = classify_tool_result(result)
|
|
tool_status = classification.status
|
|
error = classification.error or None
|
|
if status == "error":
|
|
tool_status = ToolCallStatus.ERROR
|
|
error = error or (str(result) if result else "tool call failed")
|
|
events.append(
|
|
ToolCallComplete(
|
|
name=name,
|
|
status=tool_status,
|
|
result=result,
|
|
error=error,
|
|
metadata={"call_id": call_id, "is_bridged": is_bridged},
|
|
)
|
|
)
|
|
return events
|
|
|
|
return events
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bridged-tool result encoding
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _tool_error_payload(text: str) -> dict[str, Any]: # type: ignore[explicit-any]
|
|
"""An SDK custom-tool *error* result.
|
|
|
|
A mapping with a ``content`` list and ``isError`` is passed through unchanged
|
|
by the SDK's ``_normalize_custom_tool_result``, so the Cursor model sees a
|
|
failure — unlike a bare string, which the SDK wraps as a *successful* result.
|
|
"""
|
|
return {"content": [{"type": "text", "text": text}], "isError": True}
|
|
|
|
|
|
def _encode_tool_result(result: Any) -> Any: # type: ignore[explicit-any]
|
|
"""Encode a bridged-tool result for the SDK custom-tool return.
|
|
|
|
A result that :func:`classify_tool_result` flags as anything other than
|
|
SUCCESS — a dispatch failure (``error``), a policy block (``blocked``), a
|
|
cancellation (``cancelled``), or any of those nested inside a
|
|
``content`` / ``result`` / ``output`` / ``text`` envelope (or under a list
|
|
element) — is surfaced as an ``isError`` payload so the model sees a
|
|
failure. This pins the encoded result to the same ``classify_tool_result``
|
|
verdict the executor already reports for the observed ``ToolCallComplete``
|
|
event (see ``_sdk_message_to_events``), rather than the top-level-only
|
|
``error`` / ``blocked`` check this used to share with the claude-sdk
|
|
handler. (The claude-sdk handler still uses that narrower top-level check,
|
|
so this is *not* parity with it.) Everything else returns its text: a
|
|
``str`` passthrough (the SDK wraps it as success), else JSON.
|
|
|
|
Trade-off: because ``classify_tool_result`` maps ``{"cancelled": True}`` to
|
|
CANCELLED (not SUCCESS), a benign cancellation result — e.g. a successful
|
|
``sys_cancel_async`` returning ``{"cancelled": True, ...}`` — is encoded as
|
|
``isError``. This is intentional: a non-SUCCESS verdict is treated as a
|
|
failure here regardless of how benign the cancellation is.
|
|
"""
|
|
if classify_tool_result(result).status != ToolCallStatus.SUCCESS:
|
|
encoded = result if isinstance(result, str) else json.dumps(result, default=str)
|
|
return _tool_error_payload(encoded)
|
|
if isinstance(result, str):
|
|
return result
|
|
try:
|
|
return json.dumps(result, default=str)
|
|
except (TypeError, ValueError):
|
|
return str(result)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CursorExecutor
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _get_conversation_id() -> str | None:
|
|
"""Extract the ``--conversation-id`` value from the CLI args.
|
|
|
|
The harness subprocess is launched by :mod:`process_manager` with
|
|
``--conversation-id conv_<hex>`` on the command line. This is the
|
|
canonical server-side conversation ID (with ``conv_`` prefix) that
|
|
the policy evaluation endpoint expects.
|
|
"""
|
|
argv = sys.argv
|
|
for i, arg in enumerate(argv):
|
|
if arg == "--conversation-id" and i + 1 < len(argv):
|
|
return argv[i + 1]
|
|
return None
|
|
|
|
|
|
def _write_cursor_hooks(cwd: str, hook_script_path: str, server_url: str, session_id: str) -> Path:
|
|
"""Write ``.cursor/hooks.json`` and a wrapper shell script for preToolUse policy enforcement.
|
|
|
|
The Cursor SDK hook executor runs the command directly (not via a shell),
|
|
so inline ``env VAR=val`` doesn't work. Instead we write a tiny shell
|
|
wrapper that exports the env vars and execs the Python hook script.
|
|
|
|
:param cwd: Workspace root directory.
|
|
:param hook_script_path: Absolute path to ``cursor_policy_hook.py``.
|
|
:param server_url: Omnigent server URL, e.g. ``"http://127.0.0.1:6767"``.
|
|
:param session_id: Conversation / session ID for policy evaluation.
|
|
:returns: The path to the written ``hooks.json`` file.
|
|
"""
|
|
hooks_dir = Path(cwd) / ".cursor"
|
|
hooks_dir.mkdir(parents=True, exist_ok=True)
|
|
hooks_file = hooks_dir / "hooks.json"
|
|
|
|
# Write a wrapper script that sets env vars and execs the hook. It bakes a
|
|
# one-shot auth token + workspace-routing header, so it is owner-only
|
|
# (0o700) — the secret is never world-readable.
|
|
from omnigent.native_policy_hook import policy_hook_wrapper_script
|
|
|
|
wrapper = hooks_dir / "omnigent-hook.sh"
|
|
wrapper.write_text(policy_hook_wrapper_script(server_url, session_id, hook_script_path))
|
|
wrapper.chmod(0o700)
|
|
command = str(wrapper)
|
|
config = {
|
|
"version": 1,
|
|
"hooks": {
|
|
"preToolUse": [
|
|
{
|
|
"command": command,
|
|
"timeout": _HOOK_APPROVAL_TIMEOUT_S,
|
|
}
|
|
]
|
|
},
|
|
}
|
|
hooks_file.write_text(json.dumps(config, indent=2))
|
|
return hooks_file
|
|
|
|
|
|
@dataclass
|
|
class _CursorSessionState:
|
|
"""Per-Omnigent-conversation SDK session state."""
|
|
|
|
client: Any = None # cursor_sdk.AsyncClient
|
|
agent: Any = None # cursor_sdk.AsyncAgent
|
|
system_prompt: str | None = None
|
|
model: str | None = None
|
|
tools_fingerprint: str | None = None
|
|
has_sent_prompt: bool = False
|
|
hooks_file: Path | None = field(default=None, repr=False)
|
|
|
|
|
|
class CursorExecutor(Executor):
|
|
"""Execute agent turns via a persistent ``cursor_sdk.AsyncAgent``."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
cwd: str | None = None,
|
|
os_env: OSEnvSpec | None = None,
|
|
model: str | None = None,
|
|
api_key: str | None = None,
|
|
bundle_dir: Path | None = None,
|
|
agent_name: str | None = None,
|
|
skills_filter: str | list[str] = "all",
|
|
) -> None:
|
|
"""Create a CursorExecutor.
|
|
|
|
:param cwd: Working directory the local agent operates in. ``None``
|
|
falls back to ``os_env.cwd`` then the process cwd.
|
|
:param os_env: Optional OS environment / sandbox spec (its ``cwd`` is
|
|
used when *cwd* is unset).
|
|
:param model: Cursor model id (e.g. ``"gpt-5"``); a gateway-routed id
|
|
or ``None`` falls back to cursor's ``auto`` select.
|
|
:param api_key: Cursor API key. ``None`` falls back to ``CURSOR_API_KEY``
|
|
in the environment.
|
|
:param bundle_dir: Reserved for future skill wiring; unused in v1.
|
|
:param agent_name: Optional agent name passed to the SDK.
|
|
:param skills_filter: Accepted for parity; cursor has no skill mechanism here.
|
|
"""
|
|
self._cwd = cwd or (os_env.cwd if os_env is not None else None)
|
|
self._os_env_spec = os_env
|
|
self._model_override = model
|
|
self._api_key = api_key or os.environ.get("CURSOR_API_KEY") or None
|
|
self._bundle_dir = bundle_dir
|
|
self._agent_name = agent_name
|
|
self._skills_filter = skills_filter
|
|
self._session_states: dict[str, _CursorSessionState] = {}
|
|
# Installed by the runtime adapter; routes a bridged-tool call back into
|
|
# Omnigent's session (policy gating, sub-agent dispatch, logging).
|
|
self._tool_executor: ToolExecutor | None = None
|
|
# Installed by the runtime adapter; evaluates PHASE_LLM_REQUEST,
|
|
# PHASE_LLM_RESPONSE, and PHASE_TOOL_CALL policies (the same round-trip
|
|
# pi / claude-sdk use). ``None`` on single-process / pre-turn paths
|
|
# (then policy is a no-op).
|
|
self._policy_evaluator: Callable[[str, dict[str, Any]], Awaitable[Any]] | None = None
|
|
# Installed by the runtime adapter; surfaces ASK verdicts to the
|
|
# user via the elicitation UI (approval prompt). ``None`` when no
|
|
# handler is wired (single-process / test paths → fail closed).
|
|
self._elicitation_handler: Callable[[str, dict[str, Any]], Awaitable[bool]] | None = None
|
|
|
|
def supports_streaming(self) -> bool:
|
|
return True
|
|
|
|
def supports_tool_calling(self) -> bool:
|
|
return True
|
|
|
|
def handles_tools_internally(self) -> bool:
|
|
# Bridged tools execute in-band via the SDK custom_tools callback (which
|
|
# calls ``_tool_executor``), so the runtime adapter must NOT re-dispatch
|
|
# the observed tool events — same contract as claude-sdk.
|
|
return True
|
|
|
|
def supports_live_message_queue(self) -> bool:
|
|
# The SDK exposes no confirmed mid-turn steer, so a message can't be
|
|
# injected into a running turn.
|
|
return False
|
|
|
|
def _session_key(self, messages: list[Message]) -> str:
|
|
if messages:
|
|
last = messages[-1]
|
|
if last.get("session_id"):
|
|
return str(last["session_id"])
|
|
meta = last.get("metadata", {})
|
|
if isinstance(meta, dict) and meta.get("session_id"):
|
|
return str(meta["session_id"])
|
|
return "__default__"
|
|
|
|
# -- native-tool policy gate --------------------------------------------
|
|
|
|
async def _evaluate_native_tool_policy(
|
|
self, name: str, args: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""Gate a Cursor native tool call via policy check + user elicitation.
|
|
|
|
Returns ``{"block": bool, "reason": str}``.
|
|
|
|
Two-stage gate that mirrors how :class:`claude_sdk_executor
|
|
<omnigent.inner.claude_sdk_executor.ClaudeSDKExecutor>` exposes
|
|
tool permission requests natively through ``ctx.elicit()``:
|
|
|
|
1. **Policy hard-deny**: if the policy evaluator returns
|
|
``POLICY_ACTION_DENY``, block immediately without prompting the
|
|
user (the admin already decided).
|
|
|
|
2. **Native elicitation**: for any other outcome (ALLOW, ASK, or no
|
|
evaluator wired), invoke ``_elicitation_handler`` so the user can
|
|
review the call and approve or abort the remainder of the turn
|
|
from the web-UI approval card.
|
|
|
|
Cursor native tools execute inside the Cursor process, so they have
|
|
already started by the time the executor observes
|
|
``ToolCallRequest(status="running")``. This gate therefore cannot
|
|
block individual tool executions; it can only cancel the remainder of
|
|
the turn on denial.
|
|
"""
|
|
# Stage 1 — hard policy deny: block immediately, no elicitation.
|
|
evaluator = self._policy_evaluator
|
|
if evaluator is not None:
|
|
verdict = await evaluator("PHASE_TOOL_CALL", {"name": name, "arguments": args})
|
|
if getattr(verdict, "action", None) == "POLICY_ACTION_DENY":
|
|
return {
|
|
"block": True,
|
|
"reason": getattr(verdict, "reason", "") or "blocked by policy",
|
|
}
|
|
|
|
# Stage 2 — native elicitation: surface an approval card so the
|
|
# user can decide whether the rest of the turn should continue.
|
|
handler = self._elicitation_handler
|
|
if handler is not None:
|
|
logger.info("surfacing elicitation for native cursor tool %s", name)
|
|
approved = await handler(name, args)
|
|
if approved:
|
|
return {"block": False, "reason": ""}
|
|
return {"block": True, "reason": "turn aborted via web-UI elicitation"}
|
|
|
|
return {"block": False, "reason": ""}
|
|
|
|
# -- custom-tool bridge -------------------------------------------------
|
|
|
|
def _make_custom_tools(
|
|
self, tools: list[ToolSpec], loop: asyncio.AbstractEventLoop
|
|
) -> dict[str, Any]: # type: ignore[explicit-any]
|
|
"""Build the SDK ``custom_tools`` mapping from Omnigent ToolSpecs.
|
|
|
|
Each tool's ``execute`` runs on the SDK callback server's daemon thread,
|
|
so it hops back to *loop* (the main event loop) to await
|
|
``_tool_executor`` — the bridge into Omnigent's tool dispatch.
|
|
"""
|
|
from cursor_sdk import CustomTool # lazy: optional dependency
|
|
|
|
custom: dict[str, Any] = {} # type: ignore[explicit-any]
|
|
for spec in tools:
|
|
name = spec.get("name")
|
|
if not isinstance(name, str) or not name:
|
|
continue
|
|
params = spec.get("parameters")
|
|
custom[name] = CustomTool(
|
|
execute=self._make_execute(name, loop),
|
|
description=spec.get("description"),
|
|
input_schema=params
|
|
if isinstance(params, dict)
|
|
else {"type": "object", "properties": {}},
|
|
)
|
|
return custom
|
|
|
|
def _make_execute(
|
|
self, tool_name: str, loop: asyncio.AbstractEventLoop
|
|
) -> Callable[[dict[str, Any], Any], Any]: # type: ignore[explicit-any]
|
|
"""Build a sync ``execute`` that bridges a cursor tool call to Omnigent.
|
|
|
|
Runs on the SDK callback server's daemon thread and blocks it on the
|
|
main-loop coroutine via ``run_coroutine_threadsafe``. The wait is bounded
|
|
by ``_TOOL_CALL_TIMEOUT_S`` (generous — ``sys_session_send`` and friends
|
|
can legitimately run for minutes) so a wedged tool surfaces as a tool
|
|
error instead of hanging the daemon thread / Cursor turn forever, and any
|
|
exception (a failed or cancelled coroutine) becomes a tool error rather
|
|
than propagating raw onto the daemon thread.
|
|
"""
|
|
|
|
def execute(args: dict[str, Any], _ctx: Any) -> Any: # type: ignore[explicit-any]
|
|
if self._tool_executor is None:
|
|
return _tool_error_payload(
|
|
f"Tool {tool_name!r} is unavailable: no tool executor wired."
|
|
)
|
|
coro = self._tool_executor(tool_name, dict(args or {}))
|
|
future = asyncio.run_coroutine_threadsafe(coro, loop)
|
|
try:
|
|
result = future.result(timeout=_TOOL_CALL_TIMEOUT_S)
|
|
except concurrent.futures.TimeoutError:
|
|
future.cancel()
|
|
return _tool_error_payload(
|
|
f"Tool {tool_name!r} timed out after {_TOOL_CALL_TIMEOUT_S:.0f}s."
|
|
)
|
|
# Exception (not BaseException) still covers a cancelled coroutine —
|
|
# future.result() raises concurrent.futures.CancelledError, an
|
|
# Exception — while letting KeyboardInterrupt / SystemExit propagate.
|
|
except Exception as exc: # noqa: BLE001 — surface as a tool error
|
|
future.cancel()
|
|
return _tool_error_payload(f"Tool {tool_name!r} failed: {exc}")
|
|
return _encode_tool_result(result)
|
|
|
|
return execute
|
|
|
|
# -- session lifecycle --------------------------------------------------
|
|
|
|
async def _ensure_session(
|
|
self,
|
|
state: _CursorSessionState,
|
|
model: str,
|
|
tools: list[ToolSpec],
|
|
) -> None:
|
|
"""Launch the local bridge and create the SDK agent if not already live.
|
|
|
|
On any bring-up failure the partially-created client is closed before
|
|
propagating, so a bad ``CURSOR_API_KEY`` / launch error can't orphan a
|
|
bridge subprocess.
|
|
|
|
Before agent creation, writes ``.cursor/hooks.json`` to the workspace
|
|
with a ``preToolUse`` hook pointing at :mod:`cursor_policy_hook` so
|
|
PHASE_TOOL_CALL policies are enforced on ALL Cursor native tools --
|
|
including those that execute silently without emitting ``tool_call``
|
|
SDK messages.
|
|
"""
|
|
if state.agent is not None:
|
|
return
|
|
try:
|
|
from cursor_sdk import AsyncAgent, AsyncClient, LocalAgentOptions
|
|
except ImportError as exc:
|
|
from omnigent.onboarding.cursor_auth import CURSOR_EXTRA
|
|
from omnigent.onboarding.extra_install import extra_install_display
|
|
|
|
raise ImportError(
|
|
"CursorExecutor requires the 'cursor-sdk' package. "
|
|
f"Install it with: {extra_install_display(CURSOR_EXTRA)}"
|
|
) from exc
|
|
|
|
loop = asyncio.get_running_loop()
|
|
cwd = os.path.abspath(self._cwd or os.getcwd())
|
|
|
|
# Write .cursor/hooks.json for preToolUse policy enforcement.
|
|
# RUNNER_SERVER_URL is inherited by the harness subprocess via
|
|
# _build_harness_spawn_env (process_manager.py).
|
|
# The conversation_id comes from the --conversation-id CLI arg
|
|
# passed by the process_manager — NOT from the executor's
|
|
# session_key (which is an internal UUID without the conv_ prefix).
|
|
server_url = os.environ.get("RUNNER_SERVER_URL", "")
|
|
conv_id = _get_conversation_id()
|
|
if server_url and conv_id:
|
|
hook_script = str(Path(__file__).with_name("cursor_policy_hook.py"))
|
|
state.hooks_file = _write_cursor_hooks(cwd, hook_script, server_url, conv_id)
|
|
|
|
client = await AsyncClient.launch_bridge(workspace=cwd)
|
|
try:
|
|
local_kwargs: dict[str, Any] = {
|
|
"cwd": cwd,
|
|
"custom_tools": self._make_custom_tools(tools, loop) or None,
|
|
# Bypass cursor's own TUI approval prompts so native tool calls
|
|
# reach the executor's event stream and can be gated via the
|
|
# web-UI elicitation card (see _evaluate_native_tool_policy).
|
|
# Without this cursor may pause internally for its own approval
|
|
# before emitting a ToolCallRequest event.
|
|
"auto_review": True,
|
|
}
|
|
# Tell the SDK to read project-level settings (including hooks.json).
|
|
if state.hooks_file is not None:
|
|
local_kwargs["setting_sources"] = ["project"]
|
|
local = LocalAgentOptions(**local_kwargs)
|
|
agent = await AsyncAgent.create(
|
|
client=client,
|
|
model=model,
|
|
api_key=self._api_key,
|
|
name=self._agent_name,
|
|
local=local,
|
|
)
|
|
except BaseException:
|
|
await _safe_close(client)
|
|
raise
|
|
state.client = client
|
|
state.agent = agent
|
|
|
|
async def run_turn(
|
|
self,
|
|
messages: list[Message],
|
|
tools: list[ToolSpec],
|
|
system_prompt: str,
|
|
config: ExecutorConfig | None = None,
|
|
) -> AsyncIterator[ExecutorEvent]:
|
|
session_key = self._session_key(messages)
|
|
model = _resolve_model((config.model if config else None) or self._model_override)
|
|
tools_fp = _tools_fingerprint(tools)
|
|
state = self._session_states.setdefault(session_key, _CursorSessionState())
|
|
|
|
# System prompt, model, and tool set are all fixed at agent creation, so
|
|
# a change to any of them means a fresh agent (otherwise a changed tool
|
|
# set would leave the initial custom_tools stale for the conversation).
|
|
if state.agent is not None and (
|
|
state.system_prompt != system_prompt
|
|
or state.model != model
|
|
or state.tools_fingerprint != tools_fp
|
|
):
|
|
await self._close_state(state)
|
|
state = _CursorSessionState()
|
|
self._session_states[session_key] = state
|
|
is_first_turn = not state.has_sent_prompt
|
|
state.system_prompt = system_prompt
|
|
state.model = model
|
|
state.tools_fingerprint = tools_fp
|
|
|
|
try:
|
|
await self._ensure_session(state, model, tools)
|
|
except Exception as exc: # noqa: BLE001 — surfaced as ExecutorError (CancelledError propagates)
|
|
await self.close_session(session_key)
|
|
yield ExecutorError(message=f"Failed to start cursor-sdk agent: {exc}")
|
|
return
|
|
|
|
prompt = _build_cursor_prompt(
|
|
messages, is_first_turn=is_first_turn, system_prompt=system_prompt
|
|
)
|
|
if not prompt:
|
|
yield TurnComplete(response=None)
|
|
return
|
|
|
|
# PHASE_LLM_REQUEST policy (parity with claude-sdk / pi): evaluate before
|
|
# the LLM call so a DENY blocks it. No-op when no evaluator is wired.
|
|
policy_eval = self._policy_evaluator
|
|
if policy_eval is not None:
|
|
req_verdict = await policy_eval(
|
|
"PHASE_LLM_REQUEST",
|
|
{
|
|
"model": model,
|
|
"messages_count": sum(1 for m in messages if m.get("role") == "user") or 1,
|
|
"tools_count": len(tools),
|
|
"system_prompt_preview": system_prompt[:200] if system_prompt else "",
|
|
"last_user_message": _latest_user_text(messages)[:500],
|
|
},
|
|
)
|
|
if getattr(req_verdict, "action", "") == "POLICY_ACTION_DENY":
|
|
reason = getattr(req_verdict, "reason", "") or "no reason given"
|
|
yield ExecutorError(message=f"LLM call denied by policy: {reason}")
|
|
return
|
|
|
|
state.has_sent_prompt = True
|
|
response_text = ""
|
|
tool_calls = 0
|
|
# A tool call between two assistant text blocks means they are distinct
|
|
# narration segments (pre- vs post-tool); insert a paragraph break so
|
|
# they don't render as one run-on string ("...by the tool.- Exit: 2").
|
|
# Streamed deltas of a single response (no tool between) still
|
|
# concatenate seamlessly, so this never splits one sentence.
|
|
separate_next_text = False
|
|
turn_usage: dict[str, Any] | None = None
|
|
try:
|
|
# Pass SendOptions with on_delta so the backend sends
|
|
# interaction updates (including TurnEndedUpdate with usage).
|
|
# Without enableDeltas the backend omits them entirely.
|
|
from cursor_sdk import SendOptions as _SendOptions # lazy: optional dep
|
|
|
|
def _capture_delta(update: Any) -> None: # type: ignore[explicit-any]
|
|
nonlocal turn_usage
|
|
if getattr(update, "type", None) == "turn-ended":
|
|
raw = getattr(update, "usage", None)
|
|
if isinstance(raw, dict) and raw:
|
|
turn_usage = _normalize_cursor_usage(raw, model)
|
|
|
|
run = await state.agent.send(prompt, options=_SendOptions(on_delta=_capture_delta))
|
|
async for stream_event in run.events():
|
|
if stream_event.sdk_message is not None:
|
|
for event in _sdk_message_to_events(stream_event.sdk_message):
|
|
if isinstance(event, TextChunk):
|
|
if separate_next_text and response_text and event.text:
|
|
# Guarantee a blank-line (paragraph) boundary between
|
|
# pre- and post-tool narration, regardless of any single
|
|
# trailing/leading newline the two blocks already carry
|
|
# (a lone space or "\n" must still become a blank line).
|
|
trailing = len(response_text) - len(response_text.rstrip("\n"))
|
|
leading = len(event.text) - len(event.text.lstrip("\n"))
|
|
if trailing + leading < 2:
|
|
pad = "\n" * (2 - trailing - leading)
|
|
event = TextChunk(text=pad + event.text)
|
|
separate_next_text = False
|
|
response_text += event.text
|
|
elif isinstance(event, ToolCallRequest):
|
|
tool_calls += 1
|
|
separate_next_text = True
|
|
# Gate non-bridged (native) tool calls through
|
|
# policy + elicitation. Bridged tools are
|
|
# already gated by the dispatch bridge.
|
|
# Trigger when either the policy evaluator or
|
|
# the elicitation handler is wired — the
|
|
# latter alone (no server connection) still
|
|
# surfaces an approval card natively.
|
|
if not event.metadata.get("is_bridged") and (
|
|
policy_eval is not None or self._elicitation_handler is not None
|
|
):
|
|
gate = await self._evaluate_native_tool_policy(
|
|
event.name, event.args if isinstance(event.args, dict) else {}
|
|
)
|
|
if gate["block"]:
|
|
# Cancel the run to prevent further
|
|
# native tool use.
|
|
with contextlib.suppress(Exception):
|
|
await run.cancel()
|
|
yield event # emit so observers see what was attempted
|
|
reason = gate["reason"]
|
|
msg = f"Native tool {event.name!r} denied: {reason}"
|
|
yield ExecutorError(message=msg)
|
|
return
|
|
elif isinstance(event, ToolCallComplete):
|
|
separate_next_text = True
|
|
yield event
|
|
# Capture usage from TurnEndedUpdate interaction updates.
|
|
iu = stream_event.interaction_update
|
|
if iu is not None and getattr(iu, "type", None) == "turn-ended":
|
|
raw_usage = getattr(iu, "usage", None)
|
|
if isinstance(raw_usage, dict) and raw_usage:
|
|
turn_usage = _normalize_cursor_usage(raw_usage, model)
|
|
result = await run.wait()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 — the SDK run failed mid-turn
|
|
await self.close_session(session_key)
|
|
yield ExecutorError(message=f"cursor-sdk turn failed: {exc}", retryable=True)
|
|
return
|
|
|
|
# RunResult.status is Literal["finished", "error", "cancelled",
|
|
# "expired"]. Only "finished" should commit a TurnComplete; the other
|
|
# terminal statuses are surfaced as errors/cancellation and tear the
|
|
# session down (the agent/bridge may be in an inconsistent state).
|
|
status = getattr(result, "status", "")
|
|
if status == "error":
|
|
await self.close_session(session_key)
|
|
detail = getattr(result, "result", "") or "cursor-sdk run reported an error"
|
|
yield ExecutorError(message=f"cursor-sdk run error: {detail}", retryable=True)
|
|
return
|
|
if status == "expired":
|
|
await self.close_session(session_key)
|
|
detail = getattr(result, "result", "") or "cursor-sdk run expired"
|
|
yield ExecutorError(message=f"cursor-sdk run expired: {detail}", retryable=True)
|
|
return
|
|
if status == "cancelled":
|
|
await self.close_session(session_key)
|
|
yield TurnCancelled(reason="cursor-sdk run cancelled")
|
|
return
|
|
if status != "finished":
|
|
await self.close_session(session_key)
|
|
detail = getattr(result, "result", "") or "cursor-sdk run finished with unknown status"
|
|
yield ExecutorError(
|
|
message=f"cursor-sdk run returned non-finished status {status!r}: {detail}",
|
|
retryable=True,
|
|
)
|
|
return
|
|
|
|
# Prefer the streamed text we accumulated (which carries the paragraph
|
|
# breaks inserted above) over the SDK's aggregate ``result`` (which does
|
|
# not) whenever any text was streamed; fall back to ``result`` only when
|
|
# nothing streamed (e.g. a tool-only turn).
|
|
final = response_text or getattr(result, "result", "") or None
|
|
# PHASE_LLM_RESPONSE policy (parity with the peer harnesses): evaluate the
|
|
# completed response before TurnComplete so a DENY blocks persistence.
|
|
if policy_eval is not None:
|
|
resp_verdict = await policy_eval(
|
|
"PHASE_LLM_RESPONSE",
|
|
{
|
|
"model": model,
|
|
"text_preview": response_text[:500] if response_text else "",
|
|
"tool_calls_count": tool_calls,
|
|
},
|
|
)
|
|
if getattr(resp_verdict, "action", "") == "POLICY_ACTION_DENY":
|
|
reason = getattr(resp_verdict, "reason", "") or "no reason given"
|
|
yield ExecutorError(message=f"LLM response denied by policy: {reason}")
|
|
return
|
|
|
|
if turn_usage:
|
|
_notify_usage_from_dict(model=model, usage=turn_usage)
|
|
yield TurnComplete(response=final, usage=turn_usage)
|
|
|
|
async def _close_state(self, state: _CursorSessionState) -> None:
|
|
if state.agent is not None:
|
|
await _safe_close(state.agent)
|
|
state.agent = None
|
|
if state.client is not None:
|
|
await _safe_close(state.client)
|
|
state.client = None
|
|
# Best-effort cleanup of hooks.json and the wrapper script.
|
|
if state.hooks_file is not None:
|
|
try:
|
|
state.hooks_file.unlink(missing_ok=True)
|
|
# Also remove the wrapper shell script alongside hooks.json.
|
|
wrapper = state.hooks_file.parent / "omnigent-hook.sh"
|
|
wrapper.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
state.hooks_file = None
|
|
|
|
async def close_session(self, session_key: str) -> None:
|
|
state = self._session_states.pop(session_key, None)
|
|
if state is not None:
|
|
await self._close_state(state)
|
|
|
|
async def interrupt_session(self, session_key: str) -> bool:
|
|
state = self._session_states.get(session_key)
|
|
if state is None:
|
|
return False
|
|
# Drop the session so the next turn starts a fresh agent — mirrors the
|
|
# pi/cursor-acp executors (a resumed turn would bypass the runner's
|
|
# interrupt marker).
|
|
try:
|
|
await self.close_session(session_key)
|
|
return True
|
|
except Exception as exc: # noqa: BLE001 — close failures surface as False
|
|
logger.debug("CursorExecutor: close after interrupt failed: %s", exc)
|
|
return False
|
|
|
|
async def close(self) -> None:
|
|
for key in list(self._session_states.keys()):
|
|
await self.close_session(key)
|
|
|
|
|
|
async def _safe_close(obj: Any) -> None: # type: ignore[explicit-any]
|
|
"""Best-effort async close of a ``cursor_sdk`` object, preferring ``aclose()``.
|
|
|
|
The SDK's :class:`cursor_sdk.AsyncClient` exposes only ``aclose()`` — and
|
|
that is the *only* path that terminates the launched bridge subprocess and
|
|
shuts down the tool-callback server's daemon HTTP thread. :class:`AsyncAgent`
|
|
exposes ``close()`` instead. Calling a method the object doesn't have raised
|
|
``AttributeError`` (swallowed below), so the client was never torn down and
|
|
every session leaked its bridge subprocess + daemon thread. Prefer ``aclose``
|
|
and fall back to ``close``; a teardown failure must not mask the original
|
|
error or leave the closer raising.
|
|
"""
|
|
closer = getattr(obj, "aclose", None) or getattr(obj, "close", None)
|
|
if closer is None:
|
|
return
|
|
try:
|
|
await closer()
|
|
except Exception as exc: # noqa: BLE001 — best-effort teardown
|
|
logger.debug("CursorExecutor: close failed: %s", exc)
|