wehub-resource-sync
479 lines
22 KiB
Python
479 lines
22 KiB
Python
"""Interaction bridge for the native Antigravity (agy) RPC harness.
|
|
|
|
This is the correctness-sensitive piece of the RPC core rework: it surfaces an
|
|
agy WAITING interaction (``ask_question`` / command ``permission``) as an
|
|
Omnigent elicitation, waits for the human's verdict, and delivers it back to agy
|
|
via ``HandleCascadeUserInteraction`` — handling agy's **WAITING-interaction
|
|
timeout gotcha** end-to-end.
|
|
|
|
The gotcha (design ``docs/antigravity-native-rpc-core-design.md`` §2.1, memory
|
|
``agy-rpc-interaction-bridge``): a WAITING interaction **times out** server-side
|
|
(→ ``CORTEX_STEP_STATUS_ERROR``), after which agy **auto-retries with a fresh
|
|
WAITING step at a HIGHER ``stepIndex``**. Omnigent elicitations wait on a human
|
|
(potentially slow), so by the time a verdict arrives the captured
|
|
``trajectoryId`` / ``stepIndex`` may be STALE. Consequences this module handles:
|
|
|
|
* The bridge **re-reads the freshest WAITING step at delivery time** and targets
|
|
THAT step's ids — never the ones captured at detection.
|
|
* ``HTTP 500 "input not registered for step N"`` is **overloaded**: it means
|
|
either a missing ``trajectoryId`` *or* a step that already timed out. After a
|
|
delivery raises it, the bridge re-reads for a NEW (higher-index) WAITING step
|
|
and re-surfaces the elicitation against it (so the retry step gets a fresh
|
|
elicitation id and the web UI re-prompts).
|
|
* The whole loop is bounded by ``max_retries`` so a pathological retry storm
|
|
cannot spin forever.
|
|
|
|
Seam design (so the timeout logic is unit-testable without a live agy):
|
|
|
|
* ``get_steps`` — re-reads the freshest trajectory steps (production wraps
|
|
:func:`omnigent.antigravity_native_rpc.get_trajectory_steps`).
|
|
* ``request_elicitation`` — publishes the elicitation under a deterministic id
|
|
and long-poll-awaits the user's result; returns ``None`` on timeout/cancel
|
|
(production, wired in Task 11, POSTs the
|
|
``antigravity-elicitation-request`` hook and long-polls — mirrors codex's
|
|
``_handle_codex_elicitation_request``).
|
|
* ``deliver`` — defaults to :func:`_deliver_via_rpc`, which offloads the blocking
|
|
:func:`omnigent.antigravity_native_rpc.handle_user_interaction` to a worker
|
|
thread (the function is synchronous); injectable so tests don't need a live agy.
|
|
* ``inject_tui`` — defaults to :func:`_inject_via_tui`, which types the verdict's
|
|
key sequence into the agy TUI pane AFTER the RPC delivery. The attended agy TUI
|
|
keeps its OWN permission/question prompt open in parallel with the RPC step
|
|
(live-verified — ``docs/claude/antigravity-rpc-spike-notes.md`` §"attended
|
|
TUI"): the RPC flips the backend step to DONE, but the TUI prompt lingers, so a
|
|
web Approve/Reject would not advance the terminal (and the next typed turn would
|
|
fold into the stale prompt's buffer) without this keystroke (#1200). Injectable
|
|
so tests assert the key sequence without a live pane; mirrors cursor-native's
|
|
``send_cursor_pane_keys`` approval drive.
|
|
|
|
The deterministic elicitation id is derived from
|
|
``(cascade_id, trajectory_id, step_index)`` (mirrors
|
|
:func:`omnigent.codex_native_elicitation.codex_elicitation_id`), so a
|
|
timeout-retry step (new ``step_index``) yields a NEW id → a fresh elicitation is
|
|
surfaced for the retry rather than silently reusing the stale card.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Protocol
|
|
|
|
from omnigent.antigravity_native_rpc import (
|
|
AntigravityRpcError,
|
|
handle_user_interaction,
|
|
)
|
|
from omnigent.antigravity_native_steps import PendingInteraction, pending_interaction
|
|
from omnigent.server.routes._antigravity_elicitation import (
|
|
to_elicitation_params,
|
|
to_interaction_payload,
|
|
to_tui_selection_keys,
|
|
)
|
|
from omnigent.server.schemas import ElicitationRequestParams, ElicitationResult
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# Length of the hex digest slice used in the deterministic elicitation id.
|
|
# Mirrors ``codex_native_elicitation._CODEX_ELICITATION_ID_DIGEST_LENGTH`` so the
|
|
# two harnesses produce ids of the same shape/cardinality.
|
|
_AGY_ELICITATION_ID_DIGEST_LENGTH = 32
|
|
|
|
# Substring agy returns (inside an HTTP 500 body) when the targeted step is no
|
|
# longer accepting input — either a missing ``trajectoryId`` or, the case this
|
|
# module retries, a step that already timed out (status ERROR) before delivery.
|
|
# Matched case-INSENSITIVELY against agy's raw body (see the delivery
|
|
# error-handling below): this substring is the SOLE discriminator between a
|
|
# retryable step-timeout race and a fatal delivery error, so a capitalization
|
|
# change in agy's message must not silently reclassify the race as fatal.
|
|
_INPUT_NOT_REGISTERED = "input not registered"
|
|
|
|
# Default bound on the detect→elicit→deliver loop. Each iteration surfaces one
|
|
# elicitation and attempts one delivery; a timeout-retry consumes one iteration.
|
|
# A handful covers any realistic chain of agy timeout-retries while guaranteeing
|
|
# the loop terminates even if every delivery keeps racing a fresh ERROR step.
|
|
_DEFAULT_MAX_RETRIES = 5
|
|
|
|
|
|
# A re-reader of the freshest trajectory steps (production wraps
|
|
# ``get_trajectory_steps(port, cascade_id)``).
|
|
GetSteps = Callable[[], Awaitable[list[dict[str, object]]]]
|
|
|
|
# Publishes one elicitation under ``elicitation_id`` and long-poll-awaits the
|
|
# user's verdict; ``None`` means timeout/cancel.
|
|
RequestElicitation = Callable[
|
|
[str, ElicitationRequestParams],
|
|
Awaitable[ElicitationResult | None],
|
|
]
|
|
|
|
|
|
class Deliver(Protocol):
|
|
"""
|
|
Delivers a built interaction payload to agy.
|
|
|
|
Matches :func:`omnigent.antigravity_native_rpc.handle_user_interaction`
|
|
exactly (``port``/``cascade_id`` positional, ids + payload keyword-only) so
|
|
that function is the drop-in default; spelled as a ``Protocol`` rather than a
|
|
``Callable[..., ...]`` alias to keep the signature precise (the package bans
|
|
explicit ``Any``).
|
|
"""
|
|
|
|
async def __call__(
|
|
self,
|
|
port: int,
|
|
cascade_id: str,
|
|
*,
|
|
trajectory_id: str,
|
|
step_index: int,
|
|
payload: dict[str, object],
|
|
) -> None:
|
|
"""Deliver one interaction answer to agy (see ``handle_user_interaction``)."""
|
|
...
|
|
|
|
|
|
# Types the verdict's tmux key sequence into the agy TUI pane to dismiss its
|
|
# in-process prompt (see ``_inject_via_tui`` / the module's "attended TUI" note).
|
|
InjectTui = Callable[[list[str]], Awaitable[None]]
|
|
|
|
|
|
async def _inject_via_tui(keys: list[str]) -> None:
|
|
"""
|
|
Default :class:`InjectTui`: type the verdict keys into the agy TUI pane.
|
|
|
|
Offloads the blocking tmux ``send-keys`` (via
|
|
:func:`omnigent.antigravity_native_bridge.send_interaction_keys_via_tui`) to a
|
|
worker thread — the same pattern :func:`_deliver_via_rpc` uses for the RPC —
|
|
so the async bridge does not stall the event loop. The bridge directory is
|
|
resolved from the harness spawn env (the reader/CLI both run with it set), so
|
|
this seam takes only the keys.
|
|
|
|
A missing tmux target / exited pane raises ``RuntimeError`` from the bridge
|
|
primitive; the caller logs it and proceeds (the RPC delivery already advanced
|
|
the backend step — the keystroke is the belt-and-braces TUI dismissal).
|
|
|
|
:param keys: Ordered tmux key arguments (e.g. ``["1", "Enter"]`` to approve).
|
|
:returns: None.
|
|
:raises RuntimeError: Propagated from the bridge primitive (no target / pane
|
|
exited / send-keys failed).
|
|
"""
|
|
# Lazy imports: keep this module importable from the lightweight CLI process
|
|
# without eagerly pulling the bridge env helpers, and resolve the bridge dir
|
|
# the same way the executor does.
|
|
from omnigent.antigravity_native_bridge import send_interaction_keys_via_tui
|
|
from omnigent.inner.antigravity_native_executor import _bridge_dir_from_env
|
|
|
|
bridge_dir = _bridge_dir_from_env()
|
|
await asyncio.to_thread(send_interaction_keys_via_tui, bridge_dir, *keys)
|
|
|
|
|
|
def agy_elicitation_id(cascade_id: str, trajectory_id: str, step_index: int) -> str:
|
|
"""
|
|
Build the Omnigent elicitation id for one agy WAITING interaction.
|
|
|
|
Deterministic over ``(cascade_id, trajectory_id, step_index)`` so a
|
|
timeout-retry step (which agy issues at a HIGHER ``step_index``) maps to a
|
|
DIFFERENT id — surfacing a fresh elicitation for the retry rather than
|
|
re-using the stale card. Mirrors
|
|
:func:`omnigent.codex_native_elicitation.codex_elicitation_id`.
|
|
|
|
:param cascade_id: agy cascade id (equal to the conversation id).
|
|
:param trajectory_id: agy trajectory id from the WAITING step.
|
|
:param step_index: Trajectory step index of the WAITING step.
|
|
:returns: Stable elicitation id beginning with ``"elicit_agy_"``.
|
|
"""
|
|
payload = json.dumps(
|
|
{
|
|
"cascade_id": cascade_id,
|
|
"trajectory_id": trajectory_id,
|
|
"step_index": step_index,
|
|
},
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
digest = hashlib.sha256(payload).hexdigest()[:_AGY_ELICITATION_ID_DIGEST_LENGTH]
|
|
return f"elicit_agy_{digest}"
|
|
|
|
|
|
async def _deliver_via_rpc(
|
|
port: int,
|
|
cascade_id: str,
|
|
*,
|
|
trajectory_id: str,
|
|
step_index: int,
|
|
payload: dict[str, object],
|
|
) -> None:
|
|
"""
|
|
Default :class:`Deliver`: offload the blocking RPC to a worker thread.
|
|
|
|
:func:`omnigent.antigravity_native_rpc.handle_user_interaction` is synchronous
|
|
and uses a blocking ``httpx.Client``, so calling it directly from the async
|
|
bridge would stall the event loop. This wraps it in
|
|
:func:`asyncio.to_thread` — the same pattern the read driver uses for
|
|
``get_trajectory_steps`` — and preserves its :class:`AntigravityRpcError`
|
|
(the bridge catches it to detect the timed-out-step race).
|
|
|
|
:param port: Validated agy connect-RPC port.
|
|
:param cascade_id: agy cascade id (equal to the conversation id).
|
|
:param trajectory_id: agy trajectory id of the target WAITING step.
|
|
:param step_index: Step index of the target WAITING step.
|
|
:param payload: The interaction variant dict (``askQuestion`` / ``permission``).
|
|
:returns: None.
|
|
:raises AntigravityRpcError: Propagated from ``handle_user_interaction``.
|
|
"""
|
|
await asyncio.to_thread(
|
|
handle_user_interaction,
|
|
port,
|
|
cascade_id,
|
|
trajectory_id=trajectory_id,
|
|
step_index=step_index,
|
|
payload=payload,
|
|
)
|
|
|
|
|
|
def _freshest_waiting(
|
|
steps: list[dict[str, object]],
|
|
*,
|
|
kind: str,
|
|
after_index: int | None = None,
|
|
) -> PendingInteraction | None:
|
|
"""
|
|
Return the highest-index WAITING interaction from a steps snapshot.
|
|
|
|
The crux of the timeout handling: when several WAITING steps are present
|
|
(agy left timed-out ones behind and issued retries), the freshest — highest
|
|
``step_index`` — is the live one to deliver against. Only steps of the
|
|
requested ``kind`` are eligible: agy keys delivery on
|
|
``trajectoryId``+``stepIndex`` with no kind check, so pairing a payload built
|
|
for one kind with a WAITING step of a different kind would mis-deliver (e.g.
|
|
answer a question against a stray permission step). When no WAITING step of
|
|
the requested kind exists, ``None`` is returned and the caller stops cleanly.
|
|
|
|
:param steps: A trajectory steps snapshot (from ``get_steps``).
|
|
:param kind: Required interaction kind (``"ask_question"`` / ``"permission"``).
|
|
:param after_index: When set, only steps with ``step_index > after_index``
|
|
are considered — used after a timeout to require a strictly NEWER step
|
|
than the one that just failed, so the loop cannot re-target the same
|
|
stale index.
|
|
:returns: The freshest same-``kind`` :class:`PendingInteraction`, or ``None``
|
|
when no WAITING step of that kind exists.
|
|
"""
|
|
same_kind: PendingInteraction | None = None
|
|
for step in steps:
|
|
pending = pending_interaction(step)
|
|
if pending is None:
|
|
continue
|
|
if after_index is not None and pending["step_index"] <= after_index:
|
|
continue
|
|
if pending["kind"] != kind:
|
|
continue
|
|
if same_kind is None or pending["step_index"] > same_kind["step_index"]:
|
|
same_kind = pending
|
|
return same_kind
|
|
|
|
|
|
def _waiting_step_at(
|
|
steps: list[dict[str, object]],
|
|
*,
|
|
trajectory_id: str,
|
|
step_index: int,
|
|
) -> PendingInteraction | None:
|
|
"""
|
|
Return the WAITING interaction at an exact ``(trajectory_id, step_index)``.
|
|
|
|
Pins verdict delivery to the step the elicitation was surfaced for rather than
|
|
the freshest WAITING step (which could be a different gate that appeared
|
|
meanwhile). Returns ``None`` when that step is no longer WAITING (timed out or
|
|
answered), letting the caller fall back to ``_freshest_waiting`` for agy's
|
|
same-gate timeout-retry.
|
|
|
|
:param steps: Trajectory steps snapshot.
|
|
:param trajectory_id: The surfaced step's trajectory id.
|
|
:param step_index: The surfaced step's index.
|
|
:returns: The matching WAITING :class:`PendingInteraction`, or ``None``.
|
|
"""
|
|
for step in steps:
|
|
pending = pending_interaction(step)
|
|
if pending is None:
|
|
continue
|
|
if pending["trajectory_id"] == trajectory_id and pending["step_index"] == step_index:
|
|
return pending
|
|
return None
|
|
|
|
|
|
async def bridge_interaction(
|
|
cascade_id: str,
|
|
pending: PendingInteraction,
|
|
*,
|
|
port: int,
|
|
get_steps: GetSteps,
|
|
request_elicitation: RequestElicitation,
|
|
deliver: Deliver = _deliver_via_rpc,
|
|
inject_tui: InjectTui = _inject_via_tui,
|
|
max_retries: int = _DEFAULT_MAX_RETRIES,
|
|
) -> None:
|
|
"""
|
|
Surface an agy WAITING interaction, await the verdict, and deliver it.
|
|
|
|
Runs the tight detect→elicit→deliver loop that absorbs agy's WAITING-timeout
|
|
gotcha (see the module docstring / design §2.1). Each iteration:
|
|
|
|
1. Publishes the elicitation under the deterministic id for ``current``'s
|
|
``(cascade_id, trajectory_id, step_index)`` and long-poll-awaits a verdict.
|
|
2. If the verdict is ``None`` (the human timed out or cancelled), **returns
|
|
without delivering** — agy's own WAITING timeout handles the dangling step;
|
|
forcing a deny here is out of scope.
|
|
3. **Re-reads the freshest WAITING step** and delivers against THAT step's
|
|
ids (never the captured ones — they may be stale because agy timed out and
|
|
retried while the human deliberated). If no WAITING step remains, returns.
|
|
4. On the overloaded ``"input not registered"`` error (the targeted step
|
|
timed out before delivery), re-reads for a NEW higher-index WAITING step;
|
|
if found, re-surfaces the elicitation against it (next iteration). Any
|
|
other :class:`AntigravityRpcError` is logged and ends the loop (no infinite
|
|
retry on a genuine shape error).
|
|
|
|
The loop is bounded by ``max_retries`` so a pathological timeout-retry storm
|
|
terminates.
|
|
|
|
:param cascade_id: agy cascade id (equal to the conversation id).
|
|
:param pending: The :class:`PendingInteraction` detected by the read driver.
|
|
:param port: Validated agy connect-RPC port.
|
|
:param get_steps: Re-reads the freshest trajectory steps (production wraps
|
|
:func:`omnigent.antigravity_native_rpc.get_trajectory_steps`).
|
|
:param request_elicitation: Publishes the elicitation under a deterministic
|
|
id and long-poll-awaits the verdict; ``None`` on timeout/cancel.
|
|
:param deliver: Delivers the built payload to agy; defaults to
|
|
:func:`_deliver_via_rpc`, which offloads the blocking
|
|
:func:`omnigent.antigravity_native_rpc.handle_user_interaction` to a
|
|
worker thread.
|
|
:param inject_tui: Types the verdict's key sequence into the agy TUI pane to
|
|
dismiss its in-process prompt AFTER a successful RPC delivery; defaults to
|
|
:func:`_inject_via_tui` (offloads ``send-keys`` to a worker thread). The
|
|
RPC flips the backend step, but the attended TUI keeps its own prompt open
|
|
in parallel — so the keystroke is required to actually advance the
|
|
terminal (#1200). A failure here is logged, not raised: the backend step
|
|
is already answered, so the turn proceeds regardless.
|
|
:param max_retries: Upper bound on detect→deliver iterations.
|
|
:returns: None. Never raises on the expected timeout/cancel/RPC-error paths —
|
|
all are logged and end the loop so the long-lived caller stays alive.
|
|
"""
|
|
current: PendingInteraction = pending
|
|
for attempt in range(max_retries):
|
|
eid = agy_elicitation_id(cascade_id, current["trajectory_id"], current["step_index"])
|
|
params = to_elicitation_params(dict(current))
|
|
|
|
result = await request_elicitation(eid, params)
|
|
if result is None:
|
|
# No verdict. Usually a human timeout/cancel (agy's own WAITING
|
|
# timeout reclaims the step; forcing a deny is a separate policy), but
|
|
# ``request_elicitation`` also returns ``None`` on a hook rejection
|
|
# (a 4xx logged distinctly by the reader's elicitation poster), so the
|
|
# cause is not exclusively timeout/cancel. Don't deliver either way.
|
|
_logger.info(
|
|
"agy elicitation %s returned no verdict (human timeout/cancel, or "
|
|
"hook rejection — see reader warnings); not delivering "
|
|
"(cascade=%s, kind=%s, step=%d)",
|
|
eid,
|
|
cascade_id,
|
|
current["kind"],
|
|
current["step_index"],
|
|
)
|
|
return
|
|
|
|
# Re-read the steps BEFORE delivering: the captured ids may be stale if agy
|
|
# timed out + retried while the human deliberated. PIN to the step we
|
|
# surfaced if it is STILL WAITING — deliver THIS verdict to THAT gate, never
|
|
# to a different higher-index gate that appeared meanwhile (#1472 review).
|
|
# Only when our captured step is gone (timed out → ERROR) do we fall back to
|
|
# the freshest WAITING, which is agy's same-gate timeout-retry (§2.1).
|
|
steps = await get_steps()
|
|
fresh = _waiting_step_at(
|
|
steps, trajectory_id=current["trajectory_id"], step_index=current["step_index"]
|
|
) or _freshest_waiting(steps, kind=current["kind"])
|
|
if fresh is None:
|
|
_logger.warning(
|
|
"agy elicitation %s resolved but no WAITING step remains to "
|
|
"deliver to (cascade=%s, kind=%s); the interaction likely timed "
|
|
"out server-side",
|
|
eid,
|
|
cascade_id,
|
|
current["kind"],
|
|
)
|
|
return
|
|
|
|
payload = to_interaction_payload(current["kind"], result, current["spec"])
|
|
try:
|
|
await deliver(
|
|
port,
|
|
cascade_id,
|
|
trajectory_id=fresh["trajectory_id"],
|
|
step_index=fresh["step_index"],
|
|
payload=payload,
|
|
)
|
|
# The RPC flipped the backend trajectory step, but the attended agy
|
|
# TUI keeps its OWN permission/question prompt open in parallel
|
|
# (live-verified — see the module note / spike doc). Type the verdict's
|
|
# selection into the pane so the terminal actually advances (#1200) and
|
|
# the next typed turn does not land in the stale prompt's buffer. The
|
|
# backend is already answered, so a TUI-typing failure is logged, not
|
|
# raised — never undo a delivered verdict over a flaky pane.
|
|
keys = to_tui_selection_keys(current["kind"], result, current["spec"])
|
|
if keys:
|
|
try:
|
|
await inject_tui(keys)
|
|
except Exception as tui_exc:
|
|
# Best-effort TUI dismissal — the backend step is already
|
|
# answered, so never undo a delivered verdict over a flaky pane.
|
|
_logger.warning(
|
|
"agy interaction delivered over RPC but TUI dismissal failed "
|
|
"(cascade=%s, kind=%s, step=%d, keys=%r): %r",
|
|
cascade_id,
|
|
current["kind"],
|
|
fresh["step_index"],
|
|
keys,
|
|
tui_exc,
|
|
)
|
|
return # delivered successfully
|
|
except AntigravityRpcError as exc:
|
|
if _INPUT_NOT_REGISTERED not in str(exc).lower():
|
|
# A genuine shape/transport error — not the timed-out-step race.
|
|
# Log and stop; retrying would not help and could loop forever.
|
|
_logger.warning(
|
|
"agy interaction delivery failed (cascade=%s, step=%d): %s",
|
|
cascade_id,
|
|
fresh["step_index"],
|
|
exc,
|
|
)
|
|
return
|
|
# The targeted step timed out (status ERROR) before delivery. agy
|
|
# auto-retries with a fresh WAITING step at a HIGHER index; find it
|
|
# and re-surface the elicitation against that step next iteration.
|
|
retry = _freshest_waiting(
|
|
await get_steps(),
|
|
kind=current["kind"],
|
|
after_index=fresh["step_index"],
|
|
)
|
|
if retry is None:
|
|
_logger.info(
|
|
"agy step %d timed out before delivery and no newer WAITING "
|
|
"step appeared (cascade=%s); giving up",
|
|
fresh["step_index"],
|
|
cascade_id,
|
|
)
|
|
return
|
|
_logger.info(
|
|
"agy step %d timed out before delivery; re-surfacing against "
|
|
"retry step %d (cascade=%s, attempt=%d)",
|
|
fresh["step_index"],
|
|
retry["step_index"],
|
|
cascade_id,
|
|
attempt + 1,
|
|
)
|
|
current = retry
|
|
_logger.warning(
|
|
"agy interaction bridge exhausted %d delivery attempts (cascade=%s); "
|
|
"giving up on the timeout-retry chain",
|
|
max_retries,
|
|
cascade_id,
|
|
)
|