"""Goose-native tool-approval mirror (TUI → web elicitation). In ``approve`` / ``smart_approve`` mode the native ``goose session`` TUI gates tool calls with an in-terminal ``cliclack`` selector (its ``prompt_tool_confirmation``). To also surface those approvals in the Omnigent web UI (so a user can answer from the chat view, not only the embedded terminal), the runner watches the Goose pane: 1. poll ``capture-pane`` and detect the confirmation block — Goose renders the question ``Goose would like to call the above tool, do you allow?`` (or, with a security message, ``Do you allow this tool call?``) followed by a cliclack radio list ``Allow`` / ``Always Allow`` / ``Deny`` / ``Cancel`` (verified against goose-cli ``session/mod.rs::prompt_tool_confirmation``), 2. POST it to the server's generic ``native-permission-request`` hook, which publishes ``response.elicitation_request`` and parks for the web verdict, 3. on the verdict, DRIVE the cliclack selector: ``Enter`` chooses the default-highlighted ``Allow``; to deny, send ``Down`` to the ``Deny`` row then ``Enter`` (the ``Deny`` index is 2 when ``Always Allow`` is offered, else 1), 4. if the prompt disappears on its own (answered in the embedded terminal), POST ``external_elicitation_resolved`` so the parked web card clears. This does NOT suppress Goose's gate — its cliclack prompt stays the source of truth and the fallback if pane detection ever fails (the user can still arrow + Enter in the terminal). Mirrors :mod:`omnigent.cursor_native_permissions`; the arrow-driven selector (vs cursor's single-key) is the fragile part and is worth confirming against a live Goose. """ from __future__ import annotations import asyncio import hashlib import logging import re from dataclasses import dataclass from pathlib import Path import httpx from omnigent.goose_native_bridge import capture_goose_pane, send_goose_pane_keys _logger = logging.getLogger(__name__) _POLL_INTERVAL_S = 0.3 _POST_TIMEOUT_S = 86400.0 # The confirmation question — both phrasings contain "do you allow"; it is only # present while cliclack is awaiting a choice, so it's the liveness signal. _PROMPT_RE = re.compile(r"do you allow", re.IGNORECASE) # cliclack box-drawing / radio prefixes to strip when reading the subject lines. _CLICLACK_PREFIX_RE = re.compile(r"^[\s│◆◇◊○●▲▶>|*-]+") _ITEM_LABELS = ("Always Allow", "Allow", "Deny", "Cancel") _SUBJECT_SCAN_LINES = 8 @dataclass(frozen=True) class GooseApprovalPrompt: """A parsed Goose cliclack tool-confirmation prompt. :param subject: The tool/command context shown above the question (best effort, for the card preview + dedupe). :param message: Human-readable card message. :param preview: Compact preview for the card. :param deny_down_count: Number of ``Down`` presses from the default- highlighted ``Allow`` to reach ``Deny`` (2 with "Always Allow", else 1). :param block_hash: Stable hash of the subject used to dedupe across polls and to mint a stable elicitation id. """ subject: str message: str preview: str deny_down_count: int block_hash: str def goose_permission_elicitation_id(session_id: str, token: str) -> str: """Return the deterministic Omnigent elicitation id for a Goose prompt. *token* identifies one approval episode (a per-session counter), not the scraped content — the rendered tool context above the cliclack widget jitters across polls, so hashing it spawned a duplicate card every poll. """ return f"elicit_goose_{session_id}_{token}" def _looks_like_item(line: str) -> bool: """Whether *line* is one of the cliclack radio item rows.""" stripped = _CLICLACK_PREFIX_RE.sub("", line).strip() return any(stripped.startswith(label) for label in _ITEM_LABELS) def parse_goose_approval_prompt(pane: str) -> GooseApprovalPrompt | None: """Parse a Goose cliclack tool-confirmation block from rendered pane text. Requires the ``do you allow`` question AND both an ``Allow`` and a ``Deny`` radio item, so unrelated text never trips it. :param pane: Visible pane text from ``capture-pane -p``. :returns: The parsed prompt, or ``None`` when no live prompt is visible. """ if not pane: return None match = _PROMPT_RE.search(pane) if match is None: return None lines = pane.splitlines() question_idx = next((i for i, ln in enumerate(lines) if _PROMPT_RE.search(ln)), None) if question_idx is None: return None # The radio items render after the question. tail = "\n".join(lines[question_idx:]) has_allow = re.search(r"\bAllow\b", tail) is not None has_deny = re.search(r"\bDeny\b", tail) is not None if not (has_allow and has_deny): return None # "Always Allow" present → Deny is the 3rd item (2 downs); else 2nd (1 down). deny_down_count = 2 if re.search(r"Always Allow", tail) else 1 # Subject = the meaningful (non-item, non-box) lines just above the question, # i.e. the tool-request context Goose rendered. Best effort; used for the card # preview and to dedupe distinct tool calls (the question text is generic). subject_lines: list[str] = [] start = max(0, question_idx - _SUBJECT_SCAN_LINES) for ln in lines[start:question_idx]: if _looks_like_item(ln): continue cleaned = _CLICLACK_PREFIX_RE.sub("", ln).strip() if cleaned: subject_lines.append(cleaned) subject = " | ".join(subject_lines[-3:])[:1024] digest_src = subject or tail block_hash = hashlib.sha256(digest_src.encode("utf-8")).hexdigest()[:16] return GooseApprovalPrompt( subject=subject, message="Goose wants to call a tool. Allow?", preview=subject or "Goose tool call", deny_down_count=deny_down_count, block_hash=block_hash, ) async def supervise_goose_approval_mirror( *, base_url: str, headers: dict[str, str], session_id: str, bridge_dir: Path, auth: httpx.Auth | None = None, poll_interval_s: float = _POLL_INTERVAL_S, ) -> None: """Poll the Goose pane and mirror its approval prompts to web elicitations. :param base_url: Server base URL. :param headers: Auth/routing headers for the runner's requests. :param session_id: Omnigent conversation id. :param bridge_dir: The goose-native bridge dir holding ``tmux.json``. :param auth: Optional httpx auth for the runner's requests. :param poll_interval_s: Pane poll cadence in seconds. """ active: dict[str, object] | None = None episode = 0 timeout = httpx.Timeout(_POST_TIMEOUT_S, connect=10.0) async with httpx.AsyncClient( base_url=base_url, headers=headers, auth=auth, timeout=timeout ) as client: while True: try: pane = await asyncio.to_thread(capture_goose_pane, bridge_dir) prompt = parse_goose_approval_prompt(pane) if pane else None if prompt is not None: # Rising edge only: ONE card per visible-prompt episode. We do # NOT re-mint while the prompt stays up — the scraped tool # context above the cliclack widget jitters across polls, and # keying on it previously parked a fresh card every poll (all # left dangling when the TUI answer cleared only the latest). if active is None: episode += 1 elicitation_id = goose_permission_elicitation_id(session_id, str(episode)) task = asyncio.create_task( _run_one_approval( client, session_id=session_id, bridge_dir=bridge_dir, prompt=prompt, elicitation_id=elicitation_id, ), name=f"goose-approval-{episode}", ) active = {"elicitation_id": elicitation_id, "task": task} elif active is not None: # Falling edge: the prompt vanished. If the web card is still # parked (answered in the TUI), release it; if the task already # finished (answered via the web verdict), nothing to do. task = active["task"] if isinstance(task, asyncio.Task) and not task.done(): await _post_external_elicitation_resolved( client, session_id, str(active["elicitation_id"]) ) active = None except asyncio.CancelledError: raise except Exception: _logger.exception( "goose approval mirror poll failed; session=%s bridge_dir=%s", session_id, bridge_dir, ) await asyncio.sleep(poll_interval_s) async def _run_one_approval( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, prompt: GooseApprovalPrompt, elicitation_id: str, ) -> None: """Park one Goose prompt on the server and drive the cliclack selector.""" payload = { "elicitation_id": elicitation_id, "agent": "Goose", "policy_name": "goose_native_permission", "operation_type": "tool", "message": prompt.message, "content_preview": prompt.preview, } try: response = await client.post( f"/v1/sessions/{session_id}/hooks/native-permission-request", json=payload, ) except httpx.HTTPError: _logger.exception("goose permission hook POST failed; session=%s", session_id) return if response.status_code >= 400: _logger.warning( "goose permission hook rejected: status=%s body=%s", response.status_code, response.text[:512], ) return if not response.content: return try: result = response.json() except ValueError: _logger.warning("goose permission hook returned non-JSON: %s", response.text[:512]) return action = result.get("action") if isinstance(result, dict) else None # Drive the cliclack radio: Enter selects the default-highlighted "Allow"; # Down×N + Enter selects "Deny". A decline/cancel verdict both map to Deny. keys: tuple[str, ...] | None = None if action == "accept": keys = ("Enter",) elif action in {"decline", "cancel"}: keys = (*(["Down"] * prompt.deny_down_count), "Enter") if keys is None: return try: await asyncio.to_thread(send_goose_pane_keys, bridge_dir, *keys) except RuntimeError: _logger.exception( "failed to send goose approval keystrokes %r; session=%s", keys, session_id ) async def _post_external_elicitation_resolved( client: httpx.AsyncClient, session_id: str, elicitation_id: str ) -> None: """Tell the server the native TUI answered a pending Goose prompt.""" try: response = await client.post( f"/v1/sessions/{session_id}/events", json={ "type": "external_elicitation_resolved", "data": {"elicitation_id": elicitation_id}, }, timeout=10.0, ) if response.status_code >= 400: _logger.warning( "goose external_elicitation_resolved rejected: status=%s body=%s", response.status_code, response.text[:512], ) except httpx.HTTPError: _logger.exception("goose external_elicitation_resolved POST failed")