wehub-resource-sync
1004 lines
45 KiB
Python
1004 lines
45 KiB
Python
"""CopilotExecutor: run agents through the GitHub Copilot SDK (``github-copilot-sdk``).
|
|
|
|
Drives GitHub Copilot via :mod:`copilot` (the ``github-copilot-sdk`` package) —
|
|
one persistent :class:`copilot.CopilotClient` + :class:`copilot.CopilotSession`
|
|
per Omnigent conversation, created once and reused turn to turn. Each
|
|
``run_turn`` issues one ``session.send_and_wait`` and translates the session's
|
|
streamed :class:`copilot.SessionEvent` objects into ExecutorEvents:
|
|
assistant text deltas → :class:`TextChunk`, reasoning deltas →
|
|
:class:`ReasoningChunk`, tool execution → :class:`ToolCallRequest` /
|
|
:class:`ToolCallComplete`, completing on ``ASSISTANT_TURN_END`` /
|
|
``SESSION_IDLE`` (when ``send_and_wait`` returns).
|
|
|
|
Crucially, Omnigent's spec-declared tools (``sys_session_send`` et al.) are
|
|
bridged into Copilot **in-process** via the SDK's ``tools``: each
|
|
:class:`~omnigent.inner.executor.ToolSpec` becomes a :class:`copilot.Tool`
|
|
whose async ``handler`` routes back to the executor's ``_tool_executor`` — the
|
|
same pattern the claude-sdk / cursor harnesses use. So a Copilot agent can call
|
|
``sys_*``, orchestrate sub-agents, and respect policies, i.e. full first-party
|
|
parity. Unlike cursor's daemon-thread bridge, the Copilot SDK awaits tool
|
|
handlers *in its own event loop* (``CopilotSession._execute_tool_and_respond``),
|
|
so the bridged handler is a plain coroutine that ``await``\\s ``_tool_executor``
|
|
directly — no ``run_coroutine_threadsafe`` hop.
|
|
|
|
Native tools — Copilot's built-in ``create`` / ``view`` / ``edit`` / ``bash`` run
|
|
*inside* the SDK rather than through Omnigent's bridged-tool dispatch. They are
|
|
gated through a two-stage check inside the SDK ``on_permission_request`` handler
|
|
(:meth:`CopilotExecutor._on_permission_request`, parity with the cursor harness):
|
|
|
|
1. **Policy hard-deny**: if the policy evaluator returns ``POLICY_ACTION_DENY``
|
|
the call is rejected immediately (the model sees the denial and continues,
|
|
rather than aborting the whole turn).
|
|
2. **User elicitation**: for any other outcome (ALLOW, ASK, or no evaluator
|
|
wired), ``_elicitation_handler`` is invoked so the user can approve or
|
|
reject from the web-UI approval card. If no handler is wired (single-process
|
|
/ pre-turn paths) the call is approved, preserving prior behavior.
|
|
|
|
Native tool calls still leave no ``function_call`` item in the persisted
|
|
transcript — only the streamed narration is recorded — so an ``on:[tool_call]``
|
|
guardrail won't *see* them as items, but the policy IS evaluated for them at
|
|
permission time. Bridged ``sys_*`` tools are gated + recorded server-side via
|
|
``_tool_executor`` (and registered ``skip_permission=True``, so they never reach
|
|
the permission handler).
|
|
|
|
Auth: a **GitHub token** that carries Copilot access — a fine-grained PAT with
|
|
the "Copilot Requests" permission, or an OAuth token from the GitHub CLI (``gh``)
|
|
/ Copilot CLI app. Resolved from a spec ``api_key`` or the ambient
|
|
``COPILOT_GITHUB_TOKEN`` / ``GH_TOKEN`` / ``GITHUB_TOKEN`` (the same precedence
|
|
the bundled CLI uses). Classic ``ghp_`` tokens are not accepted by Copilot.
|
|
|
|
Requirements:
|
|
The ``github-copilot-sdk`` package must be installed (it bundles the
|
|
Copilot CLI binary it drives as a backing server). Installed via the
|
|
``copilot`` extra; imported lazily so a missing install surfaces as a
|
|
request-time error, not an app-boot crash.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
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 omnigent.reasoning_effort import COPILOT_EFFORTS, validate_effort
|
|
|
|
from .datamodel import OSEnvSpec
|
|
from .executor import (
|
|
CompactionComplete,
|
|
Executor,
|
|
ExecutorConfig,
|
|
ExecutorError,
|
|
ExecutorEvent,
|
|
Message,
|
|
ReasoningChunk,
|
|
TextChunk,
|
|
ToolCallComplete,
|
|
ToolCallRequest,
|
|
ToolCallStatus,
|
|
ToolSpec,
|
|
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 / cursor executors' ``ToolExecutor``.
|
|
ToolExecutor: TypeAlias = Callable[[str, dict[str, Any]], Awaitable[Any]] # type: ignore[explicit-any]
|
|
|
|
# Ambient GitHub-token env vars, in the precedence the Copilot CLI/SDK itself
|
|
# honors (``copilot login --help``): a fine-grained PAT with the "Copilot
|
|
# Requests" permission, or an OAuth token from the gh / Copilot CLI app.
|
|
GITHUB_TOKEN_ENV_VARS = ("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN")
|
|
|
|
# Upper bound (seconds) on one ``send_and_wait``. The SDK default is 60s, far
|
|
# too short for an agentic turn (sub-agent dispatches, long tool calls), so we
|
|
# pass a generous finite bound: a wedged turn surfaces a timeout error instead
|
|
# of hanging forever. The runner enforces its own per-turn timeout on top.
|
|
_SEND_TIMEOUT_S = 3600.0
|
|
|
|
|
|
def _resolve_model(model: str | None) -> str | None:
|
|
"""Resolve the Copilot model id, dropping ids Copilot can't honor.
|
|
|
|
The Copilot SDK accepts only ids in the account's catalog (``auto``,
|
|
``claude-haiku-4.5``, ``gpt-5-mini``, ...), so a gateway-routed model id
|
|
(carried by a spec authored for another harness) falls back to Copilot's
|
|
own auto-select (``None`` → the SDK picks). ``None`` likewise lets the SDK
|
|
choose.
|
|
"""
|
|
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-Copilot model would otherwise have no idea it was dropped.
|
|
logger.warning(
|
|
"CopilotExecutor: requested model %r is not a Copilot model id; "
|
|
"falling back to Copilot's auto-select.",
|
|
model,
|
|
)
|
|
return None
|
|
return model
|
|
|
|
|
|
def _resolve_reasoning_effort(config: ExecutorConfig | None) -> str | None:
|
|
"""Resolve the per-turn Copilot reasoning effort from ``config.extra``.
|
|
|
|
The runtime adapter threads a web ``/reasoning`` pick into
|
|
``config.extra["reasoning_effort"]`` (see
|
|
:class:`~omnigent.runtime.harnesses._executor_adapter.ExecutorAdapter`).
|
|
The Copilot SDK exposes it only as ``create_session(reasoning_effort=...)``,
|
|
so like the model it is fixed at session creation (a change recreates the
|
|
session in :meth:`run_turn`). ``None`` lets the model use its default.
|
|
|
|
A value Copilot can't honor is dropped with a warning rather than raising:
|
|
an unsupported effort must not sink the whole turn (parity with the codex
|
|
native path). Per-model support is enforced by the Copilot backend.
|
|
|
|
Note: ``config.extra`` may also carry ``max_tokens`` (set by the adapter),
|
|
but the Copilot SDK has no per-turn output-token cap — the only
|
|
``max_output_tokens`` lever is a model *capability* override folded into
|
|
context-window math, not a generation limit — so it is intentionally not
|
|
forwarded here.
|
|
"""
|
|
if config is None:
|
|
return None
|
|
raw_effort = config.extra.get("reasoning_effort")
|
|
try:
|
|
return validate_effort(raw_effort, "copilot", COPILOT_EFFORTS)
|
|
except ValueError:
|
|
logger.warning("Ignoring unsupported copilot reasoning effort: %r", raw_effort)
|
|
return None
|
|
|
|
|
|
def _tools_fingerprint(tools: list[ToolSpec]) -> str:
|
|
"""A stable fingerprint of the tool set (names + parameter schemas).
|
|
|
|
``tools`` are fixed at session creation, so a changed tool set must
|
|
invalidate the persistent session — 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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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_copilot_prompt(messages: list[Message], *, is_first_turn: bool) -> str:
|
|
"""Build the prompt text for a ``send_and_wait``.
|
|
|
|
The SDK session persists conversation history across ``send`` calls and the
|
|
Omnigent system prompt is delivered separately (``system_message``), so on
|
|
the first turn any prior history (e.g. a ``pass_history=True`` sub-agent that
|
|
handed a single user message plus assistant / tool context) is serialized for
|
|
context. On subsequent turns the session already holds the history, so only
|
|
the latest user message is sent.
|
|
|
|
:returns: The prompt string (empty when there is nothing to send).
|
|
"""
|
|
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."
|
|
)
|
|
return "\n".join(lines)
|
|
return _latest_user_text(messages)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bridged-tool result encoding
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _encode_tool_result(result: Any) -> Any: # type: ignore[explicit-any]
|
|
"""Encode a bridged-tool result as a :class:`copilot.ToolResult`.
|
|
|
|
A dict carrying a truthy ``error`` or ``blocked`` is a dispatch failure or a
|
|
policy block (the shapes ``_bridge_one_dispatch`` / the policy layer return):
|
|
surface it as a ``failure`` result so the Copilot model sees the failure —
|
|
parity with the claude-sdk / cursor handlers. Everything else is a success:
|
|
a ``str`` is passed through; anything else is JSON-encoded.
|
|
"""
|
|
from copilot import ToolResult # lazy: optional dependency
|
|
|
|
if isinstance(result, dict) and (result.get("error") or result.get("blocked")):
|
|
text = json.dumps(result, default=str)
|
|
return ToolResult(text_result_for_llm=text, result_type="failure", error=text)
|
|
if isinstance(result, str):
|
|
return ToolResult(text_result_for_llm=result)
|
|
try:
|
|
text = json.dumps(result, default=str)
|
|
except (TypeError, ValueError):
|
|
text = str(result)
|
|
return ToolResult(text_result_for_llm=text)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CopilotExecutor
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class _CopilotSessionState:
|
|
"""Per-Omnigent-conversation SDK session state."""
|
|
|
|
client: Any = None # copilot.CopilotClient
|
|
session: Any = None # copilot.CopilotSession
|
|
system_prompt: str | None = None
|
|
model: str | None = None
|
|
reasoning_effort: str | None = None
|
|
tools_fingerprint: str | None = None
|
|
has_sent_prompt: bool = False
|
|
# call_id -> tool name, populated on TOOL_EXECUTION_START so the matching
|
|
# TOOL_EXECUTION_COMPLETE (which carries only the id) can name its tool.
|
|
call_names: dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
class CopilotExecutor(Executor):
|
|
"""Execute agent turns via a persistent GitHub Copilot SDK session."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
cwd: str | None = None,
|
|
os_env: OSEnvSpec | None = None,
|
|
model: str | None = None,
|
|
github_token: str | None = None,
|
|
bundle_dir: Path | None = None,
|
|
agent_name: str | None = None,
|
|
skills_filter: str | list[str] = "all",
|
|
) -> None:
|
|
"""Create a CopilotExecutor.
|
|
|
|
:param cwd: Working directory the Copilot session 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: Copilot model id (e.g. ``"claude-haiku-4.5"``); a
|
|
gateway-routed id or ``None`` falls back to Copilot's auto-select.
|
|
:param github_token: GitHub token carrying Copilot access. ``None``
|
|
falls back to ``COPILOT_GITHUB_TOKEN`` / ``GH_TOKEN`` /
|
|
``GITHUB_TOKEN`` in the environment.
|
|
:param bundle_dir: Reserved for future skill wiring; unused in v1.
|
|
:param agent_name: Optional agent name (reserved for parity).
|
|
:param skills_filter: Accepted for parity; copilot has no skill
|
|
mechanism wired 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._github_token = github_token or _ambient_github_token()
|
|
self._bundle_dir = bundle_dir
|
|
self._agent_name = agent_name
|
|
self._skills_filter = skills_filter
|
|
self._session_states: dict[str, _CopilotSessionState] = {}
|
|
# 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 policies (the same round-trip pi / claude-sdk /
|
|
# cursor use). ``None`` on single-process / pre-turn paths (no-op).
|
|
self._policy_evaluator: Callable[[str, dict[str, Any]], Awaitable[Any]] | None = None
|
|
# Installed by the runtime adapter; surfaces native-tool calls to the
|
|
# user via the web-UI elicitation approval card. ``None`` when no
|
|
# handler is wired (single-process / test paths → default approve).
|
|
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 tool handler (which calls
|
|
# ``_tool_executor``), so the runtime adapter must NOT re-dispatch the
|
|
# observed tool events — same contract as claude-sdk / cursor.
|
|
return True
|
|
|
|
def supports_live_message_queue(self) -> bool:
|
|
# The SDK exposes no confirmed mid-turn steer wired here, 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__"
|
|
|
|
# -- tool bridge --------------------------------------------------------
|
|
|
|
def _make_tools(self, tools: list[ToolSpec]) -> list[Any]: # type: ignore[explicit-any]
|
|
"""Build the SDK ``tools`` list from Omnigent ToolSpecs.
|
|
|
|
Each tool's async ``handler`` is awaited by the SDK in its own event
|
|
loop, so it bridges straight into ``_tool_executor`` — no thread hop.
|
|
``skip_permission=True`` because Omnigent's policy layer (and the
|
|
``_tool_executor`` round-trip) already governs these tools; the SDK
|
|
permission prompt would be redundant and would stall a headless turn.
|
|
"""
|
|
from copilot import Tool # lazy: optional dependency
|
|
|
|
built: list[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")
|
|
built.append(
|
|
Tool(
|
|
name=name,
|
|
description=spec.get("description") or "",
|
|
handler=self._make_handler(name),
|
|
parameters=params
|
|
if isinstance(params, dict)
|
|
else {"type": "object", "properties": {}},
|
|
skip_permission=True,
|
|
)
|
|
)
|
|
return built
|
|
|
|
def _make_handler(self, tool_name: str) -> Callable[[Any], Awaitable[Any]]: # type: ignore[explicit-any]
|
|
"""Build an async ``handler`` that bridges a Copilot tool call to Omnigent.
|
|
|
|
Awaited by the SDK in its event loop, so it ``await``\\s the main-loop
|
|
``_tool_executor`` directly. Any exception becomes a tool *failure*
|
|
result rather than propagating into the SDK turn.
|
|
"""
|
|
|
|
async def handler(invocation: Any) -> Any: # type: ignore[explicit-any]
|
|
from copilot import ToolResult # lazy: optional dependency
|
|
|
|
if self._tool_executor is None:
|
|
return ToolResult(
|
|
text_result_for_llm=(
|
|
f"Tool {tool_name!r} is unavailable: no tool executor wired."
|
|
),
|
|
result_type="failure",
|
|
error="no tool executor wired",
|
|
)
|
|
raw_args = getattr(invocation, "arguments", None)
|
|
args = _coerce_args(raw_args)
|
|
try:
|
|
result = await self._tool_executor(tool_name, dict(args))
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 — surface as a tool failure
|
|
return ToolResult(
|
|
text_result_for_llm=f"Tool {tool_name!r} failed: {exc}",
|
|
result_type="failure",
|
|
error=str(exc),
|
|
)
|
|
return _encode_tool_result(result)
|
|
|
|
return handler
|
|
|
|
# -- session lifecycle --------------------------------------------------
|
|
|
|
async def _on_permission_request(self, request: Any, _invocation: dict[str, str]) -> Any: # type: ignore[explicit-any]
|
|
"""Gate a Copilot NATIVE-tool permission request through policy + elicitation.
|
|
|
|
Installed as ``create_session(on_permission_request=...)``. The SDK awaits
|
|
this in its own event loop, so it ``await``\\s ``_policy_evaluator`` and
|
|
``_elicitation_handler`` directly (no thread hop). Only Copilot's built-in
|
|
tools (``bash`` / ``edit`` / ``view`` / ``create`` / ``url``) reach here;
|
|
bridged ``sys_*`` tools are registered ``skip_permission=True`` and are
|
|
gated + recorded server-side via ``_tool_executor`` instead.
|
|
|
|
Two-stage gate (parity with the cursor harness):
|
|
|
|
1. **Policy hard-deny**: if the policy evaluator returns
|
|
``POLICY_ACTION_DENY``, reject immediately without prompting the user.
|
|
The model sees the denial and continues (individual call blocked, not
|
|
the whole turn).
|
|
2. **User elicitation**: for any other outcome (ALLOW, ASK, or no
|
|
evaluator wired), invoke ``_elicitation_handler`` so the user can
|
|
approve or reject from the web-UI card. When no handler is wired
|
|
(single-process / pre-turn paths) the call is approved by default,
|
|
preserving prior behavior.
|
|
"""
|
|
from copilot.rpc import ( # lazy: optional dependency
|
|
PermissionDecisionApproveOnce,
|
|
PermissionDecisionReject,
|
|
)
|
|
|
|
name, args = _permission_policy_input(request)
|
|
|
|
# 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", "") == "POLICY_ACTION_DENY":
|
|
reason = getattr(verdict, "reason", "") or "blocked by policy"
|
|
return PermissionDecisionReject(feedback=f"Denied by Omnigent policy: {reason}")
|
|
|
|
# Stage 2 — user elicitation: surface an approval card.
|
|
handler = self._elicitation_handler
|
|
if handler is not None:
|
|
approved = await handler(name, args)
|
|
if not approved:
|
|
return PermissionDecisionReject(feedback="Denied via Omnigent approval UI")
|
|
|
|
return PermissionDecisionApproveOnce()
|
|
|
|
async def _ensure_session(
|
|
self,
|
|
state: _CopilotSessionState,
|
|
model: str | None,
|
|
tools: list[ToolSpec],
|
|
system_prompt: str,
|
|
reasoning_effort: str | None = None,
|
|
) -> None:
|
|
"""Start the SDK client and create the session if not already live.
|
|
|
|
On any bring-up failure the partially-started client is stopped before
|
|
propagating, so a bad token / launch error can't orphan the bundled
|
|
CLI subprocess.
|
|
"""
|
|
if state.session is not None:
|
|
return
|
|
try:
|
|
from copilot import CopilotClient
|
|
except ImportError as exc:
|
|
from omnigent.onboarding.copilot_auth import COPILOT_EXTRA
|
|
from omnigent.onboarding.extra_install import extra_install_display
|
|
|
|
raise ImportError(
|
|
"CopilotExecutor requires the 'github-copilot-sdk' package. "
|
|
f"Install it with: {extra_install_display(COPILOT_EXTRA)}"
|
|
) from exc
|
|
|
|
# The Copilot SDK rejects a relative working_directory ("Directory path
|
|
# must be absolute"), and a spec / os_env can hand us a relative cwd
|
|
# (e.g. ``.``), so always resolve to an absolute path.
|
|
cwd = os.path.abspath(self._cwd or os.getcwd())
|
|
client = CopilotClient(
|
|
github_token=self._github_token,
|
|
working_directory=cwd,
|
|
log_level="error",
|
|
)
|
|
try:
|
|
# ``start()`` is inside the try: it spawns the bundled Copilot CLI
|
|
# subprocess *before* connecting/verifying, and the SDK's own error
|
|
# path re-raises without terminating that subprocess — only
|
|
# ``client.stop()`` reaps it. So a start failure (bad token, version
|
|
# skew) must still hit ``_safe_stop`` below, or it orphans the CLI.
|
|
await client.start()
|
|
# ``append`` keeps Copilot's own operating instructions (how to use
|
|
# its built-in tools) and layers the Omnigent agent's system prompt
|
|
# on top — a ``replace`` would strip the tool-use guidance the model
|
|
# relies on. ``None`` when the spec carries no system prompt.
|
|
system_message = (
|
|
{"mode": "append", "content": system_prompt} if system_prompt else None
|
|
)
|
|
session = await client.create_session(
|
|
model=model,
|
|
streaming=True,
|
|
system_message=system_message,
|
|
tools=self._make_tools(tools) or None,
|
|
on_permission_request=self._on_permission_request,
|
|
working_directory=cwd,
|
|
reasoning_effort=reasoning_effort or None,
|
|
)
|
|
except BaseException:
|
|
await _safe_stop(client)
|
|
raise
|
|
state.client = client
|
|
state.session = session
|
|
|
|
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)
|
|
reasoning_effort = _resolve_reasoning_effort(config)
|
|
tools_fp = _tools_fingerprint(tools)
|
|
state = self._session_states.setdefault(session_key, _CopilotSessionState())
|
|
|
|
# System prompt, model, reasoning effort, and tool set are all fixed at
|
|
# session creation, so a change to any of them means a fresh session
|
|
# (otherwise a changed tool set would leave the initial ``tools`` stale
|
|
# for the conversation, and a new ``/reasoning`` pick would never apply).
|
|
if state.session is not None and (
|
|
state.system_prompt != system_prompt
|
|
or state.model != model
|
|
or state.reasoning_effort != reasoning_effort
|
|
or state.tools_fingerprint != tools_fp
|
|
):
|
|
await self._close_state(state)
|
|
state = _CopilotSessionState()
|
|
self._session_states[session_key] = state
|
|
is_first_turn = not state.has_sent_prompt
|
|
state.system_prompt = system_prompt
|
|
state.model = model
|
|
state.reasoning_effort = reasoning_effort
|
|
state.tools_fingerprint = tools_fp
|
|
|
|
try:
|
|
await self._ensure_session(state, model, tools, system_prompt, reasoning_effort)
|
|
except Exception as exc: # noqa: BLE001 — surfaced as ExecutorError (CancelledError propagates)
|
|
await self.close_session(session_key)
|
|
yield ExecutorError(message=f"Failed to start copilot-sdk session: {exc}")
|
|
return
|
|
|
|
prompt = _build_copilot_prompt(messages, is_first_turn=is_first_turn)
|
|
if not prompt:
|
|
yield TurnComplete(response=None)
|
|
return
|
|
|
|
# PHASE_LLM_REQUEST policy (parity with claude-sdk / cursor / pi):
|
|
# evaluate before the LLM call so a DENY blocks it. No-op when unwired.
|
|
policy_eval = self._policy_evaluator
|
|
if policy_eval is not None:
|
|
req_verdict = await policy_eval(
|
|
"PHASE_LLM_REQUEST",
|
|
{
|
|
"model": model or "auto",
|
|
"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
|
|
session = state.session
|
|
|
|
# Stream the session's events into a queue from the SDK's ``on`` callback
|
|
# (invoked in the SDK loop), then drain-and-translate them here while the
|
|
# ``send_and_wait`` task runs. Unsubscribe + drain when the send completes.
|
|
queue: asyncio.Queue[Any] = asyncio.Queue() # type: ignore[explicit-any]
|
|
|
|
def _on_event(event: Any) -> None: # type: ignore[explicit-any]
|
|
queue.put_nowait(event)
|
|
|
|
unsubscribe = session.on(_on_event)
|
|
send_task: asyncio.Task[Any] = asyncio.ensure_future( # type: ignore[explicit-any]
|
|
session.send_and_wait(prompt, timeout=_SEND_TIMEOUT_S)
|
|
)
|
|
|
|
response_text = ""
|
|
tool_calls = 0
|
|
usage_acc: dict[str, int] = {}
|
|
usage_model: str | None = None
|
|
turn_error: str | None = None
|
|
# 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. Streamed deltas of a single
|
|
# response (no tool between) still concatenate seamlessly.
|
|
separate_next_text = False
|
|
|
|
async def _drain(event: Any) -> AsyncIterator[ExecutorEvent]: # type: ignore[explicit-any]
|
|
nonlocal response_text, tool_calls, separate_next_text, usage_model, turn_error
|
|
etype = str(getattr(event, "type", ""))
|
|
data = _event_data(event)
|
|
if etype.endswith("ASSISTANT_MESSAGE_DELTA"):
|
|
text = str(data.get("deltaContent") or "")
|
|
if text:
|
|
if separate_next_text and response_text:
|
|
trailing = len(response_text) - len(response_text.rstrip("\n"))
|
|
leading = len(text) - len(text.lstrip("\n"))
|
|
if trailing + leading < 2:
|
|
text = "\n" * (2 - trailing - leading) + text
|
|
separate_next_text = False
|
|
response_text += text
|
|
yield TextChunk(text=text)
|
|
elif etype.endswith("ASSISTANT_REASONING_DELTA"):
|
|
text = str(data.get("deltaContent") or "")
|
|
if text:
|
|
yield ReasoningChunk(delta=text, event_type="reasoning_text")
|
|
elif etype.endswith("TOOL_EXECUTION_START"):
|
|
name = str(data.get("toolName") or "tool")
|
|
call_id = data.get("toolCallId")
|
|
if call_id:
|
|
state.call_names[str(call_id)] = name
|
|
tool_calls += 1
|
|
separate_next_text = True
|
|
yield ToolCallRequest(
|
|
name=name,
|
|
args=_coerce_args(data.get("arguments")),
|
|
metadata={"call_id": call_id},
|
|
)
|
|
elif etype.endswith("TOOL_EXECUTION_COMPLETE"):
|
|
call_id = data.get("toolCallId")
|
|
name = state.call_names.get(str(call_id), "tool") if call_id else "tool"
|
|
result = _unwrap_tool_result(data.get("result"))
|
|
classification = classify_tool_result(result)
|
|
status = classification.status
|
|
error = classification.error or None
|
|
raw_error = data.get("error")
|
|
if data.get("success") is False or raw_error:
|
|
status = ToolCallStatus.ERROR
|
|
error = error or _unwrap_tool_error(raw_error) or "tool call failed"
|
|
separate_next_text = True
|
|
yield ToolCallComplete(
|
|
name=name,
|
|
status=status,
|
|
result=result,
|
|
error=error,
|
|
metadata={"call_id": call_id},
|
|
)
|
|
elif etype.endswith("ASSISTANT_USAGE"):
|
|
usage_model = data.get("model") or usage_model
|
|
_accumulate_usage(usage_acc, data)
|
|
elif etype.endswith(("SESSION_ERROR", "MODEL_CALL_FAILURE")):
|
|
turn_error = str(
|
|
data.get("message") or data.get("errorMessage") or "copilot session error"
|
|
)
|
|
elif etype.endswith("SESSION_COMPACTION_COMPLETE"):
|
|
# The Copilot SDK auto-compacted the session's context window.
|
|
# Surface it as a CompactionComplete so the runner persists a
|
|
# compaction item and a resumed session gets the pre-compacted
|
|
# summary instead of replaying the full transcript (parity with
|
|
# claude-sdk / openai-agents). Only a *successful* compaction
|
|
# counts; ``success`` is False on a failed/aborted attempt.
|
|
if data.get("success"):
|
|
post_tokens = data.get("postCompactionTokens")
|
|
yield CompactionComplete(
|
|
# Copilot uniquely reports the real summary text; fall
|
|
# back to a synthetic placeholder like the peers do.
|
|
summary=str(
|
|
data.get("summaryContent")
|
|
or "[GitHub Copilot compaction: context was automatically compacted]"
|
|
),
|
|
# Post-compaction context size (same semantic as the
|
|
# peers' ``context_tokens``).
|
|
token_count=(
|
|
int(post_tokens) if isinstance(post_tokens, (int, float)) else 0
|
|
),
|
|
model=usage_model or model,
|
|
# The event carries no message list; resume falls back to
|
|
# the summary (allowed by the field's contract).
|
|
compacted_messages=None,
|
|
)
|
|
|
|
try:
|
|
while True:
|
|
getter: asyncio.Task[Any] = asyncio.ensure_future(queue.get()) # type: ignore[explicit-any]
|
|
done, _pending = await asyncio.wait(
|
|
{getter, send_task}, return_when=asyncio.FIRST_COMPLETED
|
|
)
|
|
if getter in done:
|
|
event = getter.result()
|
|
async for ev in _drain(event):
|
|
yield ev
|
|
else:
|
|
getter.cancel()
|
|
if send_task.done():
|
|
# The turn ended — drain whatever is still queued, then stop.
|
|
while not queue.empty():
|
|
async for ev in _drain(queue.get_nowait()):
|
|
yield ev
|
|
break
|
|
final_event = send_task.result()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 — the SDK turn failed mid-stream
|
|
await self.close_session(session_key)
|
|
yield ExecutorError(message=f"copilot-sdk turn failed: {exc}", retryable=True)
|
|
return
|
|
finally:
|
|
unsubscribe()
|
|
if not send_task.done():
|
|
send_task.cancel()
|
|
|
|
# Surface a mid-turn error event (``SESSION_ERROR`` / ``MODEL_CALL_FAILURE``)
|
|
# whenever the turn did NOT produce a successful final message — even if
|
|
# some text streamed first. Reporting a failed turn as a clean
|
|
# ``TurnComplete`` with partial text would mask the failure (a turn that
|
|
# errored after partial output would look successful). A real final
|
|
# ASSISTANT_MESSAGE (``final_event`` set) means the SDK completed the turn,
|
|
# so a stray earlier event is not treated as fatal.
|
|
if turn_error and final_event is None:
|
|
await self.close_session(session_key)
|
|
yield ExecutorError(message=f"copilot-sdk run error: {turn_error}", retryable=True)
|
|
return
|
|
|
|
# Prefer the streamed text (which carries the paragraph breaks inserted
|
|
# above); fall back to the final ASSISTANT_MESSAGE content only when
|
|
# nothing streamed (e.g. a tool-only turn).
|
|
final = response_text or _final_message_text(final_event) or None
|
|
|
|
# PHASE_LLM_RESPONSE policy (parity with the peer harnesses).
|
|
if policy_eval is not None:
|
|
resp_verdict = await policy_eval(
|
|
"PHASE_LLM_RESPONSE",
|
|
{
|
|
"model": model or "auto",
|
|
"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
|
|
|
|
usage = _finalize_usage(usage_acc)
|
|
if usage is not None:
|
|
_notify_usage_from_dict(model=usage_model or model or "copilot", usage=usage)
|
|
yield TurnComplete(response=final, usage=usage)
|
|
|
|
async def _close_state(self, state: _CopilotSessionState) -> None:
|
|
if state.session is not None:
|
|
await _safe_stop(state.session)
|
|
state.session = None
|
|
if state.client is not None:
|
|
await _safe_stop(state.client)
|
|
state.client = None
|
|
state.call_names.clear()
|
|
|
|
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
|
|
# Best-effort SDK abort first, to cleanly cancel the in-flight turn on
|
|
# the bundled CLI *before* teardown. ``session.abort()`` is the SDK's
|
|
# blessed cancel (it keeps the session valid), so it stops the next
|
|
# ``stop()`` from racing a live generation — which can orphan the CLI's
|
|
# tool subprocesses or dump a post-cancel stream. Mirrors pi / claude-sdk.
|
|
if state.session is not None:
|
|
try:
|
|
await asyncio.wait_for(state.session.abort(), timeout=0.5)
|
|
except Exception as exc: # noqa: BLE001 — abort is best-effort
|
|
logger.debug("CopilotExecutor: interrupt abort failed: %s", exc)
|
|
# Always drop the session so the next turn starts a fresh one — mirrors
|
|
# the cursor / pi executors. A resumed Copilot session sends only the
|
|
# latest user message (see ``_build_copilot_prompt``), which would bypass
|
|
# the runner's "[System: interrupted]" marker and silently continue the
|
|
# abandoned request; a fresh session replays full history (marker
|
|
# included). See ``claude_sdk_executor.interrupt_session`` for the rationale.
|
|
try:
|
|
await self.close_session(session_key)
|
|
return True
|
|
except Exception as exc: # noqa: BLE001 — close failures surface as False
|
|
logger.debug("CopilotExecutor: 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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _ambient_github_token() -> str | None:
|
|
"""Return the first set ambient GitHub token, in CLI precedence order."""
|
|
for var in GITHUB_TOKEN_ENV_VARS:
|
|
value = os.environ.get(var)
|
|
if value:
|
|
return value
|
|
return None
|
|
|
|
|
|
def _coerce_args(raw: Any) -> dict[str, Any]: # type: ignore[explicit-any]
|
|
"""Coerce a tool-call ``arguments`` payload to a dict.
|
|
|
|
The SDK delivers arguments as a parsed dict, but tolerate a JSON string (or
|
|
anything else) by best-effort parsing, falling back to an empty mapping.
|
|
"""
|
|
if isinstance(raw, dict):
|
|
return raw
|
|
if isinstance(raw, str) and raw.strip():
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
return {}
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
return {}
|
|
|
|
|
|
def _permission_policy_input(request: Any) -> tuple[str, dict[str, Any]]: # type: ignore[explicit-any]
|
|
"""Map a Copilot ``PermissionRequest`` variant to a (name, arguments) policy input.
|
|
|
|
The permission-request union is non-uniform: only the ``mcp`` / ``custom-tool``
|
|
/ ``hook`` variants carry ``tool_name``, while ``shell`` / ``read`` / ``write``
|
|
/ ``url`` identify themselves by their ``kind`` discriminator. Fall back to
|
|
``kind`` for the name, and pass the variant's ``to_dict()`` (camelCase wire
|
|
form, carrying all fields) as the policy ``arguments``.
|
|
"""
|
|
name = getattr(request, "tool_name", None) or getattr(request, "kind", None) or "tool"
|
|
try:
|
|
args = request.to_dict()
|
|
except Exception: # noqa: BLE001 — defensive: hand the policy a best-effort dict
|
|
args = {}
|
|
return str(name), args if isinstance(args, dict) else {}
|
|
|
|
|
|
def _event_data(event: Any) -> dict[str, Any]: # type: ignore[explicit-any]
|
|
"""Return an event's ``data`` payload as a (camelCase-keyed) dict.
|
|
|
|
Uses ``to_dict()`` (the wire form) so reads are schema-stable across the
|
|
SDK's snake_case attributes vs. camelCase wire keys.
|
|
"""
|
|
try:
|
|
payload = event.to_dict()
|
|
except Exception: # noqa: BLE001 — fall back to attribute access below
|
|
payload = None
|
|
if isinstance(payload, dict):
|
|
data = payload.get("data")
|
|
if isinstance(data, dict):
|
|
return data
|
|
data_attr = getattr(event, "data", None)
|
|
if isinstance(data_attr, dict):
|
|
return data_attr
|
|
return {}
|
|
|
|
|
|
# SDK ``ToolExecutionCompleteResult.to_dict`` wraps the payload as
|
|
# ``{"content": ..., "contents": [...], "detailedContent": ..., "uiResource": ...}``
|
|
# and ``ToolExecutionCompleteError.to_dict`` as ``{"message": ..., "code": ...}``.
|
|
_TOOL_RESULT_WRAPPER_KEYS = ("content", "contents", "detailedContent", "uiResource")
|
|
|
|
|
|
def _unwrap_tool_result(raw: Any) -> Any: # type: ignore[explicit-any]
|
|
"""Unwrap the SDK ``ToolExecutionCompleteResult`` wrapper to its content payload.
|
|
|
|
A ``TOOL_EXECUTION_COMPLETE`` ``result`` arrives in wire form as a wrapper
|
|
dict (``{"content": ..., "detailedContent": ..., ...}``). Carry the bare
|
|
content downstream (classification, tracing, the persisted ``ToolCallComplete``
|
|
result) rather than the wrapper — matching the bare-payload convention the
|
|
peer executors use. Only a recognized wrapper shape is unwrapped, so an
|
|
Omnigent-convention result (e.g. ``{"blocked": True}``) passes through
|
|
untouched for :func:`classify_tool_result`.
|
|
"""
|
|
if isinstance(raw, dict) and any(k in raw for k in _TOOL_RESULT_WRAPPER_KEYS):
|
|
for key in ("content", "detailedContent"):
|
|
value = raw.get(key)
|
|
if value is not None:
|
|
return value
|
|
return raw
|
|
|
|
|
|
def _unwrap_tool_error(raw: Any) -> str | None: # type: ignore[explicit-any]
|
|
"""Extract the message from the SDK's structured tool error.
|
|
|
|
A failed ``TOOL_EXECUTION_COMPLETE`` ``error`` arrives as
|
|
``{"message": ..., "code": ...}`` (``ToolExecutionCompleteError.to_dict``);
|
|
surface the ``message`` rather than the dict's Python repr (which would leak
|
|
``"{'message': ..., 'code': ...}"`` into the tool error shown to the
|
|
model / recorded on the span). A bare string passes through; anything else
|
|
yields ``None`` so the caller can fall back to a generic message.
|
|
"""
|
|
if isinstance(raw, dict):
|
|
message = raw.get("message")
|
|
if isinstance(message, str) and message:
|
|
return message
|
|
if isinstance(raw, str) and raw:
|
|
return raw
|
|
return None
|
|
|
|
|
|
def _final_message_text(event: Any) -> str: # type: ignore[explicit-any]
|
|
"""Extract the aggregate assistant text from the final ASSISTANT_MESSAGE event."""
|
|
if event is None:
|
|
return ""
|
|
content = _event_data(event).get("content")
|
|
return content if isinstance(content, str) else ""
|
|
|
|
|
|
def _accumulate_usage(acc: dict[str, int], data: dict[str, Any]) -> None: # type: ignore[explicit-any]
|
|
"""Sum the token counts from one ASSISTANT_USAGE event into *acc*.
|
|
|
|
Copilot emits one usage event per underlying model call, so a turn with
|
|
tool round-trips reports usage several times; we accumulate. The
|
|
authoritative AI-credit cost (``copilotUsage.totalNanoAiu``, in nano-AIU) is
|
|
summed too and converted to ``cost_usd`` in :func:`_finalize_usage`.
|
|
"""
|
|
mapping = {
|
|
"inputTokens": "input_tokens",
|
|
"outputTokens": "output_tokens",
|
|
"cacheReadTokens": "cache_read_input_tokens",
|
|
"cacheWriteTokens": "cache_creation_input_tokens",
|
|
}
|
|
for wire_key, usage_key in mapping.items():
|
|
value = data.get(wire_key)
|
|
if isinstance(value, (int, float)):
|
|
acc[usage_key] = acc.get(usage_key, 0) + int(value)
|
|
copilot_usage = data.get("copilotUsage")
|
|
if isinstance(copilot_usage, dict):
|
|
nano_aiu = copilot_usage.get("totalNanoAiu")
|
|
if isinstance(nano_aiu, (int, float)):
|
|
acc["_cost_nano_aiu"] = acc.get("_cost_nano_aiu", 0) + int(nano_aiu)
|
|
|
|
|
|
def _finalize_usage(acc: dict[str, int]) -> dict[str, Any] | None: # type: ignore[explicit-any]
|
|
"""Build the TurnComplete usage dict from accumulated counts, or ``None``."""
|
|
if not acc:
|
|
return None
|
|
usage: dict[str, Any] = dict(acc) # type: ignore[explicit-any]
|
|
nano_aiu = usage.pop("_cost_nano_aiu", 0)
|
|
usage["total_tokens"] = usage.get("input_tokens", 0) + usage.get("output_tokens", 0)
|
|
if nano_aiu:
|
|
# 1 AIC = 1e9 nano-AIU = $0.01, so nano-AIU / 1e11 = USD.
|
|
usage["cost_usd"] = nano_aiu / 1e11
|
|
return usage
|
|
|
|
|
|
async def _safe_stop(obj: Any) -> None: # type: ignore[explicit-any]
|
|
"""Best-effort async teardown of a ``copilot`` client / session.
|
|
|
|
:class:`copilot.CopilotClient` exposes ``stop()`` (terminates the bundled
|
|
CLI subprocess it launched); :class:`copilot.CopilotSession` exposes
|
|
``disconnect()``. Prefer ``stop`` then ``disconnect``; a teardown failure
|
|
must not mask the original error or leave the closer raising.
|
|
"""
|
|
closer = (
|
|
getattr(obj, "stop", None)
|
|
or getattr(obj, "disconnect", None)
|
|
or getattr(obj, "aclose", None)
|
|
or getattr(obj, "close", None)
|
|
)
|
|
if closer is None:
|
|
return
|
|
try:
|
|
result = closer()
|
|
if asyncio.iscoroutine(result):
|
|
await result
|
|
except Exception as exc: # noqa: BLE001 — best-effort teardown
|
|
logger.debug("CopilotExecutor: stop failed: %s", exc)
|