omnigent-ai--omnigent/omnigent/antigravity_native_rpc.py

"""connect-RPC client for a running native Antigravity (agy) process.

This is the **discovery / read / control helper** for the native agy paths: the
RPC read driver (:mod:`omnigent.antigravity_native_reader`, which polls/streams
trajectory steps to mirror agy's conversation) and the native executor
(:mod:`omnigent.inner.antigravity_native_executor`, which delivers web/mobile
turns). It provides, over agy's connect-RPC surface:

* **conversation-ownership discovery** (:func:`resolve_language_server_port` /
  :func:`get_trajectory_steps`) — the read side the reader binds to;
* **turn delivery** via ``SendUserCascadeMessage``
  (:func:`send_user_cascade_message`), which agy records as a real ``USER_INPUT``
  turn — used by the executor instead of agy's ``SendAgentMessage`` RPC (which
  agy records as a ``SYSTEM_MESSAGE``, "not actually sent by the user", and would
  never mirror as a user turn) and instead of typing into the TUI over tmux;
* **turn interrupt** via ``CancelCascadeSteps`` (:func:`cancel_cascade_steps`),
  used by the executor's interrupt path.

How agy exposes a control surface (verified end-to-end; see
``docs/claude/antigravity-sidecar-spike.md``):

* A running ``agy`` process opens **two** ephemeral ``127.0.0.1`` TCP LISTEN
  ports. The **lower** one is a TLS HTTP/2 **connect-RPC** server hosting
  ``exa.language_server_pb.LanguageServerService``; the higher one is a plain
  HTTP surface that 404s. The TLS cert is self-signed, so the client uses
  ``verify=False``.
* The ports are ephemeral and not configurable
  (``ANTIGRAVITY_SIDECAR_WEB_PORT`` is a sidecar-plugin no-op), so they are
  discovered from the loopback socket table — ``lsof`` per agy pid, falling back
  to ``/proc/net/tcp`` on hosts where ``lsof`` cannot attribute the socket.
* Ownership probe: ``POST .../GetConversationMetadata`` with REQUEST body
  ``{"conversationId": "<id>"}`` returns HTTP 200 whose RESPONSE echoes that id at
  ``metadata.rootConversationId`` for a hosted conversation, and HTTP 500
  ("trajectory not found") for an unknown one — so a caller can confirm which
  live agy owns a conversation before binding its port. (Request and response
  shapes differ: the id is sent flat as ``conversationId`` and echoed nested
  under ``metadata``.)

Port discovery is **port-first**: it enumerates candidate loopback connect-RPC
ports (see :func:`_candidate_agy_rpc_ports`) and, for each, checks whether its
server reports the target ``conversation_id`` via ``GetConversationMetadata`` —
so the right port is found even when several agy instances run. There is no pid
key: agy is launched under ``tmux_start_on_attach`` (CLI) and does not exist at
launch, and on some hosts agy's listening socket is owned by a backend that is
neither the agy pid nor ``lsof``-attributable. The conversation-ownership check
is what makes a discovered port trustworthy, rejecting a recycled/foreign port
that hosts a different agy.

Everything that touches the OS (``lsof``, process enumeration) or the network
(httpx) is funnelled through small module-level seams so the unit tests can mock
them without real subprocesses or sockets.
"""

from __future__ import annotations

import ipaddress
import json
import logging
import os
import struct
import subprocess
from collections.abc import AsyncIterator, Iterable
from pathlib import Path
from typing import NamedTuple
from urllib.parse import urlparse

import httpx

_logger = logging.getLogger(__name__)

# connect-RPC service + methods on agy's TLS port.
_LS_SERVICE = "exa.language_server_pb.LanguageServerService"
_METHOD_HEARTBEAT = "Heartbeat"
_METHOD_GET_CONVERSATION_METADATA = "GetConversationMetadata"
# Turn-cancel methods. The NAMES are verified present in the agy 1.0.8 binary
# (``strings`` → ``LanguageServerService/{CancelCascadeInvocation,
# CancelCascadeSteps,ForceStopCascadeTree}``), but their request CONTRACTS are
# NOT verified: the proto field tags show ``CancelCascadeInvocationRequest`` and
# friends key on an internal ``cascade_id`` / ``invocation_id`` (agy's per-turn
# identifiers) that the transcript forwarder does NOT have — it only knows the
# *conversation* id, not the live cascade/invocation id — and the stop semantics
# are unconfirmed on a live process. The best-effort interrupt
# (:func:`interrupt_turn`) is therefore wired OFF by default; see its TODO.
_METHOD_FORCE_STOP_CASCADE_TREE = "ForceStopCascadeTree"
_METHOD_GET_CASCADE_TRAJECTORY_STEPS = "GetCascadeTrajectorySteps"
_METHOD_CANCEL_CASCADE_STEPS = "CancelCascadeSteps"
_METHOD_HANDLE_CASCADE_USER_INTERACTION = "HandleCascadeUserInteraction"
_METHOD_SEND_USER_CASCADE_MESSAGE = "SendUserCascadeMessage"
_METHOD_START_CASCADE = "StartCascade"
_METHOD_GET_AVAILABLE_MODELS = "GetAvailableModels"
_METHOD_GET_ALL_CASCADE_TRAJECTORIES = "GetAllCascadeTrajectories"
_METHOD_STREAM_AGENT_STATE_UPDATES = "StreamAgentStateUpdates"

_LOOPBACK = "127.0.0.1"

# Hostnames that are unconditionally loopback. Any other host is checked
# numerically via :func:`ipaddress.ip_address` in :func:`_assert_loopback_url`.
_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})

# Timeout for the liveness/validation probes used during discovery (Heartbeat +
# GetConversationMetadata), kept tight so scanning several candidate ports stays
# fast.
_PROBE_TIMEOUT_S = 2.0

# Timeout for the FUNCTIONAL connect-RPCs (cold-start, turn delivery, interaction
# answer, step reads, cancel) — generous headroom over the tight discovery-probe
# budget. The functional calls hit a known, owned, loopback port, so the only
# reason one would exceed a couple of seconds is a momentarily busy agy; the tight
# ``_PROBE_TIMEOUT_S`` would then raise a transport ``TimeoutException`` that the
# caller does not retry — the interaction bridge only retries on "input not
# registered" (abandoning a human-answer delivery), and the cold-start does NOT
# retry ``StartCascade`` at all (a slow ack would leave the placeholder id
# deadlocked). This deadline still bounds a truly-hung agy so a functional call
# cannot block forever.
_RPC_CALL_TIMEOUT_S = 30.0

# Timeout policy for the persistent ``StreamAgentStateUpdates`` long-poll. The
# connect-stream stays open across the whole turn (and idles between turns), so
# the READ timeout is disabled (``None``) — a tight per-read deadline would abort
# the stream during normal think-time. The connect/write/pool deadlines stay
# short so a port that never accepts the connection fails fast (the reader is
# expected to reconnect / fall back to polling) rather than hanging the caller.
_STREAM_TIMEOUT_S = 10.0
_STREAM_TIMEOUT = httpx.Timeout(_STREAM_TIMEOUT_S, read=None)

# connect-protocol stream framing: each frame is ``[flag: 1B][len: 4B BE][payload]``.
# The 5-byte fixed header precedes a ``len``-byte payload. The flag is a bitfield:
# bit 0 (``0x01``) marks a COMPRESSED payload (agy sends uncompressed — a set bit
# means the JSON decode would be wrong, so the client raises); bit 1 (``0x02``)
# marks the end-of-stream TRAILER (stop iterating; the payload may carry trailer
# metadata or an error). ``flag == 0x00`` is a DATA message (payload is the JSON
# ``update`` object the generator yields).
_FRAME_HEADER_LEN = 5
_FRAME_FLAG_COMPRESSED = 0x01
_FRAME_FLAG_TRAILER = 0x02

# ``lsof`` flags: ``-a`` ANDs the filters (without it ``-p`` and ``-i`` are ORed
# and the pid filter is ignored), ``-nP`` skips name/port resolution (fast),
# ``-p`` restricts to the pid, ``-iTCP -sTCP:LISTEN`` selects TCP listeners.
_LSOF_TIMEOUT_S = 5.0

# Root of the Linux proc filesystem, scanned by ``_list_agy_pids_from_proc`` when
# ``pgrep`` is unavailable, and by ``_list_loopback_listen_ports`` for the socket
# table. A module constant so tests can repoint it at a fake tree without
# monkeypatching ``os``.
_PROC_FS = "/proc"

# Upper bound on how many loopback listeners ``_candidate_agy_rpc_ports`` will
# Heartbeat-probe in the ``/proc/net/tcp`` fallback, so a host with an unusual
# number of loopback services cannot make turn injection block for
# ``N * _PROBE_TIMEOUT_S``. Far above any real host's loopback listener count
# (the agy host pods have ~3); a breach is logged, never silent.
_MAX_FALLBACK_PROBE_PORTS = 32

# Cap on how many conversation ids are echoed into the ambiguity-refusal warning
# in ``conversation_id_owned_by_pid`` so a pathological candidate set cannot blow
# up a log line.
_MAX_LOGGED_AMBIGUOUS_IDS = 10


def _assert_loopback_url(url: str) -> None:
    """
    Refuse any connect-RPC URL whose host is not loopback.

    The connect-RPC clients disable TLS verification (agy's cert is self-signed),
    which is only safe because the endpoint is loopback-only. The port is
    discovered dynamically per session, so this guards every request against a
    URL that ever resolved to a non-loopback host — there, ``verify=False`` would
    silently trust any cert.

    :param url: Full request URL, e.g. ``"https://127.0.0.1:52548/svc/Method"``.
    :returns: None.
    :raises ValueError: When the URL's host is not a loopback address.
    """
    host = urlparse(url).hostname or ""
    if host in _LOOPBACK_HOSTS:
        return
    try:
        if ipaddress.ip_address(host).is_loopback:
            return
    except ValueError:
        pass
    raise ValueError(f"refusing non-loopback connect-RPC URL (verify is disabled): {url!r}")


def _rpc_url(port: int, method: str) -> str:
    """
    Build the connect-RPC URL for a LanguageServerService method.

    :param port: agy connect-RPC (TLS) port, e.g. ``52548``.
    :param method: RPC method name, e.g. ``"SendAgentMessage"``.
    :returns: Full ``https://127.0.0.1:<port>/<service>/<method>`` URL.
    """
    return f"https://{_LOOPBACK}:{port}/{_LS_SERVICE}/{method}"


# httpx transport seam. ``None`` (production) lets httpx use its real loopback
# TLS transport with cert verification disabled (agy's cert is self-signed and
# the endpoint is loopback-only). Tests set this to an ``httpx.MockTransport``
# to assert the URL / headers / body of each RPC without a real socket.
#
# The sync seam backs the live discovery probes (Heartbeat +
# GetConversationMetadata). The async seam is reserved for :func:`interrupt_turn`'s
# future POST — that function is async (the forwarder ``await``s it) but is wired
# OFF pending request-contract verification, so ``_async_client`` has no live
# caller yet; the async seam is exercised by its guard test, which asserts no
# async RPC fires while the interrupt is off.
_HTTP_TRANSPORT: httpx.BaseTransport | None = None
_ASYNC_HTTP_TRANSPORT: httpx.AsyncBaseTransport | None = None


def _sync_client(timeout: float) -> httpx.Client:
    """
    Build a sync httpx client for a connect-RPC probe.

    Cert verification is disabled because agy's loopback cert is self-signed;
    this is safe only because every request URL is checked by
    :func:`_assert_loopback_url` before it is sent, so the client never trusts an
    unverified cert from a non-loopback host.

    :param timeout: Per-request timeout in seconds.
    :returns: An ``httpx.Client`` with cert verification disabled (loopback,
        self-signed) and the test transport when one is installed.
    """
    return httpx.Client(verify=False, timeout=timeout, transport=_HTTP_TRANSPORT)


def _async_client(timeout: httpx.Timeout | float) -> httpx.AsyncClient:
    """
    Build an async httpx client for a connect-RPC call.

    Backs the :func:`stream_agent_state_updates` connect server-stream (a
    persistent long-poll, which is why callers pass an :class:`httpx.Timeout`
    with the read deadline disabled rather than a flat float). Cert verification
    is disabled because agy's loopback cert is self-signed; this is safe only
    because every request URL is checked by :func:`_assert_loopback_url` before
    it is sent, so the client never trusts an unverified cert from a non-loopback
    host.

    :param timeout: Per-request timeout — a flat seconds float, or an
        :class:`httpx.Timeout` for finer control (e.g. the streaming client
        disables the read deadline so the long-poll is not aborted mid-turn).
    :returns: An ``httpx.AsyncClient`` with cert verification disabled
        (loopback, self-signed) and the test transport when one is installed.
    """
    return httpx.AsyncClient(verify=False, timeout=timeout, transport=_ASYNC_HTTP_TRANSPORT)


def _run_lsof_listen_ports(pid: int) -> str:
    """
    Return raw ``lsof`` output listing a pid's TCP LISTEN sockets.

    Isolated as a seam so tests can stub the subprocess. A non-zero exit (e.g.
    the process is gone) or a missing ``lsof`` yields ``""`` rather than raising
    — discovery treats "no ports" the same as "lsof unavailable".

    :param pid: agy process id, e.g. ``72753``.
    :returns: ``lsof`` stdout, or ``""`` on any failure.
    """
    try:
        completed = subprocess.run(
            ["lsof", "-nP", "-a", "-p", str(pid), "-iTCP", "-sTCP:LISTEN"],
            capture_output=True,
            text=True,
            timeout=_LSOF_TIMEOUT_S,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        _logger.warning("lsof failed for agy pid=%s", pid, exc_info=True)
        return ""
    return completed.stdout


def _parse_loopback_listen_ports(lsof_output: str) -> list[int]:
    """
    Parse ascending unique ``127.0.0.1`` LISTEN ports from ``lsof`` output.

    Only IPv4 loopback (``127.0.0.1:<port>``) listeners are considered — agy
    binds its connect-RPC + HTTP ports there. The NAME column looks like
    ``127.0.0.1:52548 (LISTEN)``.

    :param lsof_output: Raw ``lsof -iTCP -sTCP:LISTEN`` stdout.
    :returns: Sorted, de-duplicated loopback port numbers (lowest first).
    """
    prefix = f"{_LOOPBACK}:"
    ports: set[int] = set()
    for line in lsof_output.splitlines():
        for token in line.split():
            if not token.startswith(prefix):
                continue
            port_text = token[len(prefix) :]
            if port_text.isdigit():
                ports.add(int(port_text))
    return sorted(ports)


# /proc/net/tcp local-address column is the little-endian hex of the bound IPv4
# address. 127.0.0.1 -> bytes [7F,00,00,01] -> u32 0x0100007F -> "0100007F".
# Matched case-insensitively against this exact form (agy binds the loopback host
# on IPv4, never a 127.0.0.0/8 alias).
_LOOPBACK_HEX_V4 = "0100007F"


def _is_loopback_hex_addr(addr_hex: str) -> bool:
    """
    Return whether a ``/proc/net/tcp`` local-address hex column is IPv4 loopback.

    :param addr_hex: The hex local address, e.g. ``"0100007F"`` (127.0.0.1).
    :returns: ``True`` for the IPv4 ``127.0.0.1`` encoding only.
    """
    return addr_hex.upper() == _LOOPBACK_HEX_V4


def _list_loopback_listen_ports() -> list[int]:
    """
    Return all IPv4 ``127.0.0.1`` TCP LISTEN ports from ``/proc/net/tcp``.

    The robust fallback for :func:`_candidate_agy_rpc_ports` on hosts where
    ``lsof -p <pid>`` cannot attribute a listening socket to its owning pid —
    e.g. the uid-1000 k8s pods where agy 1.0.10 holds its connect-RPC listener
    in a backend the agy process does not own as a file descriptor, so neither
    ``lsof`` nor a ``/proc/<pid>/fd`` scan finds it. The kernel's
    network-namespace socket table (``/proc/net/tcp``) lists every listener
    regardless of fd ownership and needs no ptrace/fd permission.

    IPv4 only: agy binds ``127.0.0.1`` and the connect-RPC client
    (:func:`_rpc_url`) dials ``127.0.0.1``, so an IPv6 ``::1`` listener would be
    unreachable regardless — enumerating it would only add a dead probe. Parses
    state ``0A`` (``TCP_LISTEN``). A missing table (non-Linux) yields ``[]``.

    :returns: Sorted, de-duplicated loopback LISTEN port numbers.
    """
    ports: set[int] = set()
    try:
        with open(os.path.join(_PROC_FS, "net", "tcp"), "rb") as handle:
            raw = handle.read().decode("ascii", "replace")
    except OSError:
        return []  # non-Linux, or no /proc
    for line in raw.splitlines()[1:]:  # row 0 is the column header
        fields = line.split()
        if len(fields) < 4 or fields[3] != "0A":  # 0A == TCP_LISTEN
            continue
        addr_hex, _sep, port_hex = fields[1].partition(":")
        if not _is_loopback_hex_addr(addr_hex):
            continue
        try:
            ports.add(int(port_hex, 16))
        except ValueError:
            continue
    return sorted(ports)


def _heartbeat_ok(port: int) -> bool:
    """
    Return whether a port answers the connect-RPC ``Heartbeat`` with HTTP 200.

    This is the canonical "is this the TLS connect-RPC port" probe: agy's lower
    port returns 200 for ``Heartbeat {}``; the higher plain-HTTP port does not
    (it 404s, or the TLS handshake fails). Any transport/TLS error counts as
    "not it".

    :param port: Candidate loopback port, e.g. ``52548``.
    :returns: ``True`` only when ``Heartbeat`` returns HTTP 200.
    """
    url = _rpc_url(port, _METHOD_HEARTBEAT)
    _assert_loopback_url(url)
    try:
        with _sync_client(_PROBE_TIMEOUT_S) as client:
            response = client.post(
                url,
                headers={"Content-Type": "application/json"},
                content=b"{}",
            )
    except httpx.HTTPError:
        return False
    return response.status_code == 200


def _conversation_matches(port: int, conversation_id: str) -> bool:
    """
    Return whether the agy on ``port`` owns ``conversation_id``.

    Used to pick the right port when several agy instances are running (or no
    pid was captured). For an id it hosts, agy's ``GetConversationMetadata``
    returns HTTP 200 with a ``metadata`` object that **echoes the resolved id**
    as ``metadata.rootConversationId`` (verified live); for an id it does not
    host it returns HTTP 500 (``"trajectory not found"``).

    The id echo — not merely a 200 with *some* metadata — is what makes the
    port-first candidate set (which, in the ``/proc/net/tcp`` fallback, spans
    every loopback listener, not just agy's) safe to write to: a non-agy service
    that happened to answer ``Heartbeat`` 200, or an agy hosting a *different*
    conversation, cannot echo this exact id and is rejected before any
    ``SendAgentMessage``. Fails closed (returns ``False``) on any shape it does
    not recognize.

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param conversation_id: agy conversation id to look for, e.g.
        ``"90468e33-..."``.
    :returns: ``True`` only when the server confirms it hosts ``conversation_id``
        (200 + ``metadata.rootConversationId == conversation_id``).
    """
    url = _rpc_url(port, _METHOD_GET_CONVERSATION_METADATA)
    _assert_loopback_url(url)
    try:
        with _sync_client(_PROBE_TIMEOUT_S) as client:
            response = client.post(
                url,
                headers={"Content-Type": "application/json"},
                content=json.dumps({"conversationId": conversation_id}).encode("utf-8"),
            )
    except httpx.HTTPError:
        return False
    if response.status_code != 200:
        return False
    try:
        body = response.json()
    except ValueError:
        return False
    if not isinstance(body, dict):
        return False
    metadata = body.get("metadata")
    if not isinstance(metadata, dict):
        return False
    return metadata.get("rootConversationId") == conversation_id


def get_trajectory_steps(port: int, cascade_id: str) -> list[dict[str, object]]:
    """
    Return the trajectory steps for ``cascade_id`` from the agy on ``port``.

    POSTs ``{"cascadeId": cascade_id}`` to ``GetCascadeTrajectorySteps`` and
    returns the ``steps`` list from the response body (empty list when the key
    is absent). Used by the read driver to poll incremental step progress.

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param cascade_id: agy cascade id (equal to the conversation id) to query.
    :returns: List of step dicts, each containing at least ``stepIndex`` and
        ``status`` (may be empty when no steps have been recorded yet).
    :raises httpx.HTTPError: On transport errors or non-2xx responses; the
        Task 6 read driver is responsible for retry/backoff. Intentionally NOT
        fail-open (unlike ``_conversation_matches`` which is a discovery probe);
        a non-2xx here is a hard read failure, not a "not found" signal.
    :raises ValueError: When the 2xx response body is not valid JSON (body is
        decoded; the Task 6 driver catches broadly).
    """
    url = _rpc_url(port, _METHOD_GET_CASCADE_TRAJECTORY_STEPS)
    _assert_loopback_url(url)
    with _sync_client(_RPC_CALL_TIMEOUT_S) as client:
        response = client.post(
            url,
            headers={"Content-Type": "application/json"},
            content=json.dumps({"cascadeId": cascade_id}).encode("utf-8"),
        )
    # Raises httpx.HTTPStatusError (subclass of httpx.HTTPError) on non-2xx so
    # the body is never decoded on error paths (which may not be JSON).
    response.raise_for_status()
    body = response.json()  # ValueError on non-JSON 200 propagates (documented)
    steps = body.get("steps") if isinstance(body, dict) else None
    return list(steps) if isinstance(steps, list) else []


def cancel_cascade_steps(port: int, cascade_id: str) -> bool:
    """
    Request cancellation of the active cascade steps for ``cascade_id``.

    POSTs ``{"cascadeId": cascade_id}`` to ``CancelCascadeSteps`` and returns
    ``True`` when the server responds with a non-error HTTP status (< 400).
    Fails open (returns ``False``) on any transport error so the executor can
    treat it as best-effort.

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param cascade_id: agy cascade id (equal to the conversation id) to cancel.
    :returns: ``True`` when the server accepted the cancel (HTTP < 400),
        ``False`` on any error or rejection.
    """
    url = _rpc_url(port, _METHOD_CANCEL_CASCADE_STEPS)
    _assert_loopback_url(url)
    try:
        with _sync_client(_RPC_CALL_TIMEOUT_S) as client:
            response = client.post(
                url,
                headers={"Content-Type": "application/json"},
                content=json.dumps({"cascadeId": cascade_id}).encode("utf-8"),
            )
    except Exception:  # deliberate fail-open: ssl.SSLError etc. outside httpx hierarchy
        return False
    return response.status_code < 400


class AntigravityRpcError(Exception):
    """
    Raised by :func:`handle_user_interaction` when the server returns a
    non-2xx status.

    The raw response body text is the exception message so callers can detect
    the overloaded ``"input not registered for step N"`` string that agy
    returns when the interaction has not yet been registered for the step
    (a race the Task 8 bridge must retry on).
    """


def _post_rpc_raising(port: int, method: str, body: dict[str, object]) -> None:
    """
    POST a JSON body to a functional connect-RPC method, raising on any failure.

    The shared request tail for the functional, raise-on-error RPCs
    (:func:`handle_user_interaction`, :func:`send_user_cascade_message`,
    :func:`start_cascade`): build + loopback-check the URL, POST ``body`` as JSON
    on a :data:`_RPC_CALL_TIMEOUT_S` sync client, and normalize failures so every
    caller has a single :class:`AntigravityRpcError` type to catch — transport
    errors are wrapped, and a non-2xx raises with the RAW response body text (NOT
    ``raise_for_status()``) so callers can detect agy's overloaded error strings
    (e.g. ``"input not registered for step N"``).

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param method: LanguageServerService method name, e.g.
        ``"SendUserCascadeMessage"``.
    :param body: The request object to JSON-encode and POST.
    :returns: ``None`` on success (HTTP < 400).
    :raises AntigravityRpcError: On a transport error (wrapped) or any HTTP
        status >= 400 (message is the raw response body text).
    """
    url = _rpc_url(port, method)
    _assert_loopback_url(url)
    try:
        with _sync_client(_RPC_CALL_TIMEOUT_S) as client:
            response = client.post(
                url,
                headers={"Content-Type": "application/json"},
                content=json.dumps(body).encode("utf-8"),
            )
    except httpx.HTTPError as e:
        raise AntigravityRpcError(f"transport error contacting agy: {e}") from e
    if response.status_code >= 400:
        raise AntigravityRpcError(response.text)


def handle_user_interaction(
    port: int,
    cascade_id: str,
    *,
    trajectory_id: str,
    step_index: int,
    payload: dict[str, object],
) -> None:
    """
    Deliver an interaction answer (question response / approval) to agy.

    POSTs ``{"cascadeId": cascade_id, "interaction": {"trajectoryId":
    trajectory_id, "stepIndex": step_index, **payload}}`` to
    ``HandleCascadeUserInteraction``. The ``trajectoryId`` and ``stepIndex``
    are nested inside ``interaction`` because the proto-JSON encoding drops
    top-level extras — they must be co-located with the payload variant dict.

    ``cascade_id`` is identical to the conversation id (agy uses the same
    UUID for both).

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param cascade_id: agy cascade id (equal to the conversation id) to
        address.
    :param trajectory_id: agy trajectory id identifying the active trajectory.
    :param step_index: Step index the interaction targets.
    :param payload: Variant dict, e.g. ``{"permission": {"allow": True}}`` or
        ``{"askQuestion": {...}}``.
    :raises AntigravityRpcError: On transport errors (e.g. connection refused)
        or any HTTP status >= 400. Transport errors are wrapped so the bridge
        has one exception type to catch, regardless of whether the failure was
        on the wire or at the application layer. On non-2xx, the raw response
        body text is the message (NOT ``raise_for_status()``) so callers can
        detect the overloaded ``"input not registered for step N"`` string.
    """
    body: dict[str, object] = {
        "cascadeId": cascade_id,
        "interaction": {"trajectoryId": trajectory_id, "stepIndex": step_index, **payload},
    }
    _post_rpc_raising(port, _METHOD_HANDLE_CASCADE_USER_INTERACTION, body)


def send_user_cascade_message(
    port: int,
    cascade_id: str,
    text: str,
    *,
    plan_model: str,
) -> None:
    """
    Send a user turn to agy via connect-RPC (replaces tmux send-keys).

    POSTs ``{"cascadeId": cascade_id, "items": [{"text": text}],
    "cascadeConfig": {"plannerConfig": {"planModel": plan_model}}}`` to
    ``SendUserCascadeMessage``. This records the turn as
    ``CORTEX_STEP_TYPE_USER_INPUT`` with
    ``metadata.source==CORTEX_STEP_SOURCE_USER_EXPLICIT``, which the read
    driver keys on — in contrast to ``SendAgentMessage`` which records as a
    ``SYSTEM_MESSAGE`` and is therefore unsuitable.

    Two shape constraints (live-verified against agy 1.0.10, see §10.1 of the
    design doc):

    * The turn text MUST be in ``items[].text`` (a list of objects) — a flat
      ``"message"`` field is not the correct schema.
    * The ``plan_model`` MUST be present at
      ``cascadeConfig.plannerConfig.planModel``; omitting it causes agy to
      error "neither PlanModel nor RequestedModel specified".

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param cascade_id: agy cascade id (equal to the conversation id) to send
        the turn to.
    :param text: The turn text to send (the user's message).
    :param plan_model: agy model enum string, resolved at runtime via
        :func:`get_available_models` or echoed from the read-side
        ``userInput.userConfig.plannerConfig.requestedModel.model``. Must not
        be empty or omitted.
    :returns: ``None`` on success (HTTP 200).
    :raises AntigravityRpcError: On transport errors (e.g. connection refused)
        or any HTTP status >= 400. Transport errors are wrapped so the executor
        has one exception type to catch. On non-2xx, the raw response body text
        is the message (NOT ``raise_for_status()``) so callers can surface agy
        model/validation errors (e.g. "neither PlanModel nor RequestedModel
        specified") directly. Mirrors :func:`handle_user_interaction`.
    """
    body: dict[str, object] = {
        "cascadeId": cascade_id,
        "items": [{"text": text}],
        "cascadeConfig": {"plannerConfig": {"planModel": plan_model}},
    }
    _post_rpc_raising(port, _METHOD_SEND_USER_CASCADE_MESSAGE, body)


def start_cascade(
    port: int,
    cascade_id: str,
    *,
    source: str = "CORTEX_TRAJECTORY_SOURCE_CLI",
) -> None:
    """
    Cold-start (create) an agy conversation over connect-RPC.

    POSTs ``{"cascadeId": cascade_id, "source": source}`` to ``StartCascade``,
    creating agy's brain dir (named after *cascade_id*) so the runner owns the
    conversation id from turn-1 instead of waiting for the TUI to lazily mint one
    on its first typed turn (live-verified against agy 1.0.10; see
    ``agy-rpc-interaction-bridge`` / ``/tmp/agy-coldstart-spike.md``).

    Two contract facts (live-verified):

    * ``source`` is the ONLY required field. ``cascadeId`` is optional — when
      omitted agy mints and returns one — but the runner PROVIDES it so it owns
      the id (the response echoes the same id back). No model is needed here; the
      model is selected per-turn by :func:`send_user_cascade_message`.
    * An RPC-created conversation does NOT surface in the agy TUI (it stays on the
      empty ``>`` banner); the brain dir / backend is the source of truth, which
      is correct for the headless RPC runner.

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param cascade_id: Runner-generated conversation/cascade id (a ``uuid4``) the
        new conversation will own.
    :param source: agy trajectory-source enum string. Defaults to
        ``"CORTEX_TRAJECTORY_SOURCE_CLI"`` (the headless-CLI source); kept a
        parameter so a future enum rename is a one-line caller change.
    :returns: ``None`` on success (HTTP 200).
    :raises AntigravityRpcError: On transport errors (e.g. connection refused) or
        any HTTP status >= 400. Transport errors are wrapped so the runner has one
        exception type to catch. On non-2xx, the raw response body text is the
        message (NOT ``raise_for_status()``) so the runner can surface agy's
        error verbatim. Mirrors :func:`send_user_cascade_message`.
    """
    body: dict[str, object] = {"cascadeId": cascade_id, "source": source}
    _post_rpc_raising(port, _METHOD_START_CASCADE, body)


def get_available_models(port: int) -> dict[str, object]:
    """
    Return the agy model catalog from ``GetAvailableModels``.

    POSTs ``{}`` to ``GetAvailableModels`` and returns the catalog object
    (unwrapped from the response envelope — see below). Used to resolve a model
    enum string at runtime for :func:`send_user_cascade_message` (the model is
    required per-turn and must not be hardcoded — ``planModel`` values are
    version-volatile enums).

    The response shape (live-verified, agy 1.0.10) wraps the catalog under a
    top-level ``"response"`` key::

        {"response": {
            "models": {
                "<key>": {
                    "model": "<enum-string>",
                    "displayName": "<human label>",
                    "recommended": <bool>,
                    "supportsThinking": <bool>,
                    "thinkingBudget": <int>
                },
                ...
            },
            "defaultAgentModelId": "<enum-string>",
            "tieredModelIds": {...},
            ...
        }}

    This function unwraps that envelope and returns the inner object, so callers
    see ``{"models": ..., "defaultAgentModelId": ...}`` at the top level and can
    key on ``catalog["models"]``. Pick the desired entry by ``recommended ==
    True`` or by matching the ``displayName`` / ``model`` string the user
    selected. The echo-from-read-side shortcut
    (``userInput.userConfig.plannerConfig.planModel``) is preferred when the
    current model is already known from a prior turn's step data — call this only
    when no prior model is available.

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :returns: The unwrapped catalog (a dict, at minimum ``{"models": {...}}``):
        ``body["response"]`` when the 200 body is a dict carrying a dict under
        ``"response"``; otherwise the body itself (defensive, for a future agy
        that drops the envelope). Returns ``{}`` if the 200 body is not a dict
        (guards against a future agy returning a non-object 200).
    :raises httpx.HTTPError: On transport errors or non-2xx responses; the
        executor is responsible for retry/backoff. Intentionally NOT fail-open
        (mirrors :func:`get_trajectory_steps`): a non-2xx is a hard read
        failure, not a "not found" signal.
    :raises ValueError: When the 2xx response body is not valid JSON (body is
        decoded; the caller catches broadly).
    """
    url = _rpc_url(port, _METHOD_GET_AVAILABLE_MODELS)
    _assert_loopback_url(url)
    with _sync_client(_RPC_CALL_TIMEOUT_S) as client:
        response = client.post(
            url,
            headers={"Content-Type": "application/json"},
            content=b"{}",
        )
    # Raises httpx.HTTPStatusError (subclass of httpx.HTTPError) on non-2xx so
    # the body is never decoded on error paths (which may not be JSON).
    response.raise_for_status()
    body = response.json()  # ValueError on non-JSON 200 propagates (documented)
    if not isinstance(body, dict):
        return {}
    # Live agy wraps the catalog under "response"; unwrap so callers read
    # ``catalog["models"]`` at the top level. Fall back to the body itself if a
    # future agy ever drops the envelope (defensive — keeps callers working).
    inner = body.get("response")
    return inner if isinstance(inner, dict) else body


def get_all_cascade_trajectories(port: int) -> dict[str, object]:
    """
    Return agy's cascade-trajectory summaries from ``GetAllCascadeTrajectories``.

    POSTs ``{}`` to ``GetAllCascadeTrajectories`` and returns the parsed response
    body. This is the PRIMARY ``/clear``-rotation signal for the Task T-G reader: a
    ``StreamAgentStateUpdates`` stream is bound to ONE cascade and only ever
    reports THAT cascade's id, so it cannot observe a sibling conversation; this
    surface, by contrast, lists EVERY live root cascade, so a freshly ``/clear``-
    minted conversation becomes visible here as a new, more-recently-active entry.

    The response shape (live-verified, agy 1.0.10) keys each summary by its ROOT
    conversation id::

        {"trajectorySummaries": {
            "<rootConversationId>": {
                "trajectoryId": "<uuid>",
                "status": "CASCADE_RUN_STATUS_IDLE",
                "createdTime": "2026-06-23T17:18:24.402333Z",
                "lastModifiedTime": "2026-06-23T17:50:32.565300Z",
                "lastUserInputTime": "2026-06-23T17:50:29.232919Z",
                "lastUserInputStepIndex": 8,
                "stepCount": 10,
                "summary": "<human label>",
                "trajectoryType": "CORTEX_TRAJECTORY_TYPE_CASCADE",
                "trajectoryMetadata": {
                    "createdAt": "...",
                    "rootConversationId": "<rootConversationId>",
                    ...
                }
            },
            ...
        }}

    A freshly ``/clear``-minted cascade appears with ``lastUserInputTime`` /
    ``lastModifiedTime`` ABSENT (``null``) until it is actually used — so the
    reader's :func:`~omnigent.antigravity_native_reader._detect_rotated_cascade`
    treats a never-used sibling as not-yet-the-current-conversation (no premature
    rotation). The caller is responsible for selecting the current cascade from
    these summaries; this function only fetches them.

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :returns: The parsed response body (at minimum ``{"trajectorySummaries":
        {...}}``), or ``{}`` when the 200 body is not a dict (guards against a
        future agy returning a non-object 200).
    :raises httpx.HTTPError: On transport errors or non-2xx responses; the Task
        T-G reader is responsible for retry/backoff. Intentionally NOT fail-open
        (mirrors :func:`get_trajectory_steps` / :func:`get_available_models`): a
        non-2xx is a hard read failure, not a "no rotation" signal.
    :raises ValueError: When the 2xx response body is not valid JSON (body is
        decoded; the caller catches broadly).
    """
    url = _rpc_url(port, _METHOD_GET_ALL_CASCADE_TRAJECTORIES)
    _assert_loopback_url(url)
    with _sync_client(_RPC_CALL_TIMEOUT_S) as client:
        response = client.post(
            url,
            headers={"Content-Type": "application/json"},
            content=b"{}",
        )
    # Raises httpx.HTTPStatusError (subclass of httpx.HTTPError) on non-2xx so
    # the body is never decoded on error paths (which may not be JSON).
    response.raise_for_status()
    body = response.json()  # ValueError on non-JSON 200 propagates (documented)
    return body if isinstance(body, dict) else {}


def _encode_connect_envelope(payload: dict[str, object]) -> bytes:
    """
    Encode a single connect-protocol request message.

    The connect server-stream request body is ONE enveloped message:
    ``[flag: 1B = 0x00][length: 4B big-endian uint32][payload]`` where the
    payload is the JSON request object and ``length`` is its byte count. This is
    the same 5-byte ``[flag][BE-len]`` framing used on the response side, with an
    uncompressed (flag ``0x00``) payload.

    :param payload: The request object, e.g. ``{"conversationId": "<id>"}``.
    :returns: The framed request body bytes.
    """
    raw = json.dumps(payload).encode("utf-8")
    return bytes([0x00]) + struct.pack(">I", len(raw)) + raw


def _connect_trailer_error(payload: bytes) -> object | None:
    """
    Return the connect end-of-stream trailer's error, or ``None`` for a clean EOS.

    A connect server-stream reports a mid-stream failure in the TRAILER PAYLOAD
    as ``{"error": {...}}`` (the HTTP 200 + headers were already flushed, so the
    status cannot carry it). This extracts that error so the stream client can
    raise instead of silently truncating the turn.

    Fails SAFE toward a clean stop: an empty payload (the common normal EOS), a
    payload that is not valid JSON, a non-object body, or one whose ``error`` is
    absent/empty/falsy all return ``None`` (no error). Only a non-empty ``error``
    value is returned.

    :param payload: The raw trailer frame payload bytes (may be ``b""``).
    :returns: The trailer's ``error`` value when present and non-empty; otherwise
        ``None`` (treat as a clean end-of-stream).
    """
    if not payload:
        return None
    try:
        parsed = json.loads(payload)
    except json.JSONDecodeError:
        return None
    if not isinstance(parsed, dict):
        return None
    error = parsed.get("error")
    return error if error else None


async def stream_agent_state_updates(
    port: int, conversation_id: str
) -> AsyncIterator[dict[str, object]]:
    """
    Stream agent-state updates for ``conversation_id`` over connect server-stream.

    Opens a persistent ``StreamAgentStateUpdates`` POST (connect protocol,
    ``Content-Type: application/connect+json``) and yields each DATA frame's
    logical update payload as it arrives, stopping cleanly on the end-of-stream
    trailer. Each DATA frame is a connect envelope ``{"update": {...}}`` (live-
    verified, agy 1.0.10); this generator unwraps it and yields the inner
    ``update`` object, so the reader receives
    ``{"conversationId", "trajectoryId", "status", "mainTrajectoryUpdate", ...}``
    at the top level. The first frame carries a snapshot
    (``update.mainTrajectoryUpdate.stepsUpdate.steps[]``); subsequent frames are
    cumulative snapshots that grow while a step is generating (the Task T-D
    reader owns prefix-diffing for ``output_text_delta`` parity).

    Framing (live-verified, agy 1.0.10 — see design §10.2). The request body is a
    single connect-enveloped message (see :func:`_encode_connect_envelope`). The
    response is a byte stream of frames, each ``[flag: 1B][length: 4B BE][payload:
    length B]``:

    * ``flag == 0x00`` — DATA: ``payload`` is the JSON envelope ``{"update":
      {...}}``; the unwrapped ``update`` object is yielded.
    * ``flag & 0x02`` (:data:`_FRAME_FLAG_TRAILER`) — end-of-stream trailer:
      iteration stops; the payload is NOT yielded. A trailer carrying a connect
      ``{"error": {...}}`` (how a mid-stream failure is reported once the 200 +
      headers have been flushed) raises :class:`AntigravityRpcError`; an empty /
      non-error / unparseable trailer is a clean stop.
    * ``flag & 0x01`` (:data:`_FRAME_FLAG_COMPRESSED`) — compressed payload: agy
      sends uncompressed, so a set bit means a decode mismatch — raises
      :class:`AntigravityRpcError` rather than feeding compressed bytes to the
      JSON parser.

    Reassembly is buffer-based because frames can be split across network chunks
    AND several frames can arrive in one chunk: a ``bytearray`` accumulates bytes
    and, each pass, a frame is sliced out only when the buffer holds the full
    5-byte header AND the full declared payload — otherwise the loop awaits more
    bytes. One chunk is never assumed to equal one frame.

    The stream uses :data:`_STREAM_TIMEOUT` (no read deadline) because the
    long-poll idles between turns; a tight per-read timeout would abort it during
    normal think-time.

    :param port: Validated connect-RPC port, e.g. ``52548``.
    :param conversation_id: agy conversation id (equal to the cascade id) to
        stream updates for.
    :returns: An async iterator over DATA frames' parsed JSON dicts, in arrival
        order, ending when the trailer frame is seen or the stream closes.
    :raises AntigravityRpcError: On a non-2xx HTTP status (httpx ``stream()`` does
        not raise on its own, so an unframed error body would otherwise look like
        a clean empty stream and reconnect forever), a compressed frame
        (unexpected from agy), a malformed-JSON DATA frame, or a connect
        end-of-stream trailer error (``{"error": ...}``). All route into the Task
        T-D reader's poll-fallback. Transport errors propagate as
        ``httpx.HTTPError`` from the underlying stream (also poll-fallback).
    """
    url = _rpc_url(port, _METHOD_STREAM_AGENT_STATE_UPDATES)
    _assert_loopback_url(url)
    body = _encode_connect_envelope({"conversationId": conversation_id})
    async with (
        _async_client(_STREAM_TIMEOUT) as client,
        client.stream(
            "POST",
            url,
            headers={"Content-Type": "application/connect+json"},
            content=body,
        ) as response,
    ):
        # httpx ``client.stream()`` does NOT raise on a non-2xx status; a 4xx/5xx
        # body is not connect-framed, so the frame loop below would yield nothing
        # and return cleanly — which the reader treats as a normal end-of-stream
        # and reconnects every backoff forever. Surface it instead so it routes
        # into the reader's poll-fallback. ``status_code`` is available without
        # reading the (lazy) streaming body, so this does not consume the stream.
        if response.status_code >= 400:
            raise AntigravityRpcError(f"agy connect-stream returned HTTP {response.status_code}")
        buffer = bytearray()
        async for chunk in response.aiter_bytes():
            buffer.extend(chunk)
            while len(buffer) >= _FRAME_HEADER_LEN:
                flag = buffer[0]
                (length,) = struct.unpack(">I", buffer[1:_FRAME_HEADER_LEN])
                frame_end = _FRAME_HEADER_LEN + length
                if len(buffer) < frame_end:
                    break  # full payload not arrived yet — await more bytes
                payload = bytes(buffer[_FRAME_HEADER_LEN:frame_end])
                del buffer[:frame_end]
                if flag & _FRAME_FLAG_COMPRESSED:
                    raise AntigravityRpcError(
                        f"agy sent a compressed connect-stream frame (flag={flag:#04x}); "
                        "compression is unsupported"
                    )
                if flag & _FRAME_FLAG_TRAILER:
                    # End-of-stream trailer. In connect streaming a mid-stream
                    # failure is reported HERE as ``{"error": {...}}`` (not via
                    # HTTP status — the 200 + headers were already flushed), so a
                    # trailer error must be raised rather than treated as a clean
                    # stop (which would silently truncate the turn). An empty /
                    # non-error / unparseable trailer is a normal end-of-stream.
                    error = _connect_trailer_error(payload)
                    if error is not None:
                        raise AntigravityRpcError(f"agy connect-stream error: {error}")
                    return
                try:
                    parsed = json.loads(payload)
                except json.JSONDecodeError as e:
                    raise AntigravityRpcError(
                        f"malformed JSON in agy connect-stream DATA frame: {e}"
                    ) from e
                if isinstance(parsed, dict):
                    # Each DATA frame is a connect envelope ``{"update": {...}}``;
                    # unwrap it so the reader's ``_frame_steps`` sees
                    # ``mainTrajectoryUpdate`` at the top level. Fall back to the
                    # parsed dict if a future agy ever drops the envelope
                    # (defensive — keeps the reader working).
                    inner = parsed.get("update")
                    yield inner if isinstance(inner, dict) else parsed


def _list_agy_pids() -> list[int]:
    """
    Return pids of running ``agy`` processes (best-effort).

    Isolated as a seam so tests can stub it. Matches the agy binary path
    (``.../bin/agy``) to avoid matching unrelated commands that merely mention
    "agy". Prefers ``pgrep -f`` (portable across Linux + macOS); when ``pgrep``
    is unavailable — e.g. a minimal container image without ``procps`` — falls
    back to a ``/proc`` cmdline scan (:func:`_list_agy_pids_from_proc`) so
    discovery still works rather than silently yielding no candidates (which
    surfaces to the user as the misleading "is the agy terminal still open?").

    :returns: Candidate agy pids, newest-not-guaranteed order.
    """
    try:
        completed = subprocess.run(
            ["pgrep", "-f", r"bin/agy"],
            capture_output=True,
            text=True,
            timeout=_LSOF_TIMEOUT_S,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        # ``pgrep`` missing (no procps), hung past its timeout, or otherwise
        # unusable — fall back to a /proc scan instead of giving up, so a turn
        # can still be injected. ``exc_info`` records the real cause.
        _logger.debug("pgrep failed; falling back to /proc scan for agy pids", exc_info=True)
        return _list_agy_pids_from_proc()
    return [int(line) for line in completed.stdout.split() if line.isdigit()]


def _list_agy_pids_from_proc() -> list[int]:
    """
    Enumerate agy pids by scanning ``/proc/<pid>/cmdline`` (no ``pgrep`` needed).

    The Linux-only fallback for :func:`_list_agy_pids`. Mirrors ``pgrep -f
    bin/agy``: a process matches when its full (NUL-joined) command line
    contains ``bin/agy``, which the launcher always satisfies because
    :func:`omnigent.antigravity_native_launch.agy_binary_path` resolves to an
    absolute ``.../bin/agy`` path. Unreadable or vanished ``/proc`` entries are
    skipped; a missing ``/proc`` (non-Linux) yields ``[]``.

    :returns: Candidate agy pids.
    """
    pids: list[int] = []
    try:
        entries = os.listdir(_PROC_FS)
    except OSError:
        return []
    for entry in entries:
        if not entry.isdigit():
            continue
        try:
            with open(os.path.join(_PROC_FS, entry, "cmdline"), "rb") as handle:
                cmdline = handle.read()
        except OSError:
            # Process exited between listdir and open, or cmdline is unreadable.
            continue
        if b"bin/agy" in cmdline.replace(b"\0", b" "):
            pids.append(int(entry))
    return pids


def discover_language_server_port(pid: int) -> int | None:
    """
    Resolve the connect-RPC (TLS) port for a known agy pid.

    ``lsof``-es the pid's ``127.0.0.1`` TCP LISTEN ports and returns the first
    (lowest) one that answers ``Heartbeat`` with HTTP 200 — agy's lower port is
    the TLS connect-RPC surface and the higher one is plain HTTP that fails the
    probe. Probing lowest-first means the connect-RPC port is found without
    assuming the two ports are exactly adjacent.

    :param pid: agy process id, e.g. ``72753``.
    :returns: The validated connect-RPC port, or ``None`` when the process has
        no loopback listeners or none answer ``Heartbeat`` (e.g. agy has exited
        or has not finished binding).
    """
    ports = _parse_loopback_listen_ports(_run_lsof_listen_ports(pid))
    for port in ports:
        if _heartbeat_ok(port):
            _logger.debug("agy connect-RPC port resolved: pid=%s port=%s", pid, port)
            return port
    return None


# Bound on how deep the pane->agy descendant walk descends, so a pathological or
# cyclic process table cannot make the walk run unbounded. The real pane subtree
# is shallow (pane shell -> [sandbox launcher -> bwrap ->] agy), far under this.
_MAX_PANE_SUBTREE_DEPTH = 32


def _pane_pid(socket_path: Path, tmux_target: str) -> int | None:
    """
    Return the pid of the process running in a tmux pane (best-effort).

    Runs ``tmux -S <socket> display-message -p -t <target> '#{pane_pid}'`` (the
    same socket-scoped ``tmux`` invocation style as the bridge capture helpers)
    and parses the single pid it prints. Isolated as a seam so tests can stub the
    subprocess.

    The pane pid is the process tmux ``exec``-ed into the pane, which is agy
    itself on the simple ``exec agy`` launch and the sandbox launcher (an agy
    ancestor) on a sandboxed launch — :func:`_agy_pid_in_pane_subtree` resolves
    the actual agy pid from it.

    :param socket_path: Private tmux socket path for this session's terminal.
    :param tmux_target: Tmux target (session name), e.g. ``"main"``.
    :returns: The pane's pid, or ``None`` when ``tmux`` is missing, the pane is
        gone (non-zero exit), or the output is not a single integer.
    """
    try:
        completed = subprocess.run(
            [
                "tmux",
                "-S",
                str(socket_path),
                "display-message",
                "-p",
                "-t",
                tmux_target,
                "#{pane_pid}",
            ],
            capture_output=True,
            text=True,
            timeout=_LSOF_TIMEOUT_S,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        _logger.debug(
            "tmux display-message failed resolving pane pid for target=%s socket=%s",
            tmux_target,
            socket_path,
            exc_info=True,
        )
        return None
    if completed.returncode != 0:
        return None
    text = completed.stdout.strip()
    return int(text) if text.isdigit() else None


def _child_pids(pid: int) -> list[int]:
    """
    Return the direct child pids of ``pid`` (best-effort).

    Prefers ``pgrep -P <pid>`` (portable across Linux + macOS); on a host without
    ``pgrep`` (e.g. a minimal container) falls back to scanning
    ``/proc/<child>/stat`` for a matching ``PPid``. Mirrors the dual-path
    strategy of :func:`_list_agy_pids`. A failure yields ``[]`` (the subtree walk
    then simply finds no agy via this branch).

    :param pid: Parent process id.
    :returns: Direct child pids, or ``[]`` when none / unresolvable.
    """
    try:
        completed = subprocess.run(
            ["pgrep", "-P", str(pid)],
            capture_output=True,
            text=True,
            timeout=_LSOF_TIMEOUT_S,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        _logger.debug("pgrep -P failed; falling back to /proc for children of %s", pid)
        return _child_pids_from_proc(pid)
    return [int(line) for line in completed.stdout.split() if line.isdigit()]


def _child_pids_from_proc(pid: int) -> list[int]:
    """
    Enumerate direct child pids of ``pid`` via ``/proc/<child>/stat`` (no pgrep).

    The Linux-only fallback for :func:`_child_pids`. ``/proc/<child>/stat`` field
    4 is the PPid; a child matches when its PPid equals ``pid``. The comm field
    (field 2) is parenthesized and may contain spaces, so PPid is read relative to
    the LAST ``)`` rather than by naive whitespace split. Unreadable or vanished
    entries are skipped; a missing ``/proc`` (non-Linux) yields ``[]``.

    :param pid: Parent process id.
    :returns: Direct child pids, or ``[]``.
    """
    children: list[int] = []
    try:
        entries = os.listdir(_PROC_FS)
    except OSError:
        return []
    for entry in entries:
        if not entry.isdigit():
            continue
        try:
            with open(os.path.join(_PROC_FS, entry, "stat"), encoding="ascii") as handle:
                stat = handle.read()
        except OSError:
            continue  # process exited between listdir and open
        close_paren = stat.rfind(")")
        if close_paren == -1:
            continue
        fields = stat[close_paren + 1 :].split()
        # After the ``)``, fields are: state(0) ppid(1) ... so ppid is index 1.
        if len(fields) > 1 and fields[1].isdigit() and int(fields[1]) == pid:
            children.append(int(entry))
    return children


def _agy_pid_in_pane_subtree(pane_pid: int) -> int | None:
    """
    Return the agy pid running at or beneath a tmux pane's process (best-effort).

    The pane process is agy itself on the simple ``exec agy`` launch (CLI /
    non-sandbox) and a wrapper (sandbox launcher -> bwrap -> ... -> agy) on a
    sandboxed launch. So: if the pane pid is itself a live agy pid, return it;
    otherwise breadth-first walk the pane pid's descendants and return the first
    that is a live agy pid. The walk is intersected with :func:`_list_agy_pids`
    (the same ``bin/agy`` cmdline match used everywhere) so a non-agy descendant
    is never mistaken for agy, and is depth- and visited-bounded so a cyclic or
    pathological process table cannot hang it.

    :param pane_pid: The tmux pane's process id (from :func:`_pane_pid`).
    :returns: The agy pid in the pane's subtree, or ``None`` when none is found
        (the caller then falls back to the host-wide candidate scan).
    """
    agy_pids = set(_list_agy_pids())
    if not agy_pids:
        return None
    if pane_pid in agy_pids:
        return pane_pid
    # Breadth-first descent, bounded by depth and visited-set so a cyclic ppid
    # graph (or a re-parented pid) cannot loop forever.
    visited: set[int] = {pane_pid}
    frontier = [pane_pid]
    for _depth in range(_MAX_PANE_SUBTREE_DEPTH):
        if not frontier:
            break
        next_frontier: list[int] = []
        for parent in frontier:
            for child in _child_pids(parent):
                if child in visited:
                    continue
                if child in agy_pids:
                    return child
                visited.add(child)
                next_frontier.append(child)
        frontier = next_frontier
    return None


class PaneAgyResolution(NamedTuple):
    """
    The outcome of scoping a cold-start port to a tmux pane's own agy.

    Distinguishes the THREE states the cold-start must treat differently — a bare
    ``int | None`` cannot, because "agy is up but its port is not lsof-attributable"
    and "no agy is running in this pane yet" both collapse to "no port" yet demand
    opposite fallbacks (candidate scan vs. keep polling). See
    :func:`resolve_pane_agy_rpc_port_state`.

    :param agy_found: Whether an agy process was found at/under the pane pid.
    :param port: That agy's validated connect-RPC port, or ``None`` when agy was
        found but its port is not yet resolvable (or no agy was found).
    """

    agy_found: bool
    port: int | None


def resolve_pane_agy_rpc_port_state(socket_path: Path, tmux_target: str) -> PaneAgyResolution:
    """
    Resolve THIS session's own agy connect-RPC port via its tmux pane (3-state).

    The cold-start's disambiguation seam: before any conversation exists there is
    nothing to confirm ownership with (:func:`resolve_language_server_port` needs
    a conversation id), so on a host running several agy instances the lowest
    Heartbeat-answering candidate (:func:`_candidate_agy_rpc_ports`) could be a
    FOREIGN agy. Scoping the port to the agy process actually running under this
    session's tmux pane removes that ambiguity: pane -> pane pid
    (:func:`_pane_pid`) -> agy pid in its subtree (:func:`_agy_pid_in_pane_subtree`)
    -> that pid's Heartbeat-validated connect-RPC port
    (:func:`discover_language_server_port`).

    Returns a :class:`PaneAgyResolution` so the caller can tell the three outcomes
    apart (best-effort, non-raising):

    * ``(agy_found=True, port=<int>)`` — our agy is up and its port is resolved.
    * ``(agy_found=True, port=None)`` — our agy IS in the pane subtree but its port
      is not (yet) lsof-attributable (e.g. a restricted ``/proc`` where the
      listener is owned by a backend the agy pid does not hold as an fd). The
      caller may safely use the candidate scan: agy exists here, so on the
      one-agy-per-pod hosts where this happens the lone candidate is ours.
    * ``(agy_found=False, port=None)`` — no agy is running in this pane yet (no
      pane pid, or agy has not been ``exec``-ed — the CLI ``tmux_start_on_attach``
      early-poll window). The caller MUST keep polling rather than fall back to
      candidates, because a FOREIGN agy could be the only candidate and binding it
      would durably cross-bind the session.

    :param socket_path: Private tmux socket path for this session's terminal.
    :param tmux_target: Tmux target (session name), e.g. ``"main"``.
    :returns: The :class:`PaneAgyResolution` for the three states above.
    """
    pane_pid = _pane_pid(socket_path, tmux_target)
    if pane_pid is None:
        # No pane pid (tmux gone / target absent) — agy is not reachable here.
        return PaneAgyResolution(agy_found=False, port=None)
    agy_pid = _agy_pid_in_pane_subtree(pane_pid)
    if agy_pid is None:
        _logger.debug(
            "no agy process found in tmux pane subtree (pane_pid=%s, target=%s)",
            pane_pid,
            tmux_target,
        )
        return PaneAgyResolution(agy_found=False, port=None)
    port = discover_language_server_port(agy_pid)
    if port is not None:
        _logger.info(
            "agy connect-RPC port scoped to pane: target=%s agy_pid=%s port=%s",
            tmux_target,
            agy_pid,
            port,
        )
    return PaneAgyResolution(agy_found=True, port=port)


def resolve_pane_agy_rpc_port(socket_path: Path, tmux_target: str) -> int | None:
    """
    Resolve THIS session's own agy connect-RPC port via its tmux pane.

    Thin ``port``-only view of :func:`resolve_pane_agy_rpc_port_state` for callers
    that do not need the agy-found / no-agy distinction.

    :param socket_path: Private tmux socket path for this session's terminal.
    :param tmux_target: Tmux target (session name), e.g. ``"main"``.
    :returns: The pane agy's validated connect-RPC port, or ``None`` when it
        cannot be scoped.
    """
    return resolve_pane_agy_rpc_port_state(socket_path, tmux_target).port


def resolve_cold_start_agy_rpc_port(
    tmux_socket: Path | None,
    tmux_target: str | None,
) -> int | None:
    """
    Pick the connect-RPC port a cold-start should ``StartCascade`` onto.

    Cold-start runs BEFORE any conversation exists, so the conversation-ownership
    check that normally disambiguates several agys
    (:func:`resolve_language_server_port`) is not yet usable. To avoid binding a
    FOREIGN agy on a host running several (sub-agent fan-out / shared runner /
    ``omnigent run --server`` multi-session), this scopes the port to THIS
    session's own agy via its tmux pane
    (:func:`resolve_pane_agy_rpc_port_state`), distinguishing three outcomes:

    1. **Pane present, our agy found, port resolved** → that scoped port.
    2. **Pane present, our agy found, port not resolvable** (e.g. restricted
       ``/proc`` where lsof cannot attribute the listener) → the lowest candidate
       (:func:`_candidate_agy_rpc_ports`). agy IS up in this pane, and the hosts
       where this happens run one agy per pod, so the lone candidate is ours.
    3. **Pane present, NO agy found yet** (the CLI ``tmux_start_on_attach``
       early-poll window: the pane is still the shell, agy not yet ``exec``-ed) →
       ``None``, so the caller keeps polling. It must NOT fall back to candidates
       here: a foreign agy could be the only candidate, and binding it would
       durably cross-bind this session — the exact defect this scoping prevents.

    With **no pane supplied** (``tmux_socket``/``tmux_target`` is ``None`` — e.g. a
    remote runner with no local socket) there is nothing to scope to, so this
    falls back to the lowest candidate (best-effort; cannot disambiguate). The
    candidate fallback is logged so a wrong-agy bind on a multi-agy host is
    diagnosable.

    :param tmux_socket: This session's tmux socket path, or ``None`` when no local
        pane is reachable (remote runner).
    :param tmux_target: This session's tmux target, or ``None`` as above.
    :returns: The port to ``StartCascade`` onto, or ``None`` when the port is not
        resolvable yet (the caller keeps polling until its deadline).
    """
    if tmux_socket is not None and tmux_target is not None:
        resolution = resolve_pane_agy_rpc_port_state(tmux_socket, tmux_target)
        if resolution.port is not None:
            return resolution.port  # state 1: our agy, scoped port
        if not resolution.agy_found:
            # state 3: our agy is not up yet (CLI early-poll). Keep polling — do
            # NOT fall back to candidates, where a foreign agy could be the only
            # one and would cross-bind this session.
            return None
        # state 2: our agy IS up but its port is not lsof-attributable (restricted
        # /proc; one-agy-per-pod) — the candidate scan below is safe and necessary.
        _logger.debug(
            "agy cold-start: pane agy found for target=%s but its port is not "
            "lsof-attributable; using the host-wide candidate scan (safe — agy is "
            "up in this pane)",
            tmux_target,
        )
    candidates = _candidate_agy_rpc_ports()
    if not candidates:
        return None
    port = candidates[0]
    if tmux_socket is None or tmux_target is None:
        _logger.debug(
            "agy cold-start: no local tmux pane to scope to; using the lowest "
            "candidate connect-RPC port %s (single-agy hosts and remote runners "
            "are unaffected; a multi-agy host risks a wrong-agy bind)",
            port,
        )
    return port


def _candidate_agy_rpc_ports() -> list[int]:
    """
    Return every live agy connect-RPC port, validated by ``Heartbeat``.

    Primary path: ``lsof`` each running agy pid's loopback LISTEN ports (precise
    — scopes to agy). Fallback: when ``lsof`` attributes no ports — a restricted
    ``/proc`` where agy's listening socket is not in its pid's fd table (verified
    on uid-1000 k8s pods, where agy 1.0.10 holds the listener in a backend the
    agy process does not own as an fd) — enumerate every loopback LISTEN port
    from ``/proc/net/tcp`` via :func:`_list_loopback_listen_ports`, which needs
    no fd/ptrace access.

    The fallback fires only when agy IS running but lsof saw none of its ports —
    not when no agy is running at all — so a turn-injection attempt against a
    dead session does not heartbeat every unrelated loopback service. The scan is
    also capped at :data:`_MAX_FALLBACK_PROBE_PORTS` (lowest-first), with the drop
    logged, to bound the probe count on a host with many loopback listeners.

    Either way the candidates are ``Heartbeat``-filtered, so only agy's TLS
    connect-RPC port(s) survive (agy's plain-HTTP port and unrelated loopback
    listeners fail the probe). Callers additionally confirm conversation
    ownership before injecting, so a stray non-agy port can never be written to.

    :returns: Sorted connect-RPC ports that answer ``Heartbeat`` with HTTP 200.
    """
    agy_pids = _list_agy_pids()
    ports: set[int] = set()
    for pid in agy_pids:
        ports.update(_parse_loopback_listen_ports(_run_lsof_listen_ports(pid)))
    if agy_pids and not ports:
        loopback = _list_loopback_listen_ports()
        if len(loopback) > _MAX_FALLBACK_PROBE_PORTS:
            _logger.warning(
                "agy port discovery fallback: %d loopback listeners exceed the "
                "%d-probe cap; probing the lowest %d only",
                len(loopback),
                _MAX_FALLBACK_PROBE_PORTS,
                _MAX_FALLBACK_PROBE_PORTS,
            )
            loopback = loopback[:_MAX_FALLBACK_PROBE_PORTS]
        ports.update(loopback)
    return [port for port in sorted(ports) if _heartbeat_ok(port)]


def conversation_id_owned_by_pid(pid: int, candidate_ids: Iterable[str]) -> str | None:
    """
    Return which candidate conversation id a specific agy pid owns.

    The deterministic counterpart to :func:`resolve_language_server_port`: that
    function answers "which port owns this *known* id"; this one answers "which
    of these candidate ids does *this* process own", binding discovery to a
    specific agy pid (e.g. the one running under this session's tmux pane).

    agy exposes no method to *list* its conversation id, only
    ``GetConversationMetadata`` which confirms a given id. So this resolves the
    pid's own connect-RPC port via :func:`discover_language_server_port` and asks
    that port to confirm each candidate (``GetConversationMetadata`` returns
    metadata only for an id that server hosts), eliminating the newest-dir guess,
    the cross-launch ambiguity, and the resulting livelock.

    Correctness over liveness, mirroring the forwarder's
    ``_discover_conversation_id``: a pid's connect-RPC server can confirm more
    than one candidate brain-dir id, so every candidate is tested and the result
    is bound only when *exactly one* matches. Zero or multiple matches return
    ``None`` (the multi-match case is refused rather than guessed, since
    first-match would depend on ``candidate_ids`` order and could bind the wrong
    transcript / external_session_id).

    On a host where ``lsof`` cannot attribute the socket to *pid* (restricted
    ``/proc``), the resolve falls back to checking the candidates against EVERY
    live agy connect-RPC port rather than just *pid*'s. Because the conversation
    id is globally unique this never mis-binds; but two concurrent same-host
    sessions sharing the HOME-global brain dir can put both their ids in
    ``candidate_ids``, in which case both are confirmed (by different ports) and
    the call refuses (returns ``None``) instead of guessing — the forwarder then
    retries. So the fallback trades the pid-scoped *liveness* of the binding for
    the same safety, never correctness. (The executor's write path,
    :func:`resolve_language_server_port`, is unaffected: it already holds the
    target conversation id, so a single port resolves unambiguously.)

    :param pid: agy process id whose conversation to resolve, e.g. ``72753``.
    :param candidate_ids: agy conversation ids to test (e.g. the in-window
        brain-dir names).
    :returns: The candidate id this pid's connect-RPC server confirms it hosts
        when exactly one matches; ``None`` when the port cannot be resolved (agy
        not bound yet / exited), no candidate matches, or — refusing to guess —
        more than one candidate matches.
    """
    candidates = list(candidate_ids)
    # Prefer the pid-scoped port (lsof — precise). When lsof cannot attribute the
    # socket to the pid (restricted /proc; see _candidate_agy_rpc_ports), fall
    # back to every live agy connect-RPC port: the conversation id is globally
    # unique, so a candidate is confirmed only by the port that actually hosts
    # it — the binding stays correct without the pid scoping.
    scoped = discover_language_server_port(pid)
    ports = [scoped] if scoped is not None else _candidate_agy_rpc_ports()
    if not ports:
        _logger.debug("agy pid=%s has no resolvable connect-RPC port yet", pid)
        return None
    matched = [
        candidate
        for candidate in candidates
        if any(_conversation_matches(port, candidate) for port in ports)
    ]
    if len(matched) == 1:
        _logger.info(
            "agy conversation resolved by pid ownership: pid=%s ports=%s conversation=%s",
            pid,
            ports,
            matched[0],
        )
        return matched[0]
    if not matched:
        _logger.debug(
            "agy pid=%s (ports=%s) owns none of the %d candidate conversation ids",
            pid,
            ports,
            len(candidates),
        )
        return None
    _logger.warning(
        "agy pid=%s (ports=%s) confirmed %d candidate conversation ids; refusing "
        "to guess which it owns: %s",
        pid,
        ports,
        len(matched),
        matched[:_MAX_LOGGED_AMBIGUOUS_IDS],
    )
    return None


def resolve_language_server_port(conversation_id: str) -> int | None:
    """
    Resolve agy's connect-RPC port for a conversation by validated discovery.

    Enumerates candidate agy connect-RPC ports (:func:`_candidate_agy_rpc_ports`
    — ``lsof`` per agy pid, or ``/proc/net/tcp`` where ``lsof`` cannot attribute
    the socket to a pid) and returns the first that owns ``conversation_id`` via
    ``GetConversationMetadata``. With one agy this is unambiguous; with several
    the conversation check picks the one actually hosting this conversation.

    Discovery is port-first (not pid-first): agy is launched under
    ``tmux_start_on_attach`` so the launcher never captures a pid, and on some
    hosts agy's listening socket is owned by a backend that is neither the agy
    pid nor ``lsof``-attributable — so a pid is not a reliable key. The
    ``GetConversationMetadata`` ownership check is what makes a port safe to use:
    a recycled/foreign port (a different live agy) is rejected because it does
    not host ``conversation_id``.

    :param conversation_id: agy conversation id the turn targets, e.g.
        ``"90468e33-..."``. Used to disambiguate when multiple agy processes
        run.
    :returns: A validated connect-RPC port that hosts ``conversation_id``, or
        ``None`` when no running agy could be resolved.
    """
    for port in _candidate_agy_rpc_ports():
        if _conversation_matches(port, conversation_id):
            _logger.info(
                "agy connect-RPC port resolved by conversation match: port=%s conversation=%s",
                port,
                conversation_id,
            )
            return port
    _logger.warning(
        "could not resolve an agy connect-RPC port for conversation=%s",
        conversation_id,
    )
    return None


async def interrupt_turn(port: int, conversation_id: str) -> bool:
    """
    Best-effort interrupt of an in-flight agy turn via connect-RPC (FAIL-OPEN).

    Intended to back the post-hoc audit's "decline → stop the turn" on a policy
    DENY/ASK (see :mod:`omnigent.antigravity_native_audit`). It is **best-effort
    and fail-open**: the audit warning is always surfaced regardless of whether
    this succeeds, and the offending tool has already run (agy writes a step only
    at ``DONE``), so a cancel can at most stop *subsequent* tools in the same
    turn — it never prevents the violation.

    .. warning:: **Wired OFF — request contract unverified.**
       agy 1.0.8 exposes ``ForceStopCascadeTree`` / ``CancelCascadeInvocation`` /
       ``CancelCascadeSteps`` on the connect-RPC surface (method names verified in
       the binary), but the request CONTRACT is **not** verified: the proto field
       tags show these key on an internal ``cascade_id`` / ``invocation_id`` —
       agy's per-turn identifiers, which are NOT exposed in the transcript and
       which this forwarder does not hold (it only knows the *conversation* id).
       The stop semantics are also unconfirmed against a live process. Until the
       request shape (and the cascade-id source) is verified end-to-end, this
       function does not issue an RPC: it logs and returns ``False`` so the gated
       caller treats the interrupt as unavailable and relies on the audit warning
       alone.

       TODO(antigravity-interrupt): verify ``ForceStopCascadeTreeRequest`` (does
       it accept the conversation id, or does it require a cascade id obtained
       from ``GetCascadeTrajectorySteps`` / a metadata RPC?) on a live agy, then
       implement the POST (a loopback connect-RPC call like the
       ``GetConversationMetadata`` probe above) and flip
       ``_INTERRUPT_ON_AUDIT_DENY`` (forwarder) to opt-in.

    :param port: agy connect-RPC (TLS) port, e.g. ``52548``.
    :param conversation_id: agy conversation id whose turn should be stopped.
    :returns: ``True`` when agy accepted the cancel; ``False`` when the interrupt
        is unavailable (currently always, pending contract verification) or the
        call failed. Never raises — fail-open.
    """
    # Intentionally not issuing an RPC: the request contract is unverified and
    # the forwarder lacks agy's internal cascade/invocation id. Returning False
    # keeps the caller fail-open (the audit warning still surfaces). The method
    # constant is referenced so the verified name is not dead and a future
    # implementation has the anchor.
    del port  # unused until the contract is verified (see warning above)
    _logger.debug(
        "agy turn interrupt requested but unavailable (RPC %s contract unverified); "
        "relying on the audit warning only: conversation=%s",
        _METHOD_FORCE_STOP_CASCADE_TREE,
        conversation_id,
    )
    return False