"""Qwen-native tool-approval mirror (TUI → web elicitation). The ``qwen`` TUI gates its own tool calls (shell, file write, …) with an in-terminal approval prompt. Unlike cursor-native — where the prompt lives only in the TUI's in-memory state and has to be scraped off the pane — qwen exposes a structured **permission control plane** on its dual-output stream: whenever a tool needs approval it emits a ``control_request`` / ``can_use_tool`` event on ``--json-file`` (coexisting with the in-terminal prompt) and accepts a ``confirmation_response`` on ``--input-file`` (whichever side answers first wins, the loser is harmlessly dropped). See qwen's ``dual-output.md``. So to surface those approvals in the Omnigent web UI (so a user on the Chat tab can answer from the chat view, not just inside the embedded terminal), the runner tails the same event stream the transcript forwarder uses: 1. read a ``control_request`` / ``can_use_tool`` off ``--json-file`` (structured, no pane scraping), 2. POST it to the server's generic ``native-permission-request`` hook (shared with the hermes-/goose-native mirrors; ``agent="qwen"`` labels the card), which publishes the standard ``response.elicitation_request`` event and parks for the web verdict (the same machinery cursor-/codex-native use), 3. on the verdict, answer qwen by appending a ``confirmation_response`` to ``--input-file`` (``allowed=True`` on accept, ``False`` on decline/cancel) — no keystrokes, 4. if instead a ``control_response`` for that ``request_id`` appears while the card is still parked (the user answered inside the embedded terminal, or qwen auto-resolved), POST ``external_elicitation_resolved`` so the parked web card clears and skip the now-stale ``confirmation_response``. This deliberately does NOT suppress qwen's native gate; qwen's own prompt remains the source of truth and the fallback if the mirror ever misses a request (the user can still answer in the terminal). It is the cleaner, structured analog of :mod:`omnigent.cursor_native_permissions`. See ``docs/QWEN_NATIVE_DESIGN.md`` and ``docs/QWEN_FOLLOWUPS.md``. """ from __future__ import annotations import asyncio import json import logging from dataclasses import dataclass from pathlib import Path import httpx from omnigent.qwen_native_bridge import events_file_path, submit_confirmation _logger = logging.getLogger(__name__) #: Event-file poll cadence. Matches the transcript forwarder so a pending #: approval surfaces in the web UI within a step of the terminal prompt. _POLL_INTERVAL_S = 0.4 # The approval hook parks server-side until a human answers; allow a day, well # past any realistic wait, so the runner's POST never abandons a live prompt. _POST_TIMEOUT_S = 86400.0 #: Cap on a preview string POSTed to the card (server truncates too). _PREVIEW_MAX = 1024 @dataclass(frozen=True) class QwenApprovalRequest: """A parsed qwen ``can_use_tool`` control request. :param request_id: qwen's ``request_id`` for the control request — the correlation key answered via ``confirmation_response`` and matched against a later ``control_response``. :param tool_name: The tool qwen wants to run, e.g. ``"run_shell_command"``. :param message: Human-readable card message. :param preview: Compact preview for the card (the shell command, or the JSON-encoded tool input). """ request_id: str tool_name: str message: str preview: str @dataclass(frozen=True) class _ControlEvent: """One parsed control-plane event from ``--json-file``. :param kind: ``"request"`` (a ``can_use_tool`` to park) or ``"response"`` (a resolution to release the loser). :param request_id: The control request's id (correlates request ↔ response). :param approval: The parsed request when ``kind == "request"``, else ``None``. """ kind: str request_id: str approval: QwenApprovalRequest | None def qwen_permission_elicitation_id(session_id: str, request_id: str) -> str: """Return the deterministic Omnigent elicitation id for a qwen control request. qwen's ``request_id`` is already unique per pending tool call, so it keys the elicitation directly — stable across polls and recomputable for the loser-release path. """ return f"elicit_qwen_{session_id}_{request_id}" def _preview_for(tool_name: str, tool_input: object) -> str: """Render a compact card preview from a qwen tool input. A shell tool's ``command`` is the most useful single line; otherwise the JSON-encoded input, falling back to the bare tool name. """ if isinstance(tool_input, dict): command = tool_input.get("command") if isinstance(command, str) and command.strip(): return command.strip()[:_PREVIEW_MAX] try: return json.dumps(tool_input, ensure_ascii=False)[:_PREVIEW_MAX] except (TypeError, ValueError): pass return tool_name[:_PREVIEW_MAX] def parse_can_use_tool(event: dict[str, object]) -> QwenApprovalRequest | None: """Parse a ``control_request`` / ``can_use_tool`` event, or ``None`` to skip. Tolerant of odd shapes: a missing ``tool_name`` degrades to ``"tool"`` and a non-dict ``input`` to an empty preview rather than dropping the request. :param event: One decoded ``--json-file`` event. :returns: The parsed approval request, or ``None`` when *event* is not a ``can_use_tool`` control request. """ if event.get("type") != "control_request": return None request = event.get("request") if not isinstance(request, dict) or request.get("subtype") != "can_use_tool": return None request_id = event.get("request_id") if not isinstance(request_id, str) or not request_id: return None raw_name = request.get("tool_name") tool_name = raw_name if isinstance(raw_name, str) and raw_name else "tool" preview = _preview_for(tool_name, request.get("input")) return QwenApprovalRequest( request_id=request_id, tool_name=tool_name, message=f"qwen wants to run {tool_name}", preview=preview, ) def _control_response_request_id(event: dict[str, object]) -> str | None: """Return the ``request_id`` of a ``control_response`` event, or ``None``. qwen emits ``control_response`` whether the decision was made in the TUI or by an external ``confirmation_response`` — either way it marks the request resolved, which is the signal to release a still-parked web card. """ if event.get("type") != "control_response": return None response = event.get("response") if not isinstance(response, dict): return None request_id = response.get("request_id") return request_id if isinstance(request_id, str) and request_id else None def _read_new_control_events(events_file: Path, offset: int) -> tuple[list[_ControlEvent], int]: """Read NDJSON lines past *offset*, returning control events + the new offset. Mirrors :func:`omnigent.qwen_native_forwarder._read_new_events`: detects a truncated/recreated file (``size < offset`` → rewind to 0), consumes only fully terminated lines, and leaves a trailing partial line for the next poll. Only control-plane events (``control_request`` / ``control_response``) are returned; user/assistant transcript events are the forwarder's concern. """ try: size = events_file.stat().st_size except OSError: return [], offset if size < offset: offset = 0 # file truncated by a relaunched terminal if size == offset: return [], offset try: with open(events_file, "rb") as fh: fh.seek(offset) data = fh.read(size - offset) except OSError: return [], offset last_nl = data.rfind(b"\n") if last_nl == -1: return [], offset # no complete line yet consumed = data[: last_nl + 1] new_offset = offset + len(consumed) events: list[_ControlEvent] = [] for raw in consumed.split(b"\n"): raw = raw.strip() if not raw: continue try: event = json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): continue # tolerate a malformed line rather than stalling the tail if not isinstance(event, dict): continue approval = parse_can_use_tool(event) if approval is not None: events.append(_ControlEvent("request", approval.request_id, approval)) continue response_id = _control_response_request_id(event) if response_id is not None: events.append(_ControlEvent("response", response_id, None)) return events, new_offset async def supervise_qwen_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: """ Tail qwen's ``--json-file`` and mirror its approval prompts to web elicitations. Runs for the session's lifetime (cancelled on teardown; any other exception is logged and the loop continues, so a transient failure never abandons the gate). Seeds the read offset at the current end of file so only *new* control requests are parked — history was already resolved, and re-parking it would flash stale cards. Several tool calls can be parked at once (qwen tags each with its own ``request_id``); a ``control_response`` for a still-parked request means it was answered in the TUI (or auto-resolved), so the web card is released via ``external_elicitation_resolved``. :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 qwen-native bridge dir holding the event/input files. :param auth: Optional httpx auth for the runner's requests. :param poll_interval_s: Event-file poll cadence in seconds. """ events_file = events_file_path(bridge_dir) # Start watching from "now": only act on prompts emitted after launch. A # request already in the file was either resolved (has a control_response) or # is still showing in the TUI as the fallback; either way re-parking it would # be wrong. Matches cursor-native's "the terminal is the fallback" stance. try: offset = events_file.stat().st_size except OSError: offset = 0 # request_id -> {"elicitation_id": str, "task": asyncio.Task} pending: dict[str, dict[str, object]] = {} 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: events, offset = await asyncio.to_thread( _read_new_control_events, events_file, offset ) # A request whose control_response is already in THIS same batch # was answered (in the TUI or auto) within one poll window, before # we could park it. Parking it now would race its own response: the # response branch runs against a freshly-created task that hasn't # POSTed yet, so it can't release the card, which would then linger # until the server-side park times out. The decision is already # made — skip the card entirely. resolved_in_batch = {ev.request_id for ev in events if ev.kind == "response"} for ev in events: if ev.kind == "request": if ( ev.request_id in pending or ev.approval is None or ev.request_id in resolved_in_batch ): continue elicitation_id = qwen_permission_elicitation_id(session_id, ev.request_id) task = asyncio.create_task( _run_one_approval( client, session_id=session_id, bridge_dir=bridge_dir, approval=ev.approval, elicitation_id=elicitation_id, ), name=f"qwen-approval-{ev.request_id}", ) pending[ev.request_id] = { "elicitation_id": elicitation_id, "task": task, } else: # "response": the request was resolved entry = pending.pop(ev.request_id, None) if entry is None: continue task = entry["task"] if isinstance(task, asyncio.Task) and not task.done(): # Resolved in the TUI (or auto) before the web card # answered → release the parked card. await _post_external_elicitation_resolved( client, session_id, str(entry["elicitation_id"]) ) except asyncio.CancelledError: raise except Exception: _logger.exception( "qwen 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, approval: QwenApprovalRequest, elicitation_id: str, ) -> None: """Park one qwen control request on the server and answer with the verdict.""" # Reuse the vendor-agnostic native-permission hook (shared with the hermes- # and goose-native mirrors); ``agent`` labels the card and ``policy_name`` # keeps the qwen flavor. payload = { "elicitation_id": elicitation_id, "agent": "qwen", "policy_name": "qwen_native_permission", "operation_type": approval.tool_name, "message": approval.message, "content_preview": approval.preview, } try: response = await client.post( f"/v1/sessions/{session_id}/hooks/native-permission-request", json=payload, ) except httpx.HTTPError: _logger.exception("qwen permission hook POST failed; session=%s", session_id) return if response.status_code >= 400: _logger.warning( "qwen permission hook rejected: status=%s body=%s", response.status_code, response.text[:512], ) return if not response.content: # Empty 2xx → resolved elsewhere (TUI answered) or timeout: no response. return try: result = response.json() except ValueError: _logger.warning("qwen permission hook returned non-JSON: %s", response.text[:512]) return action = result.get("action") if isinstance(result, dict) else None if action == "accept": allowed = True elif action in {"decline", "cancel"}: allowed = False else: return try: await asyncio.to_thread( submit_confirmation, bridge_dir, request_id=approval.request_id, allowed=allowed ) except RuntimeError: _logger.exception( "failed to write qwen confirmation_response for %s; session=%s", approval.request_id, 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 qwen 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( "qwen external_elicitation_resolved rejected: status=%s body=%s", response.status_code, response.text[:512], ) except httpx.HTTPError: _logger.exception("qwen external_elicitation_resolved POST failed")