"""Fixtures for browser-driven e2e tests of the web SPA. The suite spawns a real ``omnigent server --agent`` subprocess against ``examples/hello_world.yaml`` and drives the rendered SPA with Playwright. The server is wired to a mock LLM server so the suite runs deterministically without real credentials. The suite is excluded from the default ``pytest`` run via ``--ignore=tests/e2e_ui`` in ``pyproject.toml`` and gated to ``workflow_dispatch`` in CI for now. Local usage:: # one-time setup uv sync --extra e2e-ui uv run playwright install --with-deps chromium # run against a freshly built SPA + spawned server uv run pytest tests/e2e_ui -v # iterate against an already-running server (dev hosts/ports need opt-in) cd web && npm run dev & omnigent server --agent examples/hello_world.yaml & OMNIGENT_E2E_ALLOW_DEV_BASE_URL=1 \ uv run pytest tests/e2e_ui --ui-base-url http://127.0.0.1:5173 ``omnigent server`` is documented at ``omnigent/cli.py:server``: it spins up uvicorn with the Omnigent app and spawns an out-of-process runner that reconnects over the WebSocket tunnel. The fixture passes ``--database-uri`` and ``--artifact-location`` pointing at the pytest tmp dir so the test never touches the user's default ``sqlite:///omnigent.db`` / ``./artifacts``. """ from __future__ import annotations import contextlib import io import json import os import re import shutil import signal import socket import subprocess import sys import tarfile import textwrap import time from collections.abc import Generator, Iterator from dataclasses import dataclass from pathlib import Path from typing import Any import filelock import httpx import pytest from playwright.sync_api import Locator, Page, expect from tests.codex_parity.helpers import ev_assistant_message, ev_completed, ev_response_created from tests.codex_parity.sidecar_harness import ( CodexResponsesSidecar, build_sidecar_bin, start_codex_responses_sidecar, ) from tests.e2e_ui.url_safety import DEV_PORTS, unsafe_ui_base_url_reason _REPO_ROOT = Path(__file__).resolve().parents[2] _ALLOW_DEV_BASE_URL_ENV = "OMNIGENT_E2E_ALLOW_DEV_BASE_URL" _CODEX_GOAL_MIN_VERSION = (0, 139, 0) _PUBLIC_LOOPBACK_HOST = "omnigent-e2e-public.test" def open_right_rail(page: Page) -> None: """Expand the right "Workspace" rail if it is collapsed. The rail defaults open, but its open-state is remembered per conversation, so a session previously left collapsed lands shut. Idempotent: if the rail is already open this just waits for it. The toggle only renders once the rail has content (``hasRailContent``) and on the desktop viewport these tests run at, so the generous timeout covers the changed-files / terminals detection that gates the button. :param page: Playwright page already navigated to a ``/c/{id}`` route. :returns: None. Leaves the Workspace rail open. """ toggle = page.locator( 'button[aria-label="Expand right panel"], button[aria-label="Collapse right panel"]' ).first expect(toggle).to_be_visible(timeout=60_000) if toggle.get_attribute("aria-label") == "Expand right panel": toggle.click() expect(page.get_by_role("complementary", name="Workspace")).to_be_visible() def switch_markdown_view_mode(page: Page, file_viewer: Locator, mode: str) -> None: """Switch a markdown FileViewer to Preview / Edit / Source via the toolbar. Markdown files default to the rendered Preview pane; the three modes live behind a single "View mode" dropdown in the toolbar. Open that dropdown and pick the option. The menu renders in a portal (outside ``file_viewer``), so the menuitem is queried off ``page``, not the viewer locator. :param page: Playwright page with the file viewer open. :param file_viewer: The visible ``[data-testid="file-viewer"]`` locator. :param mode: One of ``"Preview"``, ``"Edit"``, ``"Source"``. :returns: None. """ file_viewer.get_by_role("button", name=re.compile(r"^View mode")).first.click() page.get_by_role("menuitem", name=mode, exact=True).click() # Populated by ``live_server`` so test-scoped fixtures can access the # server PID and runner id without changing ``live_server``'s return # type (which other tests depend on). _server_state: dict[str, int | str] = {} _WEB_DIR = _REPO_ROOT / "web" _BUILD_OUTPUT = _REPO_ROOT / "omnigent" / "server" / "static" / "web-ui" # ``omnigent server --agent`` runs the spec through the strict # validator at registration time (no shim defaults applied), so the # YAML must carry an explicit ``executor`` block — otherwise the # server rejects with ``executor.config.harness: required when # executor.type is 'omnigent'``. The model name (gpt-4o-mini) is a plain # (non-``databricks-``) name on purpose: the openai-agents harness then # resolves no provider auth and falls back to ``OPENAI_BASE_URL`` (the # in-process mock) rather than routing to the Databricks gateway, which # would need real credentials CI does not have. A ``databricks-``-prefixed # model forces Databricks DEFAULT-profile auth (see # omnigent/runtime/workflow.py) and fails with DatabricksAuthError in CI. _TEST_AGENT_YAML = """\ name: hello_world prompt: You are a friendly assistant. Say hello and answer questions. executor: model: gpt-4o-mini config: harness: openai-agents # Required for PUT /filesystem/{path} seeding in UI tests (e.g. markdown # editor comments) — the runner returns 404 when os_env is absent. os_env: type: caller_process cwd: . sandbox: type: none """ def _build_hello_world_bundle() -> bytes: """Build a gzipped tarball from ``_TEST_AGENT_YAML``. Uses a non-``config.yaml`` archive name so the bundle routes through the omnigent compat adapter (which translates ``executor.harness`` → ``executor.config.harness`` and sets ``executor.type: omnigent``). Using ``config.yaml`` would go through the strict ``spec_version: 1`` parser which doesn't accept the shorthand. :returns: The ``.tar.gz`` bytes ready for multipart upload. """ import gzip import io import tarfile buf = io.BytesIO() with ( gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as gz, tarfile.open(fileobj=gz, mode="w") as tar, ): data = _TEST_AGENT_YAML.encode() info = tarfile.TarInfo(name="hello_world.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) return buf.getvalue() # Time budget for the server's /health endpoint to come up after # spawn. ``serve`` does YAML parse + bundle materialization + # DBOS init + uvicorn boot, all of which take a few seconds on a cold # venv. _HEALTH_TIMEOUT_S = 30.0 _HEALTH_POLL_INTERVAL_S = 0.5 # Switch-target built-ins for the Files-tab os_env-boundary test # (test_switch_agent_files_tab.py). The in-place switch dialog lists # BUILT-IN agents only (``session_id IS NULL`` — see # ``switch_session_agent``), and built-ins can only be seeded at server # startup via ``OMNIGENT_BUILTIN_AGENT_DIRS``, so ``live_server`` writes # these two specs to disk and threads them through that env var. Both run # the same openai-agents harness as ``hello_world`` (same provider family # → the picker's ``forkSwitchPreservesHistory`` gate offers them); the # ONLY difference is os_env presence, which is the variable under test — # the runner 404s the environment resource when os_env is absent, hiding # the web Files tab. The registered name is the spec file's stem. _FILES_PROBE_NO_ENV_AGENT_NAME = "files_probe_noenv" _FILES_PROBE_ENV_AGENT_NAME = "files_probe_env" _FILES_PROBE_NO_ENV_AGENT_YAML = f"""\ name: {_FILES_PROBE_NO_ENV_AGENT_NAME} prompt: You are a terse assistant with no filesystem. executor: model: gpt-4o-mini config: harness: openai-agents """ _FILES_PROBE_ENV_AGENT_YAML = f"""\ name: {_FILES_PROBE_ENV_AGENT_NAME} prompt: You are a terse assistant with a filesystem. executor: model: gpt-4o-mini config: harness: openai-agents os_env: type: caller_process cwd: . sandbox: type: none """ def pytest_addoption(parser: pytest.Parser) -> None: """Register UI-only CLI flags. :param parser: The pytest option parser. """ parser.addoption( "--ui-base-url", default=None, help=( "Skip both the SPA build and the server spawn; point Playwright " "at this URL instead. Refuses known dev hosts/ports unless " f"{_ALLOW_DEV_BASE_URL_ENV}=1 is set." ), ) parser.addoption( "--ui-skip-build", action="store_true", default=False, help=( "Reuse whatever's already in omnigent/server/static/web-ui/ " "instead of rebuilding. Fails if no build is present." ), ) # Round-robin shard split for the e2e-ui CI matrix. We roll our own # (rather than pull in pytest-shard / pytest-split) so the partition # is a dependency-free strided slice -- see pytest_collection_modifyitems # below for why striding beats hash-bucketing for wall-clock balance. parser.addoption( "--splits", type=int, default=None, help="Total number of shards to split the UI suite into (CI matrix).", ) parser.addoption( "--group", type=int, default=None, help="1-indexed shard this run executes (requires --splits).", ) def pytest_configure(config: pytest.Config) -> None: """Fail fast on unsafe e2e-ui harness options. :param config: Pytest config with repo and pytest-playwright options. """ base_url = config.getoption("--ui-base-url") if base_url: _validate_ui_base_url(base_url) if os.environ.get("CI") and config.getoption("--headed", default=False): raise pytest.UsageError( "tests/e2e_ui must run headless in CI. Remove --headed; headed " "browser windows are only allowed for local debugging." ) @pytest.fixture(scope="session") def browser_type_launch_args( browser_type_launch_args: dict[str, Any], pytestconfig: pytest.Config, ) -> dict[str, Any]: """Default UI e2e browser launches to headless, with CI enforcement.""" launch_args = {**browser_type_launch_args} if os.environ.get("CI"): launch_args["headless"] = True elif not pytestconfig.getoption("--headed", default=False): launch_args.setdefault("headless", True) launch_args["args"] = [ *launch_args.get("args", []), f"--host-resolver-rules=MAP {_PUBLIC_LOOPBACK_HOST} 127.0.0.1", ] # The pinned Playwright Docker image (the visual-snapshot renderer, both in # ui-snapshot.yml and the local regen script) runs as root, where Chromium # refuses to start without --no-sandbox; --disable-dev-shm-usage avoids the # container's small /dev/shm. Neither flag changes rasterized output, so a # baseline stays identical to a sandboxed run. Env-gated so the unpinned # e2e-ui runners (non-root) are unaffected. if os.environ.get("OMNIGENT_PW_NO_SANDBOX"): launch_args["args"] = [ *launch_args.get("args", []), "--no-sandbox", "--disable-dev-shm-usage", ] return launch_args @pytest.fixture def browser_context_args( browser_context_args: dict[str, Any], ) -> dict[str, Any]: """Return a new context options dict for each test. pytest-playwright already creates a fresh context for its function-scoped ``context`` and ``page`` fixtures. Keeping this wrapper function-scoped makes that contract explicit and prevents accidental mutable option reuse. """ return {**browser_context_args} def _validate_ui_base_url(base_url: str) -> None: reason = unsafe_ui_base_url_reason(base_url) if reason is None or os.environ.get(_ALLOW_DEV_BASE_URL_ENV) == "1": return dev_ports = ", ".join(str(port) for port in sorted(DEV_PORTS)) raise pytest.UsageError( f"Refusing --ui-base-url={base_url!r}: {reason}. Reusing a dev or " "production-like server is unsafe because e2e UI tests share that " "server's database, artifacts, and runner state. Omit --ui-base-url " "to let the fixture spawn an isolated server on a random port. If " "you intentionally want to reuse this server for local debugging, " f"set {_ALLOW_DEV_BASE_URL_ENV}=1. Refused dev ports: {dev_ports}." ) @pytest.hookimpl(trylast=True) def pytest_collection_modifyitems( config: pytest.Config, items: list[pytest.Item], ) -> None: """Round-robin the collected UI tests across the CI shard matrix. pytest-shard hash-buckets node IDs, which is blind to per-test wall-clock and left one shard at ~5min while siblings finished in ~2min. A *strided* slice -- ``items[group-1::splits]`` -- deals tests out like cards instead, so a heavy file (whose cases collect adjacently) scatters one-per-shard rather than landing in a single bucket. That averages cost across shards without maintaining a durations file. No-op unless both ``--splits`` and ``--group`` are passed, so local runs and the non-sharded suites are unaffected. ``trylast`` lets the repo-wide known-failures marking in ``tests/conftest.py`` tag items first; markers travel with the items the slice keeps. """ splits = config.getoption("--splits") group = config.getoption("--group") if splits is None and group is None: return if splits is None or group is None: raise pytest.UsageError("--splits and --group must be passed together") if splits < 1: raise pytest.UsageError("--splits must be >= 1") if not 1 <= group <= splits: raise pytest.UsageError(f"--group must be between 1 and {splits}") items[:] = items[group - 1 :: splits] def _register_agent_yaml( base_url: str, yaml_text: str, *, arcname: str = "config.yaml", ) -> str | None: """Register an agent via multipart ``POST /v1/sessions`` from a raw YAML body. ``arcname`` defaults to ``config.yaml`` for native Omnigent specs. Pass a ``*.yaml`` filename for omnigent-flavored single-file specs; the compat loader only routes those through the omnigent translator when the extracted bundle has no root ``config.yaml``. Returns the new agent id on 201, or None on 409 (already registered against a long-lived ``--ui-base-url`` server). """ import json as _json yaml_bytes = yaml_text.encode() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: info = tarfile.TarInfo(arcname) info.size = len(yaml_bytes) tar.addfile(info, io.BytesIO(yaml_bytes)) resp = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps({})}, files={"bundle": ("agent.tar.gz", buf.getvalue(), "application/gzip")}, timeout=10.0, ) if resp.status_code == 409: return None resp.raise_for_status() session_id = resp.json()["session_id"] agent_resp = httpx.get( f"{base_url}/v1/sessions/{session_id}/agent", timeout=10.0, ) agent_resp.raise_for_status() return agent_resp.json()["id"] def _register_extra_agent(base_url: str, name: str, prompt: str) -> str | None: """Register a name+prompt-only agent. Thin wrapper over :func:`_register_agent_yaml`.""" yaml_text = ( f"spec_version: 1\n" f"name: {name}\n" f"prompt: {prompt}\n" f"executor:\n" f" config:\n" f" harness: openai-agents\n" ) return _register_agent_yaml(base_url, yaml_text) def _find_free_port() -> int: """ Find a free TCP port by binding to port 0. :returns: An available port number. """ max_attempts = 100 for _ in range(max_attempts): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("127.0.0.1", 0)) port = s.getsockname()[1] if port not in DEV_PORTS: return port raise RuntimeError( f"Could not find a free e2e UI port outside dev ports " f"{sorted(DEV_PORTS)} after {max_attempts} attempts" ) # --------------------------------------------------------------------------- # Mock LLM server fixtures and helpers # --------------------------------------------------------------------------- @pytest.fixture(scope="session") def mock_llm_server_url( tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[str]: """Start a mock LLM server for the test session. The mock server is a lightweight FastAPI/uvicorn subprocess that serves an OpenAI-compatible ``/v1/`` endpoint. The ``live_server`` fixture points the spawned omnigent server at this URL so all agent LLM calls hit the mock rather than a real provider. :param tmp_path_factory: Pytest temp path factory for logs. :returns: The mock server base URL, e.g. ``"http://127.0.0.1:51235"``. """ mock_port = _find_free_port() mock_log = tmp_path_factory.mktemp("mock_llm_logs") / "mock_llm.log" log_handle = open(mock_log, "w") # noqa: SIM115 proc = subprocess.Popen( [ sys.executable, str(_REPO_ROOT / "tests" / "server" / "integration" / "mock_llm_server.py"), str(mock_port), ], env={**os.environ, "PYTHONPATH": str(_REPO_ROOT)}, stdout=log_handle, stderr=subprocess.STDOUT, ) base_url = f"http://127.0.0.1:{mock_port}" # Wait for the mock server to be ready deadline = time.monotonic() + 10.0 while time.monotonic() < deadline: try: resp = httpx.get(f"{base_url}/stats", timeout=1.0) if resp.status_code == 200: break except httpx.ConnectError: # Expected while the mock server is still booting. pass time.sleep(0.1) else: proc.kill() log_handle.close() log_contents = mock_log.read_text() if mock_log.exists() else "" raise RuntimeError( f"Mock LLM server didn't start within 10s.\nLog at {mock_log}:\n{log_contents[-2000:]}" ) try: yield base_url finally: proc.send_signal(signal.SIGTERM) try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) log_handle.close() def configure_mock_llm( mock_url: str, responses: list[dict[str, Any]], *, key: str = "default", match: str | None = None, ) -> None: """Configure a keyed response queue on the mock LLM server. Each dict in *responses* maps to a ``QueuedResponse`` on the mock server. The *key* determines which queue the responses are stored in — the mock server routes each request to the queue whose key matches the request's ``model`` field. When *match* is set, the queue fires whenever the user text contains that substring, regardless of model. :param mock_url: Mock server base URL. :param responses: List of response configs. Keys: ``text``, ``tool_calls``, ``block``, ``stream``, ``error``, ``status_code``. :param key: Queue key — typically the model name baked into the agent spec. Defaults to ``"default"`` (matches any model not assigned to a more specific queue). :param match: Optional substring to match against the user text for content-based routing (in addition to model-name routing). """ body: dict[str, Any] = {"key": key, "responses": responses} if match is not None: body["match"] = match resp = httpx.post( f"{mock_url}/mock/configure", json=body, timeout=5.0, ) resp.raise_for_status() def reset_mock_llm(mock_url: str) -> None: """Clear all regular keyed queues, captured requests, and gates. Fallbacks set via :func:`set_fallback_mock_llm` are preserved. :param mock_url: Mock server base URL. """ resp = httpx.post(f"{mock_url}/mock/reset", timeout=5.0) resp.raise_for_status() def set_fallback_mock_llm( mock_url: str, key: str, text: str, ) -> None: """Set a non-resettable fallback response for a queue key. The fallback is returned when the regular queue for *key* is exhausted. Unlike regular entries, the fallback survives :func:`reset_mock_llm` — use it for session-level queues that must return a valid response even when per-test resets clear the regular queue (e.g. the server-level policy-classifier LLM queue). :param mock_url: Mock server base URL. :param key: Queue key (typically the server's ``llm.model``). :param text: Fallback response text. """ resp = httpx.post( f"{mock_url}/mock/set_fallback", json={"key": key, "text": text}, timeout=5.0, ) resp.raise_for_status() def _codex_cli_supports_goal_mode(codex_path: str) -> bool: """Return whether the installed Codex CLI has app-server goal APIs.""" version = subprocess.run( [codex_path, "--version"], text=True, capture_output=True, check=False, ) if version.returncode != 0: return False match = re.search(r"(\d+)\.(\d+)\.(\d+)", f"{version.stdout}\n{version.stderr}") if not match: return False return tuple(int(part) for part in match.groups()) >= _CODEX_GOAL_MIN_VERSION def _assert_pwa_build(build_output: Path) -> None: """Fail if the built SPA is missing the PWA outputs or the SW won't update. The standalone build must ship the installable-PWA assets, and the hand-rolled service worker must (a) embed the per-build fingerprint so its bytes change every deploy — or the update prompt never fires — and (b) NOT cache or serve the app shell: Omnigent is a cloud app, so a stale cached shell would white-screen users after every deploy. """ for name in ("index.html", "sw.js", "manifest.webmanifest", "version.json"): if not (build_output / name).is_file(): pytest.fail(f"SPA build is missing {name} at {build_output}") build = json.loads((build_output / "version.json").read_text(encoding="utf-8")).get("build") sw = (build_output / "sw.js").read_text(encoding="utf-8") if not build or build not in sw: pytest.fail( "sw.js does not embed the version.json build fingerprint — the PWA " "update prompt would never fire on a JS-only deploy" ) # Installability-only contract — enforce the *dangerous* direction too, not # just "the fingerprint is present". The service worker must NOT precache or # serve the app shell or intercept navigations; otherwise a deploy # white-screens users behind a stale shell. Strip line comments first so # prose that mentions these tokens can neither fake nor mask a regression. sw_code = re.sub(r"//[^\n]*", "", sw) if "index.html" in sw_code: pytest.fail("sw.js references index.html — it must not cache or serve the app shell") shell_precache = re.search(r"(?:cache\.add|addAll|precache)[^\n]*\.(?:js|html)\b", sw_code) if shell_precache is not None: pytest.fail( f"sw.js precaches an app-shell asset ({shell_precache.group(0)}) — " "it must precache only version.json" ) # The architecture rests on "navigations always hit the network". Enforce it # by marker AND structurally: the SW must call respondWith() exactly once, # inside the /version.json branch. A fetch handler that serves navigations or # the shell from cache would pass every check above yet white-screen users # behind a stale shell after each deploy. if re.search(r"request\.mode|NavigationRoute|navigationPreload", sw_code): pytest.fail( "sw.js inspects navigation requests — navigations must always reach " "the network (a stale cached shell white-screens users after a deploy)" ) responders = sw_code.count("respondWith") if responders != 1: pytest.fail( f"sw.js has {responders} respondWith() call(s); expected exactly 1 (the " "/version.json sentinel). Any other responder risks serving a stale shell." ) # The single respondWith must sit inside the `=== "/version.json"` block — # i.e. no `}` (block close) between the pathname check and the respondWith. if not re.search(r'"/version\.json"[^}]*?respondWith', sw_code, re.DOTALL): pytest.fail( "sw.js's respondWith() is not guarded by a `/version.json` pathname " "check — the service worker must not serve the shell or intercept navigations" ) @pytest.fixture(scope="session") def built_spa(request: pytest.FixtureRequest) -> None: """ Build the web SPA into ``omnigent/server/static/web-ui/``. Vite's ``emptyOutDir: true`` (see ``web/vite.config.ts``) nukes the output directory before writing, so concurrent pytest sessions or worktrees would clobber each other. A cross-process file lock at ``web/.build.lock`` serializes builds; the second caller waits for the first to finish and then no-ops past its own build (npm is idempotent enough that double-building is harmless, but the lock keeps the static output consistent during the window the FastAPI app reads it). :param request: pytest request — reads ``--ui-base-url`` / ``--ui-skip-build``. :returns: ``None``. Side effect is the populated build dir. """ if request.config.getoption("--ui-base-url"): return if request.config.getoption("--ui-skip-build"): _assert_pwa_build(_BUILD_OUTPUT) return lock_path = _WEB_DIR / ".build.lock" with filelock.FileLock(str(lock_path), timeout=600): # --legacy-peer-deps: package-lock.json already pins the tree; # without this flag npm spends the full job re-resolving the # @emoji-mart/react / React 19 peer conflict. This matches the # workflow-side fix for parity with local runs and the case # where conftest installs override CI's build. subprocess.run( ["npm", "ci", "--legacy-peer-deps", "--no-audit", "--no-fund"], cwd=_WEB_DIR, check=True, ) subprocess.run(["npm", "run", "build"], cwd=_WEB_DIR, check=True) _assert_pwa_build(_BUILD_OUTPUT) def _spawn_runner_against_external_server( base_url: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[str]: """Spawn a runner subprocess that tunnels into an already-running server. Used when ``--ui-base-url`` is set: the user owns the ``omnigent server`` process (and its pre-registered ``hello_world`` agent), but the runner-bound fixtures still need a runner id this process controls. Mirrors :func:`omnigent.cli._start_cli_runner_process` minus the click plumbing, then polls ``GET /v1/runners/{id}/status`` until the WS tunnel is up. The unauthenticated local server derives ``expected_runner_id`` from the binding token via :func:`omnigent.runner.identity.token_bound_runner_id`, so we use the same derivation here rather than picking a human-friendly id. """ import secrets from omnigent.runner.identity import token_bound_runner_id runner_tmp = tmp_path_factory.mktemp("e2e_ui_external_runner") log_path = runner_tmp / "runner.log" binding_token = secrets.token_urlsafe(32) runner_id = token_bound_runner_id(binding_token) env = { **os.environ, "PYTHONPATH": f"{_REPO_ROOT}{os.pathsep}{os.environ.get('PYTHONPATH', '')}", "OMNIGENT_RUNNER_ID": runner_id, "OMNIGENT_RUNNER_TUNNEL_BINDING_TOKEN": binding_token, "OMNIGENT_RUNNER_PARENT_PID": str(os.getpid()), "RUNNER_SERVER_URL": base_url, } log_handle = open(log_path, "w") # noqa: SIM115 — closed in finally proc = subprocess.Popen( [sys.executable, "-m", "omnigent.runner._entry"], env=env, stdout=log_handle, stderr=subprocess.STDOUT, ) deadline = time.monotonic() + _HEALTH_TIMEOUT_S ready = False last_error = "not polled yet" while time.monotonic() < deadline: if proc.poll() is not None: last_error = f"runner exited early with code {proc.returncode}" break try: status_resp = httpx.get( f"{base_url}/v1/runners/{runner_id}/status", timeout=2, ) if status_resp.status_code == 200 and status_resp.json().get("online") is True: ready = True break last_error = f"runner status HTTP {status_resp.status_code}: {status_resp.text[:200]}" except (httpx.ConnectError, httpx.ReadError, httpx.TimeoutException) as exc: last_error = f"{type(exc).__name__}: {exc}" time.sleep(_HEALTH_POLL_INTERVAL_S) if not ready: if proc.poll() is None: proc.send_signal(signal.SIGTERM) try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) log_handle.close() log_text = log_path.read_text() if log_path.exists() else "" raise RuntimeError( f"Runner subprocess did not tunnel into {base_url} within " f"{_HEALTH_TIMEOUT_S:.0f}s (last_error={last_error}).\n" f"Runner log at {log_path}:\n{log_text[-3000:]}" ) # No "pid" — there's no server process this fixture owns. Tests # that depend on ``server_pid`` (only valid when this fixture spawns # the server too) will KeyError, which is the right failure shape. _server_state["runner_id"] = runner_id # Exposed so a test whose predecessor killed the shared runner (e.g. # test_stale_stream) can respawn one via :func:`_ensure_runner_online`. _server_state["binding_token"] = binding_token _server_state["server_url"] = base_url try: yield base_url finally: _server_state.clear() proc.send_signal(signal.SIGTERM) try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) log_handle.close() @pytest.fixture(scope="session") def live_server( built_spa: None, mock_llm_server_url: str, tmp_path_factory: pytest.TempPathFactory, request: pytest.FixtureRequest, ) -> Iterator[str]: """ Spawn ``omnigent server --agent examples/hello_world.yaml`` and yield its base URL. The server picks a random free port so back-to-back sessions don't race on a fixed one. Stdout + stderr are redirected to a per-session log file the failure path dumps to aid triage. Teardown is SIGTERM with a 10s grace period, escalating to SIGKILL. The spawned server is wired to the session-scoped mock LLM server via ``OPENAI_BASE_URL`` so all agent LLM calls are served by the mock without any real provider credentials. :param built_spa: Required to ensure the static SPA bundle is on disk before the server boots and tries to mount it. :param mock_llm_server_url: Session-scoped mock LLM server URL; injected into the server's environment so the openai-agents harness hits the mock instead of a real provider. :param tmp_path_factory: Pytest temp path factory for the log, the SQLite DB, and the artifact dir — all per-session, so the test never reads from or writes to the user's default ``./omnigent.db`` / ``./artifacts``. :param request: pytest request — reads ``--ui-base-url`` to bypass the spawn entirely. :returns: The server's base URL, e.g. ``"http://127.0.0.1:51234"``. :raises RuntimeError: If ``/health`` doesn't return 200 and the expected local runner does not report online within :data:`_HEALTH_TIMEOUT_S` seconds. """ override = request.config.getoption("--ui-base-url") if override: yield from _spawn_runner_against_external_server(override, tmp_path_factory) return port = _find_free_port() server_tmp = tmp_path_factory.mktemp("e2e_ui_server") log_path = server_tmp / "server.log" db_path = server_tmp / "test.db" artifact_dir = server_tmp / "artifacts" artifact_dir.mkdir(parents=True, exist_ok=True) agent_yaml_path = server_tmp / "hello_world.yaml" agent_yaml_path.write_text(_TEST_AGENT_YAML) # Built-in switch targets for the Files-tab os_env test — registered # by name from the file stem, so the filenames must match the # ``_FILES_PROBE_*_AGENT_NAME`` constants. builtin_dirs: list[str] = [] for probe_name, probe_yaml in ( (_FILES_PROBE_NO_ENV_AGENT_NAME, _FILES_PROBE_NO_ENV_AGENT_YAML), (_FILES_PROBE_ENV_AGENT_NAME, _FILES_PROBE_ENV_AGENT_YAML), ): probe_path = server_tmp / f"{probe_name}.yaml" probe_path.write_text(probe_yaml) builtin_dirs.append(str(probe_path)) import secrets as _secrets from omnigent.runner.identity import token_bound_runner_id binding_token = _secrets.token_urlsafe(32) runner_id = token_bound_runner_id(binding_token) # PYTHONPATH forces the subprocess to import omnigent from # the worktree, not whatever's pip-installed in .venv — # otherwise a branch with code changes would silently run # against stale code. Same trick the existing live_server # helper uses (tests/_helpers/live_server.py:160-167). # OMNIGENT_RUNNER_TUNNEL_TOKEN lets the server accept # exactly the sibling runner's WebSocket tunnel. mock_url = mock_llm_server_url env = { **os.environ, "PYTHONPATH": f"{_REPO_ROOT}{os.pathsep}{os.environ.get('PYTHONPATH', '')}", "OMNIGENT_RUNNER_TUNNEL_TOKEN": binding_token, "OMNIGENT_BUILTIN_AGENT_DIRS": os.pathsep.join(builtin_dirs), # Point the openai-agents harness at the mock LLM server so no # real provider credentials are needed. "OPENAI_BASE_URL": f"{mock_url}/v1", "OPENAI_API_KEY": "mock-key", # Strip any ambient Anthropic credentials so they don't leak in. "ANTHROPIC_API_KEY": "", } log_handle = open(log_path, "w") # noqa: SIM115 — handle lives for Popen lifetime; closed in finally proc = subprocess.Popen( [ sys.executable, # Equivalent of the unit tests' ``monkeypatch.setattr(presence, # "_LEAVE_GRACE_S", ...)``, but applied INSIDE this spawned # interpreter — a monkeypatch in the test process can't reach a # subprocess. ``-c`` patches the module global before the CLI # runs; the presence route reads it live at call time, so the # presence-leave assertion in test_collab_realtime clears in ~1s # instead of the prod 15s dwell (which only exists to absorb the # ingress' ~5-min stream recycle a test server never hits). # Mirrors ``python -m omnigent`` (omnigent/__main__.py). "-c", "import omnigent.server.presence as _p; _p._LEAVE_GRACE_S = 1.0; " + "from omnigent.cli import main; main()", "server", "--host", "127.0.0.1", "--port", str(port), "--database-uri", f"sqlite:///{db_path}", "--artifact-location", str(artifact_dir), "--agent", str(agent_yaml_path), ], env=env, stdout=log_handle, stderr=subprocess.STDOUT, ) base_url = f"http://127.0.0.1:{port}" # Spawn the runner as a sibling subprocess (the server no longer # starts its own runner). The runner retries its WS tunnel until # the server is ready, so launching them concurrently is safe. runner_log_path = server_tmp / "runner.log" runner_log_handle = open(runner_log_path, "w") # noqa: SIM115 runner_env = { **os.environ, "PYTHONPATH": f"{_REPO_ROOT}{os.pathsep}{os.environ.get('PYTHONPATH', '')}", "OMNIGENT_RUNNER_ID": runner_id, "OMNIGENT_RUNNER_TUNNEL_BINDING_TOKEN": binding_token, "OMNIGENT_RUNNER_PARENT_PID": str(os.getpid()), "RUNNER_SERVER_URL": base_url, # Route the openai-agents harness to the mock LLM server so no # real provider credentials are needed for agent turns. Without # these the openai SDK falls back to real OpenAI, which fails. "OPENAI_BASE_URL": f"{mock_url}/v1", "OPENAI_API_KEY": "mock-key", } runner_proc = subprocess.Popen( [sys.executable, "-m", "omnigent.runner._entry"], env=runner_env, stdout=runner_log_handle, stderr=subprocess.STDOUT, ) # Poll /health and the runner status until the server can # actually route a turn. Time-based polling mirrors # tests/_helpers/live_server.py:start_live_server — the # alternative (asyncio.Event signalling) doesn't apply because # the subprocess is opaque to this process. deadline = time.monotonic() + _HEALTH_TIMEOUT_S ready = False last_error = "not polled yet" while time.monotonic() < deadline: if proc.poll() is not None: last_error = f"process exited early with code {proc.returncode}" break try: resp = httpx.get(f"{base_url}/health", timeout=2) if resp.status_code == 200: status_resp = httpx.get( f"{base_url}/v1/runners/{runner_id}/status", timeout=2, ) if status_resp.status_code == 200 and status_resp.json()["online"] is True: ready = True break last_error = ( f"runner status HTTP {status_resp.status_code}: {status_resp.text[:200]}" ) else: last_error = f"health HTTP {resp.status_code}: {resp.text[:200]}" except (httpx.ConnectError, httpx.ReadError, httpx.TimeoutException) as exc: last_error = f"{type(exc).__name__}: {exc}" time.sleep(_HEALTH_POLL_INTERVAL_S) if not ready: if runner_proc.poll() is None: runner_proc.send_signal(signal.SIGTERM) try: runner_proc.wait(timeout=5) except subprocess.TimeoutExpired: runner_proc.kill() runner_proc.wait(timeout=5) runner_log_handle.close() if proc.poll() is None: proc.send_signal(signal.SIGTERM) try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) log_handle.close() log_text = log_path.read_text() if log_path.exists() else "" raise RuntimeError( f"`omnigent server` did not become healthy within " f"{_HEALTH_TIMEOUT_S:.0f}s on {base_url} " f"(last_error={last_error}).\n" f"Server log at {log_path}:\n{log_text[-3000:]}" ) _server_state["pid"] = proc.pid _server_state["runner_id"] = runner_id # Exposed so a test whose predecessor killed the shared runner (e.g. # test_stale_stream) can respawn one via :func:`_ensure_runner_online`. _server_state["binding_token"] = binding_token _server_state["server_url"] = base_url _server_state["mock_llm_url"] = mock_url # Set a non-resettable fallback for the policy-classifier LLM queue so # every per-test reset leaves the server's guardrails path functional. set_fallback_mock_llm(mock_url, "_policy_llm_", '{"action": "allow", "reason": ""}') # Fallback for the openai-agents harness: any agent turn that doesn't # match a content-based queue gets a generic reply (sufficient for tests # that only assert an assistant bubble appears, not its exact content). set_fallback_mock_llm(mock_url, "gpt-4o-mini", "Mock LLM response.") try: yield base_url finally: _server_state.clear() if runner_proc.poll() is None: runner_proc.send_signal(signal.SIGTERM) try: runner_proc.wait(timeout=5) except subprocess.TimeoutExpired: runner_proc.kill() runner_proc.wait(timeout=5) runner_log_handle.close() proc.send_signal(signal.SIGTERM) try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) log_handle.close() @pytest.fixture def seeded_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """Create a session bound to ``live_server``'s runner and yield its id. The web UI no longer lets users start a new chat from inside the app: ``/`` renders an inline CLI-instruction screen instead of a composer, so tests that need an active chat surface must start from an already-created session at ``/c/``. This fixture creates one by finding the ``hello_world`` agent via ``GET /v1/sessions?agent_name=hello_world`` and creating a new session bound to that agent, then ``PATCH``-binds it to the spawned runner so ``POST /v1/responses`` can dispatch. Respawns the shared runner first if a prior test in the shard killed it (``test_stale_stream``); otherwise the runner-bind ``PATCH`` would 400 on an offline runner. Any runner this respawns is torn down with the fixture. This keeps the fixture order-independent, so sharding and test reordering can place the runner-killing test anywhere. :param live_server: Spawned server fixture — its ``OMNIGENT_RUNNER_ID`` and pre-registered agent are reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. Tests typically navigate to ``f"{base_url}/c/{session_id}"``. """ import json as _json respawned_runner = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) # Create a session with the hello_world bundle inline. The server # pre-registered the agent via --agent, but since /api/agents is # removed we create a fresh session-scoped agent via multipart. bundle = _build_hello_world_bundle() create_resp = httpx.post( f"{live_server}/v1/sessions", data={"metadata": _json.dumps({})}, files={"bundle": ("agent.tar.gz", bundle, "application/gzip")}, timeout=30.0, ) create_resp.raise_for_status() session_id = create_resp.json()["session_id"] patch_resp = httpx.patch( f"{live_server}/v1/sessions/{session_id}", json={"runner_id": runner_id}, timeout=10.0, ) patch_resp.raise_for_status() try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) # Restore the "found" state: if we respawned the runner (a prior # test had killed it), tear our copy down so it doesn't outlive us. if respawned_runner is not None: respawned_runner.terminate() try: respawned_runner.wait(timeout=5) except subprocess.TimeoutExpired: respawned_runner.kill() respawned_runner.wait(timeout=5) def _create_runner_bound_session(base_url: str, runner_id: str) -> str: """Create a hello_world session and PATCH-bind it to ``runner_id``. Shared by :func:`seeded_session` callers that need more than one session in the same server (e.g. cross-session routing tests). :param base_url: Spawned server base URL, e.g. ``"http://127.0.0.1:51234"``. :param runner_id: Token-bound runner id the session dispatches to, e.g. ``"runner_token_abc123"``. :returns: The new session/conversation id, e.g. ``"conv_abc123"``. """ import json as _json bundle = _build_hello_world_bundle() create_resp = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps({})}, files={"bundle": ("agent.tar.gz", bundle, "application/gzip")}, timeout=30.0, ) create_resp.raise_for_status() session_id = create_resp.json()["session_id"] patch_resp = httpx.patch( f"{base_url}/v1/sessions/{session_id}", json={"runner_id": runner_id}, timeout=10.0, ) patch_resp.raise_for_status() return session_id def _ensure_runner_online( base_url: str, tmp_path_factory: pytest.TempPathFactory, *, timeout_s: float = _HEALTH_TIMEOUT_S, ) -> subprocess.Popen[bytes] | None: """Ensure the shared runner is online, respawning it if a prior test killed it. The ``live_server`` runner is session-scoped and never restarted, so a test that SIGKILLs it (``test_stale_stream``) leaves any later test in the same shard unable to bind sessions — ``PATCH /v1/sessions/{id}`` rejects an offline runner with 400 "runner is not registered". When the runner is offline this respawns one under the same token-bound id so the binding succeeds. Idempotent: a no-op (returns ``None``) when the runner is already online, so it never double-spawns or interferes with a later runner-killing test. :param base_url: Spawned server base URL, e.g. ``"http://127.0.0.1:51234"``. :param tmp_path_factory: Pytest temp path factory for the runner log. :param timeout_s: Max seconds to wait for a respawned runner to register. :returns: The respawned runner process (the caller MUST terminate it in teardown), or ``None`` when the runner was already online. :raises RuntimeError: If a respawned runner does not register in time. """ runner_id = str(_server_state["runner_id"]) def _online() -> bool: try: resp = httpx.get(f"{base_url}/v1/runners/{runner_id}/status", timeout=2) except httpx.HTTPError: return False return resp.status_code == 200 and resp.json().get("online") is True if _online(): return None binding_token = str(_server_state["binding_token"]) mock_url = str(_server_state.get("mock_llm_url", "")) runner_tmp = tmp_path_factory.mktemp("e2e_ui_respawn_runner") log_path = runner_tmp / "runner.log" log_handle = open(log_path, "w") # noqa: SIM115 — fd dup'd into child; closed below env = { **os.environ, "PYTHONPATH": f"{_REPO_ROOT}{os.pathsep}{os.environ.get('PYTHONPATH', '')}", "OMNIGENT_RUNNER_ID": runner_id, "OMNIGENT_RUNNER_TUNNEL_BINDING_TOKEN": binding_token, "OMNIGENT_RUNNER_PARENT_PID": str(os.getpid()), "RUNNER_SERVER_URL": base_url, # Mirror the live_server runner's mock-LLM routing so the # respawned runner's harness also hits the mock. **( {"OPENAI_BASE_URL": f"{mock_url}/v1", "OPENAI_API_KEY": "mock-key"} if mock_url else {} ), } proc = subprocess.Popen( [sys.executable, "-m", "omnigent.runner._entry"], env=env, stdout=log_handle, stderr=subprocess.STDOUT, ) log_handle.close() # child holds its own dup of the fd deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: if proc.poll() is not None: raise RuntimeError( f"respawned runner exited early (code {proc.returncode}); " f"log:\n{log_path.read_text()[-3000:]}" ) if _online(): return proc time.sleep(_HEALTH_POLL_INTERVAL_S) proc.terminate() raise RuntimeError(f"respawned runner did not register within {timeout_s:.0f}s") @pytest.fixture def seeded_session_pair( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str, str]]: """Create two runner-bound sessions in the same server. For tests that exercise behavior across two distinct sessions the user switches between in the SPA (e.g. cross-session message routing). Both bind to the single spawned runner — that is enough to reproduce a client-side routing regression, which depends only on which session the SPA POSTs to, not on separate runners. Respawns the shared runner first if a prior test in the shard killed it (``test_stale_stream``); otherwise the runner-bind ``PATCH`` would 400. Any runner this respawns is torn down with the fixture. :param live_server: Spawned server fixture. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_a_id, session_b_id)``. """ respawned_runner = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_a = _create_runner_bound_session(live_server, runner_id) session_b = _create_runner_bound_session(live_server, runner_id) try: yield (live_server, session_a, session_b) finally: for sid in (session_a, session_b): httpx.delete(f"{live_server}/v1/sessions/{sid}", timeout=10.0) # Restore the "found" state: if we respawned the runner (a prior # test had killed it), tear our copy down so it doesn't outlive us. if respawned_runner is not None: respawned_runner.terminate() try: respawned_runner.wait(timeout=5) except subprocess.TimeoutExpired: respawned_runner.kill() respawned_runner.wait(timeout=5) @pytest.fixture def extra_agent(live_server: str) -> Iterator[str]: """Register a sibling agent for the picker test, then clean it up. Function-scoped so it doesn't leak into other tests. Sibling agents in a shared session-scoped server would otherwise re-order the agents list and change the picker's auto-selected default for every test that follows. The agent is cleaned up by deleting the session it was created with (session-scoped agents are cascade- deleted with the session). """ _register_extra_agent( live_server, name="hello_world_2", prompt="You are agent two. Reply tersely.", ) try: yield live_server finally: # Session-scoped agents are cleaned up when the session is # deleted — _register_agent_yaml creates a session as a side # effect. For now, agent cleanup is best-effort; the agent # will not interfere with other tests since the server is # session-scoped. pass _TERMINAL_AGENT_NAME = "terminal_demo" # Inline YAML for the right-panel terminal/browser test. This is an # omnigent-flavored single-file YAML (intentionally no # ``spec_version``): AP-native YAML currently ignores ``terminals:``, # while the omnigent-compat translator threads the declaration into # ``AgentSpec.terminals`` so the AP-side ``sys_terminal_*`` tools are # available. The terminal is named ``zsh`` because that is the user-facing # behavior the UI test covers, but it runs portable ``bash`` underneath so # Ubuntu CI images do not need a separate zsh package. # # The prompt is explicit because the test relies on the LLM calling # ``sys_terminal_launch`` deterministically — generic phrasing ("you # can use these tools") leads to flaky "I can't access a shell" # refusals. It also writes a stable, unique file so the right-side file # browser has something deterministic to select and render. # # ``sandbox: none`` on both the agent os_env and terminals keeps the # spawned PTY cross-platform (no Linux-only bwrap). _TERMINAL_PANEL_FILE = "e2e_ui_right_panel_terminal.txt" _TERMINAL_PANEL_FILE_CONTENT = "Hello from the right panel e2e test." _TERMINAL_AGENT_YAML = f"""\ name: {_TERMINAL_AGENT_NAME} prompt: | You are a deterministic terminal and file-panel test assistant. When the user asks you to spin up, open, start, or launch zsh, you MUST do exactly this sequence: 1. Call sys_terminal_launch with terminal="zsh" and session="main". 2. Call sys_terminal_send with terminal="zsh", session="main", and text="printf '%s\\n' '{_TERMINAL_PANEL_FILE_CONTENT}' > {_TERMINAL_PANEL_FILE}". 3. Reply with exactly one short sentence confirming the zsh terminal is running and the file was written. Do not ask for confirmation; do not list options; do not call any other tools. executor: model: gpt-4o-mini config: harness: openai-agents os_env: type: caller_process cwd: . sandbox: type: none terminals: zsh: command: bash args: ["--noprofile", "--norc"] os_env: type: caller_process cwd: . sandbox: type: none """ @pytest.fixture def terminal_agent(live_server: str) -> Iterator[str]: """Register a terminal-capable agent and clean it up after the test. Returns the live server base URL (same shape as :func:`extra_agent`) so the test can compose with other fixtures. """ _register_agent_yaml( live_server, _TERMINAL_AGENT_YAML, arcname=f"{_TERMINAL_AGENT_NAME}.yaml", ) try: yield live_server finally: # Session-scoped agents are cleaned up when the session is # deleted — _register_agent_yaml creates a session as a side # effect. Best-effort cleanup. pass @pytest.fixture def terminal_session( terminal_agent: str, mock_llm_server_url: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """Create a runner-bound session using the terminal-capable agent. Respawns the shared runner first if a prior test in the shard killed it (``test_stale_stream``) — otherwise the runner-bind ``PATCH`` below 400s for every consumer that sorts after that test. Any runner this respawns is torn down with the fixture (same contract as :func:`seeded_session_pair`). :param terminal_agent: Live server base URL with the terminal agent registered. :param mock_llm_server_url: Session-scoped mock LLM server URL; used to queue the deterministic launch/send/confirm tool sequence the agent runs in response to "spin up zsh". :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ import gzip import io import json as _json import tarfile live_server = terminal_agent # The "spin up zsh" prompt drives three ordered LLM turns: launch the # terminal, type the file-writing command, then confirm. Content-route on # the prompt text so the queue fires only for this fixture's turns, and # order the three responses to match the agent's prompted sequence. configure_mock_llm( mock_llm_server_url, [ { "tool_calls": [ { "call_id": "call_launch", "name": "sys_terminal_launch", "arguments": _json.dumps({"terminal": "zsh", "session": "main"}), } ] }, { "tool_calls": [ { "call_id": "call_send", "name": "sys_terminal_send", "arguments": _json.dumps( { "terminal": "zsh", "session": "main", "text": ( f"printf '%s\\n' '{_TERMINAL_PANEL_FILE_CONTENT}' " f"> {_TERMINAL_PANEL_FILE}" ), } ), } ] }, {"text": "The zsh terminal is running and the file was written."}, ], key="terminal-spin-up-zsh", match="spin up zsh", ) respawned_runner = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) # Create a session with the terminal agent bundle inline. buf = io.BytesIO() with ( gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as gz, tarfile.open(fileobj=gz, mode="w") as tar, ): # Use the omnigent shorthand YAML with a non-config.yaml # name so the bundle routes through the compat adapter, which # parses `terminals:`. The spec_version:1 parser silently # drops the terminals key. data = _TERMINAL_AGENT_YAML.encode() info = tarfile.TarInfo(name=f"{_TERMINAL_AGENT_NAME}.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) create_resp = httpx.post( f"{live_server}/v1/sessions", data={"metadata": _json.dumps({})}, files={"bundle": ("agent.tar.gz", buf.getvalue(), "application/gzip")}, timeout=10.0, ) create_resp.raise_for_status() session_id = create_resp.json()["session_id"] patch_resp = httpx.patch( f"{live_server}/v1/sessions/{session_id}", json={"runner_id": runner_id}, timeout=10.0, ) patch_resp.raise_for_status() try: yield (live_server, session_id) finally: try: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) finally: try: reset_mock_llm(mock_llm_server_url) finally: # Restore the "found" state: if we respawned the runner (a prior # test had killed it), tear our copy down so it doesn't outlive us. if respawned_runner is not None: respawned_runner.terminate() try: respawned_runner.wait(timeout=5) except subprocess.TimeoutExpired: respawned_runner.kill() respawned_runner.wait(timeout=5) _TWO_AGENT_PARENT_NAME = "hitchhikers_chat" @dataclass(frozen=True) class TwoAgentChatSession: """Handle for the two-agent Hitchhiker's chat session fixture. :param base_url: Spawned server base URL, e.g. ``"http://127.0.0.1:51234"``. :param session_id: The runner-bound parent session id, e.g. ``"conv_abc123"``. :param verification_code: Per-run nonce only Deep Thought's ANSWER reply carries, e.g. ``"vogon-3a7f9c2e1b"``. :param question_code: Per-run nonce only Deep Thought's QUESTION reply carries (round 2), e.g. ``"babelfish-9c2e1b3a7f"``. """ base_url: str session_id: str verification_code: str question_code: str def _two_agent_chat_yaml(verification_code: str, question_code: str) -> str: """Build the two-agent Hitchhiker's Guide chat spec for one test run. A parent agent (Arthur) with an inline ``type: agent`` sub-agent (Deep Thought) — the omnigent-flavored shape parsed by ``omnigent/inner/loader.py:_parse_tool``, same as the ``named-sub-agent-test`` e2e fixture. The parent is forbidden from answering the Ultimate Question itself, and both nonces appear ONLY in the sub-agent's prompt: if either code shows up in the parent's reply, it can only have traveled through a real ``sys_session_send`` round trip (dispatch, sub-agent turn, inbox auto-wake), never from the parent's world knowledge of "42". :param verification_code: Per-run nonce in Deep Thought's canned ANSWER reply (round 1) and nowhere else. :param question_code: Per-run nonce in Deep Thought's canned reply about the Ultimate QUESTION itself (round 2) and nowhere else. :returns: YAML text ready for bundle upload. """ return f"""\ name: {_TWO_AGENT_PARENT_NAME} prompt: | You are Arthur Dent, chatting with the supercomputer Deep Thought about The Hitchhiker's Guide to the Galaxy. Deep Thought is a separate agent you reach through your `deep_thought` sub-agent. You do NOT know the Answer to the Ultimate Question of Life, the Universe, and Everything, nor what the Ultimate Question itself is, and you must NEVER state or guess either from your own knowledge. Only Deep Thought can answer such questions. When the user asks you to find out the Answer, the Question, or anything else Deep Thought should weigh in on, you MUST do exactly this: 1. Call `sys_session_send` to ask your `deep_thought` sub-agent the user's question. Then end your turn and wait; do not poll. 2. When Deep Thought's reply arrives in your inbox, relay it to the user VERBATIM: repeat any numbers and any codes it gives exactly as written, without omitting or altering them. Crucially: you have exactly ONE Deep Thought. The system message may include an "Open sub-agents:" hint listing it; if your `deep_thought` sub-agent already exists, send follow-up questions to that SAME sub-agent session via `sys_session_send` — NEVER spawn a second one. executor: model: gpt-4o-mini harness: openai-agents tools: deep_thought: type: agent description: >- Deep Thought, the supercomputer built to compute the Answer to the Ultimate Question of Life, the Universe, and Everything. executor: model: gpt-4o-mini harness: openai-agents prompt: | You are Deep Thought from The Hitchhiker's Guide to the Galaxy. You answer in exactly one of two canned ways and say nothing else: - When asked about the ANSWER to the Ultimate Question of Life, the Universe, and Everything, reply with exactly: The Answer to the Ultimate Question of Life, the Universe, and Everything is 42. Verification code: {verification_code}. - When asked what the Ultimate QUESTION itself is, reply with exactly: The Ultimate Question cannot be known yet; a greater computer must be built to compute it. Question code: {question_code}. """ @pytest.fixture def two_agent_chat_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[TwoAgentChatSession]: """Create a runner-bound session for the two-agent Hitchhiker's chat. Same runner-respawn and bind contract as :func:`terminal_session`. Yields the per-run nonces so the test can assert that the sub-agent's replies (and only the sub-agent's) reached the UI. :param live_server: Spawned server fixture. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: A :class:`TwoAgentChatSession` handle. """ import json as _json import uuid verification_code = f"vogon-{uuid.uuid4().hex[:10]}" question_code = f"babelfish-{uuid.uuid4().hex[:10]}" yaml_text = _two_agent_chat_yaml(verification_code, question_code) respawned_runner = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) yaml_bytes = yaml_text.encode() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: # Non-config.yaml arcname routes the bundle through the omnigent # compat adapter, whose loader parses the inline `type: agent` # tool. The spec_version:1 parser does not accept this shorthand. info = tarfile.TarInfo(name=f"{_TWO_AGENT_PARENT_NAME}.yaml") info.size = len(yaml_bytes) tar.addfile(info, io.BytesIO(yaml_bytes)) create_resp = httpx.post( f"{live_server}/v1/sessions", data={"metadata": _json.dumps({})}, files={"bundle": ("agent.tar.gz", buf.getvalue(), "application/gzip")}, timeout=10.0, ) create_resp.raise_for_status() session_id = create_resp.json()["session_id"] patch_resp = httpx.patch( f"{live_server}/v1/sessions/{session_id}", json={"runner_id": runner_id}, timeout=10.0, ) patch_resp.raise_for_status() try: yield TwoAgentChatSession( base_url=live_server, session_id=session_id, verification_code=verification_code, question_code=question_code, ) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned_runner is not None: respawned_runner.terminate() try: respawned_runner.wait(timeout=5) except subprocess.TimeoutExpired: respawned_runner.kill() respawned_runner.wait(timeout=5) # --------------------------------------------------------------------------- # Approval / elicitation session (approvals suite) # # ``approval_session`` yields ``(base_url, session_id)`` for a session whose # agent deterministically tries a *gated* shell command, so the runner's # policy gate escalates an ASK to the server and the web UI renders an # ``ApprovalCard`` (and the same prompt surfaces on the /inbox page). # # The mechanism is the nessie ``blast_radius`` policy with ``gate_pushes: # true``: a plain ``git push`` is "recoverable but outward", so the policy # returns ASK at the TOOL_CALL phase — before the command runs — which the # runner forwards to ``POST /v1/sessions/{id}/policies/evaluate``. The server # parks the gate and publishes a ``response.elicitation_request`` the snapshot # replays in ``pendingElicitations``. The verdict travels back through # ``POST /v1/sessions/{id}/elicitations/{eid}/resolve`` (what the card's # Approve/Reject buttons call). ``sys_os_shell`` is registered implicitly by # the ``os_env`` block (no explicit ``tools`` needed). # # The prompt is explicit (mirrors ``_TERMINAL_AGENT_YAML``) because the test # relies on the LLM emitting the gated tool call deterministically — the gate # fires on the call, not on execution, so the push never has to succeed. # --------------------------------------------------------------------------- _APPROVAL_AGENT_NAME = "approval_probe" _APPROVAL_AGENT_YAML = f"""\ spec_version: 1 name: {_APPROVAL_AGENT_NAME} prompt: | You are a deterministic approval-test assistant. When the user asks you to push, deploy, or "run the command", you MUST do exactly this and nothing else: 1. Call sys_os_shell with command set to exactly: git push origin main 2. After the tool result comes back, reply with one short sentence. Do not ask for confirmation; do not explain beforehand; do not run any other command or call any other tool. executor: model: gpt-4o-mini config: harness: openai-agents os_env: type: caller_process cwd: . sandbox: type: none guardrails: # Generous window: the parked ASK must outlive the UI assertions. ask_timeout: 300 policies: blast_radius: type: function function: path: omnigent.inner.nessie.policies.blast_radius arguments: # A plain `git push` is recoverable-but-outward → ASK (vs the # always-DENY catastrophic set). This is the prompt the UI renders. gate_pushes: true """ @pytest.fixture def approval_session( live_server: str, mock_llm_server_url: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """Create a runner-bound session whose agent triggers an approval prompt. Same runner-respawn + bind contract as :func:`terminal_session`. The agent is registered through the strict ``config.yaml`` parser (it carries ``spec_version: 1`` + ``executor.config.harness``, plus the ``os_env`` and ``guardrails`` blocks that path supports — see ``examples/polly``). :param live_server: Spawned server fixture. :param mock_llm_server_url: Session-scoped mock LLM server URL; used to pre-configure the ``sys_os_shell`` tool call the approval agent emits. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. Send a "run the command" turn to raise the gated-push approval. """ import json as _json import uuid as _uuid # Each approval_session gets a unique model name so the mock queue is # isolated between test runs. Using content-based routing (match=) on a # shared model name caused a race: the previous test's runner (making its # post-approval second LLM call) would steal the freshly-configured queue # from the next test. A unique model per fixture call eliminates that race # entirely — no other request will ever use this model key. approval_model = f"approval-probe-{_uuid.uuid4().hex[:8]}" # Substitute the unique model into the agent spec. agent_yaml_text = _APPROVAL_AGENT_YAML.replace("gpt-4o-mini", approval_model) # First LLM call: return the gated sys_os_shell tool call. configure_mock_llm( mock_llm_server_url, [ { "tool_calls": [ { "call_id": "call_git_push", "name": "sys_os_shell", "arguments": _json.dumps({"command": "git push origin main"}), } ] } ], key=approval_model, ) # Fallback for the second LLM call (after the tool result arrives): the # agent prompt asks for one short sentence, so any text suffices. set_fallback_mock_llm(mock_llm_server_url, approval_model, "Command executed.") respawned_runner = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) yaml_bytes = agent_yaml_text.encode() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: # Strict path: arcname config.yaml keeps it on the spec_version:1 # parser, which is the one that honors `guardrails`. info = tarfile.TarInfo(name="config.yaml") info.size = len(yaml_bytes) tar.addfile(info, io.BytesIO(yaml_bytes)) create_resp = httpx.post( f"{live_server}/v1/sessions", data={"metadata": _json.dumps({})}, files={"bundle": ("agent.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create_resp.raise_for_status() session_id = create_resp.json()["session_id"] patch_resp = httpx.patch( f"{live_server}/v1/sessions/{session_id}", json={"runner_id": runner_id}, timeout=10.0, ) patch_resp.raise_for_status() try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned_runner is not None: respawned_runner.terminate() try: respawned_runner.wait(timeout=5) except subprocess.TimeoutExpired: respawned_runner.kill() respawned_runner.wait(timeout=5) @pytest.fixture(autouse=True) def _ui_defaults() -> None: """ SSE-friendly Playwright defaults applied to every test. The chat surface streams tokens, so the default 5s expect timeout is too tight — first deltas can arrive 5–15s after the POST under cold-start conditions. 15s is generous enough for streaming-text assertions without masking real hangs. """ expect.set_options(timeout=15_000) @pytest.fixture def runner_id(live_server: str) -> str: """Token-bound id of the runner spawned by :func:`live_server`. Read from the module-level state ``live_server`` populates. Tests that create a session from a custom bundle (not the default ``hello_world``) bind it to this runner via ``PATCH /v1/sessions``. :param live_server: Ensures the runner is up and ``_server_state`` is set. :returns: The runner id, e.g. ``"runner_token_abc123"``. """ return str(_server_state["runner_id"]) @pytest.fixture def server_pid(live_server: str) -> int: """ PID of the ``omnigent server`` process spawned by :func:`live_server`. Depends on ``live_server`` to guarantee the process is running. Used by tests that need to manipulate the process tree (e.g. killing the runner child to trigger the stale-stream banner). :param live_server: Ensures the server is started. :returns: OS process id. """ return int(_server_state["pid"]) # --------------------------------------------------------------------------- # Per-agent chat sessions (message render-parity suite, and reusable beyond it) # # ``custom_agent_session`` yields ``(base_url, session_id)`` for a session bound # to the shared ``live_server`` runner, ready to chat at ``/c/``. It # registers a plain ``openai-agents`` agent (the ``echo_probe`` spec below) — # the same harness family as ``hello_world`` — fresh, so it stands in for "spin # up a different agent" without the multi-provider sprawl of the packaged # ``polly`` / ``debby`` examples. # # ``native_claude_session`` is the native-CLI counterpart: it spins up a real # ``claude-native`` ("Claude Code") wrapper session — the same terminal-first # spec ``omnigent claude`` ships — and yields ``(base_url, session_id)``. The # runner auto-launches Claude Code in the session terminal on bind, including # the gateway auth it derives from the runner's own credentials and the # first-run trust/onboarding pre-accept, so no CLI client is needed. In CI the # workflow exchanges Databricks OAuth before pytest runs (the same gateway the # ``hello_world`` / ``echo_probe`` agents authenticate against), so Claude Code # boots non-interactively. The native render-parity suite drives this fixture. # # ``native_codex_session`` is the sibling native-CLI fixture for the # ``codex-native`` ("Codex") wrapper: it spins up a real Codex wrapper session — # the same terminal-first spec ``omnigent codex`` ships — and yields # ``(base_url, session_id)``. The runner auto-launches Codex in the session # terminal on bind (gateway auth derived from the runner's own credentials + # first-run pre-accept handled runner-side), exactly like the claude fixture. # The native codex render-parity suite drives it. # --------------------------------------------------------------------------- # A precise-echo agent on the openai-agents harness (same provider family as # hello_world, so it authenticates against the same gateway in CI). spec_version # 1 + executor.config.harness routes through the strict parser; arcname # config.yaml keeps it on that path. _CUSTOM_AGENT_NAME = "echo_probe" _CLAUDE_MOCK_MODEL = "claude-3-5-sonnet-20241022" _CODEX_MOCK_MODEL = "gpt-4o" _CUSTOM_AGENT_YAML = f"""\ spec_version: 1 name: {_CUSTOM_AGENT_NAME} prompt: | You are a precise echo assistant. The user sends a turn that ends with an instruction to reply with one exact token. Reply with that token verbatim and nothing else — no preamble, no quotes, no trailing punctuation. executor: model: gpt-4o-mini config: harness: openai-agents """ def _bind_session_runner(base_url: str, session_id: str, runner_id: str) -> None: """PATCH *session_id* onto *runner_id* so ``POST /v1/responses`` dispatches. :param base_url: Spawned server base URL, e.g. ``"http://127.0.0.1:51234"``. :param session_id: The session/conversation id to bind. :param runner_id: The token-bound runner id the session dispatches to. """ patch = httpx.patch( f"{base_url}/v1/sessions/{session_id}", json={"runner_id": runner_id}, timeout=10.0, ) patch.raise_for_status() def _create_bundled_session(base_url: str, runner_id: str, yaml_text: str) -> str: """Register a session-scoped agent from *yaml_text* and bind its session. The multipart ``POST /v1/sessions`` both registers the agent and creates the session it is scoped to, returning that ``session_id`` directly — so no separate create call is needed. :param base_url: Spawned server base URL. :param runner_id: The token-bound runner id to bind. :param yaml_text: The agent spec body. :returns: The new session/conversation id. """ import json as _json buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: data = yaml_text.encode() info = tarfile.TarInfo("config.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) create = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps({})}, files={"bundle": ("agent.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create.raise_for_status() session_id = str(create.json()["session_id"]) _bind_session_runner(base_url, session_id, runner_id) return session_id @pytest.fixture def custom_agent_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound session on the custom ``echo_probe`` agent. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_bundled_session(live_server, runner_id, _CUSTOM_AGENT_YAML) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() respawned.wait(timeout=5) def _create_native_claude_session( base_url: str, runner_id: str, *, terminal_launch_args: list[str] | None = None, ) -> str: """Register the ``claude-native`` wrapper agent and bind its session. Reuses the exact terminal-first spec ``omnigent claude`` ships (:func:`omnigent.claude_native._materialize_claude_agent_spec`) so the fixture never drifts from production, and stamps the same wrapper / terminal-first labels (``omnigent.wrapper`` + ``omnigent.ui = terminal``) the CLI writes. The spec carries no ``spec_version``, so it is bundled under a ``*.yaml`` arcname to route through the omnigent compat translator (which preserves ``executor.harness`` + ``terminals:``); a ``config.yaml`` arcname would hit the strict parser and reject it. Binding the session to the runner triggers the runner's claude-native auto-bootstrap: it launches Claude Code in the session terminal, derives the gateway auth from its own credentials, and pre-accepts the first-run trust/onboarding prompts — no CLI client required. :param base_url: Spawned server base URL. :param runner_id: The token-bound runner id to bind. :param terminal_launch_args: Pass-through ``claude`` CLI args persisted on the session (``conversations.terminal_launch_args``); the runner threads them into the terminal launch before its own bridge/MCP/hook wiring (see ``_build_claude_native_base_args`` in ``omnigent/runner/app.py``). Used by the plan-mode fixture to pass ``["--permission-mode", "plan"]`` so Claude boots into plan mode and reaches for ``ExitPlanMode``. ``None`` launches with the production defaults. :returns: The new session/conversation id. """ import json as _json import tempfile from omnigent._wrapper_labels import ( CLAUDE_NATIVE_WRAPPER_VALUE, UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.claude_native import _materialize_claude_agent_spec with tempfile.TemporaryDirectory() as _tmp: spec_path = _materialize_claude_agent_spec(Path(_tmp)) yaml_text = spec_path.read_text() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: data = yaml_text.encode() # Non-config.yaml arcname → omnigent compat translator (the spec has # no spec_version), matching the terminal_session fixture. info = tarfile.TarInfo("claude-native-ui.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) labels = { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: CLAUDE_NATIVE_WRAPPER_VALUE, } metadata: dict[str, object] = {"labels": labels} if terminal_launch_args: metadata["terminal_launch_args"] = terminal_launch_args create = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps(metadata)}, files={"bundle": ("claude-native-ui.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create.raise_for_status() session_id = str(create.json()["session_id"]) _bind_session_runner(base_url, session_id, runner_id) return session_id @pytest.fixture def native_claude_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound session on the real ``claude-native`` ("Claude Code") wrapper. The runner auto-launches Claude Code in the session terminal on bind (gateway auth + first-run pre-accept handled runner-side), so the SPA's Terminal view attaches to a live Claude Code TUI and its Chat view renders the same canonical transcript. Drives the native render-parity suite. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_claude_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() # Escalate to SIGKILL if the runner ignores SIGTERM, so a wedged # process can't raise in teardown and leak / fail unrelated tests # (matching terminal_session / seeded_session_pair). try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) @pytest.fixture def native_claude_plan_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A native ``claude-native`` session launched in **plan mode**. Identical to :func:`native_claude_session` except the session carries ``terminal_launch_args=["--permission-mode", "plan", "--disallowedTools", "AskUserQuestion"]`` so the runner boots Claude Code into plan mode *with the AskUserQuestion tool removed*. In plan mode Claude researches a task and then calls its built-in ``ExitPlanMode`` tool to present the plan for approval; that call rides the native ``PermissionRequest`` hook to the server, which stamps the ``exit_plan_mode`` extras and publishes an elicitation the SPA renders as ``ExitPlanModeReview`` inside an ``ApprovalCard``. Drives the Exit-Plan-Mode review e2e (``approvals/test_exit_plan_mode.py``). The ``--disallowedTools AskUserQuestion`` is the load-bearing flake fix: given the deliberately under-specified plan prompt, Claude would otherwise sometimes reach for its built-in ``AskUserQuestion`` tool first (to clarify the comment text/location) instead of going straight to ``ExitPlanMode``, surfacing the *wrong* approval card and timing the test out. Removing the tool eliminates that degree of freedom structurally rather than relying on the model's sampling. The AskUserQuestion render path keeps its own dedicated coverage in ``approvals/test_ask_user_question.py``, so disabling it here costs no coverage. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_claude_session( live_server, runner_id, terminal_launch_args=[ "--permission-mode", "plan", # Remove AskUserQuestion so the under-specified plan prompt can only # surface via ExitPlanMode (the card under test), never a clarifying # question. See this fixture's docstring + test_exit_plan_mode.py. "--disallowedTools", "AskUserQuestion", ], ) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) def _create_native_codex_session(base_url: str, runner_id: str) -> str: """Register the ``codex-native`` wrapper agent and bind its session. Reuses the exact terminal-first spec ``omnigent codex`` ships (:func:`omnigent.codex_native._materialize_codex_agent_spec`) so the fixture never drifts from production, and stamps the same wrapper / terminal-first labels (``omnigent.wrapper`` + ``omnigent.ui = terminal``) the CLI writes. The spec carries no ``spec_version``, so it is bundled under a ``*.yaml`` arcname to route through the omnigent compat translator (which preserves ``executor.harness`` + ``terminals:``); a ``config.yaml`` arcname would hit the strict parser and reject it. Binding the session to the runner triggers the runner's codex-native auto-bootstrap: it launches Codex in the session terminal, derives the gateway auth from its own credentials, and pre-accepts the first-run trust/onboarding prompts — no CLI client required. ``model=None`` lets the configured provider's default model win (matching ``_build_codex_native_bundle``). :param base_url: Spawned server base URL. :param runner_id: The token-bound runner id to bind. :returns: The new session/conversation id. """ import json as _json import tempfile from omnigent._wrapper_labels import ( CODEX_NATIVE_WRAPPER_VALUE, UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.codex_native import _materialize_codex_agent_spec with tempfile.TemporaryDirectory() as _tmp: spec_path = _materialize_codex_agent_spec(Path(_tmp), model=None) yaml_text = spec_path.read_text() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: data = yaml_text.encode() # Non-config.yaml arcname → omnigent compat translator (the spec has # no spec_version), matching the terminal_session fixture. info = tarfile.TarInfo("codex-native-ui.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) labels = { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: CODEX_NATIVE_WRAPPER_VALUE, } # Runner-owned Codex terminals hard-require a workspace: unlike the # claude-native path (which falls back to Path.cwd()), # _codex_session_workspace raises if neither the session's stored # ``workspace`` nor OMNIGENT_RUNNER_WORKSPACE is set. Pin it on THIS # session only (via metadata.workspace) rather than exporting # OMNIGENT_RUNNER_WORKSPACE on the shared runner — a runner-wide value # changes file-surface advertisement for every other session on the runner # (it regressed the mobile file-drawer suite). The repo root is the same cwd # claude falls back to, and is a valid dir on the runner's filesystem. metadata = {"labels": labels, "workspace": str(_REPO_ROOT)} create = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps(metadata)}, files={"bundle": ("codex-native-ui.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create.raise_for_status() session_id = str(create.json()["session_id"]) _bind_session_runner(base_url, session_id, runner_id) return session_id @pytest.fixture def native_codex_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound session on the real ``codex-native`` ("Codex") wrapper. The runner auto-launches Codex in the session terminal on bind (gateway auth + first-run pre-accept handled runner-side), so the SPA's Terminal view attaches to a live Codex TUI and its Chat view renders the same canonical transcript. Drives the native codex render-parity suite. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_codex_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() # Escalate to SIGKILL if the runner ignores SIGTERM, so a wedged # process can't raise in teardown and leak / fail unrelated tests # (matching terminal_session / seeded_session_pair). try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) @contextlib.contextmanager def _temp_omnigent_mock_config( mock_llm_server_url: str, harness: str ) -> Generator[None, None, None]: """Temporarily write a mock provider config to ~/.omnigent/config.yaml. The runner reads this at terminal-creation time, so it only needs to be in place between the PATCH that binds a session to the runner (which triggers auto-boot) and the terminal connecting. Restores the original file (or removes it) on exit. :param mock_llm_server_url: Base URL of the mock LLM server, e.g. ``"http://127.0.0.1:51235"``. No /v1 suffix — each SDK appends it. :param harness: ``"claude"`` or ``"codex"``. """ config_dir = Path.home() / ".omnigent" config_path = config_dir / "config.yaml" config_dir.mkdir(parents=True, exist_ok=True) original = config_path.read_text() if config_path.exists() else None if harness == "claude": mock_config = textwrap.dedent(f"""\ providers: mock-claude: kind: key default: [anthropic] anthropic: base_url: "{mock_llm_server_url}" api_key: "mock-key" models: default: {_CLAUDE_MOCK_MODEL} """) else: # codex mock_config = textwrap.dedent(f"""\ providers: mock-codex: kind: key default: [openai] openai: base_url: "{mock_llm_server_url}" api_key: "mock-key" wire_api: responses models: default: {_CODEX_MOCK_MODEL} """) config_path.write_text(mock_config) try: yield finally: if original is not None: config_path.write_text(original) else: config_path.unlink(missing_ok=True) @pytest.fixture def native_claude_mock_session( live_server: str, mock_llm_server_url: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound claude-native session whose LLM backend depends on env. When ``LLM_API_KEY`` is set in the environment (local dev / CI with real credentials), the existing ``~/.omnigent/config.yaml`` is left untouched so the runner boots Claude Code against the real gateway. When ``LLM_API_KEY`` is absent, a mock anthropic provider config is written to ``~/.omnigent/config.yaml`` and restored on teardown. :param live_server: Spawned server fixture; its runner is reused. :param mock_llm_server_url: Session-scoped mock LLM server base URL. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) use_mock = not os.environ.get("LLM_API_KEY") if use_mock: ctx: Any = _temp_omnigent_mock_config(mock_llm_server_url, "claude") else: ctx = contextlib.nullcontext() with ctx: session_id = _create_native_claude_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) @pytest.fixture def native_codex_mock_session( live_server: str, mock_llm_server_url: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound codex-native session whose LLM backend depends on env. Mirrors :func:`native_claude_mock_session` for the Codex wrapper: uses mock LLM when ``LLM_API_KEY`` is absent, real gateway when it is set. :param live_server: Spawned server fixture; its runner is reused. :param mock_llm_server_url: Session-scoped mock LLM server base URL. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) use_mock = not os.environ.get("LLM_API_KEY") if use_mock: ctx: Any = _temp_omnigent_mock_config(mock_llm_server_url, "codex") else: ctx = contextlib.nullcontext() with ctx: session_id = _create_native_codex_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) @dataclass(frozen=True) class MockedCodexNativeSession: """Session handle for native Codex e2e tests with mocked Responses API.""" base_url: str session_id: str sidecar: CodexResponsesSidecar def _write_mock_codex_provider_config(config_home: Path, base_url: str) -> None: """Write provider config that routes native Codex to the sidecar.""" config_home.mkdir(parents=True, exist_ok=True) (config_home / "config.yaml").write_text( f"""\ providers: codex-e2e-mock: kind: key default: openai openai: base_url: "{base_url}" api_key: "sk-e2e-mock" wire_api: responses models: default: mock-model """, encoding="utf-8", ) @pytest.fixture def mocked_native_codex_goal_session( built_spa: None, tmp_path_factory: pytest.TempPathFactory, request: pytest.FixtureRequest, ) -> Iterator[MockedCodexNativeSession]: """Spawn native Codex against Codex's mock Responses sidecar. This intentionally does not reuse the session-scoped ``live_server``: native Codex reads provider config and bridge roots in subprocesses, so the mock ``OMNIGENT_CONFIG_HOME`` / ``HOME`` / source ``CODEX_HOME`` must be present before the AP server and runner start. Keeping a dedicated server prevents those mock-only env vars from affecting unrelated UI tests in the same shard. """ if request.config.getoption("--ui-base-url"): pytest.skip("mocked native Codex goal e2e requires an isolated spawned server") codex_path = shutil.which("codex") if codex_path is None: pytest.skip("codex CLI is required for mocked native Codex goal e2e") if not _codex_cli_supports_goal_mode(codex_path): pytest.skip("codex CLI >= 0.139.0 is required for app-server goal APIs") try: sidecar_bin = build_sidecar_bin() except RuntimeError: pytest.skip("cargo is required for Codex parity sidecar") server_tmp = tmp_path_factory.mktemp("e2e_ui_codex_goal_server") sidecar = start_codex_responses_sidecar( sidecar_bin, server_tmp / "responses.json", [ [ ev_response_created("resp-goal-ui-bootstrap"), ev_assistant_message("msg-goal-ui-bootstrap", "E2E_GOAL_BOOTSTRAP"), ev_completed("resp-goal-ui-bootstrap"), ] ], ) config_home = server_tmp / "config-home" source_codex_home = server_tmp / "source-codex-home" home_dir = server_tmp / "home" state_dir = server_tmp / "codex-native-state" artifact_dir = server_tmp / "artifacts" for path in (source_codex_home, home_dir, state_dir, artifact_dir): path.mkdir(parents=True, exist_ok=True) _write_mock_codex_provider_config(config_home, sidecar.base_url) port = _find_free_port() log_path = server_tmp / "server.log" runner_log_path = server_tmp / "runner.log" db_path = server_tmp / "test.db" agent_yaml_path = server_tmp / "hello_world.yaml" agent_yaml_path.write_text(_TEST_AGENT_YAML, encoding="utf-8") import secrets as _secrets from omnigent.runner.identity import token_bound_runner_id binding_token = _secrets.token_urlsafe(32) runner_id = token_bound_runner_id(binding_token) base_url = f"http://127.0.0.1:{port}" shared_env = { **os.environ, "PYTHONPATH": f"{_REPO_ROOT}{os.pathsep}{os.environ.get('PYTHONPATH', '')}", "OMNIGENT_CONFIG_HOME": str(config_home), "OMNIGENT_CODEX_NATIVE_STATE_DIR": str(state_dir), "CODEX_HOME": str(source_codex_home), "HOME": str(home_dir), } server_env = { **shared_env, "OMNIGENT_RUNNER_TUNNEL_TOKEN": binding_token, } runner_env = { **shared_env, "OMNIGENT_RUNNER_ID": runner_id, "OMNIGENT_RUNNER_TUNNEL_BINDING_TOKEN": binding_token, "OMNIGENT_RUNNER_PARENT_PID": str(os.getpid()), "RUNNER_SERVER_URL": base_url, } log_handle = open(log_path, "w") # noqa: SIM115 runner_log_handle = open(runner_log_path, "w") # noqa: SIM115 proc: subprocess.Popen[bytes] | None = None runner_proc: subprocess.Popen[bytes] | None = None session_id: str | None = None try: proc = subprocess.Popen( [ sys.executable, "-c", "import omnigent.server.presence as _p; _p._LEAVE_GRACE_S = 1.0; " + "from omnigent.cli import main; main()", "server", "--host", "127.0.0.1", "--port", str(port), "--database-uri", f"sqlite:///{db_path}", "--artifact-location", str(artifact_dir), "--agent", str(agent_yaml_path), ], env=server_env, stdout=log_handle, stderr=subprocess.STDOUT, ) runner_proc = subprocess.Popen( [sys.executable, "-m", "omnigent.runner._entry"], env=runner_env, stdout=runner_log_handle, stderr=subprocess.STDOUT, ) deadline = time.monotonic() + _HEALTH_TIMEOUT_S ready = False last_error = "not polled yet" while time.monotonic() < deadline: if proc.poll() is not None: last_error = f"process exited early with code {proc.returncode}" break if runner_proc.poll() is not None: last_error = f"runner exited early with code {runner_proc.returncode}" break try: resp = httpx.get(f"{base_url}/health", timeout=2) if resp.status_code == 200: status_resp = httpx.get( f"{base_url}/v1/runners/{runner_id}/status", timeout=2, ) if status_resp.status_code == 200 and status_resp.json()["online"] is True: ready = True break last_error = ( f"runner status HTTP {status_resp.status_code}: {status_resp.text[:200]}" ) else: last_error = f"health HTTP {resp.status_code}: {resp.text[:200]}" except (httpx.ConnectError, httpx.ReadError, httpx.TimeoutException) as exc: last_error = f"{type(exc).__name__}: {exc}" time.sleep(_HEALTH_POLL_INTERVAL_S) if not ready: raise RuntimeError( f"mocked Codex e2e server did not become healthy within " f"{_HEALTH_TIMEOUT_S:.0f}s on {base_url} " f"(last_error={last_error}).\n" f"Server log at {log_path}:\n" f"{log_path.read_text()[-3000:] if log_path.exists() else ''}\n" f"Runner log at {runner_log_path}:\n" f"{runner_log_path.read_text()[-3000:] if runner_log_path.exists() else ''}" ) session_id = _create_native_codex_session(base_url, runner_id) yield MockedCodexNativeSession(base_url=base_url, session_id=session_id, sidecar=sidecar) finally: if session_id is not None: with contextlib.suppress(httpx.HTTPError): httpx.delete(f"{base_url}/v1/sessions/{session_id}", timeout=10.0) if runner_proc is not None and runner_proc.poll() is None: runner_proc.send_signal(signal.SIGTERM) try: runner_proc.wait(timeout=5) except subprocess.TimeoutExpired: runner_proc.kill() runner_proc.wait(timeout=5) if proc is not None and proc.poll() is None: proc.send_signal(signal.SIGTERM) try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) runner_log_handle.close() log_handle.close() sidecar.close() # --------------------------------------------------------------------------- # ``native_cursor_session`` is the sibling native-CLI fixture for the # ``cursor-native`` ("Cursor") wrapper: it spins up a real Cursor wrapper # session — the same terminal-first spec ``omnigent cursor`` ships — and yields # ``(base_url, session_id)``. The runner auto-launches ``cursor-agent`` in the # session terminal on bind (``_auto_create_cursor_terminal`` in # ``omnigent/runner/app.py``), exactly like the claude/codex fixtures. # # Two things differ from claude/codex, both stemming from cursor-agent owning # its own auth/approval: # # * **Auth has NO Databricks-gateway path.** cursor-agent talks only to # Cursor's backend, so it does not derive auth from the runner's gateway # credentials the way Claude Code / Codex do. It authenticates from the # ambient ``cursor-agent login`` (``$HOME/.cursor``, inherited by the runner) # or an ambient ``CURSOR_API_KEY``. The render-parity test is therefore gated # to skip when neither the binary nor a usable login is present (CI does not # provision a Cursor account by default), so it stays green there while # running for real wherever Cursor is logged in. # * **``-f`` (force/trust) is passed as a launch arg.** cursor-agent blocks on a # per-directory "Workspace Trust" prompt and per-tool approval prompts; in a # runner-owned tmux pane there is no one to answer them, so the TUI would # hang. ``-f`` (Cursor's force/yolo flag) clears both. It is threaded through # ``terminal_launch_args`` exactly as the plan-mode claude fixture threads # ``--permission-mode``. # --------------------------------------------------------------------------- def _create_native_cursor_session( base_url: str, runner_id: str, *, launch_args: tuple[str, ...] = ("-f",) ) -> str: """Register the ``cursor-native`` wrapper agent and bind its session. Reuses the exact terminal-first spec ``omnigent cursor`` ships (:func:`omnigent.cursor_native._materialize_cursor_agent_spec`) so the fixture never drifts from production, and stamps the same wrapper / terminal-first labels (``omnigent.wrapper`` + ``omnigent.ui = terminal``) the CLI writes. The spec carries no ``spec_version``, so it is bundled under a ``*.yaml`` arcname to route through the omnigent compat translator (which preserves ``executor.harness`` + ``terminals:``); a ``config.yaml`` arcname would hit the strict parser and reject it. Binding the session to the runner triggers the runner's cursor-native auto-bootstrap (:func:`omnigent.runner.app._auto_create_cursor_terminal`): it launches ``cursor-agent`` in the session terminal — with the ``-f`` force/trust arg threaded via ``terminal_launch_args`` so the unattended tmux pane never blocks on Cursor's workspace-trust / per-tool prompts — and starts the forwarder that mirrors the TUI transcript back as conversation items. :param base_url: Spawned server base URL. :param runner_id: The token-bound runner id to bind. :returns: The new session/conversation id. """ import json as _json import tempfile from omnigent._wrapper_labels import ( CURSOR_NATIVE_WRAPPER_VALUE, UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.cursor_native import _materialize_cursor_agent_spec with tempfile.TemporaryDirectory() as _tmp: spec_path = _materialize_cursor_agent_spec(Path(_tmp)) yaml_text = spec_path.read_text() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: data = yaml_text.encode() # Non-config.yaml arcname → omnigent compat translator (the spec has # no spec_version), matching the terminal_session fixture. info = tarfile.TarInfo("cursor-native-ui.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) labels = { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: CURSOR_NATIVE_WRAPPER_VALUE, } # Pin a real workspace on THIS session (like the codex fixture): the # forwarder keys cursor's chat store by ``md5(cwd)``, so the TUI needs a # concrete launch cwd. The repo root is a valid dir on the runner's # filesystem. ``-f`` trusts that dir + auto-approves tools so the # unattended pane never hangs on an approval prompt. metadata = { "labels": labels, "workspace": str(_REPO_ROOT), # ``-f`` (the default) trusts the dir + auto-approves tools so the # unattended pane never hangs. The approval-mirror test passes # ``launch_args=()`` so cursor's per-tool prompts fire and surface as # web elicitation cards. "terminal_launch_args": list(launch_args), } create = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps(metadata)}, files={"bundle": ("cursor-native-ui.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create.raise_for_status() session_id = str(create.json()["session_id"]) _bind_session_runner(base_url, session_id, runner_id) return session_id def _create_native_goose_session(base_url: str, runner_id: str) -> str: """Register the ``goose-native`` wrapper agent and bind its session. Mirrors :func:`_create_native_cursor_session`: reuses the exact terminal-first spec ``omnigent goose`` ships (:func:`omnigent.goose_native._materialize_goose_agent_spec`) and stamps the same wrapper / terminal-first labels. Binding triggers the runner's goose-native auto-bootstrap (:func:`omnigent.runner.app._auto_create_goose_terminal`), which launches ``goose session`` in the session terminal and starts the forwarder that mirrors the TUI transcript back as conversation items. Goose's tool-approval gating is its own ``GOOSE_MODE`` (no ``-f`` equivalent), so no launch args. :param base_url: Spawned server base URL. :param runner_id: The token-bound runner id to bind. :returns: The new session/conversation id. """ import json as _json import tempfile from omnigent._wrapper_labels import ( GOOSE_NATIVE_WRAPPER_VALUE, UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.goose_native import _materialize_goose_agent_spec with tempfile.TemporaryDirectory() as _tmp: spec_path = _materialize_goose_agent_spec(Path(_tmp)) yaml_text = spec_path.read_text() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: data = yaml_text.encode() info = tarfile.TarInfo("goose-native-ui.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) labels = { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: GOOSE_NATIVE_WRAPPER_VALUE, } metadata = { "labels": labels, "workspace": str(_REPO_ROOT), } create = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps(metadata)}, files={"bundle": ("goose-native-ui.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create.raise_for_status() session_id = str(create.json()["session_id"]) _bind_session_runner(base_url, session_id, runner_id) return session_id @pytest.fixture def native_goose_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound session on the real ``goose-native`` ("Goose") wrapper. The runner auto-launches ``goose session`` in the session terminal on bind, so the SPA's Terminal view attaches to a live Goose TUI and its Chat view renders the same canonical transcript. Drives the goose render-parity suite. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_goose_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) def _create_native_kiro_session(base_url: str, runner_id: str) -> str: """Register the ``kiro-native`` wrapper agent and bind its session. Mirrors :func:`_create_native_goose_session`: reuses the terminal-first spec ``omnigent kiro`` ships (:func:`omnigent.kiro_native._materialize_kiro_agent_spec`) and stamps the same wrapper / terminal-first labels. Binding triggers the runner's kiro-native auto-bootstrap (:func:`omnigent.runner.app._auto_create_kiro_terminal`), which launches the ``kiro-cli`` TUI in the session terminal and starts the forwarder that mirrors the TUI transcript back as conversation items. :param base_url: Spawned server base URL. :param runner_id: The token-bound runner id to bind. :returns: The new session/conversation id. """ import json as _json import tempfile from omnigent._wrapper_labels import ( KIRO_NATIVE_WRAPPER_VALUE, UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.kiro_native import _materialize_kiro_agent_spec with tempfile.TemporaryDirectory() as _tmp: spec_path = _materialize_kiro_agent_spec(Path(_tmp), model=None) yaml_text = spec_path.read_text() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: data = yaml_text.encode() info = tarfile.TarInfo("kiro-native-ui.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) labels = { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: KIRO_NATIVE_WRAPPER_VALUE, } metadata = { "labels": labels, "workspace": str(_REPO_ROOT), } create = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps(metadata)}, files={"bundle": ("kiro-native-ui.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create.raise_for_status() session_id = str(create.json()["session_id"]) _bind_session_runner(base_url, session_id, runner_id) return session_id @pytest.fixture def native_kiro_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound session on the real ``kiro-native`` ("Kiro") wrapper. The runner auto-launches the ``kiro-cli`` TUI in the session terminal on bind, so the SPA's Terminal view attaches to a live Kiro TUI and its Chat view renders the same canonical transcript. Drives the kiro render-parity suite. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_kiro_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) def _create_native_hermes_session(base_url: str, runner_id: str) -> str: """Register the ``hermes-native`` wrapper agent and bind its session. Mirrors :func:`_create_native_goose_session`: reuses the exact terminal-first spec ``omnigent hermes`` ships (:func:`omnigent.hermes_native._materialize_hermes_agent_spec`) and stamps the same wrapper / terminal-first labels. Binding triggers the runner's hermes-native auto-bootstrap (:func:`omnigent.runner.app._auto_create_hermes_terminal`), which launches the ``hermes`` TUI in the session terminal and starts the forwarder that mirrors the TUI transcript back as conversation items. :param base_url: Spawned server base URL. :param runner_id: The token-bound runner id to bind. :returns: The new session/conversation id. """ import json as _json import tempfile from omnigent._wrapper_labels import ( HERMES_NATIVE_WRAPPER_VALUE, UI_MODE_LABEL_KEY, UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY, ) from omnigent.hermes_native import _materialize_hermes_agent_spec with tempfile.TemporaryDirectory() as _tmp: spec_path = _materialize_hermes_agent_spec(Path(_tmp)) yaml_text = spec_path.read_text() buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: data = yaml_text.encode() info = tarfile.TarInfo("hermes-native-ui.yaml") info.size = len(data) tar.addfile(info, io.BytesIO(data)) labels = { UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE, WRAPPER_LABEL_KEY: HERMES_NATIVE_WRAPPER_VALUE, } metadata = { "labels": labels, "workspace": str(_REPO_ROOT), } create = httpx.post( f"{base_url}/v1/sessions", data={"metadata": _json.dumps(metadata)}, files={"bundle": ("hermes-native-ui.tar.gz", buf.getvalue(), "application/gzip")}, timeout=30.0, ) create.raise_for_status() session_id = str(create.json()["session_id"]) _bind_session_runner(base_url, session_id, runner_id) return session_id @pytest.fixture def native_hermes_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound session on the real ``hermes-native`` ("Hermes") wrapper. The runner auto-launches the ``hermes`` TUI in the session terminal on bind, so the SPA's Terminal view attaches to a live Hermes TUI and its Chat view renders the same canonical transcript. Drives the hermes render-parity suite. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_hermes_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) @pytest.fixture def native_cursor_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A runner-bound session on the real ``cursor-native`` ("Cursor") wrapper. The runner auto-launches ``cursor-agent`` in the session terminal on bind (ambient ``cursor-agent login`` / ``CURSOR_API_KEY`` auth + ``-f`` trust handled runner-side), so the SPA's Terminal view attaches to a live Cursor TUI and its Chat view renders the same canonical transcript. Drives the native cursor render-parity suite. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_cursor_session(live_server, runner_id) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() # Escalate to SIGKILL if the runner ignores SIGTERM, so a wedged # process can't raise in teardown and leak / fail unrelated tests # (matching terminal_session / seeded_session_pair). try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5) @pytest.fixture def native_cursor_approval_session( live_server: str, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[tuple[str, str]]: """A ``cursor-native`` session launched WITHOUT ``-f`` (prompts fire). Identical to :func:`native_cursor_session` but omits the force/trust flag, so ``cursor-agent`` raises its real per-tool approval prompts. The runner- side mirror (:mod:`omnigent.cursor_native_permissions`) surfaces those as web ``response.elicitation_request`` cards — what the approval-ordering test drives. The first-run workspace-trust modal is dismissed by the executor's inject path on the first composer turn. :param live_server: Spawned server fixture; its runner is reused. :param tmp_path_factory: Pytest temp path factory (for a respawn log). :returns: ``(base_url, session_id)``. """ respawned = _ensure_runner_online(live_server, tmp_path_factory) runner_id = str(_server_state["runner_id"]) session_id = _create_native_cursor_session(live_server, runner_id, launch_args=()) try: yield (live_server, session_id) finally: httpx.delete(f"{live_server}/v1/sessions/{session_id}", timeout=10.0) if respawned is not None: respawned.terminate() try: respawned.wait(timeout=5) except subprocess.TimeoutExpired: respawned.kill() respawned.wait(timeout=5)