"""File + tmux bridge for the qwen-native terminal harness. Unlike goose-/cursor-native (which simulate keystrokes via ``tmux send-keys``), ``qwen`` ships a built-in remote-control protocol that we drive through two files the TUI is launched against: - ``--input-file`` (:func:`input_file_path`): ``qwen`` ``watchFile``\\s it and parses appended JSONL commands. We append ``{"type":"submit","text":...}`` to deliver a web-UI turn (qwen routes it through the *same* ``submitQuery`` path the keyboard uses, so it renders in the TUI transcript) and ``{"type":"confirmation_response","request_id":...,"allowed":...}`` to answer a tool-approval request. - ``--json-file`` (:func:`events_file_path`): ``qwen`` streams structured JSON events here while the TUI renders normally; :mod:`omnigent.qwen_native_forwarder` tails it. The runner launches the TUI in a runner-owned tmux pane (for the embedded display) and records that pane via :func:`write_tmux_target`. Message injection is file-based, but two affordances still go through the pane because qwen's input-file watcher has no command for them: **interrupt** (Stop → ``Escape``, see :func:`inject_interrupt`) and **hard stop** (:func:`kill_session`). Verified against ``qwen`` v0.18.1 (``RemoteInputWatcher`` + dual-output). See ``docs/QWEN_NATIVE_DESIGN.md``. """ from __future__ import annotations import contextlib import hashlib import json import os import re import secrets import socket import subprocess import sys import time import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any #: Env var carrying the bridge dir into the harness executor process. BRIDGE_DIR_ENV_VAR = "HARNESS_QWEN_NATIVE_BRIDGE_DIR" #: Fixed namespace for deriving a stable qwen ``--session-id`` from an Omnigent #: conversation id (UUIDv5). Never change it — it would orphan every existing #: qwen recording (resume would mint a new id and lose history). _QWEN_SESSION_NAMESPACE = uuid.UUID("6b6f3d2e-9a1c-5e84-bf0a-1d7c5a2e9f43") _BRIDGE_ROOT = Path(os.environ.get("TMPDIR", "/tmp")) / f"omnigent-{os.getuid()}" / "qwen-native" _TMUX_FILE = "tmux.json" #: JSONL command file qwen watches (``--input-file``); we append to it. _INPUT_FILE = "qwen_in.jsonl" #: NDJSON event file qwen writes (``--json-file``); the forwarder tails it. _EVENTS_FILE = "qwen_out.ndjson" _TMUX_READY_TIMEOUT_S = 30.0 _TMUX_SEND_TIMEOUT_S = 10.0 _POLL_INTERVAL_S = 0.2 #: Token config the shared Omnigent MCP relay (``serve-mcp``) reads from the #: bridge dir. Mirrors cursor-/claude-native (``cursor_native_bridge.py``). _BRIDGE_CONFIG_FILE = "bridge.json" #: Name qwen lists the Omnigent MCP server under (shows in ``/mcp``). _MCP_SERVER_NAME = "omnigent" #: Per-session MCP config passed to qwen via ``--mcp-config ``. Lives in #: the bridge dir (NOT the workspace), so we never drop a file in the user's repo #: and concurrent same-workspace sessions can't collide. CLI-provided MCP servers #: are also ungated (no "Untrusted MCP server" prompt), unlike a project #: ``.mcp.json`` / ``.qwen/settings.json``. The claude-native ``--mcp-config`` #: model (it writes no workspace file either). _MCP_CONFIG_FILE = "mcp_config.json" #: qwen version string stamped on synthesized recording records + sidecars. Must #: be a version qwen's resume loader accepts; verified loadable on qwen v0.18.2. _QWEN_SYNTH_VERSION = "0.18.2" #: ``contextWindowSize`` stamped on synthesized assistant records. Informational #: only — the live resume uses the resolved model's real window. _QWEN_SYNTH_CONTEXT_WINDOW = 131072 def bridge_dir_for_session_id(session_id: str) -> Path: """Return the per-session bridge dir, e.g. ``/tmp/omnigent-/qwen-native/``.""" digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest()[:32] return _BRIDGE_ROOT / digest def bridge_root() -> Path: """Return the configured qwen-native bridge root.""" return _BRIDGE_ROOT def qwen_session_id_for_conversation(conversation_id: str) -> str: """Return the deterministic qwen ``--session-id`` for an Omnigent conversation. UUIDv5 of the conversation id: stable across resumes (recomputable, never stored) and a valid UUID (qwen requires one). The runner launches a fresh session with ``--session-id `` and later restores it with ``--resume `` so the qwen TUI shows the prior conversation on resume. :param conversation_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :returns: A stable UUID string usable as qwen's session id. """ return str(uuid.uuid5(_QWEN_SESSION_NAMESPACE, conversation_id)) def _qwen_project_slug(workspace: Path | str) -> str: """Return qwen's per-project directory slug for *workspace*. qwen keys its on-disk session store by project: the cwd's real path with every non-alphanumeric character replaced by ``-`` (verified against qwen v0.18.1 — e.g. ``/private/tmp/qwen_x`` → ``-private-tmp-qwen-x``). Uses ``realpath`` because the runner launches qwen with ``cwd=realpath(workspace)`` and qwen records under ``process.cwd()``. """ real = os.path.realpath(str(workspace)) return re.sub(r"[^A-Za-z0-9]", "-", real) def qwen_session_recording_path(session_id: str, workspace: Path | str) -> Path: """Return the path to qwen's on-disk chat recording for *session_id*. ``~/.qwen/projects//chats/.jsonl`` — the JSONL qwen appends interactive-session events to (``--chat-recording``, on by default), scoped to *workspace*'s project slug. The file may not exist yet (a fresh session creates it on first event). Used both to gate ``--resume`` (:func:`qwen_session_recording_exists`) and to tail for the ``chat_compression`` marker (see :mod:`omnigent.qwen_native_forwarder`). """ return ( Path.home() / ".qwen" / "projects" / _qwen_project_slug(workspace) / "chats" / f"{session_id}.jsonl" ) def qwen_session_recording_exists(session_id: str, workspace: Path | str) -> bool: """Return whether qwen has an on-disk chat recording for *session_id* in *workspace*. qwen records interactive sessions (``--chat-recording``, on by default) to ``~/.qwen/projects//chats/.jsonl`` and resolves ``--resume `` **relative to the current project** (cwd slug) — not globally. So the check must be scoped to the *launch workspace's* slug: a glob across all projects would report a recording made under workspace A as present when resuming from workspace B, choosing ``--resume`` and landing the user on qwen's blocking "No saved session found with ID" error screen — the exact failure this guard exists to prevent (moved/renamed repo, or resume from a different cwd). The runner uses this to choose ``--resume`` (recording present here) vs ``--session-id`` (fresh). A false negative (slug drift) only degrades to a clean fresh launch, never the blocking error. :param session_id: A qwen session id (see :func:`qwen_session_id_for_conversation`). :param workspace: The cwd qwen will be (re)launched in. :returns: ``True`` if a recording for *session_id* exists under *workspace*'s qwen project dir. """ try: return qwen_session_recording_path(session_id, workspace).is_file() except OSError: return False def _qwen_iso_now() -> str: """Return the current UTC time as a qwen-style ISO-8601 millisecond stamp.""" return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" def _qwen_synth_uuid(qwen_session_id: str, index: int) -> str: """Return a deterministic record uuid for a synthesized qwen recording. UUIDv5 over the session id + position so re-running the fork rebuild produces a byte-identical recording (idempotent rebuilds, stable tests) instead of fresh random ids each launch. """ return str(uuid.uuid5(_QWEN_SESSION_NAMESPACE, f"{qwen_session_id}:{index}")) def _qwen_text_from_api_content(content: object, api_type: str) -> str: """Concatenate the text of an Omnigent content array's blocks of *api_type*. :param content: Omnigent content array, e.g. ``[{"type":"input_text","text":"hi"}]``. :param api_type: Block type to include, ``"input_text"`` or ``"output_text"``. :returns: The joined text, or ``""`` when there is none. """ if not isinstance(content, list): return "" parts: list[str] = [] for block in content: if isinstance(block, dict) and block.get("type") == api_type: text = block.get("text") if isinstance(text, str) and text: parts.append(text) return "".join(parts) def qwen_session_records_from_session_items( items: list[dict[str, Any]], *, qwen_session_id: str, cwd: Path | str, model: str = "", timestamp: str | None = None, ) -> list[dict[str, Any]]: """Convert Omnigent session items into qwen chat-recording JSONL records. qwen's recording is a linked list chained by ``uuid`` / ``parentUuid``. We emit only the ``user`` / ``assistant`` message records qwen reconstructs history from; the ``system`` snapshot records it writes live are telemetry and not required for ``--resume`` (verified on v0.18.2). Tool calls are dropped — text turns carry the context a cross-harness fork needs. Omnigent items map as: - user ``message`` → ``{"type":"user","message":{"role":"user","parts":[{"text"}]}}`` - assistant ``message`` → ``{"type":"assistant","message":{"role":"model",...}}`` Cancelled turns aren't restored: an interrupted assistant turn and its response group are skipped (claude/codex/pi share a ``response_id`` across the turn), and a trailing unanswered user prompt is dropped (covers a qwen-native source, whose per-event ``response_id`` doesn't group a turn). :param items: Flat Omnigent item dicts in chronological order. :param qwen_session_id: qwen session id stamped on every record. :param cwd: Working directory stamped on records (realpath'd to match the project slug qwen records under). :param model: Default model id for assistant records; overridden per-item by the item's own ``model`` when present. :param timestamp: ISO stamp for all records; defaults to now. Pass a fixed value for deterministic output in tests. :returns: qwen recording record dicts in order (empty if nothing carryable). """ ts = timestamp or _qwen_iso_now() cwd_str = os.path.realpath(str(cwd)) skip_response_ids = { item.get("response_id") for item in items if item.get("type") == "message" and item.get("role") == "assistant" and item.get("interrupted") is True and isinstance(item.get("response_id"), str) and item.get("response_id") } records: list[dict[str, Any]] = [] parent_uuid: str | None = None for index, item in enumerate(items): if item.get("type") != "message": continue response_id = item.get("response_id") if isinstance(response_id, str) and response_id in skip_response_ids: continue role = item.get("role") if role == "user": text = _qwen_text_from_api_content(item.get("content"), "input_text") if not text: continue rec_uuid = _qwen_synth_uuid(qwen_session_id, index) records.append( { "uuid": rec_uuid, "parentUuid": parent_uuid, "sessionId": qwen_session_id, "timestamp": ts, "type": "user", "cwd": cwd_str, "version": _QWEN_SYNTH_VERSION, "message": {"role": "user", "parts": [{"text": text}]}, } ) parent_uuid = rec_uuid elif role == "assistant": text = _qwen_text_from_api_content(item.get("content"), "output_text") if not text: continue item_model = item.get("model") eff_model = item_model if isinstance(item_model, str) and item_model else model rec_uuid = _qwen_synth_uuid(qwen_session_id, index) records.append( { "uuid": rec_uuid, "parentUuid": parent_uuid, "sessionId": qwen_session_id, "timestamp": ts, "type": "assistant", "cwd": cwd_str, "version": _QWEN_SYNTH_VERSION, "model": eff_model, "message": {"role": "model", "parts": [{"text": text}]}, "usageMetadata": { "promptTokenCount": 0, "candidatesTokenCount": 0, "totalTokenCount": 0, }, "contextWindowSize": _QWEN_SYNTH_CONTEXT_WINDOW, } ) parent_uuid = rec_uuid # Drop a trailing unanswered user prompt (a turn cancelled before any reply). # The response-group skip can't catch it for a qwen-native source, whose # per-event ``response_id`` (``qwen:``) doesn't group a turn; a # committed transcript otherwise ends on a completed assistant turn. while records and records[-1]["type"] == "user": records.pop() return records def write_qwen_session_recording( qwen_session_id: str, workspace: Path | str, records: list[dict[str, Any]], *, timestamp: str | None = None, ) -> Path: """Write a synthesized qwen chat recording (+ discovery sidecars) to disk. qwen resolves ``--resume `` from THREE files under its per-project dir, not the ``.jsonl`` alone (verified on v0.18.2 — a bare recording lands the user on the blocking "No saved session found" screen): - ``chats/.jsonl`` — the conversation records (*records*). - ``chats/.runtime.json`` — the session index entry ``sessions list`` / ``--resume`` read to discover the session. - ``meta.json`` — the project-level marker (created if absent; an existing one is left untouched so we don't reset another session's ``createdAt``). All three are written atomically, and the ``.jsonl`` is committed LAST (after both sidecars): :func:`qwen_session_recording_exists` keys on the ``.jsonl``, so a failed sidecar write leaves no ``.jsonl`` and the launch degrades to a clean fresh start, never the blocking "No saved session found" screen. :param qwen_session_id: qwen session id (file stem + ``session_id`` field). :param workspace: cwd qwen will resume in; its realpath drives the project slug. :param records: qwen recording records (see :func:`qwen_session_records_from_session_items`). :param timestamp: ISO stamp for the project ``meta.json``; defaults to now. :returns: The written ``chats/.jsonl`` recording path. :raises RuntimeError: If the recording cannot be written. """ recording = qwen_session_recording_path(qwen_session_id, workspace) chats_dir = recording.parent chats_dir.mkdir(mode=0o700, parents=True, exist_ok=True) try: hostname = socket.gethostname() or "omnigent" except OSError: hostname = "omnigent" # Sidecars first so the gate file (``.jsonl``) lands last: a sidecar failure # then leaves no ``.jsonl`` and the resume gate cleanly picks a fresh launch. meta_path = chats_dir.parent / "meta.json" if not meta_path.exists(): stamp = timestamp or _qwen_iso_now() _atomic_write_text( meta_path, json.dumps({"version": 1, "createdAt": stamp, "updatedAt": stamp}) ) _atomic_write_text( chats_dir / f"{qwen_session_id}.runtime.json", json.dumps( { "schema_version": 1, "pid": os.getpid(), "session_id": qwen_session_id, "work_dir": os.path.realpath(str(workspace)), "hostname": hostname, "started_at": time.time(), "qwen_version": _QWEN_SYNTH_VERSION, } ), ) _atomic_write_text( recording, "".join(json.dumps(r, separators=(",", ":")) + "\n" for r in records) ) return recording def _atomic_write_text(target: Path, text: str) -> None: """Atomically write *text* to *target* (temp file + ``os.replace``). :param target: Destination path. :param text: Full file contents to write. :raises RuntimeError: If the file cannot be written; the temp is cleaned up. """ tmp = target.with_suffix(target.suffix + ".tmp") try: tmp.write_text(text, encoding="utf-8") os.replace(tmp, target) except OSError as exc: with contextlib.suppress(FileNotFoundError): tmp.unlink() raise RuntimeError(f"Failed to write {target}: {exc}") from exc def input_file_path(bridge_dir: Path) -> Path: """Return the ``--input-file`` path qwen watches for JSONL commands.""" return bridge_dir / _INPUT_FILE def events_file_path(bridge_dir: Path) -> Path: """Return the ``--json-file`` path qwen writes structured events to.""" return bridge_dir / _EVENTS_FILE def _ensure_dir(path: Path) -> None: """Create *path* (and parents) with owner-only permissions.""" path.mkdir(parents=True, exist_ok=True) with contextlib.suppress(OSError): os.chmod(path, 0o700) def _ensure_secure_bridge_dir(bridge_dir: Path) -> None: """Create/validate *bridge_dir* as an owner-only chain before writing secrets. ``_ensure_dir`` only ``mkdir(parents=True, exist_ok=True)`` + a suppressed ``chmod`` on the leaf: it trusts pre-existing ancestors, so on a shared host an attacker could pre-create ``$TMPDIR/omnigent-`` (or a deeper ancestor) as a symlink / world-writable dir and redirect the bridge tree. That tree now holds ``bridge.json`` — a bearer token for the relay's localhost control endpoint — so its directory must be hardened. Delegate to the same ``_ensure_secure_dir`` the shared relay (``start_tool_relay``) already applies to token-bearing trees; it rejects symlinked / non-owned / group-or-other accessible ancestors (the qwen-native root is in its allowlist). Lazy import avoids a cycle (``claude_native_bridge`` resolves qwen's ``bridge_root`` lazily in turn). :raises RuntimeError: If any ancestor fails owner-only validation. """ from omnigent.claude_native_bridge import _ensure_secure_dir _ensure_secure_dir(bridge_dir) def prepare_bridge_files(bridge_dir: Path) -> None: """Create the bridge dir and a fresh, empty input file before launch. qwen ``watchFile``\\s the ``--input-file`` and reads from a byte offset, so the path must exist when the TUI starts. We truncate it (and the events file) so a relaunched terminal can't replay a prior process's queued commands. """ _ensure_dir(bridge_dir) # Truncate both so a re-created terminal starts from a clean slate. input_file_path(bridge_dir).write_text("", encoding="utf-8") events_file_path(bridge_dir).write_text("", encoding="utf-8") def build_qwen_native_spawn_env(session_id: str) -> dict[str, str]: """Build the ``HARNESS_QWEN_NATIVE_*`` env the harness executor reads. The executor only needs the bridge dir (to locate the input file it appends to). qwen's model / auth / dual-output flags are set by the runner when it launches the TUI (``_auto_create_qwen_terminal``), not here. :param session_id: The Omnigent session id (keys the bridge dir). :returns: Env-var overrides for the harness spawn. """ bridge_dir = bridge_dir_for_session_id(session_id) _ensure_dir(bridge_dir) return {BRIDGE_DIR_ENV_VAR: str(bridge_dir)} # --------------------------------------------------------------------------- # Omnigent MCP server config — expose Omnigent's builtin tools (sys_*, # load_skill, web_fetch, …) to the qwen TUI so it can call them and ``/mcp`` # lists them. Reuses the shared stdio relay implemented in # ``omnigent.claude_native_bridge serve-mcp`` (same server cursor-/claude-/ # opencode-native point at); only the *registration* surface differs per CLI. # --------------------------------------------------------------------------- def write_mcp_bridge_config(bridge_dir: Path) -> None: """Write the token config the shared Omnigent MCP relay reads at startup. The ``serve-mcp`` relay (spawned by qwen) reads ``bridge.json`` for a bearer token and exits if it's missing, so this must exist *before* qwen launches. ``bridge.json`` only ever holds ``{token}``; the live tool surface is advertised separately via the ``tool_relay.json`` that the runner's comment relay (``ensure_comment_relay`` → ``_ensure_comment_relay_started``) writes into this same bridge dir when it starts. Mirrors :func:`omnigent.cursor_native_bridge.write_mcp_bridge_config`. :raises RuntimeError: If the bridge dir fails owner-only validation (:func:`_ensure_secure_bridge_dir`) — the token is not written. """ _ensure_secure_bridge_dir(bridge_dir) config_path = bridge_dir / _BRIDGE_CONFIG_FILE if config_path.exists(): return payload = {"token": secrets.token_urlsafe(32)} tmp = bridge_dir / (_BRIDGE_CONFIG_FILE + ".tmp") tmp.write_text(json.dumps(payload, sort_keys=True) + "\n", encoding="utf-8") os.replace(tmp, config_path) def build_mcp_server_entry( bridge_dir: Path, *, python_executable: str | None = None, ) -> dict[str, Any]: """Build qwen's ``mcpServers.omnigent`` entry for the Omnigent relay. ``trust: true`` auto-approves the qwen-side MCP tool gate so the TUI doesn't add a second in-terminal prompt: Omnigent already gates these calls through its own policy/elicitation engine (surfaced as web cards by the approval mirror), so qwen's prompt would only be a hidden duplicate. Same rationale as cursor's ``autoApprove`` (see ``cursor_native_bridge.build_mcp_config``). """ python = python_executable or sys.executable return { "command": python, "args": [ "-I", "-m", "omnigent.claude_native_bridge", "serve-mcp", "--bridge-dir", str(bridge_dir), ], "env": { "PYTHONUNBUFFERED": "1", "TMPDIR": os.environ.get("TMPDIR", "/tmp"), }, "trust": True, } def mcp_config_path(bridge_dir: Path) -> Path: """Return the per-session ``--mcp-config`` file path inside the bridge dir.""" return bridge_dir / _MCP_CONFIG_FILE def write_mcp_config( bridge_dir: Path, *, python_executable: str | None = None, ) -> Path: """Write the per-session Omnigent MCP config for qwen's ``--mcp-config`` flag. Writes ``{"mcpServers": {"omnigent": ...}}`` to a file *inside the bridge dir* (never the workspace) and the relay token (:func:`write_mcp_bridge_config`). The runner passes the returned path to qwen via ``--mcp-config ``. Unlike a project ``.mcp.json`` / ``.qwen/settings.json``, a CLI-provided MCP server: - drops no file in the user's (often git) workspace — nothing to accidentally commit, nothing left behind pointing at a dead bridge dir; - is per-session by construction (the file and its ``--bridge-dir`` live in this session's bridge dir), so concurrent same-workspace sessions can't collide on a shared file; - is **not** gated behind qwen's "Untrusted MCP server" prompt (CLI servers carry no project/workspace scope), so no pre-approval step is needed. The claude-native ``--mcp-config`` model (it writes no workspace file either). :returns: Path to the written ``--mcp-config`` file. """ write_mcp_bridge_config(bridge_dir) path = mcp_config_path(bridge_dir) payload = { "mcpServers": { _MCP_SERVER_NAME: build_mcp_server_entry( bridge_dir, python_executable=python_executable ) } } tmp = path.with_name(f"{_MCP_CONFIG_FILE}.{secrets.token_hex(8)}.tmp") tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") os.replace(tmp, path) return path def _append_command(bridge_dir: Path, command: dict[str, Any]) -> None: """Append one JSONL command line to the input file qwen watches. A single ``write`` of one ``\\n``-terminated line is atomic enough for qwen's incremental ``readNewLines`` reader (it splits on newlines), so concurrent appends from ``run_turn`` and a confirmation response don't interleave. :raises RuntimeError: If the input file can't be written. """ line = json.dumps(command, ensure_ascii=False) + "\n" try: _ensure_dir(bridge_dir) with open(input_file_path(bridge_dir), "a", encoding="utf-8") as fh: fh.write(line) fh.flush() except OSError as exc: raise RuntimeError(f"qwen-native could not write input command: {exc}") from exc def wait_for_ready( bridge_dir: Path, *, timeout_s: float = 30.0, poll_interval_s: float = _POLL_INTERVAL_S, ) -> bool: """Block until the qwen TUI has booted its dual-output stream. qwen's ``RemoteInputWatcher`` initializes its read offset (``bytesRead``) to the **current size of the input file** when it starts watching, synchronously during TUI boot (before the React app renders). If we append a ``submit`` *before* that runs, qwen initializes ``bytesRead`` past our line and never reads it — the message is silently dropped. The first turn of a freshly launched session hits exactly this race, since the harness turn fires while ``qwen`` is still starting up (it takes seconds). qwen emits its first event — a ``{"type":"system","subtype":"session_start"}`` on the ``--json-file`` stream — only *after* the watcher's constructor (and thus ``startWatching``) has run. So the appearance of a ``system`` event in the events file is a safe "the watcher is active, ``bytesRead`` was taken on the still-empty input file" signal: appending after it is reliably detected. :param bridge_dir: The qwen-native bridge dir holding the events file. :param timeout_s: Max seconds to wait for the boot signal. :param poll_interval_s: Seconds between polls of the events file. :returns: ``True`` once the boot signal is seen; ``False`` on timeout (the caller submits anyway — best effort beats hanging the turn). """ events_file = events_file_path(bridge_dir) deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: if _events_file_has_system_event(events_file): return True time.sleep(poll_interval_s) return False def _events_file_has_system_event(events_file: Path) -> bool: """Return whether the NDJSON events file contains a parsed ``system`` event. Parses line by line and checks ``event["type"] == "system"`` rather than a raw substring scan: a substring like ``"type":"system"`` could appear inside another event's payload (latching ready early and re-opening the boot race), and the first ``system`` event isn't guaranteed to sit within a fixed byte window. The boot ``session_start`` is qwen's first emitted line, so this returns on the first line in practice. """ try: with open(events_file, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: event = json.loads(line) except ValueError: continue # tolerate a partial/garbled line mid-write if isinstance(event, dict) and event.get("type") == "system": return True except OSError: return False return False def submit_user_message(bridge_dir: Path, *, content: str) -> None: """Deliver a web-UI user message into the qwen TUI via the input file. Appends ``{"type":"submit","text":content}``; qwen's ``RemoteInputWatcher`` routes it through ``submitQuery`` (the keyboard's submit path), so the message renders in the TUI transcript exactly like typed input. :param bridge_dir: The qwen-native bridge dir holding the input file. :param content: User text (non-empty). :raises RuntimeError: If *content* is empty or the input file can't be written. """ if not content: raise RuntimeError("qwen-native submit requires non-empty content") _append_command(bridge_dir, {"type": "submit", "text": content}) def submit_confirmation(bridge_dir: Path, *, request_id: str, allowed: bool) -> None: """Answer a qwen ``can_use_tool`` control request via the input file. :param bridge_dir: The qwen-native bridge dir holding the input file. :param request_id: The ``request_id`` from the ``control_request`` event. :param allowed: Whether the tool call is permitted. :raises RuntimeError: If the input file can't be written. """ _append_command( bridge_dir, {"type": "confirmation_response", "request_id": request_id, "allowed": allowed}, ) # --------------------------------------------------------------------------- # tmux target (display pane) — used only for interrupt / hard-stop. # --------------------------------------------------------------------------- def write_tmux_target( bridge_dir: Path, *, socket_path: Path, tmux_target: str, pid: int | None = None, ) -> None: """Advertise the tmux socket + target for the running qwen terminal.""" _ensure_dir(bridge_dir) payload: dict[str, Any] = { "socket_path": str(socket_path), "tmux_target": tmux_target, "updated_at": time.time(), } if pid is not None: payload["pid"] = pid tmp = bridge_dir / (_TMUX_FILE + ".tmp") tmp.write_text(json.dumps(payload), encoding="utf-8") os.replace(tmp, bridge_dir / _TMUX_FILE) def read_tmux_info(bridge_dir: Path) -> dict[str, str] | None: """Return ``{socket_path, tmux_target}`` from ``tmux.json``, or ``None``.""" try: raw = (bridge_dir / _TMUX_FILE).read_text(encoding="utf-8") except OSError: return None try: data = json.loads(raw) except ValueError: return None socket_path = data.get("socket_path") tmux_target = data.get("tmux_target") if ( isinstance(socket_path, str) and socket_path and isinstance(tmux_target, str) and tmux_target ): return {"socket_path": socket_path, "tmux_target": tmux_target} return None def _wait_for_tmux_info(bridge_dir: Path, *, timeout_s: float) -> dict[str, str]: """Block until ``tmux.json`` is advertised, or raise on timeout.""" deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: info = read_tmux_info(bridge_dir) if info is not None: return info time.sleep(_POLL_INTERVAL_S) raise RuntimeError(f"qwen-native tmux target was not advertised within {timeout_s:.0f}s") def _run_tmux(socket_path: str, *args: str) -> None: """Invoke ``tmux -S `` and raise on failure.""" try: proc = subprocess.run( ["tmux", "-S", socket_path, *args], check=False, capture_output=True, text=True, timeout=_TMUX_SEND_TIMEOUT_S, ) except subprocess.TimeoutExpired as exc: raise RuntimeError(f"tmux command timed out after {_TMUX_SEND_TIMEOUT_S}s") from exc if proc.returncode != 0: detail = proc.stderr.strip() or proc.stdout.strip() or "" raise RuntimeError(f"tmux command failed (rc={proc.returncode}): {detail}") def inject_interrupt(bridge_dir: Path, *, timeout_s: float = _TMUX_READY_TIMEOUT_S) -> None: """Cancel the in-flight qwen turn by sending ``Escape`` to the pane. qwen's input-file watcher accepts only ``submit`` / ``confirmation_response``, so the web UI's Stop button drives interrupt through the display pane — the analog of :func:`submit_user_message` for cancellation. The harness ``run_turn`` returns right after appending the submit line, so the runner's in-process cancel floor can't reach the turn. :raises RuntimeError: If the tmux target is not advertised or send-keys fails. """ info = _wait_for_tmux_info(bridge_dir, timeout_s=timeout_s) # No ``-l``: tmux must interpret ``Escape`` as a key name. _run_tmux(info["socket_path"], "send-keys", "-t", info["tmux_target"], "Escape") def kill_session(bridge_dir: Path, *, timeout_s: float = _TMUX_READY_TIMEOUT_S) -> None: """Hard-stop the qwen session by killing its tmux session. Terminates ``qwen`` and the pane outright — the analog of the user manually exiting the attached TUI, for the web UI's "Stop session" affordance. :raises RuntimeError: If the tmux target is not advertised or kill-session fails. """ info = _wait_for_tmux_info(bridge_dir, timeout_s=timeout_s) _run_tmux(info["socket_path"], "kill-session", "-t", info["tmux_target"])