"""Tests for utility endpoints on the FastAPI app (health + version). These endpoints are defined inline in ``create_app()`` in ``omnigent/server/app.py`` rather than in a route sub-module, so they live here following the source ↔ test directory mirroring rule. """ from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path import httpx import pytest from fastapi import FastAPI from omnigent.runtime.agent_cache import AgentCache from omnigent.server import app as server_app from omnigent.stores.agent_store.sqlalchemy_store import SqlAlchemyAgentStore from omnigent.stores.artifact_store.local import LocalArtifactStore from omnigent.stores.conversation_store.sqlalchemy_store import SqlAlchemyConversationStore @pytest.mark.asyncio async def test_health_returns_ok(client: httpx.AsyncClient) -> None: """GET /health returns HTTP 200 and ``{"status": "ok"}``.""" resp = await client.get("/health") assert resp.status_code == 200 # Exact shape — a regression that changes the key name or value # would break health-check integrations that parse this response. assert resp.json() == {"status": "ok"} @pytest.mark.asyncio async def test_version_returns_source_of_truth_version( client: httpx.AsyncClient, ) -> None: """GET /api/version returns ``omnigent.version.VERSION``. The endpoint surfaces the shared source-of-truth constant, authoritative regardless of how the package was installed. We deliberately do NOT assert against ``importlib.metadata.version`` here: that is a frozen build-time snapshot which can legitimately differ from ``VERSION`` (stale editable install, or ``"source"`` placeholder metadata) — asserting equality would re-couple to exactly the metadata this change moved off of. """ from omnigent.version import VERSION resp = await client.get("/api/version") assert resp.status_code == 200 body = resp.json() # "version" key must be present — a missing key means the UI's # fetchVersion() falls back to "unknown" in every bug report. assert "version" in body assert body["version"] == VERSION, ( f"Expected version {VERSION!r} from omnigent.version.VERSION, got {body['version']!r}." ) def test_server_version_reads_version_constant() -> None: """The server version is the shared ``omnigent.version.VERSION`` constant.""" from omnigent.version import VERSION assert server_app._server_version() == VERSION class _StubWebSocket: """ Minimal real ``WebSocketLike`` for registering a runner tunnel. The tunnel registry only stores this object; ``_bulk_session_liveness`` reads tunnel presence via ``TunnelRegistry.get(...) is not None`` and never sends or receives on it. A real class (not ``MagicMock``) is used so any unexpected I/O call raises loudly rather than silently succeeding — but in this test the socket is never driven. """ async def send_text(self, data: str) -> None: """Unused — the liveness path never sends. Fails loud if reached.""" raise AssertionError("the liveness test must not send on the tunnel socket") async def receive_text(self) -> str: """Unused — the liveness path never receives. Fails loud if reached.""" raise AssertionError("the liveness test must not receive on the tunnel socket") def _register_live_runner(app: FastAPI, runner_id: str) -> None: """ Register a live runner tunnel on the app's registry. Mirrors what the runner-tunnel WS route does on connect: it adds a :class:`RunnerSession` to ``app.state.tunnel_registry`` so ``_bulk_session_liveness`` reads the runner as up (``runner_online=True``). Uses the registry's real ``register`` API with a minimal real WS stub and a real :class:`HelloFrame`. :param app: The FastAPI app returned by ``create_app`` (carries the registry on ``app.state.tunnel_registry``). :param runner_id: Runner id to register, e.g. ``"rnr_live"``. """ from omnigent.runner.transports.ws_tunnel.frames import HelloFrame app.state.tunnel_registry.register( runner_id, _StubWebSocket(), HelloFrame(runner_version="0.0.0-test", frame_protocol_version=1), ) @dataclass(frozen=True) class _LivenessApp: """ A wired app plus the store that seeds its conversations. :param app: The FastAPI app to drive ``/health`` against. :param conversation_store: The store used to seed conversations the liveness lookups read. """ app: FastAPI conversation_store: SqlAlchemyConversationStore def _build_liveness_app( db_uri: str, tmp_path: Path, ) -> _LivenessApp: """ Build a real app + conversation store wired for liveness tests. :param db_uri: SQLite URI for the shared test db (the ``db_uri`` fixture). :param tmp_path: Per-test temp dir for artifacts and cache. :returns: A :class:`_LivenessApp` carrying the app to drive ``/health`` against and the store to seed conversations in. """ from omnigent.server.app import create_app from omnigent.stores.file_store.sqlalchemy_store import SqlAlchemyFileStore from omnigent.stores.host_store import HostStore conversation_store = SqlAlchemyConversationStore(db_uri) host_store = HostStore(db_uri) artifact_store = LocalArtifactStore(str(tmp_path / "artifacts")) # Online host so a host-bound session reports host_online=True. host_store.upsert_on_connect("host_live", "laptop", "alice@example.com") # A host that EXISTS (so the conversations.host_id FK is satisfied) but # is offline, for the runner-down + host-offline state. host_store.upsert_on_connect("host_offline", "old-laptop", "alice@example.com") host_store.set_offline("host_offline") app = create_app( agent_store=SqlAlchemyAgentStore(db_uri), file_store=SqlAlchemyFileStore(db_uri), conversation_store=conversation_store, artifact_store=artifact_store, host_store=host_store, agent_cache=AgentCache( artifact_store=artifact_store, cache_dir=tmp_path / "cache", ), ) return _LivenessApp(app=app, conversation_store=conversation_store) @pytest.mark.asyncio async def test_health_batch_reports_strict_runner_and_host_liveness( db_uri: str, tmp_path: Path, ) -> None: """ ``GET /health?session_ids=`` reports the strict 4-state liveness matrix, with both ``runner_online`` and ``host_online`` per session. Exercises the batched online-dot path the sidebar polls every 1-2s through the public ``/health`` endpoint, which routes to ``_bulk_session_liveness``. The four states the open-session view must distinguish: (a) runner tunnel up → runner_online True; (b) runner down + host online → runner_online False, host_online True (the runner-down-but-host-can-relaunch state — strict ``runner_online`` does NOT fold in host optimism); (c) runner down + host offline → runner_online False, host_online False; (d) no host_id, runner down → runner_online False, host_online None. Also pins that a deliberately-stopped session is NOT special-cased (the stopped marker is retired in this workstream): it reads purely by tunnel/host state, same as any other host-bound session. """ wired = _build_liveness_app(db_uri, tmp_path) app = wired.app conversation_store = wired.conversation_store # (a) runner tunnel up: bind a runner and register its tunnel. runner_up = conversation_store.create_conversation(runner_id="rnr_live") _register_live_runner(app, "rnr_live") # (b) runner bound but tunnel NOT registered, host online. runner_down_host_up = conversation_store.create_conversation( runner_id="rnr_dead", host_id="host_live", workspace="/tmp/ws" ) # (c) runner bound but tunnel down, host offline (unknown host id). runner_down_host_down = conversation_store.create_conversation( runner_id="rnr_dead2", host_id="host_offline", workspace="/tmp/ws" ) # (d) runner bound but tunnel down, NO host binding. runner_down_no_host = conversation_store.create_conversation(runner_id="rnr_dead3") # A stopped, host-bound session with a dead runner must read by # tunnel/host state only — the stopped marker is no longer consulted by # liveness. Bind a (down) runner so this isn't the no-runner terminal. stopped = conversation_store.create_conversation( runner_id="rnr_dead4", host_id="host_live", workspace="/tmp/ws" ) conversation_store.set_labels(stopped.id, {"omnigent.stopped": "true"}) ids = ",".join( [ runner_up.id, runner_down_host_up.id, runner_down_host_down.id, runner_down_no_host.id, stopped.id, "conv_unknown", ] ) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: resp = await c.get(f"/health?session_ids={ids}") assert resp.status_code == 200 body = resp.json() assert body["status"] == "ok" sessions = body["sessions"] # host_version is None for every row here: the hosts are seeded in the # host_store (DB) for the online gate but have no live tunnel in the # in-memory host_registry, which is the only source of the version (the # cross-replica / DB-only degradation path). The live-registry path is # covered by test_health_reports_host_version_from_live_registry. # # (a) Live runner tunnel ⇒ reachable. host_online None (no host_id). # A failure here means the strict _runner_up check stopped reading the # tunnel registry, or the registration path changed. assert sessions[runner_up.id] == { "runner_online": True, "host_online": None, "host_version": None, } # (b) Dead runner, live host: strict runner_online is False (no # host-relaunch optimism), but host_online surfaces the live host so # the open view can offer "send a message to wake the runner". A True # runner_online here would be the old conflated behavior regressing. assert sessions[runner_down_host_up.id] == { "runner_online": False, "host_online": True, "host_version": None, } # (c) Dead runner, offline host ⇒ both False. assert sessions[runner_down_host_down.id] == { "runner_online": False, "host_online": False, "host_version": None, } # (d) Dead runner, no host ⇒ runner_online False, host_online None. assert sessions[runner_down_no_host.id] == { "runner_online": False, "host_online": None, "host_version": None, } # Stopped + live host: NOT special-cased — reads exactly like (b). If # this regressed to False/True-with-stopped-collapse, the stopped marker # would still be leaking into liveness (it must not — WS-S2 retires it). assert sessions[stopped.id] == { "runner_online": False, "host_online": True, "host_version": None, } # Unknown id (no conversation row) ⇒ reachable, no host. assert sessions["conv_unknown"] == { "runner_online": True, "host_online": None, "host_version": None, } @pytest.mark.asyncio async def test_health_single_session_reports_both_liveness_fields( db_uri: str, tmp_path: Path, ) -> None: """ ``GET /health?session_id=`` returns a single ``session`` object that carries the id plus both ``runner_online`` and ``host_online``. Guards the single-id branch of the endpoint (distinct from the ``session_ids`` batch branch) so a client probing one session still gets the host-vs-runner split the open view needs. """ wired = _build_liveness_app(db_uri, tmp_path) # Dead runner on a live host: the runner-down-but-host-alive state. conv = wired.conversation_store.create_conversation( runner_id="rnr_dead", host_id="host_live", workspace="/tmp/ws" ) app = wired.app transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: resp = await c.get(f"/health?session_id={conv.id}") assert resp.status_code == 200 body = resp.json() # The single object echoes the id and the liveness fields — strict # runner_online False with host_online True. host_version is None: the # host is online in the DB store but has no live tunnel in the in-memory # registry the version is read from. A missing host_online/host_version # key would mean the single-id branch wasn't updated with the batch one. assert body["session"] == { "id": conv.id, "runner_online": False, "host_online": True, "host_version": None, } @pytest.mark.asyncio async def test_health_reports_host_version_from_live_registry( db_uri: str, tmp_path: Path, ) -> None: """ ``GET /health`` surfaces the bound host's version when that host has a live tunnel on this replica. ``host_version`` is read from the in-memory host registry (the host's ``host.hello`` frame), not the hosts table — so a session bound to a host with a live local tunnel reports that host's version, which the session info popover renders next to the server version. This is the non-``None`` counterpart to the DB-only rows in the batch test. """ from omnigent.host.frames import HostHelloFrame wired = _build_liveness_app(db_uri, tmp_path) app = wired.app # host_live is already online in the host_store (DB) via the builder; # registering it in the in-memory registry is what carries the version. app.state.host_registry.register( "host_live", _StubWebSocket(), HostHelloFrame(version="9.9.9-test", frame_protocol_version=1, name="laptop"), "alice@example.com", ) conv = wired.conversation_store.create_conversation( runner_id="rnr_dead", host_id="host_live", workspace="/tmp/ws" ) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: resp = await c.get(f"/health?session_id={conv.id}") assert resp.status_code == 200 # host_online True (DB store) AND host_version from the live registry. assert resp.json()["session"] == { "id": conv.id, "runner_online": False, "host_online": True, "host_version": "9.9.9-test", } @pytest.mark.asyncio async def test_info_includes_server_version( client: httpx.AsyncClient, ) -> None: """ ``GET /v1/info`` includes ``server_version`` — the shared ``VERSION`` constant (same source as ``/api/version``) — so the web UI can show it in the session info popover's version footer without a second fetch. """ from omnigent.version import VERSION resp = await client.get("/v1/info") assert resp.status_code == 200 body = resp.json() # Matches the source-of-truth constant, not importlib.metadata: the latter # is a frozen build-time snapshot that can drift from VERSION in a stale # editable or "source"-placeholder install. assert body["server_version"] == VERSION @pytest.mark.asyncio async def test_health_bare_returns_status_ok(db_uri: str, tmp_path: Path) -> None: """ ``GET /health`` with no session params still returns the bare ``{"status": "ok"}`` — the liveness rearchitecture must not break the plain health-check integrations that parse this exact shape. """ app = _build_liveness_app(db_uri, tmp_path).app transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: resp = await c.get("/health") assert resp.status_code == 200 # Exact shape — no session/sessions keys leak in when none were asked for. assert resp.json() == {"status": "ok"} @pytest.mark.asyncio async def test_health_unbound_fork_of_coding_session_reads_offline( db_uri: str, tmp_path: Path, ) -> None: """ An unbound fork of a coding session reads offline; a chat fork online. Both sessions are unbound (no runner, no host) — the case that previously short-circuited to "online" unconditionally. The fork-source label (set when the source had a workspace) flips that: - a fork carrying ``omnigent.fork.source_id`` reads offline, so the first message routes into the directory picker instead of dropping against a runner that can't start; - a fork without it (chat-only source, CUJ 2) stays online and resumes in-process like a brand-new chat session. This is the regression guard for the ``needs_workspace`` branch of ``_bulk_runner_online``: if that branch reverts to ``True`` for all unbound sessions, the coding-fork assertion fails. """ from omnigent.server.app import create_app from omnigent.stores.conversation_store.sqlalchemy_store import ( SqlAlchemyConversationStore, ) from omnigent.stores.file_store.sqlalchemy_store import SqlAlchemyFileStore from omnigent.stores.host_store import HostStore conversation_store = SqlAlchemyConversationStore(db_uri) host_store = HostStore(db_uri) artifact_store = LocalArtifactStore(str(tmp_path / "artifacts")) # Unbound forks: no runner_id, no host_id. The label is the only # difference between them. coding_fork = conversation_store.create_conversation() conversation_store.set_labels(coding_fork.id, {"omnigent.fork.source_id": "conv_src"}) chat_fork = conversation_store.create_conversation() app = create_app( agent_store=SqlAlchemyAgentStore(db_uri), file_store=SqlAlchemyFileStore(db_uri), conversation_store=conversation_store, artifact_store=artifact_store, host_store=host_store, agent_cache=AgentCache( artifact_store=artifact_store, cache_dir=tmp_path / "cache", ), ) ids = f"{coding_fork.id},{chat_fork.id}" transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: resp = await c.get(f"/health?session_ids={ids}") assert resp.status_code == 200 sessions = resp.json()["sessions"] # needs_workspace fork → offline. A True here means the unbound # branch ignored the flag (the pre-fix behavior). assert sessions[coding_fork.id]["runner_online"] is False # Chat-only fork → still reachable in-process. assert sessions[chat_fork.id]["runner_online"] is True @dataclass(frozen=True) class _SeedStores: """ The three stores the default-agent seeders take. :param agent_store: Store the seeder writes the agent row into. :param artifact_store: Store the seeder writes the bundle blob into. :param agent_cache: Cache the seeder evicts after registering. """ agent_store: SqlAlchemyAgentStore artifact_store: LocalArtifactStore agent_cache: AgentCache @pytest.fixture() def seed_stores(tmp_path: Path, db_uri: str) -> _SeedStores: """ Real stores wired for the default-agent seeders, backed by the shared test SQLite db and a per-test temp artifact dir. :param tmp_path: Per-test temp dir for the artifact store and cache. :param db_uri: SQLite URI for the agent store (the ``db_uri`` fixture), e.g. ``"sqlite:////tmp/.../test.db"``. :returns: The bundled stores as a :class:`_SeedStores`. """ artifact_store = LocalArtifactStore(str(tmp_path / "artifacts")) return _SeedStores( agent_store=SqlAlchemyAgentStore(db_uri), artifact_store=artifact_store, agent_cache=AgentCache( artifact_store=artifact_store, cache_dir=tmp_path / "cache", ), ) @pytest.fixture() def polly_src_copy(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: """ A writable copy of the packaged polly bundle, wired as the seed source. The seeder reads from ``server_app._POLLY_BUNDLE_SOURCE`` (a read-only path inside the install). Tests that need to mutate the spec (change the subtitle) or its file mtimes (simulate a wheel reinstall) copy it here and point the seeder at the copy. :param tmp_path: Per-test temp dir for the copy. :param monkeypatch: Used to repoint ``_POLLY_BUNDLE_SOURCE``. :returns: Path to the copied polly bundle directory. """ import shutil dest = tmp_path / "polly_src" shutil.copytree(server_app._POLLY_BUNDLE_SOURCE, dest) monkeypatch.setattr(server_app, "_POLLY_BUNDLE_SOURCE", dest) return dest def _independent_seed_stores(tmp_path: Path, label: str) -> _SeedStores: """A fresh, migrated, independent set of seed stores under ``tmp_path``.""" from omnigent.db.utils import get_or_create_engine uri = f"sqlite:///{tmp_path / f'{label}.db'}" get_or_create_engine(uri) # run migrations, same path as production artifact_store = LocalArtifactStore(str(tmp_path / f"{label}-artifacts")) return _SeedStores( agent_store=SqlAlchemyAgentStore(uri), artifact_store=artifact_store, agent_cache=AgentCache( artifact_store=artifact_store, cache_dir=tmp_path / f"{label}-cache", ), ) def test_builtin_agent_id_is_stable_across_independent_stores(tmp_path: Path) -> None: """A built-in's id is identical across two independent fresh stores — the contract the multi-tenant deployment needs. A revert to the random ``generate_agent_id()`` makes the two ids differ and fails this test.""" from omnigent.db.utils import builtin_agent_id a = _independent_seed_stores(tmp_path, "a") b = _independent_seed_stores(tmp_path, "b") server_app._ensure_default_polly_agent(a.agent_store, a.artifact_store, a.agent_cache) server_app._ensure_default_polly_agent(b.agent_store, b.artifact_store, b.agent_cache) seeded_a = a.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) seeded_b = b.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert seeded_a is not None and seeded_b is not None assert seeded_a.id == seeded_b.id == builtin_agent_id(server_app._POLLY_AGENT_NAME) def test_ensure_extra_builtin_agents_skips_bad_path_and_seeds_good( seed_stores: _SeedStores, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """A bad entry in OMNIGENT_BUILTIN_AGENT_DIRS is logged + skipped, not fatal. Operator-supplied paths may be wrong (typo, stale mount). One bad entry must not crash server startup nor block a valid entry from registering. A regression that lets ``materialize_bundle``'s ``FileNotFoundError`` propagate would raise here (server wouldn't start) and the good agent would never seed. """ good = tmp_path / "extra-good.yaml" good.write_text( "name: extra-good\n" "executor:\n" " harness: claude-sdk\n" " model: claude-sonnet-4-20250514\n" "prompt: hi\n" ) missing = tmp_path / "does-not-exist.yaml" # Bad path first so the good one only seeds if the loop survives the bad. monkeypatch.setenv(server_app._EXTRA_BUILTIN_AGENTS_ENV, f"{missing}{os.pathsep}{good}") # Must not raise despite the missing path. server_app._ensure_extra_builtin_agents( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache ) # The valid spec registered as a built-in (its file stem is the name); # the missing one is absent — proving skip-and-continue, not abort. seeded = seed_stores.agent_store.get_by_name("extra-good") assert seeded is not None, "valid extra built-in must seed even after a bad entry" assert seeded.session_id is None, "extra built-ins must be session-scope NULL" assert seed_stores.agent_store.get_by_name("does-not-exist") is None def test_ensure_default_qwen_agent_seeds_card(seed_stores: _SeedStores) -> None: """ Seeding registers qwen-native-ui as a built-in the picker can render. The new-session picker reads built-ins from ``GET /v1/agents``; without this seeder Qwen Code only appears after the ``omnigent qwen`` CLI first registers it, so it was absent from the Web UI dropdown. """ server_app._ensure_default_qwen_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) seeded = seed_stores.agent_store.get_by_name(server_app._QWEN_NATIVE_AGENT_NAME) assert seeded is not None, "qwen-native-ui was not registered" assert seeded.name == "qwen-native-ui" # The bundle must be retrievable, not just referenced. assert seed_stores.artifact_store.get(seeded.bundle_location) is not None def test_ensure_default_qwen_agent_is_idempotent(seed_stores: _SeedStores) -> None: """A second seed call is a no-op — startup runs the seeder every boot.""" server_app._ensure_default_qwen_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) first = seed_stores.agent_store.get_by_name(server_app._QWEN_NATIVE_AGENT_NAME) assert first is not None server_app._ensure_default_qwen_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) page = seed_stores.agent_store.list(limit=100) qwen_rows = [a for a in page.data if a.name == "qwen-native-ui"] assert len(qwen_rows) == 1 assert qwen_rows[0].id == first.id assert qwen_rows[0].version == first.version == 1 def test_ensure_default_polly_agent_seeds_card(seed_stores: _SeedStores) -> None: """ Seeding registers polly as a built-in the picker can render. The new-session picker reads built-ins from ``GET /v1/agents`` and renders each as a card; this is what makes polly launchable next to Claude Code. """ server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) seeded = seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert seeded is not None, "polly was not registered" assert seeded.name == "polly" # The bundle must be retrievable, not just referenced. assert seed_stores.artifact_store.get(seeded.bundle_location) is not None def test_ensure_default_antigravity_agent_seeds_card(seed_stores: _SeedStores) -> None: """ Seeding registers antigravity-native-ui as a built-in the picker renders. This is what makes "Antigravity" launchable from the web-UI new-session picker next to Claude Code and Codex. The agent must be session-scope NULL (a built-in) and carry the ``antigravity-native`` harness so the runner boots the agy native terminal rather than an SDK harness. """ server_app._ensure_default_antigravity_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) seeded = seed_stores.agent_store.get_by_name(server_app._ANTIGRAVITY_NATIVE_AGENT_NAME) assert seeded is not None, "antigravity-native-ui was not registered" assert seeded.name == "antigravity-native-ui" # Built-ins are session-scope NULL so ``GET /v1/agents`` (which filters on # ``session_id IS NULL``) returns them to the picker. assert seeded.session_id is None # The bundle must be retrievable, not just referenced. assert seed_stores.artifact_store.get(seeded.bundle_location) is not None # The materialized spec must carry the native harness — the contract the # server's _native_terminal_runtime + runner auto-create key off. loaded = seed_stores.agent_cache.load(seeded.id, seeded.bundle_location, expand_env=False) assert loaded.spec.executor.config.get("harness") == "antigravity-native" def test_ensure_default_agents_includes_antigravity(seed_stores: _SeedStores) -> None: """ The startup seeder registers the antigravity built-in alongside the others. ``_ensure_default_agents`` is the single call the server lifespan makes; a regression that drops the antigravity line would silently remove it from the picker even though its helper still works. """ server_app._ensure_default_agents( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) assert ( seed_stores.agent_store.get_by_name(server_app._ANTIGRAVITY_NATIVE_AGENT_NAME) is not None ) def test_ensure_default_polly_agent_is_idempotent(seed_stores: _SeedStores) -> None: """ A second seed call is a no-op — it must not register a duplicate. Startup runs the seeder every boot; a non-idempotent seeder would accumulate a new polly agent on each restart and break the unique-name invariant the store enforces. """ server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) first = seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert first is not None server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) # Exactly one polly row, and it's the same agent id as the first # seed (no replacement, no duplicate). page = seed_stores.agent_store.list(limit=100) polly_rows = [a for a in page.data if a.name == "polly"] assert len(polly_rows) == 1 assert polly_rows[0].id == first.id # Unchanged re-seed must not bump version (a bump = nondeterministic bundle). assert polly_rows[0].version == first.version == 1 def test_ensure_default_polly_agent_refreshes_on_spec_change( seed_stores: _SeedStores, polly_src_copy: Path ) -> None: """ A changed on-disk bundle refreshes the existing row in place. This is the regression guard for the seed-once bug: when a new wheel shipped a changed polly spec, the old seeder saw the row already existed and returned early, so the deployed app kept serving the stale bundle forever. The fix re-bundles and updates in place. """ server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) first = seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert first is not None assert first.version == 1 # Simulate a new wheel shipping changed bundle content. (polly_src_copy / "NEW_FILE.md").write_text("shipped in the new wheel") server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) refreshed = seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert refreshed is not None # Same agent_id keeps task history (a delete would cascade the tasks FK). assert refreshed.id == first.id # Bundle + version refreshed in place; under seed-once both stayed stale (the bug). assert refreshed.version == 2 assert refreshed.bundle_location != first.bundle_location # One row (refresh, not duplicate); new bundle is retrievable. page = seed_stores.agent_store.list(limit=100) assert len([a for a in page.data if a.name == "polly"]) == 1 assert seed_stores.artifact_store.get(refreshed.bundle_location) is not None def test_ensure_default_polly_agent_no_churn_on_mtime_change( seed_stores: _SeedStores, polly_src_copy: Path ) -> None: """ A redeploy with unchanged content does NOT refresh the row. A wheel reinstall stamps fresh mtimes on identical files. Because :func:`server_app._tar_gz_dir` normalizes tar member metadata, the bundle bytes (and thus the content hash) are unchanged, so the seeder is a no-op. Without that normalization the hash would shift on every deploy and the version would climb forever. """ import os server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) first = seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert first is not None assert first.version == 1 # Simulate a wheel reinstall: same content, brand-new mtimes. bumped = 1_700_000_000 # fixed mtime, != the copy's os.utime(polly_src_copy, (bumped, bumped)) for path in polly_src_copy.rglob("*"): os.utime(path, (bumped, bumped)) server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) after = seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert after is not None # Unchanged content → no-op; a v2 here means tar mtimes leaked into the hash. assert after.version == 1 assert after.bundle_location == first.bundle_location def test_ensure_default_polly_agent_repairs_stale_cache( seed_stores: _SeedStores, polly_src_copy: Path ) -> None: """ A matching-hash re-seed repairs a stale local agent cache. ``AgentCache.load`` is keyed by ``agent_id`` and trusts its cached entry without checking ``bundle_location``. If a replica boots with a cache that lags the (already-current) DB row — or a prior boot's cache swap failed after the row update — the seeder must not just early-return and keep serving the stale spec; it evicts so the next load re-fetches the row's bundle. Without the evict this test fails: ``load`` would return the poisoned spec. """ import shutil import yaml server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) agent = seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) assert agent is not None real_desc = seed_stores.agent_cache.load(agent.id, agent.bundle_location).spec.description # Poison the local cache with a different spec (a lagging replica's stale cache). stale_src = polly_src_copy.parent / "stale" shutil.copytree(polly_src_copy, stale_src) config = stale_src / "config.yaml" raw = yaml.safe_load(config.read_text()) raw["description"] = "STALE CACHED SPEC" config.write_text(yaml.safe_dump(raw, sort_keys=False)) seed_stores.agent_cache.replace( agent.id, agent.bundle_location, server_app._tar_gz_dir(stale_src) ) poisoned = seed_stores.agent_cache.load(agent.id, agent.bundle_location).spec.description assert poisoned == "STALE CACHED SPEC" # Re-seed: source unchanged → hash matches the row (the early-return path). server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) # Cache was evicted, so load now returns the row's real bundle, not the poison. repaired = seed_stores.agent_cache.load(agent.id, agent.bundle_location).spec.description assert repaired == real_desc assert repaired != "STALE CACHED SPEC" def test_tar_gz_dir_is_order_independent(tmp_path: Path) -> None: """ Same content, different file-creation order → identical bundle bytes. Content-addressing only works if the bundle hash depends solely on content, not on the order entries happen to land on disk across machines. Guards the determinism the seeder's no-op detection relies on; would fail if the packer stopped sorting members. """ import os def build(order: list[str]) -> bytes: root = tmp_path / f"bundle-{'-'.join(order)}" (root / "sub").mkdir(parents=True) for name in order: (root / name).write_text(f"content-of-{name}") (root / "sub" / "nested.txt").write_text("nested") # Vary mtimes too, so this also pins content-only hashing. for path in root.rglob("*"): os.utime(path, (1, 1)) return server_app._tar_gz_dir(root) assert build(["z.yaml", "a.yaml", "m.yaml"]) == build(["a.yaml", "m.yaml", "z.yaml"]) def test_tar_gz_dir_ignores_mode_bits(tmp_path: Path) -> None: """ A chmod-only difference must not change the bundle bytes. Package vs install environments can hand the same file different mode bits; :func:`server_app._normalize_tarinfo` pins a canonical mode so that doesn't bump the content hash. Safe because extraction uses ``set_attrs=False``, so on-disk permissions don't depend on the tar mode. Would fail if mode were left in the hash input. """ import os def build(mode: int) -> bytes: root = tmp_path / f"bundle-{mode:o}" root.mkdir() config = root / "config.yaml" config.write_text("name: x") os.chmod(config, mode) return server_app._tar_gz_dir(root) assert build(0o600) == build(0o644) def test_ensure_default_polly_agent_skips_when_bundle_absent( seed_stores: _SeedStores, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """ No bundle on disk → no card. Seeding is skipped, not errored. On a deployment that didn't package the ``examples/polly`` bundle, seeding must skip silently so startup doesn't fail and no broken card (an agent that can't launch here) appears. """ monkeypatch.setattr(server_app, "_POLLY_BUNDLE_SOURCE", tmp_path / "no-such-polly") server_app._ensure_default_polly_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) assert seed_stores.agent_store.get_by_name(server_app._POLLY_AGENT_NAME) is None def test_ensure_default_debby_agent_seeds_card(seed_stores: _SeedStores) -> None: """ Seeding registers debby as a built-in the picker can render. The new-session picker reads built-ins from ``GET /v1/agents`` and renders each as a card; this is what makes debby launchable next to Claude Code, Codex, and polly. The deeper refresh/idempotency behavior lives in the shared ``_ensure_builtin_agent`` and is covered by the polly tests above — this verifies debby's wiring (name constant, packaged bundle source) specifically. """ server_app._ensure_default_debby_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) seeded = seed_stores.agent_store.get_by_name(server_app._DEBBY_AGENT_NAME) assert seeded is not None, "debby was not registered" assert seeded.name == "debby" # The bundle must be retrievable, not just referenced. assert seed_stores.artifact_store.get(seeded.bundle_location) is not None def test_ensure_default_debby_agent_skips_when_bundle_absent( seed_stores: _SeedStores, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """ No bundle on disk → no card. Seeding is skipped, not errored. On a deployment that didn't package the ``examples/debby`` bundle, seeding must skip silently so startup doesn't fail and no broken card (an agent that can't launch here) appears. """ monkeypatch.setattr(server_app, "_DEBBY_BUNDLE_SOURCE", tmp_path / "no-such-debby") server_app._ensure_default_debby_agent( seed_stores.agent_store, seed_stores.artifact_store, seed_stores.agent_cache, ) assert seed_stores.agent_store.get_by_name(server_app._DEBBY_AGENT_NAME) is None def _build_api_only_app(db_uri: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> FastAPI: """Build an app with the web UI bundle ABSENT (the API-only branch). The dev checkout has a built ``static/web-ui`` (so ``create_app`` would mount the SPA), so point ``_WEB_UI_DIST`` at an empty path to force the API-only fallback branch under test. """ from omnigent.server.app import create_app from omnigent.stores.file_store.sqlalchemy_store import SqlAlchemyFileStore from omnigent.stores.host_store import HostStore monkeypatch.setattr(server_app, "_WEB_UI_DIST", tmp_path / "no-web-ui") artifact_store = LocalArtifactStore(str(tmp_path / "artifacts")) return create_app( agent_store=SqlAlchemyAgentStore(db_uri), file_store=SqlAlchemyFileStore(db_uri), conversation_store=SqlAlchemyConversationStore(db_uri), artifact_store=artifact_store, host_store=HostStore(db_uri), agent_cache=AgentCache(artifact_store=artifact_store, cache_dir=tmp_path / "cache"), ) @pytest.mark.asyncio async def test_api_only_root_serves_html_200_to_any_client( db_uri: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """On a no-web-UI server, ``GET /`` always returns the HTML explainer with a 200 — no content negotiation. A browser navigation and a plain JSON client get the same page (``/`` is not used for anything else).""" app = _build_api_only_app(db_uri, tmp_path, monkeypatch) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: for headers in ({"accept": "text/html"}, {"accept": "application/json"}): resp = await c.get("/", headers=headers) assert resp.status_code == 200, headers assert resp.headers["content-type"].startswith("text/html"), headers assert "web UI" in resp.text assert "OMNIGENT_SKIP_WEB_UI" in resp.text @pytest.mark.asyncio async def test_api_only_unknown_path_gets_json_404( db_uri: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """An unknown path still returns the exact default ``404 {"detail": "Not Found"}`` for every client — the landing is served only at ``/``, never as a catch-all, so API consumers that parse that body are unaffected.""" app = _build_api_only_app(db_uri, tmp_path, monkeypatch) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: for headers in ({"accept": "text/html"}, {"accept": "application/json"}): resp = await c.get("/c/conv_abc", headers=headers) assert resp.status_code == 404, headers assert resp.json() == {"detail": "Not Found"}, headers @pytest.mark.asyncio async def test_api_only_root_does_not_shadow_real_routes( db_uri: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The ``/`` landing is an exact-path route, so real routes like ``/health`` still serve their normal JSON even when the client prefers HTML.""" app = _build_api_only_app(db_uri, tmp_path, monkeypatch) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: resp = await c.get("/health", headers={"accept": "text/html"}) assert resp.status_code == 200 assert resp.json() == {"status": "ok"}