omnigent-ai--omnigent/omnigent/inner/copilot_executor.py

"""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, Protocol, 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")

# The Copilot CLI's GHE hostname override (``copilot help environment``). Spelled
# here rather than imported from ``omnigent.onboarding.copilot_auth``: onboarding
# pulls in the secret store / keyring, which this path imports lazily. A test
# pins the two spellings equal.
COPILOT_HOST_ENV_VAR = "COPILOT_GH_HOST"

# The user's ambient host, captured before we ever write the var ourselves.
_AMBIENT_HOST = os.environ.get(COPILOT_HOST_ENV_VAR) or None

# 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: object) -> object:
    """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
# ---------------------------------------------------------------------------


class _CopilotSession(Protocol):
    """SDK session methods used by the executor."""

    def on(self, callback: Callable[[object], None]) -> Callable[[], None]:
        pass

    async def send_and_wait(self, prompt: str, *, timeout: float) -> object:
        pass

    async def abort(self) -> object:
        pass


@dataclass
class _CopilotSessionState:
    """Per-Omnigent-conversation SDK session state."""

    client: object | None = None
    session: _CopilotSession | None = None
    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,
        github_host: 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, then the ``gh`` CLI login.
        :param github_host: GitHub Enterprise hostname (e.g. ``"acme.ghe.com"``)
            to authenticate against. ``None`` falls back to the configured
            ``copilot.github_host``, then Copilot's default ``github.com``.
        :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
        # Resolved before the token: a GHE user's token must come from that host.
        self._github_host = github_host or _configured_github_host()
        self._github_token = github_token or _ambient_github_token(self._github_host)
        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, object]], Awaitable[object]] | 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, object]], 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: object,
        _invocation: dict[str, str],
    ) -> object:
        """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())
        # A GHE hostname reaches the bundled CLI only as an env var; the SDK has
        # no host parameter. Set it in our own environment rather than passing
        # ``env=`` — the SDK inherits ``os.environ`` only when ``env`` is None,
        # so handing it a dict would strip everything else from the subprocess.
        # Assign both ways so a hostless executor can't inherit a host another
        # one left behind.
        if self._github_host:
            os.environ[COPILOT_HOST_ENV_VAR] = self._github_host
        else:
            os.environ.pop(COPILOT_HOST_ENV_VAR, None)
        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
        assert session is not None

        # 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(host: str | None = None) -> str | None:
    """Return an ambient GitHub token: env vars first, then the ``gh`` CLI login.

    The env vars come first because the Copilot CLI honors them in this same
    precedence. The ``gh`` fallback covers the common "I ran ``gh auth login``"
    case: Copilot only picks a ``gh`` login up by reading ``oauth_token`` from
    ``~/.config/gh/hosts.yml``, which is absent whenever ``gh`` stores the token
    in an OS keychain instead (the default on macOS).

    :param host: GHE hostname whose ``gh`` token to fetch; ``None`` resolves the
        configured host, so a GHE user's token comes from their own instance.
    """
    for var in GITHUB_TOKEN_ENV_VARS:
        value = os.environ.get(var)
        if value:
            return value
    from omnigent.onboarding.copilot_auth import gh_cli_github_token

    return gh_cli_github_token(host if host is not None else _configured_github_host())


def _configured_github_host() -> str | None:
    """Return the configured GHE hostname: env var first, then the config block.

    Reads the *ambient* env var captured at import, not the live one — the
    executor writes ``COPILOT_HOST_ENV_VAR`` to hand the host to the bundled CLI,
    and reading that back would let one executor's host leak into a later
    hostless one.
    """
    from omnigent.onboarding.copilot_auth import copilot_github_host

    return _AMBIENT_HOST or copilot_github_host()


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)