wehub/omnigent-ai--omnigent
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
omnigent-ai--omnigent/tests/runner/test_runner_dispatch.py
T
wehub-resource-sync 3de48288cb chore: import upstream snapshot with attribution
2026-07-13 13:12:00 +08:00

7194 lines
279 KiB
Python
Raw Permalink Blame History

"""End-to-end runner-dispatch tests: server → runner → spawned harness.
The load-bearing assertion: the runner FastAPI app, when given a
real :class:`HarnessProcessManager`, accepts a
POST /v1/sessions/{conversation_id}/events?stream=true,
spawns a harness subprocess (using the existing
``omnigent/runtime/harnesses/`` machinery — NOT a parallel impl),
forwards the request to the harness via UDS, and streams the
harness's SSE response back through the runner's own SSE response.
Architecture verified end-to-end:
- A real ``HarnessProcessManager`` is started (writes its instance
dir, runs orphan sweep, starts the idle reaper).
- A test-only harness module is registered in ``_HARNESS_MODULES``.
- The runner FastAPI app is built with ``process_manager=mgr``.
- The test posts to the runner's
/v1/sessions/{conversation_id}/events?stream=true with a message
body + harness name.
- The runner calls ``mgr.get_client()`` → spawns a uvicorn
subprocess running the test harness on a UDS → returns the
per-conversation httpx client.
- The runner POSTs the request to the harness via that client.
- The harness drives an LLM call via ``run_turn`` and streams SSE
back.
- The runner relays each SSE chunk through to the test client.
The OpenAI-key-gated test runs the full chain against gpt-4o-mini.
The unkeyed test asserts on plumbing only (handler is reached,
503 on bad harness name).
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import tempfile
from collections.abc import AsyncIterator, Callable
from contextlib import asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace, TracebackType
from typing import Any, cast
import httpx
import pytest
from fastapi import FastAPI
from omnigent.runner import create_runner_app
from omnigent.runner.app import (
_build_spawn_env_from_spec,
_forward_harness_response,
_resolve_harness_config,
)
from omnigent.runtime.harnesses import _HARNESS_MODULES
from omnigent.runtime.harnesses.process_manager import HarnessProcessManager
from omnigent.session_lifecycle import CLOSED_LABEL_KEY, CLOSED_LABEL_VALUE
from omnigent.spec.types import AgentSpec, ExecutorSpec, SharePolicy
from tests.runner.helpers import NullServerClient
_TEST_HARNESS_NAME = "runner-test-default"
_TEST_HARNESS_MODULE = "tests._fixtures.runner_test_harness"
@pytest.fixture(autouse=True)
def _assume_harness_clis_installed(monkeypatch: pytest.MonkeyPatch) -> None:
"""Neutralize the sub-agent dispatch CLI preflight for hermetic tests.
The named-mode child-create path refuses to spawn a sub-agent whose
harness CLI (``claude`` / ``codex`` / ``pi``) is absent from ``PATH``
(see ``missing_harness_cli``, dispatched from ``tool_dispatch``). These
dispatch tests run in a hermetic environment where those binaries may be
absent (e.g. CI), so without this stub they would fail at the preflight
instead of exercising the create / continue logic under test. Tests that
specifically assert the preflight re-patch ``missing_harness_cli`` in
their own body, which wins over this autouse default.
:param monkeypatch: Pytest monkeypatch fixture.
"""
monkeypatch.setattr(
"omnigent.onboarding.harness_install.missing_harness_cli",
lambda harness: None,
)
@asynccontextmanager
async def _runner_test_client(app: FastAPI) -> AsyncIterator[httpx.AsyncClient]:
"""Create a test client against a runner ASGI app.
:param app: Runner app under test.
:returns: Async context manager yielding an ``httpx.AsyncClient``.
"""
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://runner") as client:
yield client
class _FakeHarnessStream:
"""
Async context manager that yields scripted harness SSE chunks.
:param chunks: SSE chunks returned by ``aiter_text``.
:param status_code: HTTP status exposed to the runner.
"""
def __init__(self, chunks: list[str], status_code: int = 200) -> None:
"""
Store scripted stream state.
:param chunks: SSE chunks returned by ``aiter_text``.
:param status_code: HTTP status exposed to the runner.
"""
self._chunks = chunks
self.status_code = status_code
async def __aenter__(self) -> _FakeHarnessStream:
"""
Enter the fake stream context.
:returns: This fake stream.
"""
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
"""
Exit the fake stream without suppressing exceptions.
:param exc_type: Exception type from the context, if any.
:param exc: Exception value from the context, if any.
:param tb: Traceback from the context, if any.
:returns: None.
"""
del exc_type, exc, tb
async def aiter_text(self) -> AsyncIterator[str]:
"""
Yield scripted text chunks.
:returns: Async iterator of SSE chunks.
"""
for chunk in self._chunks:
yield chunk
async def _await_bg_turn_task(conv: str, *, timeout: float = 10.0) -> None:
"""Await the fire-and-forget background turn task for *conv* before draining.
The ``POST /events`` background path returns 202 before its turn task
(named ``turn-{conv}``) finishes publishing the terminal ``session.status``.
Awaiting that task by name removes the race where a status-queue drain's
timeout expires under heavy CI load before the task completes. A task that
already finished is absent from ``asyncio.all_tasks()`` (it published its
terminal status synchronously on the way out), so a ``None`` lookup is a
safe no-op.
:param conv: Session/conversation identifier, e.g. ``"conv_abc123"``.
:param timeout: Hard cap in seconds for awaiting the task.
"""
turn_task = next(
(t for t in asyncio.all_tasks() if t.get_name() == f"turn-{conv}"),
None,
)
if turn_task is not None:
await asyncio.wait_for(turn_task, timeout=timeout)
async def _drain_published_statuses(
conv: str,
*,
until: str,
timeout: float,
) -> list[str]:
"""Collect ``session.status`` values a runner published for a session.
Reads the runner's module-level per-session event queue
(``omnigent.runner.app._session_event_queues_ref``) — the same queue
the SSE ``/stream`` endpoint drains — and returns the ordered list of
``session.status`` values seen, stopping once *until* is published. This
polls the in-process queue rather than a concurrent SSE ``GET`` because
``httpx.ASGITransport`` does not interleave a streaming response with a
concurrent ``POST`` on the same client, so a live SSE subscriber would
never observe the background turn's events.
:param conv: Session/conversation identifier, e.g. ``"conv_abc123"``.
:param until: Stop once this ``session.status`` value is observed,
e.g. ``"failed"``.
:param timeout: Hard cap in seconds — if *until* never arrives the poll
gives up and returns what it saw, so a hang regression fails the
assertion instead of spinning forever.
:returns: Ordered ``session.status`` values published for *conv*.
"""
from omnigent.runner.app import _session_event_queues_ref
statuses: list[str] = []
deadline = asyncio.get_running_loop().time() + timeout
while asyncio.get_running_loop().time() < deadline:
queue = _session_event_queues_ref.get(conv)
drained = False
while queue is not None and not queue.empty():
event = queue.get_nowait()
drained = True
if isinstance(event, dict) and event.get("type") == "session.status":
status = event.get("status")
if isinstance(status, str):
statuses.append(status)
if until in statuses:
return statuses
if not drained:
# Let the background turn task make progress before re-polling.
await asyncio.sleep(0.02)
return statuses
async def _drain_failed_status_event(
conv: str,
*,
timeout: float,
) -> dict[str, Any] | None:
"""Return the first ``session.status: failed`` event a runner published.
Mirrors :func:`_drain_published_statuses` but returns the full event
dict (not just the status string) so a test can assert the carried
``error`` payload. Used to prove a SETUP-phase failure forwards its
error message on the terminal ``failed`` event instead of dropping it.
:param conv: Session/conversation identifier, e.g. ``"conv_abc123"``.
:param timeout: Hard cap in seconds; returns ``None`` if no failed
event arrives so a regression fails the assertion rather than
hanging.
:returns: The ``session.status: failed`` event dict, or ``None``.
"""
from omnigent.runner.app import _session_event_queues_ref
deadline = asyncio.get_running_loop().time() + timeout
while asyncio.get_running_loop().time() < deadline:
queue = _session_event_queues_ref.get(conv)
drained = False
while queue is not None and not queue.empty():
event = queue.get_nowait()
drained = True
if (
isinstance(event, dict)
and event.get("type") == "session.status"
and event.get("status") == "failed"
):
return event
if not drained:
await asyncio.sleep(0.02)
return None
class _FakeHarnessClient:
"""
Harness client stub exposing ``stream`` for runner proxy tests.
:param chunks: SSE chunks returned by the fake stream.
"""
def __init__(self, chunks: list[str]) -> None:
"""
Store scripted stream chunks.
:param chunks: SSE chunks returned by the fake stream.
"""
self._chunks = chunks
def stream(
self,
method: str,
url: str,
*,
json: dict[str, object],
timeout: float | None,
) -> _FakeHarnessStream:
"""
Return a fake streaming response.
:param method: HTTP method, e.g. ``"POST"``.
:param url: Harness endpoint path.
:param json: JSON body sent to the harness.
:param timeout: Request timeout.
:returns: Fake stream context manager.
"""
del method, url, json, timeout
return _FakeHarnessStream(self._chunks)
class _FakeProcessManager:
"""
Process manager stub for runner dispatch tests.
:param harness_client: Optional harness client to return.
"""
def __init__(self, harness_client: _FakeHarnessClient | None = None) -> None:
"""
Store the optional harness client.
:param harness_client: Optional harness client to return.
"""
self._harness_client = harness_client
async def get_client(
self,
conversation_id: str,
harness_name: str,
*,
env: dict[str, str] | None = None,
) -> _FakeHarnessClient:
"""
Return the configured fake harness client.
:param conversation_id: Omnigent conversation id.
:param harness_name: Harness name requested by the runner.
:param env: Optional spawn environment.
:returns: Configured fake harness client.
:raises AssertionError: If no fake client was configured.
"""
del conversation_id, harness_name, env
if self._harness_client is None:
raise AssertionError("get_client should not be called")
return self._harness_client
def mark_in_flight(self, conversation_id: str, response_id: str) -> None:
"""Reaper in-flight marker — no-op for this stub (issue #1414)."""
del conversation_id, response_id
def clear_in_flight(self, conversation_id: str) -> None:
"""Reaper in-flight clear — no-op for this stub (issue #1414)."""
del conversation_id
@pytest.fixture
async def started_manager() -> AsyncIterator[HarnessProcessManager]:
"""A real, started HarnessProcessManager with the test harness registered.
Uses a short ``/tmp/oa-rtest`` parent rather than pytest's
``tmp_path`` because UDS paths on Linux are capped at 108 chars
and the manager's per-conversation socket layout
(``<parent>/ap-<uuid32>/<conv_id>.sock``) blows past that when
nested under pytest's already-long ``/tmp/pytest-of-.../...``
tree.
Yields the started manager; on teardown, shuts it down so any
spawned subprocesses are reaped before the test ends.
"""
import shutil
import uuid
short_parent = Path(f"/tmp/oa-rtest-{uuid.uuid4().hex[:8]}")
short_parent.mkdir(mode=0o700, parents=True, exist_ok=True)
# Inject the test-only harness module into the registry. We
# mutate the dict directly per the registry's documented test-
# injection pattern; restore on teardown to avoid leaking into
# other tests.
_HARNESS_MODULES[_TEST_HARNESS_NAME] = _TEST_HARNESS_MODULE
mgr = HarnessProcessManager(tmp_parent=short_parent)
await mgr.start()
try:
yield mgr
finally:
await mgr.shutdown()
_HARNESS_MODULES.pop(_TEST_HARNESS_NAME, None)
shutil.rmtree(short_parent, ignore_errors=True)
# ── Plumbing tests (no LLM key required) ─────────────────
def test_forward_harness_response_preserves_no_body_responses() -> None:
"""204/304 harness side-channel responses must not serialize JSON null.
Returning ``JSONResponse(status_code=204, content=None)`` writes ``b"null"``
even though Uvicorn/HTTP semantics require an empty body for 204. That
manifests under uvicorn as ``Response content longer than Content-Length``.
"""
response = _forward_harness_response(httpx.Response(204, content=b""))
assert response.status_code == 204
assert response.body == b""
assert b"content-length" not in dict(response.raw_headers)
def test_forward_harness_response_preserves_json_body() -> None:
response = _forward_harness_response(httpx.Response(404, json={"error": "not_found"}))
assert response.status_code == 404
assert response.body == b'{"error":"not_found"}'
assert dict(response.raw_headers)[b"content-length"] == b"21"
@pytest.mark.asyncio
async def test_runner_post_without_manager_returns_501() -> None:
"""Scaffold-mode preserved when no manager is wired up."""
app = create_runner_app(server_client=NullServerClient()) # type: ignore[arg-type] # no process_manager → scaffold
async with _runner_test_client(app) as http:
response = await http.post(
"/v1/sessions/conv_x/events?stream=true",
json={
"type": "message",
"role": "user",
"harness": _TEST_HARNESS_NAME,
"model": "fake/model",
"content": [],
},
)
assert response.status_code == 501
assert "HarnessProcessManager" in response.json()["detail"]
@pytest.mark.asyncio
async def test_runner_resolves_harness_from_fallback_when_no_agent_id(
started_manager: HarnessProcessManager,
) -> None:
"""Without agent_id or server_base_url, runner falls back to the
test-default harness. Verifies the fallback path doesn't crash."""
app = create_runner_app(
process_manager=started_manager,
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
# No agent_id → runner falls back to "runner-test-default"
# harness. With that registered in _HARNESS_MODULES, the
# runner must spawn the harness and return its SSE stream.
# Missing LLM credentials are represented inside that stream
# as ``response.failed``, not as a runner spawn failure.
response = await http.post(
"/v1/sessions/c_fallback/events?stream=true",
json={
"type": "message",
"role": "user",
"model": "x",
"content": [{"role": "user", "content": "test"}],
},
)
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/event-stream")
assert "event: response.created" in response.text
class _RecordingProcessManager:
"""
Process manager stub that records the harness name get_client saw.
Unlike :class:`_FakeProcessManager`, this captures the resolved
harness name so a test can assert which harness the runner chose,
and signals an event when the background turn reaches dispatch.
:param captured: Dict the recorded harness name is written into
under the ``"harness"`` key.
:param reached: Event set once ``get_client`` is called.
"""
def __init__(self, captured: dict[str, str], reached: asyncio.Event) -> None:
"""
Store the capture sink and the reached-dispatch event.
:param captured: Dict the harness name is written into.
:param reached: Event set once ``get_client`` is called.
"""
self._captured = captured
self._reached = reached
async def get_client(
self,
conversation_id: str,
harness_name: str,
*,
env: dict[str, str] | None = None,
) -> _FakeHarnessClient:
"""
Record the harness name and return an empty fake harness client.
:param conversation_id: Omnigent conversation id.
:param harness_name: Harness name the runner resolved — the
value under test.
:param env: Optional spawn environment (ignored).
:returns: A fake harness client with an empty SSE stream so the
background turn completes immediately.
"""
del conversation_id, env
self._captured["harness"] = harness_name
self._reached.set()
return _FakeHarnessClient([])
def mark_in_flight(self, conversation_id: str, response_id: str) -> None:
"""Reaper in-flight marker — no-op for this stub (issue #1414)."""
del conversation_id, response_id
def clear_in_flight(self, conversation_id: str) -> None:
"""Reaper in-flight clear — no-op for this stub (issue #1414)."""
del conversation_id
@pytest.mark.asyncio
async def test_runner_resolves_agent_from_server_snapshot_when_msg_lacks_agent_id() -> None:
"""A turn-triggering message that races ahead of session assignment
arrives with no ``agent_id`` and an empty spec cache. The runner must
resolve the agent from the authoritative server snapshot
(``GET /v1/sessions/{id}``) rather than falling through to the
test-only ``runner-test-default`` harness, which would silently drop
the turn (the first-message race).
"""
conv = "conv_ondemand_race"
resolved_agent_id = "ag_resolved_from_snapshot"
resolved_harness = "runner-test-resolved"
def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Stub Omnigent server: the session snapshot carries the agent_id.
:param request: Outbound request from the runner.
:returns: Snapshot with ``agent_id`` for the session GET; benign
payloads otherwise so the background turn can proceed.
"""
if request.method == "GET" and request.url.path == f"/v1/sessions/{conv}":
return httpx.Response(200, json={"id": conv, "agent_id": resolved_agent_id})
if request.url.path.endswith("/items"):
return httpx.Response(
200,
json={
"object": "list",
"data": [],
"first_id": None,
"last_id": None,
"has_more": False,
},
)
return httpx.Response(200, json={})
server_client = httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
)
async def _snapshot_spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Resolve the spec for the agent_id read from the snapshot.
:param agent_id: Agent id the runner resolved. MUST equal the
snapshot's agent_id — the message body carried none, so any
other value means the on-demand snapshot path didn't run.
:param session_id: Session id (unused).
:returns: A minimal spec whose harness is ``resolved_harness``.
"""
# The agent_id can only come from the server snapshot here — the
# POST body below omits it. If this fires with a different value
# (or not at all), the on-demand resolution path is broken.
assert agent_id == resolved_agent_id
return AgentSpec(
spec_version=1,
name="ondemand-agent",
executor=ExecutorSpec(type="omnigent", config={"harness": resolved_harness}),
)
captured: dict[str, str] = {}
reached_dispatch = asyncio.Event()
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_RecordingProcessManager(captured, reached_dispatch),
),
spec_resolver=_snapshot_spec_resolver,
server_client=server_client,
)
try:
async with _runner_test_client(app) as http:
response = await http.post(
# No ``?stream=true`` → background turn, the production
# path the Omnigent server uses to forward session messages.
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"model": "x",
"content": [{"role": "user", "content": "hi"}],
# No agent_id — this is the race condition under test.
},
)
# Background turn accepted; dispatch happens asynchronously.
assert response.status_code == 202
await asyncio.wait_for(reached_dispatch.wait(), timeout=10.0)
finally:
await server_client.aclose()
# The harness came from the snapshot-resolved spec, proving the
# runner fetched agent_id from the server when the message lacked it.
# Without the fix this is "runner-test-default" (the fallback) and the
# real turn never dispatches.
assert captured["harness"] == resolved_harness
class _ContentCapturingProcessManager:
"""
Process manager stub that captures the body sent to the harness.
Returns a harness client whose ``stream`` records the JSON body
(which carries the turn's ``content`` history) into a shared sink
and yields an empty SSE stream so the background turn completes
immediately.
:param captured: Dict the harness request body is written into
under the ``"body"`` key.
:param reached: Event set once the harness stream is opened.
"""
def __init__(self, captured: dict[str, Any], reached: asyncio.Event) -> None:
"""
Store the capture sink and the reached-dispatch event.
:param captured: Dict the harness request body is written into.
:param reached: Event set once the harness stream is opened.
"""
self._captured = captured
self._reached = reached
async def get_client(
self,
conversation_id: str,
harness_name: str,
*,
env: dict[str, str] | None = None,
) -> _ContentCapturingHarnessClient:
"""
Return a harness client that records the body it is sent.
:param conversation_id: Omnigent conversation id (unused).
:param harness_name: Harness name the runner resolved (unused).
:param env: Optional spawn environment (unused).
:returns: A capturing harness client.
"""
del conversation_id, harness_name, env
return _ContentCapturingHarnessClient(self._captured, self._reached)
def mark_in_flight(self, conversation_id: str, response_id: str) -> None:
"""Reaper in-flight marker — no-op for this stub (issue #1414)."""
del conversation_id, response_id
def clear_in_flight(self, conversation_id: str) -> None:
"""Reaper in-flight clear — no-op for this stub (issue #1414)."""
del conversation_id
class _ContentCapturingHarnessClient:
"""
Harness client stub that records the JSON body of each stream.
:param captured: Dict the request body is written into under
``"body"``.
:param reached: Event set once ``stream`` is invoked.
"""
def __init__(self, captured: dict[str, Any], reached: asyncio.Event) -> None:
"""
Store the capture sink and reached event.
:param captured: Dict the request body is written into.
:param reached: Event set once ``stream`` is invoked.
"""
self._captured = captured
self._reached = reached
def stream(
self,
method: str,
url: str,
*,
json: dict[str, Any],
timeout: float | None,
) -> _FakeHarnessStream:
"""
Record the body and return an empty SSE stream.
:param method: HTTP method (unused).
:param url: Harness endpoint path (unused).
:param json: JSON body sent to the harness — captured here.
:param timeout: Request timeout (unused).
:returns: An empty fake stream so the turn completes at once.
"""
del method, url, timeout
self._captured["body"] = json
self._reached.set()
return _FakeHarnessStream([])
@pytest.mark.asyncio
async def test_runner_reloads_full_history_on_cold_cache_after_restart() -> None:
"""A message to a cold session reloads prior history, not just itself.
Regression for an agent (e.g. nessie) losing all chat context after a
server/runner restart. On restart the runner's in-memory
``_session_histories`` cache is empty; the old code seeded it with ONLY
the incoming message (``setdefault(conv, []).append(...)``), so the
harness ran the turn with no prior context. This is acute for the
claude-sdk harness, which on a cold SDK session replays the in-memory
history verbatim as the prompt — a one-message cache erased the whole
conversation.
The fix rehydrates the full history from the store on the first touch of
a conversation. The stub server models invariant I1 (persist-before-
forward): its ``GET /items`` returns the prior turns AND the just-posted
message (``item_3``), and the forwarded body carries
``persisted_item_id="item_3"`` — so the reload drops that exact item by
id and appends the runner's copy, proving no duplication.
"""
from omnigent.runner import app as runner_app
conv = "conv_restart_history_reload"
prior_user = "what is the capital of France?"
prior_assistant = "Paris."
new_user = "and of Germany?"
def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Stub Omnigent server: snapshot + full persisted history on ``/items``.
:param request: Outbound request from the runner.
:returns: Snapshot for the session GET; the persisted history
(prior turns + the new message, per invariant I1) for
``/items``; benign payloads otherwise.
"""
if request.method == "GET" and request.url.path == f"/v1/sessions/{conv}":
return httpx.Response(200, json={"id": conv, "agent_id": "ag_restart"})
if request.url.path.endswith("/items"):
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "item_1",
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": prior_user}],
},
{
"id": "item_2",
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": prior_assistant}],
},
# Persist-before-forward (I1): the new message is
# already in the store when the runner reloads.
{
"id": "item_3",
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": new_user}],
},
],
"first_id": "item_1",
"last_id": "item_3",
"has_more": False,
},
)
return httpx.Response(200, json={})
server_client = httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
)
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Resolve a minimal spec for the restarted session.
:param agent_id: Agent id resolved from the snapshot (unused).
:param session_id: Session id (unused).
:returns: A minimal spec on a benign test harness.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="restart-agent",
executor=ExecutorSpec(
type="omnigent",
config={"harness": "runner-test-resolved"},
),
)
captured: dict[str, Any] = {}
reached = asyncio.Event()
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_ContentCapturingProcessManager(captured, reached),
),
spec_resolver=_spec_resolver,
server_client=server_client,
)
# Simulate a fresh runner process: no cached history for this conv.
runner_app._session_histories_ref.pop(conv, None)
try:
async with _runner_test_client(app) as http:
response = await http.post(
# No ``?stream=true`` → background turn, the production path
# the Omnigent server uses to forward session messages.
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"model": "x",
# The store id the Omnigent server persisted for this turn
# (matches ``item_3`` from the stub ``/items``), so the
# cold-cache reload drops that exact item and the dedup
# fires (no duplicate).
"persisted_item_id": "item_3",
"content": [{"type": "input_text", "text": new_user}],
},
)
assert response.status_code == 202
await asyncio.wait_for(reached.wait(), timeout=10.0)
finally:
await server_client.aclose()
runner_app._session_histories_ref.pop(conv, None)
content = captured["body"]["content"]
texts = [
block.get("text")
for item in content
if isinstance(item, dict)
for block in item.get("content", [])
if isinstance(block, dict)
]
# Prior context survived the restart...
assert prior_user in texts, f"prior user turn missing from reloaded history: {texts}"
assert prior_assistant in texts, f"prior assistant turn missing: {texts}"
# ...and the new message is present exactly once (reload didn't dup it).
assert texts.count(new_user) == 1, f"new message not delivered exactly once: {texts}"
# The full 3-item history reached the harness, not just the new message.
assert len(content) == 3, f"expected full history, got {len(content)} items: {content}"
@pytest.mark.asyncio
async def test_runner_cold_cache_appends_message_when_store_lacks_it() -> None:
"""A cold-cache message NOT yet in the store is appended, not dropped.
Not every forward is persist-before-forward (invariant I1): native-
terminal web injections (claude-native/codex-native) are forwarded
WITHOUT persisting first, so a fresh ``GET /items`` returns the prior
turns but NOT the just-posted message. If the cold-cache reload simply
overwrote ``_session_histories`` with that load, the new input would be
dropped — and the native executor, which types only the LATEST user
message into its pane, would inject stale text.
This drives the cold path with a stub server whose history reload
excludes the new message and asserts the harness still receives it,
appended as the latest turn, with prior context preserved.
"""
from omnigent.runner import app as runner_app
conv = "conv_cold_cache_append"
prior_user = "first question"
prior_assistant = "first answer"
new_user = "second question not yet persisted"
def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Stub Omnigent server: history reload that does NOT include the new message.
:param request: Outbound request from the runner.
:returns: Snapshot for the session GET; prior turns only (no
new message, modeling a forward-without-persist) for
``/items``; benign payloads otherwise.
"""
if request.method == "GET" and request.url.path == f"/v1/sessions/{conv}":
return httpx.Response(200, json={"id": conv, "agent_id": "ag_cold"})
if request.url.path.endswith("/items"):
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "item_1",
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": prior_user}],
},
{
"id": "item_2",
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": prior_assistant}],
},
# No item for ``new_user`` — the forward did not
# persist it before reaching the runner.
],
"first_id": "item_1",
"last_id": "item_2",
"has_more": False,
},
)
return httpx.Response(200, json={})
server_client = httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
)
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Resolve a minimal spec for the session.
:param agent_id: Agent id resolved from the snapshot (unused).
:param session_id: Session id (unused).
:returns: A minimal spec on a benign test harness.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="cold-cache-agent",
executor=ExecutorSpec(
type="omnigent",
config={"harness": "runner-test-resolved"},
),
)
captured: dict[str, Any] = {}
reached = asyncio.Event()
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_ContentCapturingProcessManager(captured, reached),
),
spec_resolver=_spec_resolver,
server_client=server_client,
)
runner_app._session_histories_ref.pop(conv, None)
try:
async with _runner_test_client(app) as http:
response = await http.post(
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"model": "x",
# No ``persisted_item_id``: the native-terminal forward
# skipped persist-before-forward, so there's nothing in the
# store to drop — the runner must append, not dedup.
"content": [{"type": "input_text", "text": new_user}],
},
)
assert response.status_code == 202
await asyncio.wait_for(reached.wait(), timeout=10.0)
finally:
await server_client.aclose()
runner_app._session_histories_ref.pop(conv, None)
content = captured["body"]["content"]
texts = [
block.get("text")
for item in content
if isinstance(item, dict)
for block in item.get("content", [])
if isinstance(block, dict)
]
# Prior context preserved, and the not-yet-persisted message was
# appended (not dropped) as the latest turn — present exactly once.
assert texts == [prior_user, prior_assistant, new_user], (
f"expected prior history + appended new message, got {texts}"
)
@pytest.mark.asyncio
async def test_runner_cold_cache_keeps_trailing_user_when_no_persisted_id() -> None:
"""A real trailing user message is kept when no ``persisted_item_id`` is sent.
Regression for the id-based dedup replacing the old role heuristic. The
earlier fix unconditionally popped a trailing *user* item, assuming it was
always this turn's persisted input. That's wrong when the forward did NOT
persist-before-forward AND the store legitimately ends with a user
message — e.g. a crash mid-turn where the prior user prompt was persisted
but its assistant reply never was, or a native-terminal injection. Popping
there deletes real history.
With id-based dedup, no ``persisted_item_id`` means nothing is dropped: the
real trailing user message survives and the new message is appended.
"""
from omnigent.runner import app as runner_app
conv = "conv_cold_cache_keep_user"
# A prior user prompt whose assistant reply was never persisted (e.g. the
# runner crashed mid-turn), so the store ends on a USER message.
prior_user = "prompt whose reply was lost to a crash"
new_user = "follow-up not persisted before forward"
def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Stub Omnigent server: history reload ending on a real prior user message.
:param request: Outbound request from the runner.
:returns: Snapshot for the session GET; a single prior user item
(no assistant reply, no new message) for ``/items``; benign
payloads otherwise.
"""
if request.method == "GET" and request.url.path == f"/v1/sessions/{conv}":
return httpx.Response(200, json={"id": conv, "agent_id": "ag_keep"})
if request.url.path.endswith("/items"):
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "item_1",
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": prior_user}],
},
],
"first_id": "item_1",
"last_id": "item_1",
"has_more": False,
},
)
return httpx.Response(200, json={})
server_client = httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
)
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Resolve a minimal spec for the session.
:param agent_id: Agent id (unused).
:param session_id: Session id (unused).
:returns: A minimal spec on a benign test harness.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="keep-user-agent",
executor=ExecutorSpec(
type="omnigent",
config={"harness": "runner-test-resolved"},
),
)
captured: dict[str, Any] = {}
reached = asyncio.Event()
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_ContentCapturingProcessManager(captured, reached),
),
spec_resolver=_spec_resolver,
server_client=server_client,
)
runner_app._session_histories_ref.pop(conv, None)
try:
async with _runner_test_client(app) as http:
response = await http.post(
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"model": "x",
# No ``persisted_item_id`` → the trailing user item is NOT
# this turn's input, so it must be kept.
"content": [{"type": "input_text", "text": new_user}],
},
)
assert response.status_code == 202
await asyncio.wait_for(reached.wait(), timeout=10.0)
finally:
await server_client.aclose()
runner_app._session_histories_ref.pop(conv, None)
content = captured["body"]["content"]
texts = [
block.get("text")
for item in content
if isinstance(item, dict)
for block in item.get("content", [])
if isinstance(block, dict)
]
# The real prior user message survived (old role heuristic would drop it)
# and the new message was appended.
assert texts == [prior_user, new_user], f"trailing user message must be preserved, got {texts}"
@pytest.mark.asyncio
async def test_runner_cold_cache_uses_resolved_message_not_stored_file_id() -> None:
"""Cold-cache reload of a media turn uses the resolved block, not the store copy.
The server persists the PRE-resolution body (``file_id`` blocks) and the
runner resolves ``file_id`` → ``image_url`` itself. So on a cold cache the
``GET /items`` tail is the just-posted message in its *unresolved* form,
while ``message_body`` is the *resolved* form. Any content-equality dedup
would never match (the blocks differ), which would both duplicate the
message AND leave an unresolved ``file_id`` block in the history forwarded
to the harness.
The fix drops the persisted item by id (``persisted_item_id``, forwarded by
the server) and appends the runner-resolved message, so the harness sees
exactly one, fully resolved copy regardless of the content mismatch.
"""
from omnigent.runner import app as runner_app
conv = "conv_cold_cache_media"
prior_user = "earlier question"
prior_assistant = "earlier answer"
prompt_text = "what is this?"
def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Stub Omnigent server: file resolution, snapshot, and a history reload
whose tail is the UNRESOLVED (``file_id``) copy of this message.
:param request: Outbound request from the runner.
:returns: File metadata / bytes for resolution; the persisted
history (ending with the unresolved media message, per I1)
for ``/items``; snapshot for the session GET.
"""
path = request.url.path
if path == f"/v1/sessions/{conv}":
return httpx.Response(200, json={"id": conv, "agent_id": "ag_media"})
if path.endswith("/resources/files/file_img/content"):
return httpx.Response(200, content=b"png-bytes", headers={"content-type": "image/png"})
if path.endswith("/resources/files/file_img"):
return httpx.Response(
200,
json={"id": "file_img", "filename": "photo.png", "content_type": "image/png"},
)
if path.endswith("/items"):
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "item_1",
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": prior_user}],
},
{
"id": "item_2",
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": prior_assistant}],
},
# I1 persisted THIS message — but in its unresolved
# (file_id) form, exactly as received.
{
"id": "item_3",
"type": "message",
"role": "user",
"content": [
{
"type": "input_image",
"file_id": "file_img",
"filename": "photo.png",
},
{"type": "input_text", "text": prompt_text},
],
},
],
"first_id": "item_1",
"last_id": "item_3",
"has_more": False,
},
)
return httpx.Response(200, json={})
server_client = httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
)
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Resolve a minimal spec for the session.
:param agent_id: Agent id (unused).
:param session_id: Session id (unused).
:returns: A minimal spec on a benign test harness.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="media-agent",
executor=ExecutorSpec(
type="omnigent",
config={"harness": "runner-test-resolved"},
),
)
captured: dict[str, Any] = {}
reached = asyncio.Event()
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_ContentCapturingProcessManager(captured, reached),
),
spec_resolver=_spec_resolver,
server_client=server_client,
)
runner_app._session_histories_ref.pop(conv, None)
try:
async with _runner_test_client(app) as http:
response = await http.post(
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"model": "x",
# Server forwards the id of the (unresolved) item it
# persisted; the runner drops it from the reload by id.
"persisted_item_id": "item_3",
"content": [
{"type": "input_image", "file_id": "file_img", "filename": "photo.png"},
{"type": "input_text", "text": prompt_text},
],
},
)
assert response.status_code == 202
await asyncio.wait_for(reached.wait(), timeout=10.0)
finally:
await server_client.aclose()
runner_app._session_histories_ref.pop(conv, None)
content = captured["body"]["content"]
# Prior context + the media turn = three items, not four (no duplicate).
assert len(content) == 3, f"expected no duplicate of the media message: {content}"
image_msg = content[-1]
image_block = image_msg["content"][0]
# The stored unresolved copy was dropped; the resolved block is used —
# no ``file_id`` leaks to the harness.
assert "file_id" not in image_block, f"unresolved file_id leaked to harness: {image_block}"
assert image_block.get("image_url", "").startswith("data:image/png;base64,"), (
f"expected resolved image_url block, got {image_block}"
)
# No unresolved file_id anywhere in the forwarded history.
flat = json.dumps(content)
assert "file_id" not in flat, f"unresolved file_id present in forwarded history: {flat}"
@pytest.mark.asyncio
async def test_runner_post_returns_503_when_spec_resolver_fails(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Spec resolver failures are surfaced as structured 503 errors.
:param caplog: Pytest log capture, used to confirm the raw cause is
logged server-side (the other half of the log-and-genericize
contract).
:returns: None.
"""
async def _failing_spec_resolver(
agent_id: str, session_id: str | None = None
) -> AgentSpec | None:
"""
Raise the resolver failure under test.
:param agent_id: Agent id requested by the runner.
:returns: Never returns.
:raises RuntimeError: Always.
"""
raise RuntimeError(f"spec resolver unavailable for {agent_id}")
app = create_runner_app(
process_manager=cast(HarnessProcessManager, _FakeProcessManager()),
spec_resolver=_failing_spec_resolver,
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
with caplog.at_level(logging.WARNING, logger="omnigent.runner.app"):
response = await http.post(
"/v1/sessions/conv_spec_resolver_failed/events?stream=true",
json={
"type": "message",
"role": "user",
"agent_id": "ag_missing",
"model": "x",
"content": [],
},
)
assert response.status_code == 503
body = response.json()
# The structured error slug is preserved for the caller; the detail is a
# fixed client-safe string. The raw resolver exception text must not leak
# into the HTTP body (it is logged on the runner instead).
assert body["error"] == "spec_resolver_failed"
assert body["detail"] == "Request failed on the runner; see runner logs for details."
assert "spec resolver unavailable" not in body["detail"]
# The other half of the contract: the raw cause IS logged for operators.
# If this fails, log-and-genericize logged nothing and the detail is the
# only record of the failure — defeating the diagnostic path.
assert "spec resolver unavailable for ag_missing" in caplog.text
@pytest.mark.asyncio
async def test_runner_stream_emits_failed_when_tool_spec_resolver_fails() -> None:
"""Streaming spec resolver failures emit ``response.failed`` SSE.
:returns: None.
"""
chunks = [
(
"event: response.created\ndata: "
'{"type":"response.created","response":{"id":"resp_1"}}\n\n'
),
(
'event: response.output_item.done\ndata: {"type":"response.output_item.done",'
'"item":{"type":"function_call","status":"action_required",'
'"name":"sys_os_read","call_id":"call_1","arguments":"{}"}}\n\n'
),
]
async def _failing_spec_resolver(
agent_id: str, session_id: str | None = None
) -> AgentSpec | None:
"""
Raise during local tool dispatch spec resolution.
:param agent_id: Agent id requested by the runner.
:returns: Never returns.
:raises RuntimeError: Always.
"""
raise RuntimeError(f"stream spec resolver unavailable for {agent_id}")
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_FakeProcessManager(_FakeHarnessClient(chunks)),
),
spec_resolver=_failing_spec_resolver,
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
response = await http.post(
"/v1/sessions/conv_stream_spec_resolver_failed/events?stream=true",
json={
"type": "message",
"role": "user",
"harness": _TEST_HARNESS_NAME,
"agent_id": "ag_stream",
"model": "x",
"content": [],
},
)
assert response.status_code == 200
assert "event: response.created" in response.text
assert "event: response.failed" in response.text
# The exception class (a safe, generic label) is still surfaced as the
# error ``type``; the failure message is a fixed client-safe string. The
# raw resolver exception text must not leak into the SSE stream (it is
# logged on the runner instead).
assert "RuntimeError" in response.text
assert "Failed to resolve the agent spec for this turn." in response.text
assert "stream spec resolver unavailable for ag_stream" not in response.text
def test_build_spawn_env_applies_model_override(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A per-session ``/model`` override overrides ``HARNESS_<H>_MODEL``.
Regression: ``/model`` was recorded in the session readout but the turn
still used the provider/catalog default, because the SDK harnesses take
their model from the spawn-env env var (which only the native CLIs'
``--model`` path honored). The override must override the baked-in
``HARNESS_CLAUDE_SDK_MODEL`` so the switch actually takes effect.
:param tmp_path: Pytest temp dir for an isolated provider config.
:param monkeypatch: Pytest monkeypatch fixture.
"""
monkeypatch.setenv("OMNIGENT_CONFIG_HOME", str(tmp_path))
monkeypatch.setenv("OMNIGENT_DISABLE_KEYRING", "1")
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
(tmp_path / "config.yaml").write_text(
"providers:\n"
" anthropic:\n"
" kind: key\n"
" default: true\n"
" anthropic:\n"
" base_url: https://api.anthropic.com\n"
" api_key: $ANTHROPIC_API_KEY\n"
)
spec = AgentSpec(
spec_version=1,
name="x",
executor=ExecutorSpec(type="omnigent", config={"harness": "claude-sdk"}),
)
base = _build_spawn_env_from_spec(spec, "claude-sdk")
overridden = _build_spawn_env_from_spec(spec, "claude-sdk", model_override="claude-sonnet-4-6")
assert base is not None and overridden is not None
# Baseline uses the provider/catalog default (not the override) …
assert base["HARNESS_CLAUDE_SDK_MODEL"] != "claude-sonnet-4-6"
# … and the override wins, landing in the model env var the SDK reads.
assert overridden["HARNESS_CLAUDE_SDK_MODEL"] == "claude-sonnet-4-6"
@pytest.mark.asyncio
async def test_resolve_harness_config_applies_harness_override(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A per-session ``harness_override`` replaces the spec's brain harness.
The web UI's new-chat picker persists the override on the session and
the server forwards it in the message body; the runner must spawn THAT
harness (with its spawn-env shape), not the spec's declared one — else
the snapshot would claim "pi" while claude-sdk actually runs.
:param tmp_path: Pytest temp dir for an isolated provider config.
:param monkeypatch: Pytest monkeypatch fixture.
"""
monkeypatch.setenv("OMNIGENT_CONFIG_HOME", str(tmp_path))
monkeypatch.setenv("OMNIGENT_DISABLE_KEYRING", "1")
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
(tmp_path / "config.yaml").write_text(
"providers:\n"
" anthropic:\n"
" kind: key\n"
" default: true\n"
" anthropic:\n"
" base_url: https://api.anthropic.com\n"
" api_key: $ANTHROPIC_API_KEY\n"
)
spec = AgentSpec(
spec_version=1,
name="x",
executor=ExecutorSpec(type="omnigent", config={"harness": "claude-sdk"}),
)
async def _resolver(_agent_id: str, _session_id: str | None) -> AgentSpec:
return spec
# Baseline: no override resolves the spec's declared harness.
harness, spawn_env = await _resolve_harness_config(
agent_id="ag_x", spec_resolver=_resolver, session_id="conv_x"
)
assert harness == "claude-sdk"
assert spawn_env is not None and "HARNESS_CLAUDE_SDK_MODEL" in spawn_env
# Override: the harness AND its spawn-env shape follow the override.
harness, spawn_env = await _resolve_harness_config(
agent_id="ag_x",
spec_resolver=_resolver,
session_id="conv_x",
harness_override="pi",
)
assert harness == "pi", (
f"harness_override='pi' resolved to {harness!r} — the override "
f"was ignored and the spec's declared harness won."
)
# The spawn env must be built FOR the overridden harness (pi env keys),
# not the spec's claude-sdk shape — a claude env here means the harness
# name and env were resolved inconsistently.
assert spawn_env is not None and "HARNESS_PI_MODEL" in spawn_env, (
f"Expected a pi spawn-env; got keys {sorted(spawn_env or {})!r}"
)
@pytest.mark.asyncio
async def test_runner_background_turn_emits_failed_when_spawn_env_build_raises(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A spawn-env build failure must end the turn, never hang on "running".
Regression for the silent-hang failure mode: when
``_build_claude_sdk_spawn_env`` raises (e.g. a generic provider routed
to the claude-sdk harness has no resolvable model, raising
``OmnigentError``), the setup phase of ``_run_turn_bg`` failed before
the streaming block's own error handling. Because the background turn
task's only done-callback was ``_background_tasks.discard``, the
exception was swallowed: ``_active_turns`` stayed set and no terminal
``session.status`` event was published, so the REPL spun on "working"
forever with no output.
The fix wraps the setup phase so any pre-stream exception routes through
``_on_proxy_stream_end``, which clears the active turn and publishes
``session.status: failed``. This test drives the background-turn path
(no ``?stream=true`` — the production path the Omnigent server uses) and
asserts the ``failed`` status reaches the session SSE stream.
:param monkeypatch: pytest fixture used to force the spawn-env build to
raise the same error class the no-model provider path produces.
"""
from omnigent.errors import ErrorCode, OmnigentError
conv = "conv_spawn_env_build_raises"
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Return a claude-sdk spec so the runner takes the claude-sdk
spawn-env builder path.
:param agent_id: Agent id requested by the runner (unused).
:param session_id: Session id (unused).
:returns: A minimal claude-sdk spec.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="claude-sdk-agent",
executor=ExecutorSpec(type="omnigent", config={"harness": "claude-sdk"}),
)
def _raising_build(spec: object, *, workdir: object = None) -> dict[str, str]:
"""
Stand in for ``_build_claude_sdk_spawn_env`` and fail the way the
no-model generic-provider path does.
:param spec: The agent spec (unused).
:param workdir: Bundle workdir (unused).
:returns: Never returns.
:raises OmnigentError: Always — mirrors the no-model provider error.
"""
del spec, workdir
raise OmnigentError(
"No model resolved for the 'claude-sdk' harness on a generic provider.",
code=ErrorCode.INVALID_INPUT,
)
# ``_build_spawn_env_from_spec`` imports this from workflow at call time,
# so patching the workflow attribute reaches the runner's call site.
monkeypatch.setattr(
"omnigent.runtime.workflow._build_claude_sdk_spawn_env",
_raising_build,
)
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_FakeProcessManager(_FakeHarnessClient([])),
),
spec_resolver=_spec_resolver,
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
response = await http.post(
# No ``?stream=true`` → background turn (the production path the
# Omnigent server uses to forward session messages).
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"agent_id": "ag_claude_sdk",
"model": "x",
"content": [{"role": "user", "content": "hi"}],
},
)
assert response.status_code == 202
await _await_bg_turn_task(conv)
statuses = await _drain_published_statuses(conv, until="failed", timeout=2.0)
# The turn published "running" then "failed" — it reached a terminal
# state and cleared. Without the fix, the setup-phase OmnigentError is
# swallowed: only "running" is published, ``_active_turns`` stays set, and
# the session hangs on "working" forever (the silent-hang regression).
assert statuses == ["running", "failed"]
@pytest.mark.asyncio
async def test_runner_failed_status_carries_setup_error_message(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A SETUP-phase failure forwards its error message on the ``failed`` event.
Regression for the silent-REPL failure mode: ending the turn on a
pre-stream error stopped the spinner but rendered no text, because the
runner published a bare ``session.status: failed`` with no error
detail — the message never left the runner. This drives the same
spawn-env-build failure as
:func:`test_runner_background_turn_emits_failed_when_spawn_env_build_raises`
and asserts the published ``failed`` event now carries the normalized
``{code, message}`` error so Omnigent and the REPL can render it.
:param monkeypatch: pytest fixture used to force the spawn-env build to
raise the no-model provider error.
"""
from omnigent.errors import ErrorCode, OmnigentError
conv = "conv_failed_status_carries_error"
raised_message = "No model resolved for the 'claude-sdk' harness on a generic provider."
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Return a claude-sdk spec so the spawn-env builder path is taken.
:param agent_id: Agent id (unused).
:param session_id: Session id (unused).
:returns: A minimal claude-sdk spec.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="claude-sdk-agent",
executor=ExecutorSpec(type="omnigent", config={"harness": "claude-sdk"}),
)
def _raising_build(spec: object, *, workdir: object = None) -> dict[str, str]:
"""
Fail the spawn-env build the way the no-model provider path does.
:param spec: The agent spec (unused).
:param workdir: Bundle workdir (unused).
:returns: Never returns.
:raises OmnigentError: Always.
"""
del spec, workdir
raise OmnigentError(raised_message, code=ErrorCode.INVALID_INPUT)
monkeypatch.setattr(
"omnigent.runtime.workflow._build_claude_sdk_spawn_env",
_raising_build,
)
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_FakeProcessManager(_FakeHarnessClient([])),
),
spec_resolver=_spec_resolver,
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
response = await http.post(
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"agent_id": "ag_claude_sdk",
"model": "x",
"content": [{"role": "user", "content": "hi"}],
},
)
assert response.status_code == 202
await _await_bg_turn_task(conv)
failed_event = await _drain_failed_status_event(conv, timeout=2.0)
# The failed event must carry the real setup error message — not a
# bare status. Without the fix ``error`` is absent and the REPL
# renders nothing.
assert failed_event is not None
error = failed_event.get("error")
assert isinstance(error, dict)
# The raised OmnigentError message is wrapped as "turn setup
# failed: <message>" by _run_turn_bg's setup-phase handler.
assert raised_message in error["message"]
# Normalized shape always has a code so the wire ErrorDetail validates.
assert error["code"]
# ── Harness-stream failure → terminal session.status ────
async def _drain_status_events(
conv: str,
*,
until: str,
timeout: float,
) -> list[dict[str, Any]]:
"""Collect full ``session.status`` events a runner published for a session.
Like :func:`_drain_published_statuses` but returns the full event dicts,
so one drain can assert both the status order and the carried ``error``
payload — the queue is consumed by reading, so a test cannot drain twice.
:param conv: Session/conversation identifier, e.g. ``"conv_abc123"``.
:param until: Stop once this ``session.status`` value is observed,
e.g. ``"failed"``.
:param timeout: Hard cap in seconds — if *until* never arrives the poll
gives up and returns what it saw, so a hang regression fails the
assertion instead of spinning forever.
:returns: Ordered ``session.status`` event dicts published for *conv*.
"""
from omnigent.runner.app import _session_event_queues_ref
events: list[dict[str, Any]] = []
deadline = asyncio.get_running_loop().time() + timeout
while asyncio.get_running_loop().time() < deadline:
queue = _session_event_queues_ref.get(conv)
drained = False
while queue is not None and not queue.empty():
event = queue.get_nowait()
drained = True
if isinstance(event, dict) and event.get("type") == "session.status":
events.append(event)
if any(event.get("status") == until for event in events):
return events
if not drained:
# Let the background turn task make progress before re-polling.
await asyncio.sleep(0.02)
return events
_STREAM_FAILURE_MESSAGE = "harness turn failed: executor stream disconnected"
_SSE_RESPONSE_CREATED = (
'event: response.created\ndata: {"type":"response.created","response":{"id":"resp_sf_1"}}\n\n'
)
_SSE_RESPONSE_FAILED = (
"event: response.failed\ndata: "
'{"type":"response.failed","response":{"status":"failed"},'
f'"error":{{"message":"{_STREAM_FAILURE_MESSAGE}","code":"executor_error"}}}}\n\n'
)
_SSE_RESPONSE_COMPLETED = (
"event: response.completed\ndata: "
'{"type":"response.completed","response":{"id":"resp_sf_1","status":"completed"}}\n\n'
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("harness", "frames", "until", "expected_statuses"),
[
pytest.param(
"codex-native",
[_SSE_RESPONSE_CREATED, _SSE_RESPONSE_FAILED],
"failed",
["running", "failed"],
id="codex-native-failed-stream",
),
pytest.param(
_TEST_HARNESS_NAME,
[_SSE_RESPONSE_CREATED, _SSE_RESPONSE_FAILED],
"failed",
["running", "failed"],
id="subprocess-harness-failed-stream",
),
pytest.param(
_TEST_HARNESS_NAME,
[_SSE_RESPONSE_CREATED, _SSE_RESPONSE_FAILED, _SSE_RESPONSE_COMPLETED],
"idle",
["running", "idle"],
id="failure-superseded-by-completion",
),
],
)
async def test_runner_publishes_terminal_failed_when_harness_stream_fails(
harness: str,
frames: list[str],
until: str,
expected_statuses: list[str],
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""A harness stream that ends after ``response.failed`` publishes ``failed``.
Regression for the stuck working indicator in codex-native sessions:
when a turn failed inside the harness, the scaffold emitted
``response.failed`` on the SSE stream, but the runner's proxy relay
called ``_on_proxy_stream_end`` with no error, so the turn ended with
``session.status: idle``. For codex-native sessions ``idle`` is
suppressed (the Codex app-server forwarder owns that edge — and posts
nothing when Codex never started a turn), so NO terminal status was
published at all: the web UI showed the error block yet spun on
"working" forever. For subprocess harnesses the terminal edge was
published with the wrong value (``idle`` instead of ``failed``).
A ``response.completed`` after an earlier in-stream ``response.failed``
supersedes the failure: the turn ended successfully, so the terminal
edge must be ``idle``.
:param harness: Harness name baked into the resolved spec, driving
the runner's native-vs-subprocess status-edge policy.
:param frames: Scripted harness SSE frames for the turn stream.
:param until: Terminal ``session.status`` value the drain waits for.
:param expected_statuses: Exact ordered ``session.status`` values the
runner must publish for the turn.
:param monkeypatch: Pytest fixture, used to isolate the codex-native
bridge directory.
:param tmp_path: Pytest temp dir receiving the isolated bridge files.
"""
conv = f"conv_stream_failed_{until}_{harness.replace('-', '_')}"
# Keep the codex-native pre-turn bridge writes (write_mcp_bridge_config)
# out of the real ``~/.omnigent/codex-native`` tree. The module documents
# this monkeypatch as the supported test isolation point.
monkeypatch.setattr("omnigent.codex_native_bridge._BRIDGE_ROOT", tmp_path)
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""
Return a spec pinned to the parametrized harness.
:param agent_id: Agent id requested by the runner (unused).
:param session_id: Session id (unused).
:returns: A minimal spec for the parametrized harness.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="stream-fail-agent",
executor=ExecutorSpec(type="omnigent", config={"harness": harness}),
)
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_FakeProcessManager(_FakeHarnessClient(frames)),
),
spec_resolver=_spec_resolver,
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
response = await http.post(
# No ``?stream=true`` → background turn (the production path the
# Omnigent server uses to forward session messages).
f"/v1/sessions/{conv}/events",
json={
"type": "message",
"role": "user",
"agent_id": "ag_stream_fail",
"model": "x",
"content": [{"role": "user", "content": "hi"}],
},
)
assert response.status_code == 202
# Await the background turn task directly so we know it has completed
# (and published its terminal status) before draining — the same race
# guard the sibling failed-status tests use.
await _await_bg_turn_task(conv)
events = await _drain_status_events(conv, until=until, timeout=2.0)
statuses = [event.get("status") for event in events]
# The turn must reach the parametrized terminal state. Without the fix,
# the failed-stream cases regress in two distinct ways: codex-native
# publishes only ["running"] (idle is suppressed, so the drain times out
# and the working indicator hangs), and the subprocess harness publishes
# ["running", "idle"] (the wrong terminal edge).
assert statuses == expected_statuses, (
f"Expected session.status sequence {expected_statuses}, got {statuses}. "
f"A missing terminal value means the turn never cleared (stuck working "
f"indicator); 'idle' in place of 'failed' means the in-stream "
f"response.failed was dropped at stream end."
)
if until == "failed":
error = events[-1].get("error")
# The terminal failed edge must carry the harness's real error so
# clients can render it — a bare ``failed`` with no payload would
# clear the spinner but tell the user nothing.
assert isinstance(error, dict)
assert _STREAM_FAILURE_MESSAGE in error["message"]
# ── Runner-local OS env dispatch ────────────────────────
@pytest.mark.asyncio
async def test_runner_os_env_tools_use_agent_spec_cwd() -> None:
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
from omnigent.runner.tool_dispatch import _execute_os_env_tool
from omnigent.spec.types import AgentSpec
with tempfile.TemporaryDirectory() as td:
root = Path(td)
spec = AgentSpec(
spec_version=1,
os_env=OSEnvSpec(
type="caller_process",
cwd=str(root),
sandbox=OSEnvSandboxSpec(type="none"),
),
)
write = await _execute_os_env_tool(
"sys_os_write",
{"path": "note.txt", "content": "hello\nplanet\n"},
agent_spec=spec,
conversation_id="conv_runner_os_env_test",
)
assert json.loads(write)["created"] is True
assert root.joinpath("note.txt").read_text() == "hello\nplanet\n"
edit = await _execute_os_env_tool(
"sys_os_edit",
{"path": "note.txt", "oldText": "planet", "newText": "world"},
agent_spec=spec,
conversation_id="conv_runner_os_env_test",
)
assert json.loads(edit)["replacements"] == 1
read = await _execute_os_env_tool(
"sys_os_read",
{"path": "note.txt"},
agent_spec=spec,
conversation_id="conv_runner_os_env_test",
)
assert json.loads(read)["content"] == "hello\nworld\n"
shell = await _execute_os_env_tool(
"sys_os_shell",
{"command": "pwd"},
agent_spec=spec,
conversation_id="conv_runner_os_env_test",
)
shell_result = json.loads(shell)
assert shell_result["exit_code"] == 0
assert Path(shell_result["stdout"].strip()).resolve() == root.resolve()
@pytest.mark.asyncio
async def test_runner_os_env_placeholder_cwd_uses_cli_workspace(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""Runner-local OS tools map ``cwd: .`` to the CLI workspace.
Remote ``run --server`` uploads the spec to an app server,
but the local runner still owns filesystem access. This pins
the contract that a placeholder cwd resolves to the local
project root the CLI passed into the runner, not to the
runner-owned temp fallback.
:param monkeypatch: Pytest environment patch fixture.
:param tmp_path: Per-test temp root for workspace and fallback paths.
:returns: None.
"""
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
from omnigent.runner.tool_dispatch import _execute_os_env_tool
workspace = tmp_path / "project"
workspace.mkdir()
monkeypatch.setenv("OMNIGENT_RUNNER_OS_ENV_ROOT", str(tmp_path / "fallback"))
spec = AgentSpec(
spec_version=1,
os_env=OSEnvSpec(
type="caller_process",
cwd=".",
sandbox=OSEnvSandboxSpec(type="none"),
),
)
out = await _execute_os_env_tool(
"sys_os_write",
{"path": "created.txt", "content": "from workspace"},
agent_spec=spec,
conversation_id="conv_runner_workspace",
runner_workspace=workspace.resolve(),
)
assert json.loads(out)["created"] is True
assert workspace.joinpath("created.txt").read_text() == "from workspace"
assert not (tmp_path / "fallback").exists()
@pytest.mark.asyncio
async def test_runner_os_env_tools_default_to_conversation_workspace(monkeypatch) -> None:
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
from omnigent.runner.tool_dispatch import _execute_os_env_tool
from omnigent.spec.types import AgentSpec
with tempfile.TemporaryDirectory() as td:
monkeypatch.setenv("OMNIGENT_RUNNER_OS_ENV_ROOT", td)
spec = AgentSpec(
spec_version=1,
os_env=OSEnvSpec(
type="caller_process",
sandbox=OSEnvSandboxSpec(type="none"),
),
)
out = await _execute_os_env_tool(
"sys_os_write",
{"path": "created.txt", "content": "hi"},
agent_spec=spec,
conversation_id="conv/default workspace",
)
assert json.loads(out)["created"] is True
assert Path(td, "conv_default_workspace", "workspace", "created.txt").read_text() == "hi"
def test_clone_os_env_spec_preserves_all_sandbox_fields() -> None:
"""Cloning an OSEnvSpec must preserve every sandbox field.
Regression guard for the same class of bug previously fixed in
:func:`omnigent.inner.terminal._clone_sandbox_spec`: hand-enumerated
field copies silently drop security-critical fields (egress_rules,
egress_allow_private_destinations, env_passthrough, etc.) when new
fields are added to :class:`OSEnvSandboxSpec`. This test asserts the
runner-side clone is field-complete by comparing the dataclass dicts
and verifying list-typed fields are not aliased with the original.
:returns: None.
"""
import dataclasses
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
from omnigent.runner.tool_dispatch import _clone_os_env_spec
sandbox = OSEnvSandboxSpec(
type="darwin_seatbelt",
read_paths=["~/.databrickscfg"],
write_paths=["."],
write_files=["~/.ssh/known_hosts"],
allow_network=True,
cwd_allow_hidden=[".git", ".venv"],
cwd_hidden_scan_max_entries=12345,
cwd_hidden_scan_overflow="warn",
env_passthrough=["DATABRICKS_HOST", "DATABRICKS_TOKEN"],
egress_rules=["GET api.github.com/repos/databricks/*/**"],
egress_allow_private_destinations=True,
)
spec = OSEnvSpec(
type="caller_process",
cwd="/tmp/work",
sandbox=sandbox,
fork=True,
start_in_scratch=True,
)
clone = _clone_os_env_spec(spec)
# Every OSEnvSpec field round-trips.
assert dataclasses.asdict(clone) == dataclasses.asdict(spec), (
"clone must preserve every OSEnvSpec / OSEnvSandboxSpec field; "
"hand-enumerated copies silently drop newly-added fields."
)
# Mutable list fields are copied, not aliased.
assert clone.sandbox is not sandbox
for name in (
"read_paths",
"write_paths",
"write_files",
"cwd_allow_hidden",
"env_passthrough",
"egress_rules",
):
original_list = getattr(sandbox, name)
cloned_list = getattr(clone.sandbox, name)
assert cloned_list == original_list
assert cloned_list is not original_list, (
f"{name} must be a new list so later mutation of the clone "
"does not leak into the original spec."
)
def test_effective_runner_os_env_defaults_when_spec_has_no_os_env(monkeypatch) -> None:
"""Agent specs without ``os_env`` get a runner-owned workspace cwd.
:param monkeypatch: Pytest environment patch fixture.
:returns: None.
"""
from omnigent.runner.tool_dispatch import _effective_runner_os_env_spec
from omnigent.spec.types import AgentSpec
with tempfile.TemporaryDirectory() as td:
monkeypatch.setenv("OMNIGENT_RUNNER_OS_ENV_ROOT", td)
spec = AgentSpec(spec_version=1)
os_env = _effective_runner_os_env_spec(spec, "conv/no os env")
assert os_env.type == "caller_process"
assert Path(os_env.cwd) == Path(td, "conv_no_os_env", "workspace")
assert Path(os_env.cwd).is_dir()
def test_effective_runner_os_env_uses_cli_workspace_when_spec_has_no_os_env(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""Agent specs without ``os_env`` use the CLI workspace when available.
:param monkeypatch: Pytest environment patch fixture.
:param tmp_path: Per-test temp root for workspace and fallback paths.
:returns: None.
"""
from omnigent.runner.tool_dispatch import _effective_runner_os_env_spec
from omnigent.spec.types import AgentSpec
workspace = tmp_path / "project"
workspace.mkdir()
monkeypatch.setenv("OMNIGENT_RUNNER_OS_ENV_ROOT", str(tmp_path / "fallback"))
spec = AgentSpec(spec_version=1)
os_env = _effective_runner_os_env_spec(
spec,
"conv/no os env",
runner_workspace=workspace.resolve(),
)
assert os_env.type == "caller_process"
assert Path(os_env.cwd) == workspace.resolve()
assert not (tmp_path / "fallback").exists()
def test_effective_runner_os_env_runner_workspace_overrides_absolute_spec_cwd(
tmp_path: Path,
) -> None:
"""
runner_workspace wins over an absolute ``os_env.cwd`` in the spec.
Per designs/SESSION_WORKSPACE_SELECTION.md "How this maps onto
runtime": absolute spec cwds are session-create-time
boundaries, not runtime overrides. When the runner is launched
with ``OMNIGENT_RUNNER_WORKSPACE`` set, it always wins —
otherwise picking ``~/universe/src/foo`` for an agent
declaring ``cwd: ~/universe`` would silently relocate up to
``~/universe`` at runtime.
"""
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
from omnigent.runner.tool_dispatch import _effective_runner_os_env_spec
from omnigent.spec.types import AgentSpec
workspace = tmp_path / "picked-subdir"
workspace.mkdir()
spec_cwd = tmp_path / "agent-spec-cwd"
spec_cwd.mkdir()
spec = AgentSpec(
spec_version=1,
os_env=OSEnvSpec(
type="caller_process",
cwd=str(spec_cwd),
sandbox=OSEnvSandboxSpec(type="none"),
),
)
os_env = _effective_runner_os_env_spec(
spec,
"conv_abs_override",
runner_workspace=workspace.resolve(),
)
# Workspace wins over the absolute spec cwd.
assert Path(os_env.cwd) == workspace.resolve()
assert Path(os_env.cwd) != spec_cwd.resolve()
def test_effective_runner_os_env_absolute_spec_cwd_used_without_runner_workspace(
tmp_path: Path,
) -> None:
"""
Without runner_workspace, an absolute ``os_env.cwd`` in the
spec is used as-is.
Pins the no-env-var fallback path so unit tests / pure local
runs that construct an agent spec directly without the env
var continue to honor whatever the spec declared.
"""
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
from omnigent.runner.tool_dispatch import _effective_runner_os_env_spec
from omnigent.spec.types import AgentSpec
spec_cwd = tmp_path / "agent-spec-cwd"
spec_cwd.mkdir()
spec = AgentSpec(
spec_version=1,
os_env=OSEnvSpec(
type="caller_process",
cwd=str(spec_cwd),
sandbox=OSEnvSandboxSpec(type="none"),
),
)
os_env = _effective_runner_os_env_spec(
spec,
"conv_no_workspace_abs",
runner_workspace=None,
)
assert Path(os_env.cwd) == spec_cwd
@pytest.mark.asyncio
async def test_runner_terminal_dispatch_passes_cli_workspace(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""Runner terminal tools receive the CLI workspace in ``ToolContext``.
:param monkeypatch: Pytest monkeypatch fixture.
:param tmp_path: Workspace path exported to the runner.
:returns: None.
"""
from omnigent.inner.datamodel import TerminalEnvSpec
from omnigent.runner.tool_dispatch import _execute_terminal_tool
from omnigent.terminals import TerminalRegistry
from omnigent.tools.base import ToolContext
from omnigent.tools.builtins.sys_terminal import SysTerminalLaunchTool
workspace = tmp_path / "project"
workspace.mkdir()
spec = AgentSpec(
spec_version=1,
terminals={"zsh": TerminalEnvSpec(command="zsh")},
)
captured: dict[str, object] = {}
def _fake_invoke(
self: SysTerminalLaunchTool,
arguments: str,
ctx: ToolContext,
) -> str:
"""
Capture the runner-built context without launching tmux.
:param self: Bound launch tool instance.
:param arguments: JSON arguments forwarded by dispatch.
:param ctx: Tool context created by runner dispatch.
:returns: JSON status payload.
"""
del self
captured["arguments"] = arguments
captured["workspace"] = ctx.workspace
return json.dumps({"status": "captured"})
monkeypatch.setattr(SysTerminalLaunchTool, "invoke", _fake_invoke)
out = await _execute_terminal_tool(
"sys_terminal_launch",
{"terminal": "zsh", "session": "s1"},
terminal_registry=TerminalRegistry(),
agent_spec=spec,
conversation_id="conv_terminal_dispatch",
task_id="task_terminal_dispatch",
agent_id="ag_terminal_dispatch",
runner_workspace=workspace.resolve(),
)
assert json.loads(out)["status"] == "captured"
assert json.loads(captured["arguments"]) == {"terminal": "zsh", "session": "s1"}
assert captured["workspace"] == workspace.resolve()
class _StubTerminalInstance:
"""Minimal stand-in for a launched ``TerminalInstance``.
``terminal_resource_view`` reads only stable ``TerminalInstance``
fields; the launch/close tool's ``invoke`` is monkeypatched in
these tests so no real tmux instance is created.
:param running: Value surfaced as ``metadata.running`` in the
resource view.
"""
def __init__(self, running: bool = True) -> None:
self.running = running
# ``None`` => the view uses the default environment id, so we
# don't need to fabricate an OSEnvironment.
self.os_env = None
self.socket_path = Path("/tmp/omnigent-test-tmux.sock")
self.tmux_target = "main"
# ``terminal_resource_view`` reads this to project the effective
# web-attach transport into metadata; ``None`` => the global default.
self.terminal_transport = None
# Records on_activity callbacks the dispatch wires up so a fresh
# launch's pane-activity watcher start is observable (and so the
# call doesn't AttributeError against this stub).
self.activity_watchers: list[Callable[[], None]] = []
def start_idle_watcher_thread(
self,
on_idle: Callable[[], None] | None = None,
*,
on_activity: Callable[[], None] | None = None,
) -> None:
"""Record the activity callback instead of polling real tmux."""
if on_activity is not None:
self.activity_watchers.append(on_activity)
class _StubTerminalRegistry:
"""Registry stub whose ``get`` returns a fixed instance.
The launch/close tool's ``invoke`` is monkeypatched, so the tool
never touches the registry; only ``_emit_terminal_resource_event``
calls ``get`` (on the fresh-launch branch) to build the resource
view. A real stub class (not ``MagicMock``) so an unexpected extra
call surfaces as a recorded entry rather than silently passing.
:param instance: Instance returned for every ``get``, or ``None``
to simulate a registry miss.
"""
def __init__(self, instance: _StubTerminalInstance | None) -> None:
self._instance = instance
self.get_calls: list[tuple[str, str, str]] = []
def get(
self,
conversation_id: str,
terminal_name: str,
session_key: str,
) -> _StubTerminalInstance | None:
"""Record the lookup and return the configured instance."""
self.get_calls.append((conversation_id, terminal_name, session_key))
return self._instance
def _capturing_publish_event(
captured: list[dict[str, Any]],
) -> Any:
"""Build a ``publish_event`` stub that records published events.
:param captured: List the returned callable appends each event
dict to (the session id is discarded — every event in these
tests targets the same conversation).
:returns: A ``(session_id, event) -> None`` callable.
"""
def _publish(session_id: str, event: dict[str, Any]) -> None:
del session_id
captured.append(event)
return _publish
@pytest.mark.asyncio
async def test_terminal_launch_dispatch_emits_resource_created(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A fresh ``sys_terminal_launch`` publishes ``session.resource.created``.
Verifies the runner dispatcher surfaces a tool-launched terminal
on the live SSE stream mid-turn (the whole point of this change),
with the same resource shape the REST path emits.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.runner.tool_dispatch import _execute_terminal_tool
from omnigent.tools.base import ToolContext
from omnigent.tools.builtins.sys_terminal import SysTerminalLaunchTool
spec = AgentSpec(spec_version=1)
def _fake_invoke(self: SysTerminalLaunchTool, arguments: str, ctx: ToolContext) -> str:
del self, arguments, ctx
return json.dumps({"terminal": "zsh", "session": "s1", "status": "launched"})
monkeypatch.setattr(SysTerminalLaunchTool, "invoke", _fake_invoke)
registry = _StubTerminalRegistry(_StubTerminalInstance(running=True))
published: list[dict[str, Any]] = []
out = await _execute_terminal_tool(
"sys_terminal_launch",
{"terminal": "zsh", "session": "s1"},
terminal_registry=registry,
agent_spec=spec,
conversation_id="conv_emit",
task_id="task_emit",
agent_id="ag_emit",
publish_event=_capturing_publish_event(published),
)
# Tool output is returned unchanged to the harness.
assert json.loads(out)["status"] == "launched"
# Exactly one live event — the fresh-launch resource.created. A
# second event would mean the close branch or a duplicate fired.
assert len(published) == 1, f"expected 1 published event, got {published}"
event = published[0]
assert event["type"] == "session.resource.created"
# Resource shape matches the REST path: deterministic id +
# terminal type so the web rail / relay handle both identically.
assert event["resource"]["id"] == "terminal_zsh_s1"
assert event["resource"]["type"] == "terminal"
assert event["resource"]["metadata"]["terminal_name"] == "zsh"
# The instance was looked up from the registry to build the view.
assert registry.get_calls == [("conv_emit", "zsh", "s1")]
@pytest.mark.asyncio
async def test_terminal_launch_idempotent_does_not_emit(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""An ``already_running`` launch publishes nothing.
Re-launching an existing ``(terminal, session)`` returns
``status: already_running``; emitting ``session.resource.created``
again would double-create the same id in the rail.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.runner.tool_dispatch import _execute_terminal_tool
from omnigent.tools.base import ToolContext
from omnigent.tools.builtins.sys_terminal import SysTerminalLaunchTool
spec = AgentSpec(spec_version=1)
def _fake_invoke(self: SysTerminalLaunchTool, arguments: str, ctx: ToolContext) -> str:
del self, arguments, ctx
return json.dumps({"terminal": "zsh", "session": "s1", "status": "already_running"})
monkeypatch.setattr(SysTerminalLaunchTool, "invoke", _fake_invoke)
registry = _StubTerminalRegistry(_StubTerminalInstance(running=True))
published: list[dict[str, Any]] = []
await _execute_terminal_tool(
"sys_terminal_launch",
{"terminal": "zsh", "session": "s1"},
terminal_registry=registry,
agent_spec=spec,
conversation_id="conv_emit",
task_id="task_emit",
agent_id="ag_emit",
publish_event=_capturing_publish_event(published),
)
# No event: the resource already existed before this launch.
assert published == []
# And we didn't even bother looking up the instance to build a view.
assert registry.get_calls == []
@pytest.mark.asyncio
async def test_terminal_close_dispatch_emits_resource_deleted(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A successful ``sys_terminal_close`` publishes ``session.resource.deleted``.
Verifies the symmetric teardown path: closing a terminal removes
it from the rail live, with the deleted-event shape the REST path
and web UI already handle.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.runner.tool_dispatch import _execute_terminal_tool
from omnigent.tools.base import ToolContext
from omnigent.tools.builtins.sys_terminal import SysTerminalCloseTool
spec = AgentSpec(spec_version=1)
def _fake_invoke(self: SysTerminalCloseTool, arguments: str, ctx: ToolContext) -> str:
del self, arguments, ctx
return json.dumps({"terminal": "zsh", "session": "s1", "status": "closed"})
monkeypatch.setattr(SysTerminalCloseTool, "invoke", _fake_invoke)
# Close doesn't look up the instance (no resource view to build),
# so a miss-returning registry is fine here.
registry = _StubTerminalRegistry(None)
published: list[dict[str, Any]] = []
await _execute_terminal_tool(
"sys_terminal_close",
{"terminal": "zsh", "session": "s1"},
terminal_registry=registry,
agent_spec=spec,
conversation_id="conv_emit",
task_id="task_emit",
agent_id="ag_emit",
publish_event=_capturing_publish_event(published),
)
assert len(published) == 1, f"expected 1 published event, got {published}"
event = published[0]
assert event["type"] == "session.resource.deleted"
assert event["resource_id"] == "terminal_zsh_s1"
assert event["resource_type"] == "terminal"
assert event["session_id"] == "conv_emit"
# Deleted carries only the id; no registry lookup needed.
assert registry.get_calls == []
@pytest.mark.asyncio
async def test_runner_read_inbox_continues_after_malformed_terminal_idle_item() -> None:
"""
Malformed terminal idle items must not abort the inbox drain.
:returns: None.
"""
from omnigent.runner.tool_dispatch import execute_tool
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
session_inbox.put_nowait(
{
"handle_id": "handle_before",
"tool_name": "sys_os_shell",
"status": "completed",
"output": "before",
}
)
session_inbox.put_nowait({"type": "terminal_idle", "source": "zsh"})
session_inbox.put_nowait(
{
"handle_id": "handle_after",
"tool_name": "sys_os_shell",
"status": "completed",
"output": "after",
}
)
inbox_output = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
session_inbox=session_inbox,
)
assert "task handle_before completed" in inbox_output
assert "sys_os_shell returned: before" in inbox_output
assert "malformed terminal_idle inbox item ignored" in inbox_output
assert "terminal-idle inbox payload requires non-empty string session" in inbox_output
assert "task handle_after completed" in inbox_output
assert "sys_os_shell returned: after" in inbox_output
assert session_inbox.empty()
# ── End-to-end with real harness subprocess + real LLM ──
@pytest.mark.skipif(
not os.environ.get("OPENAI_API_KEY"),
reason="OPENAI_API_KEY not set; runner→harness→LLM e2e skipped",
)
@pytest.mark.asyncio
async def test_runner_dispatches_to_spawned_harness_with_real_llm(
started_manager: HarnessProcessManager,
) -> None:
"""The flagship architectural test.
Server-side httpx → runner FastAPI's
``POST /v1/sessions/{conversation_id}/events?stream=true`` →
runner asks ``HarnessProcessManager`` to spawn / fetch the
per-conversation harness → uvicorn subprocess on a UDS →
test harness's ``run_turn`` → real OpenAI gpt-4o-mini → SSE
chunks stream back through harness UDS → runner's proxy_stream
→ test httpx client.
Successful streaming with the expected event sequence proves:
1. The runner package uses the existing HarnessProcessManager
(no parallel impl).
2. A real harness subprocess is spawned per (conversation, harness).
3. The runner correctly relays the harness's SSE bytes.
4. The runner is exercising the architectural shape from
designs/RUNNER.md §4 — not just calling an LLM directly.
"""
app = create_runner_app(
process_manager=started_manager,
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
async with http.stream(
"POST",
"/v1/sessions/conv_runner_e2e_test/events?stream=true",
json={
"type": "message",
"role": "user",
"harness": _TEST_HARNESS_NAME,
"content": [{"role": "user", "content": "Reply with the single word 'pong'."}],
"model": "openai/gpt-4o-mini",
"instructions": "You echo single words.",
"connection_params": {"api_key": os.environ["OPENAI_API_KEY"]},
},
timeout=120.0,
) as response:
assert response.status_code == 200, (
f"runner dispatch failed: {response.status_code} "
f"{(await response.aread()).decode('utf-8', errors='replace')}"
)
assert response.headers["content-type"].startswith("text/event-stream")
body = b"".join([chunk async for chunk in response.aiter_bytes()])
# Parse the SSE bytes back into events.
events = _parse_sse(body)
types = [t for t, _ in events]
# Must START with response.created and END with response.completed —
# proves both ends of the harness's emitted stream made it through
# the runner's proxy. A real LLM produces at least one delta in
# between.
assert types[0] == "response.created", (
f"first event must be response.created; got types={types}"
)
assert types[-1] == "response.completed", (
f"last event must be response.completed; got types={types}"
)
assert any(t == "response.output_text.delta" for t in types), (
f"expected text deltas from real LLM; got types={types}"
)
# The subprocess actually got spawned — verify by checking the
# manager's internal state. (Direct attribute access; the manager
# is a test-fixture instance so this is fine.)
assert "conv_runner_e2e_test" in started_manager._entries, (
"manager should have a spawned entry for the test conversation; "
"if missing, the runner didn't actually dispatch through "
"HarnessProcessManager.get_client"
)
def _parse_sse(raw: bytes) -> list[tuple[str, dict]]:
"""Decode SSE bytes into ``[(event_type, payload), ...]``."""
text = raw.decode("utf-8", errors="replace")
events: list[tuple[str, dict]] = []
current_type: str | None = None
current_data: list[str] = []
for line in text.split("\n"):
if line.startswith("event: "):
current_type = line[len("event: ") :]
elif line.startswith("data: "):
current_data.append(line[len("data: ") :])
elif line == "" and current_type is not None and current_data:
try:
payload = json.loads("\n".join(current_data))
except json.JSONDecodeError:
payload = {"_raw": "\n".join(current_data)}
events.append((current_type, payload))
current_type = None
current_data = []
return events
def test_maybe_signal_changed_files_throttles_within_window() -> None:
"""
``_maybe_signal_changed_files`` emits at most one
``session.changed_files.invalidated`` per throttle window per
session, then re-emits after the window elapses.
The monotonic ``now`` is injected so this is deterministic with no
real sleep. Regression target: a multi-file turn must collapse to
one refetch trigger (leading-edge throttle), not fire per write.
"""
from omnigent.runner.tool_dispatch import (
_CHANGED_FILES_SIGNAL_THROTTLE_S,
_maybe_signal_changed_files,
)
published: list[dict[str, object]] = []
def _pub(_sid: str, event: dict[str, Any]) -> None:
published.append(event)
sid = "conv_changed_files_throttle_unique"
# First call emits immediately (leading edge).
_maybe_signal_changed_files(sid, _pub, now=100.0)
# Within the window — suppressed.
_maybe_signal_changed_files(sid, _pub, now=100.0 + _CHANGED_FILES_SIGNAL_THROTTLE_S / 2)
# After the window — emits again.
_maybe_signal_changed_files(sid, _pub, now=100.0 + _CHANGED_FILES_SIGNAL_THROTTLE_S + 0.01)
# 2 = first (leading) + post-window; the middle call was throttled.
# If 3, the throttle window is not being honored; if 1, the
# post-window re-emit is broken.
assert len(published) == 2, f"expected 2 signals (leading + post-window), got {len(published)}"
for event in published:
assert event["type"] == "session.changed_files.invalidated"
assert event["session_id"] == sid
assert event["environment_id"] == "default"
# A missing publisher or session id is a no-op (no crash, no emit).
_maybe_signal_changed_files(None, _pub, now=200.0)
_maybe_signal_changed_files(sid, None, now=200.0)
assert len(published) == 2
def test_subagent_read_tools_are_runner_local() -> None:
"""
``sys_session_list`` and ``sys_session_get_history`` dispatch locally in the runner.
If this regresses, native harnesses that call these tools fall through to
spec-callable resolution and the user sees "not in local dispatch table"
instead of sub-agent recovery data.
"""
from omnigent.runner.tool_dispatch import should_dispatch_locally
assert should_dispatch_locally("sys_session_list") is True
assert should_dispatch_locally("sys_session_get_history") is True
# get_info is a runner-local read like list/get_history; if it falls out of
# the local dispatch table the orchestrator's status checks break.
assert should_dispatch_locally("sys_session_get_info") is True
@pytest.mark.asyncio
@pytest.mark.parametrize(
"subagent_args",
[
pytest.param("continue", id="plain-string-contract"),
pytest.param({"input": "continue"}, id="object-input-contract"),
],
)
async def test_sys_session_send_reuses_existing_child_session(
monkeypatch: pytest.MonkeyPatch,
subagent_args: str | dict[str, str],
) -> None:
"""
Re-sending to the same ``(agent, title)`` continues the existing child.
This catches the duplicate-create regression behind unreliable
continuation: if the runner POSTs ``/v1/sessions`` despite the existing
child row, the test fails and the user would see a duplicate-title server
error instead of a continuation. The parameterized ``args`` values also
prove the runner preserves the public plain-string contract while accepting
Nessie's object form.
:param monkeypatch: Pytest monkeypatch fixture.
:param subagent_args: ``sys_session_send`` ``args`` payload.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
create_posts = 0
event_posts: list[dict[str, Any]] = []
published: list[dict[str, Any]] = []
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
nonlocal create_posts
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent/child_sessions"
):
return httpx.Response(
200,
json={
"data": [
{
"id": "conv_existing",
"tool": "claude",
"session_name": "issue-1756",
"busy": False,
}
]
},
)
if request.method == "POST" and request.url.path == "/v1/sessions":
create_posts += 1
return httpx.Response(500, json={"error": "duplicate"})
if request.method == "POST" and request.url.path == "/v1/sessions/conv_existing/events":
event_posts.append(json.loads(request.content))
return httpx.Response(200, json={"ok": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "claude",
"title": "issue-1756",
"args": subagent_args,
}
),
server_client=server_client,
conversation_id="conv_parent",
agent_spec=SimpleNamespace(sub_agents=[SimpleNamespace(name="claude")]),
session_inbox=session_inbox,
publish_event=_capturing_publish_event(published),
)
finally:
runner_app.unregister_subagent_work("conv_existing")
runner_app._session_inboxes_ref.pop("conv_parent", None)
payload = json.loads(output)
assert create_posts == 0, "continuation must not create a duplicate child session"
assert payload["conversation_id"] == "conv_existing"
assert payload["status"] == "launching"
assert "continued ok" not in payload["message"]
assert event_posts[0]["data"]["content"][0]["text"] == "continue"
assert published[-1]["type"] == "session.child_session.updated"
assert published[-1]["child"]["current_task_status"] == "launching"
assert published[-1]["child"]["busy"] is False
def _spec_with_subagent_harness(harness: str) -> SimpleNamespace:
"""
Build a parent-spec stub declaring one ``worker`` sub-agent.
Mirrors the AP-style ``sub_agents`` shape ``_subagent_harness``
walks: ``executor.config["harness"]`` falling back to
``executor.type``.
:param harness: The sub-agent's declared harness, e.g.
``"codex-native"``.
:returns: A structural parent-spec stub for ``execute_tool``.
"""
return SimpleNamespace(
sub_agents=[
SimpleNamespace(
name="worker",
executor=SimpleNamespace(type="omnigent", config={"harness": harness}),
)
]
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("harness", "model"),
[
pytest.param("claude-native", "databricks-claude-sonnet-4-6", id="claude-native"),
pytest.param("codex-native", "databricks-gpt-5-4", id="codex-native"),
pytest.param("claude-sdk", "databricks-claude-sonnet-4-6", id="claude-sdk"),
],
)
async def test_sys_session_send_model_lands_in_child_create_body(
monkeypatch: pytest.MonkeyPatch,
harness: str,
model: str,
) -> None:
"""
A per-dispatch ``model`` reaches the child create as ``model_override``.
The server persists ``model_override`` on the child row, where the
native launch paths read it as ``--model`` and the SDK harness path
as ``HARNESS_<H>_MODEL``. If the create body drops the field, the
child silently runs on the harness default — the exact silent-drop
failure this feature forbids.
:param monkeypatch: Pytest monkeypatch fixture.
:param harness: Declared sub-agent harness under test.
:param model: Family-appropriate model id for *harness*.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
create_bodies: list[dict[str, Any]] = []
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve fresh-create child lookup, create, and message POSTs."""
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent_model/child_sessions"
):
return httpx.Response(200, json={"data": []})
if request.method == "POST" and request.url.path == "/v1/sessions":
create_bodies.append(json.loads(request.content))
return httpx.Response(201, json={"id": "conv_child_model"})
if request.method == "POST" and request.url.path == "/v1/sessions/conv_child_model/events":
return httpx.Response(202, json={"queued": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "fix-auth",
"args": {
"input": "fix the auth bug",
"model": model,
},
}
),
server_client=server_client,
conversation_id="conv_parent_model",
agent_spec=_spec_with_subagent_harness(harness),
session_inbox=session_inbox,
)
finally:
runner_app.unregister_subagent_work("conv_child_model")
runner_app._session_inboxes_ref.pop("conv_parent_model", None)
payload = json.loads(output)
assert payload["status"] == "launching"
# Exactly one create, carrying the override verbatim — the value the
# server persists and the harness launch consumes.
assert len(create_bodies) == 1, "fresh named send must create exactly one child"
assert create_bodies[0]["model_override"] == model
assert create_bodies[0]["sub_agent_name"] == "worker"
@pytest.mark.asyncio
async def test_sys_session_send_blocks_fresh_dispatch_when_harness_cli_missing(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A fresh dispatch whose harness CLI is absent fails loud, creates nothing.
Without this preflight a missing CLI surfaces only as a lazy first-turn
boot failure (the pi harness raises ImportError, which the parent sees as
a generic "turn failed" inbox item), and the orchestrator may re-dispatch
into the same wall. The tool must instead return an actionable error
naming the missing binary and install command, BEFORE creating any child
session.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.onboarding.harness_install import HarnessInstallSpec
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
# Override the autouse "all CLIs present" stub: pi's CLI is absent here.
monkeypatch.setattr(
"omnigent.onboarding.harness_install.missing_harness_cli",
lambda harness: HarnessInstallSpec("Pi", "pi", "@earendil-works/pi-coding-agent"),
)
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
create_posts = 0
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve the fresh-create child lookup; count (and reject) any create."""
nonlocal create_posts
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent_nopi/child_sessions"
):
return httpx.Response(200, json={"data": []})
if request.method == "POST" and request.url.path == "/v1/sessions":
create_posts += 1
return httpx.Response(201, json={"id": "conv_should_not_exist"})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "review-auth",
"args": {"input": "review the diff"},
}
),
server_client=server_client,
conversation_id="conv_parent_nopi",
agent_spec=_spec_with_subagent_harness("pi"),
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent_nopi", None)
# The output is a plain error string (not a JSON status payload) naming
# the missing binary and how to install it — what the orchestrator/human
# needs to unblock. If this regresses, the dispatch would instead create a
# child that can never boot.
assert output.startswith("Error:")
assert "'pi' CLI" in output
assert "npm install -g @earendil-works/pi-coding-agent" in output
# No child was created — the guard returned before the create POST. A
# nonzero count would mean we spawned a worker doomed to fail at boot.
assert create_posts == 0
@pytest.mark.asyncio
async def test_sys_session_send_model_rejected_for_existing_child(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
Passing ``model`` on a continuation send fails loud, sends nothing.
A native child bakes ``--model`` in at terminal launch, so applying
a new model to an existing session would be silently ignored. The
tool must return an actionable error (continue without ``model`` or
close and respawn) instead of continuing on the wrong model.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
create_posts = 0
event_posts = 0
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve the existing-child lookup; count any writes."""
nonlocal create_posts, event_posts
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent_model_cont/child_sessions"
):
return httpx.Response(
200,
json={
"data": [
{
"id": "conv_existing_model",
"tool": "worker",
"session_name": "fix-auth",
"busy": False,
}
]
},
)
if request.method == "POST" and request.url.path == "/v1/sessions":
create_posts += 1
return httpx.Response(201, json={"id": "conv_dup"})
if request.method == "POST" and request.url.path.endswith("/events"):
event_posts += 1
return httpx.Response(202, json={"queued": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "fix-auth",
"args": {"input": "continue", "model": "claude-opus-4-8"},
}
),
server_client=server_client,
conversation_id="conv_parent_model_cont",
agent_spec=_spec_with_subagent_harness("claude-native"),
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent_model_cont", None)
assert output.startswith("Error:"), output
# The error must name the existing session and the recovery paths.
assert "conv_existing_model" in output
assert "sys_session_close" in output
# No write happened: the wrong-model continuation never started.
assert create_posts == 0
assert event_posts == 0
@pytest.mark.asyncio
async def test_sys_session_send_model_rejected_in_by_id_mode() -> None:
"""
``model`` plus ``session_id`` fails loud before any server call.
By-id mode always targets an existing session, where a model
override cannot take effect — the tool must reject it instead of
silently dropping the field.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
requests_seen = 0
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Count every request — none is expected."""
nonlocal requests_seen
requests_seen += 1
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"session_id": "conv_some_child",
"args": {"input": "continue", "model": "claude-opus-4-8"},
}
),
server_client=server_client,
conversation_id="conv_parent_by_id_model",
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent_by_id_model", None)
assert output.startswith("Error:"), output
assert "model" in output
# Rejected before lookup: a misaddressed override must not even read
# the target session.
assert requests_seen == 0
@pytest.mark.asyncio
async def test_sys_session_send_model_rejected_for_unplumbed_harness(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
A ``model`` for a harness without override plumbing fails loud.
Unknown harnesses have no runner-side model-override path, so
the persisted value would be silently ignored. The error must name
the harness so the orchestrator understands why the dispatch failed.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
create_posts = 0
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve the empty child lookup; count creates."""
nonlocal create_posts
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent_unplumbed/child_sessions"
):
return httpx.Response(200, json={"data": []})
if request.method == "POST" and request.url.path == "/v1/sessions":
create_posts += 1
return httpx.Response(201, json={"id": "conv_never"})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "summarize",
"args": {"input": "summarize this", "model": "some-model"},
}
),
server_client=server_client,
conversation_id="conv_parent_unplumbed",
agent_spec=_spec_with_subagent_harness("unknown-harness"),
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent_unplumbed", None)
assert output.startswith("Error:"), output
assert "unknown-harness" in output
# The unsupported dispatch must not create a child that would then
# silently run on the harness default.
assert create_posts == 0
@pytest.mark.asyncio
@pytest.mark.parametrize(
("harness", "model", "expected_rule"),
[
pytest.param(
"claude-native", "databricks-gpt-5-4", "only runs Claude models", id="gpt-on-claude"
),
pytest.param(
"codex-native",
"databricks-claude-sonnet-4-6",
"only runs GPT models",
id="claude-on-codex",
),
pytest.param(
"claude-native",
"databricks-meta-llama-3.3-70b-instruct",
"only runs Claude models",
id="unknown-family-on-claude",
),
],
)
async def test_sys_session_send_model_rejected_for_wrong_family(
monkeypatch: pytest.MonkeyPatch,
harness: str,
model: str,
expected_rule: str,
) -> None:
"""
A cross-family ``model`` fails loud at dispatch, before any create.
The single-vendor workers can only run their own vendor's models; a
wrong-family id would otherwise spawn a child that errors opaquely
at the harness/gateway. The error names the rule so the orchestrator
can re-dispatch with a compatible model or the pi worker.
:param monkeypatch: Pytest monkeypatch fixture.
:param harness: Sub-agent harness under test.
:param model: Cross-family or undeterminable model id.
:param expected_rule: Rule text the error must contain.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
create_posts = 0
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve the empty child lookup; count creates."""
nonlocal create_posts
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent_family/child_sessions"
):
return httpx.Response(200, json={"data": []})
if request.method == "POST" and request.url.path == "/v1/sessions":
create_posts += 1
return httpx.Response(201, json={"id": "conv_never"})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "task",
"args": {"input": "do the task", "model": model},
}
),
server_client=server_client,
conversation_id="conv_parent_family",
agent_spec=_spec_with_subagent_harness(harness),
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent_family", None)
assert output.startswith("Error:"), output
assert expected_rule in output
assert model in output
# The rejected dispatch must not create a child that would then fail
# opaquely on its first turn.
assert create_posts == 0
@pytest.mark.asyncio
@pytest.mark.parametrize(
"bad_model",
[
pytest.param("claude; rm -rf /", id="shell-metacharacters"),
pytest.param(" ", id="whitespace-only"),
pytest.param(42, id="non-string"),
],
)
async def test_sys_session_send_model_invalid_rejected_before_any_server_call(
bad_model: object,
) -> None:
"""
Malformed ``model`` values fail loud before any server traffic.
The override eventually lands on a command line (``--model``), so
shell-shaped or non-string values must be rejected at the tool
boundary — not persisted and not silently dropped.
:param bad_model: The invalid ``model`` payload under test.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
requests_seen = 0
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Count every request — none is expected."""
nonlocal requests_seen
requests_seen += 1
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "fix-auth",
"args": {"input": "fix it", "model": bad_model},
}
),
server_client=server_client,
conversation_id="conv_parent_bad_model",
agent_spec=_spec_with_subagent_harness("claude-native"),
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent_bad_model", None)
assert output.startswith("Error:"), output
assert "model" in output
# Validation precedes every lookup/create — nothing reached the server.
assert requests_seen == 0
def _spec_with_real_subagent(harness: str) -> AgentSpec:
"""
Build a real parent :class:`AgentSpec` with one ``worker`` sub-agent.
Unlike :func:`_spec_with_subagent_harness` (a structural stub), this
is a fully-typed spec the model-provider resolution can walk — the
normalization gate resolves ``executor.auth`` / ``profile`` / config
providers on the sub-spec.
:param harness: The sub-agent's declared harness, e.g.
``"claude-native"``.
:returns: The parent spec.
"""
return AgentSpec(
spec_version=1,
name="parent",
sub_agents=[
AgentSpec(
spec_version=1,
name="worker",
executor=ExecutorSpec(type="omnigent", config={"harness": harness}),
)
],
)
def _isolate_model_providers(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, yaml_text: str
) -> None:
"""
Point provider resolution at an isolated config, no ambient creds.
:param monkeypatch: Pytest monkeypatch fixture.
:param tmp_path: Per-test temp dir holding ``config.yaml``.
:param yaml_text: The config contents, e.g. a ``providers:`` block.
"""
monkeypatch.setenv("OMNIGENT_CONFIG_HOME", str(tmp_path))
monkeypatch.setenv("OMNIGENT_DISABLE_KEYRING", "1")
monkeypatch.delenv("DATABRICKS_CONFIG_PROFILE", raising=False)
monkeypatch.setattr("omnigent.onboarding.detected.detect_providers", list)
(tmp_path / "config.yaml").write_text(yaml_text)
@dataclass
class _ModelSendResult:
"""
Outcome of one fresh-create ``sys_session_send`` model dispatch.
:param output: The tool output string (JSON handle or ``Error:``).
:param create_bodies: The ``POST /v1/sessions`` bodies the mock
server captured — the persisted ``model_override`` lives here.
"""
output: str
create_bodies: list[dict[str, Any]]
async def _dispatch_model_send(
monkeypatch: pytest.MonkeyPatch,
*,
agent_spec: Any,
model: str,
conv_id: str,
) -> _ModelSendResult:
"""
Drive one fresh-create ``sys_session_send`` carrying ``args.model``.
:param monkeypatch: Pytest monkeypatch fixture.
:param agent_spec: The parent spec under test.
:param model: The requested per-dispatch model id.
:param conv_id: A unique parent conversation id per test.
:returns: The tool output and the captured create bodies.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
create_bodies: list[dict[str, Any]] = []
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve fresh-create child lookup, create, and message POSTs."""
if (
request.method == "GET"
and request.url.path == f"/v1/sessions/{conv_id}/child_sessions"
):
return httpx.Response(200, json={"data": []})
if request.method == "POST" and request.url.path == "/v1/sessions":
create_bodies.append(json.loads(request.content))
return httpx.Response(201, json={"id": "conv_child_norm"})
if request.method == "POST" and request.url.path == "/v1/sessions/conv_child_norm/events":
return httpx.Response(202, json={"queued": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "task",
"args": {"input": "do the task", "model": model},
}
),
server_client=server_client,
conversation_id=conv_id,
agent_spec=agent_spec,
session_inbox=session_inbox,
)
finally:
runner_app.unregister_subagent_work("conv_child_norm")
runner_app._session_inboxes_ref.pop(conv_id, None)
return _ModelSendResult(output=output, create_bodies=create_bodies)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("harness", "model", "expected"),
[
pytest.param(
"claude-native",
"claude-sonnet-4-6",
"databricks-claude-sonnet-4-6",
id="canonical-claude-localized",
),
pytest.param(
"codex-native", "gpt-5-4", "databricks-gpt-5-4", id="canonical-gpt-localized"
),
pytest.param(
"claude-native",
"databricks-claude-sonnet-4-6",
"databricks-claude-sonnet-4-6",
id="already-local-unchanged",
),
pytest.param(
"claude-native",
"us.anthropic.claude-sonnet-4-6",
"us.anthropic.claude-sonnet-4-6",
id="non-mechanical-passthrough",
),
],
)
async def test_sys_session_send_localizes_canonical_model_for_gateway_child(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
harness: str,
model: str,
expected: str,
) -> None:
"""
A gateway-routed child persists the gateway-local spelling.
With a Databricks default provider, a bare canonical vendor id
(``claude-sonnet-4-6``) would die at the gateway ("model not
found"); the gate must persist the ``databricks-``-prefixed
spelling as ``model_override`` — and ONLY for mechanical ids:
already-local and vendor-prefixed shapes pass through verbatim.
:param monkeypatch: Pytest monkeypatch fixture.
:param tmp_path: Per-test temp dir for the isolated provider config.
:param harness: The sub-agent harness under test.
:param model: The requested per-dispatch model id.
:param expected: The id the create body must persist.
"""
_isolate_model_providers(
monkeypatch,
tmp_path,
"providers:\n workspace:\n kind: databricks\n profile: prof-a\n default: true\n",
)
result = await _dispatch_model_send(
monkeypatch,
agent_spec=_spec_with_real_subagent(harness),
model=model,
conv_id="conv_parent_norm_gateway",
)
payload = json.loads(result.output)
assert payload["status"] == "launching"
# The persisted override is the localized id — this is the value the
# server stores and the harness launch consumes.
assert len(result.create_bodies) == 1
assert result.create_bodies[0]["model_override"] == expected
@pytest.mark.asyncio
async def test_sys_session_send_strips_gateway_prefix_for_vendor_direct_child(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""
A vendor-direct child persists the bare canonical spelling.
With an Anthropic API-key default provider, a ``databricks-``
prefixed id would be rejected by the vendor API; the gate must
strip the prefix before persisting.
:param monkeypatch: Pytest monkeypatch fixture.
:param tmp_path: Per-test temp dir for the isolated provider config.
"""
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
_isolate_model_providers(
monkeypatch,
tmp_path,
"providers:\n"
" anthropic:\n"
" kind: key\n"
" default: true\n"
" anthropic:\n"
" base_url: https://api.anthropic.com\n"
" api_key: $ANTHROPIC_API_KEY\n",
)
result = await _dispatch_model_send(
monkeypatch,
agent_spec=_spec_with_real_subagent("claude-native"),
model="databricks-claude-opus-4-8",
conv_id="conv_parent_norm_direct",
)
payload = json.loads(result.output)
assert payload["status"] == "launching"
assert len(result.create_bodies) == 1
# Stripped: the vendor API only routes the bare canonical id.
assert result.create_bodies[0]["model_override"] == "claude-opus-4-8"
@pytest.mark.asyncio
async def test_sys_session_send_passes_model_through_when_provider_undeterminable(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""
An undeterminable child provider leaves the requested id untouched.
The structural sub-spec stub has no auth/profile attributes, so
provider resolution degrades to "none" — the gate must neither
crash the dispatch nor guess a transform.
:param monkeypatch: Pytest monkeypatch fixture.
:param tmp_path: Per-test temp dir for the isolated provider config.
"""
_isolate_model_providers(monkeypatch, tmp_path, "")
result = await _dispatch_model_send(
monkeypatch,
agent_spec=_spec_with_subagent_harness("claude-native"),
model="claude-sonnet-4-6",
conv_id="conv_parent_norm_unknown",
)
payload = json.loads(result.output)
assert payload["status"] == "launching"
assert len(result.create_bodies) == 1
# Pass-through: no provider kind, no transform — fail-loud at the
# harness remains the safety net.
assert result.create_bodies[0]["model_override"] == "claude-sonnet-4-6"
@pytest.mark.asyncio
async def test_sys_session_send_family_guard_runs_before_normalization(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""
The family guard fires on the RAW requested id, before any localize.
A GPT id on a claude worker must be rejected quoting exactly what
the caller sent (``gpt-5-4``, not ``databricks-gpt-5-4``) and no
child may be created — even though the gateway provider would have
localized the id had the guard passed.
:param monkeypatch: Pytest monkeypatch fixture.
:param tmp_path: Per-test temp dir for the isolated provider config.
"""
_isolate_model_providers(
monkeypatch,
tmp_path,
"providers:\n workspace:\n kind: databricks\n profile: prof-a\n default: true\n",
)
result = await _dispatch_model_send(
monkeypatch,
agent_spec=_spec_with_real_subagent("claude-native"),
model="gpt-5-4",
conv_id="conv_parent_norm_family",
)
assert result.output.startswith("Error:"), result.output
assert "only runs Claude models" in result.output
# The error quotes the caller's raw id, proving the guard ran first.
assert "'gpt-5-4'" in result.output
assert result.create_bodies == []
@pytest.mark.asyncio
async def test_sys_list_models_dispatches_locally_with_static_provider(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""
``execute_tool`` routes ``sys_list_models`` to the catalog enumerator.
With a subscription default (static — no HTTP), the payload must
carry one row per declared sub-agent plus ``self``, each in the
documented ``{source, verified, models, note}`` shape with the
curated claude ids surviving the claude-family filter.
:param monkeypatch: Pytest monkeypatch fixture.
:param tmp_path: Per-test temp dir for the isolated provider config.
"""
from omnigent.runner.tool_dispatch import execute_tool
_isolate_model_providers(
monkeypatch,
tmp_path,
"providers:\n claude:\n kind: subscription\n cli: claude\n default: true\n",
)
output = await execute_tool(
tool_name="sys_list_models",
arguments="{}",
agent_spec=_spec_with_real_subagent("claude-native"),
conversation_id="conv_list_models",
)
payload = json.loads(output)
assert set(payload) == {"worker", "self"}
worker = payload["worker"]
assert worker["source"] == "static"
assert worker["verified"] is False
# The curated claude aliases survive the claude-family filter — the
# exact ids an orchestrator may pass back as args.model.
assert [m["id"] for m in worker["models"]] == [
"claude-fable-5",
"claude-opus-4-8",
"claude-sonnet-5",
"claude-sonnet-4-6",
"claude-haiku-4-5",
]
assert worker["note"]
@pytest.mark.asyncio
async def test_sys_list_models_requires_agent_spec() -> None:
"""
``sys_list_models`` with no resolvable spec fails loud, not empty.
A silent ``{}`` would read as "no workers exist" — the error string
tells the orchestrator the runner couldn't resolve its spec.
"""
from omnigent.runner.tool_dispatch import execute_tool
output = await execute_tool(
tool_name="sys_list_models",
arguments="{}",
agent_spec=None,
conversation_id="conv_list_models_nospec",
)
assert output.startswith("Error:")
assert "agent spec" in output
@pytest.mark.asyncio
async def test_sys_session_send_by_id_rejects_closed_child(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
By-id ``sys_session_send`` refuses closed direct children.
The close tool hands orchestrators a durable ``conversation_id``.
Without checking ``omnigent.closed=true`` in by-id mode, an
orchestrator could keep chatting with the exact child it had just
closed, bypassing the named lookup that skips closed rows.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
event_posts = 0
registrations: list[str] = []
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
monkeypatch.setattr(
runner_app,
"register_child_session",
lambda child_id, **_kwargs: registrations.append(child_id),
)
async def _server_handler(request: httpx.Request) -> httpx.Response:
nonlocal event_posts
if request.method == "GET" and request.url.path == "/v1/sessions/conv_closed":
return httpx.Response(
200,
json={
"id": "conv_closed",
"title": "researcher:auth",
"parent_session_id": "conv_parent",
"labels": {CLOSED_LABEL_KEY: CLOSED_LABEL_VALUE},
"busy": False,
},
)
if request.method == "POST" and request.url.path == "/v1/sessions/conv_closed/events":
event_posts += 1
return httpx.Response(202, json={"queued": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"session_id": "conv_closed",
"args": "please continue",
}
),
server_client=server_client,
conversation_id="conv_parent",
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent", None)
payload = json.loads(output)
assert payload["error"] == "session_closed"
assert payload["conversation_id"] == "conv_closed"
assert event_posts == 0
assert registrations == []
@pytest.mark.asyncio
async def test_sys_session_send_completion_drains_from_parent_inbox(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
A completed async sub-agent turn arrives through ``sys_read_inbox``.
This proves ``sys_session_send`` no longer inlines the child result in the
tool result. If completion delivery is not wired, the drain returns the
empty-inbox sentinel instead of the child marker.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve child-session create, lookup, and message POST requests."""
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent_inbox/child_sessions"
):
return httpx.Response(200, json={"data": []})
if request.method == "POST" and request.url.path == "/v1/sessions":
return httpx.Response(201, json={"id": "conv_child_inbox"})
if request.method == "POST" and request.url.path == "/v1/sessions/conv_child_inbox/events":
return httpx.Response(202, json={"queued": True})
if (
request.method == "POST"
and request.url.path == "/v1/sessions/conv_parent_inbox/policies/evaluate"
):
return httpx.Response(200, json={"result": "POLICY_ACTION_UNSPECIFIED"})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "phase-a",
"args": "run phase a",
}
),
server_client=server_client,
conversation_id="conv_parent_inbox",
agent_spec=SimpleNamespace(sub_agents=[SimpleNamespace(name="worker")]),
session_inbox=session_inbox,
)
payload = json.loads(output)
assert payload["status"] == "launching"
assert "CHILD_MARKER" not in payload["message"]
runner_app.mark_subagent_work_terminal(
"conv_child_inbox",
status="completed",
output="CHILD_MARKER",
)
inbox_output = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
server_client=server_client,
conversation_id="conv_parent_inbox",
session_inbox=session_inbox,
)
finally:
runner_app.unregister_subagent_work("conv_child_inbox")
runner_app._session_inboxes_ref.pop("conv_parent_inbox", None)
assert "sub-agent task conv_child_inbox completed" in inbox_output
assert "worker:phase-a returned: CHILD_MARKER" in inbox_output
@pytest.mark.asyncio
async def test_subagent_inbox_cleanup_does_not_unregister_next_turn(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
Draining an old child result must not delete a newer turn's work entry.
Named sub-agents reuse the same child session. A parent can start a second
turn after the first has completed but before reading the first result. The
first inbox item's cleanup is guarded by the per-dispatch work id so it
cannot unregister the second running turn.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
parent_id = "conv_parent_reused_child"
child_id = "conv_reused_child"
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
lookup_count = 0
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve two sends to the same child and policy checks for drains."""
nonlocal lookup_count
if (
request.method == "GET"
and request.url.path == f"/v1/sessions/{parent_id}/child_sessions"
):
lookup_count += 1
if lookup_count == 1:
return httpx.Response(200, json={"data": []})
return httpx.Response(
200,
json={
"data": [
{
"id": child_id,
"title": "worker:repeat",
"sub_agent_name": "worker",
"busy": False,
}
]
},
)
if request.method == "POST" and request.url.path == "/v1/sessions":
return httpx.Response(201, json={"id": child_id})
if request.method == "POST" and request.url.path == f"/v1/sessions/{child_id}/events":
return httpx.Response(202, json={"queued": True})
if (
request.method == "POST"
and request.url.path == f"/v1/sessions/{parent_id}/policies/evaluate"
):
return httpx.Response(200, json={"result": "POLICY_ACTION_UNSPECIFIED"})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
for prompt in ("first", "second"):
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "worker",
"title": "repeat",
"args": prompt,
}
),
server_client=server_client,
conversation_id=parent_id,
agent_spec=SimpleNamespace(sub_agents=[SimpleNamespace(name="worker")]),
session_inbox=session_inbox,
)
assert json.loads(output)["status"] == "launching"
if prompt == "first":
runner_app.mark_subagent_work_terminal(
child_id,
status="completed",
output="FIRST_RESULT",
)
first_drain = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
server_client=server_client,
conversation_id=parent_id,
session_inbox=session_inbox,
)
current = runner_app.get_subagent_work(child_id)
assert current is not None, (
"Draining the first turn must not unregister the second turn's "
"active work entry for the reused child session."
)
assert current.status == "launching"
runner_app.mark_subagent_work_terminal(
child_id,
status="completed",
output="SECOND_RESULT",
)
second_drain = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
server_client=server_client,
conversation_id=parent_id,
session_inbox=session_inbox,
)
assert runner_app.get_subagent_work(child_id) is None, (
"Draining the matching second turn must unregister terminal "
"work; otherwise the registry leaks completed child entries."
)
finally:
runner_app.unregister_subagent_work(child_id)
runner_app._session_inboxes_ref.pop(parent_id, None)
assert "worker:repeat returned: FIRST_RESULT" in first_drain
assert "SECOND_RESULT" not in first_drain
assert "worker:repeat returned: SECOND_RESULT" in second_drain
def _sse_text_turn(text: str) -> list[str]:
"""Script one scaffold turn's SSE frames carrying ``text`` as output.
The runner's ``proxy_stream`` accumulates ``response.output_text.delta``
chunks and, on ``response.completed``, commits an assistant message to
``_session_histories`` — which is what ``_extract_last_assistant_text``
reads for the sub-agent's delivered output. So a turn that emits
``text`` here is the turn whose result the parent should receive.
:param text: Assistant text this turn streams, e.g. ``"FINAL"``.
:returns: SSE frames for a created → one-delta → completed turn.
"""
return [
'event: response.created\ndata: {"type":"response.created",'
'"response":{"id":"resp_multiturn"}}\n\n',
"event: response.output_text.delta\ndata: "
+ json.dumps({"type": "response.output_text.delta", "delta": text})
+ "\n\n",
'event: response.completed\ndata: {"type":"response.completed",'
'"response":{"id":"resp_multiturn","status":"completed"}}\n\n',
]
class _GatedTwoTurnHarnessStream:
"""Per-turn-scripted harness stream that blocks turn 1 mid-flight.
Turn 1 yields ``response.created`` + the intermediate text delta, sets
``started`` (the sync gate so the test knows the turn is live and in
``_active_turns``), then awaits ``release`` before yielding
``response.completed``. This holds turn 1 active deterministically while
the test posts a second message (which buffers a continuation) — no
sleeps, no polling of runner internals. Turn 2 (and any later turn)
streams its scripted frames straight through.
:param turns: Per-turn SSE frame lists, indexed by call order.
:param call_index: Shared 1-based turn counter (mutated per ``stream``).
:param started: Set once turn 1 has emitted its intermediate delta.
:param release: Awaited by turn 1 before emitting ``response.completed``.
"""
def __init__(
self,
turns: list[list[str]],
call_index: list[int],
started: asyncio.Event,
release: asyncio.Event,
) -> None:
"""Store scripted turns and the turn-1 synchronization events.
:param turns: Per-turn SSE frame lists, indexed by call order.
:param call_index: Shared 1-based turn counter; this stream reads
and advances it so each turn serves the next script.
:param started: Set once turn 1 has emitted its intermediate delta.
:param release: Gates turn 1's terminal ``response.completed``.
"""
self._call_index = call_index
self._call_index[0] += 1
self._turn_number = self._call_index[0]
self._frames = turns[min(self._turn_number - 1, len(turns) - 1)]
self._started = started
self._release = release
self.status_code = 200
async def __aenter__(self) -> _GatedTwoTurnHarnessStream:
"""Enter the stream context.
:returns: This stream.
"""
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
"""Exit without suppressing exceptions.
:param exc_type: Exception type from the context, if any.
:param exc: Exception value from the context, if any.
:param tb: Traceback from the context, if any.
:returns: None.
"""
del exc_type, exc, tb
async def aiter_text(self) -> AsyncIterator[str]:
"""Yield scripted frames, blocking turn 1 before it completes.
:returns: Async iterator of SSE frames.
"""
if self._turn_number == 1:
# Emit created + the intermediate delta, then hand control back to
# the test (started) and wait (release) so turn 1 stays the active
# turn while the test buffers a continuation message.
yield self._frames[0]
yield self._frames[1]
self._started.set()
await self._release.wait()
yield self._frames[2]
return
for frame in self._frames:
yield frame
class _GatedTwoTurnHarnessClient:
"""Harness client whose ``stream`` returns the gated two-turn stream.
Also implements ``post`` (202) so the runner's best-effort mid-turn
injection forward — fired when the second message buffers — succeeds
quietly instead of raising into the buffering path.
:param turns: Per-turn SSE frame lists.
:param started: Set once turn 1 has emitted its intermediate delta.
:param release: Gates turn 1's terminal ``response.completed``.
"""
def __init__(
self,
turns: list[list[str]],
started: asyncio.Event,
release: asyncio.Event,
) -> None:
"""Store the scripts and synchronization events.
:param turns: Per-turn SSE frame lists.
:param started: Set once turn 1 emits its intermediate delta.
:param release: Gates turn 1's terminal ``response.completed``.
"""
self._turns = turns
self._started = started
self._release = release
self._call_index = [0]
def stream(
self,
method: str,
url: str,
*,
json: dict[str, object],
timeout: float | None,
) -> _GatedTwoTurnHarnessStream:
"""Return the next gated turn stream.
:param method: HTTP method (ignored).
:param url: Harness endpoint path (ignored).
:param json: JSON body (ignored).
:param timeout: Request timeout (ignored).
:returns: Gated two-turn stream for the current turn.
"""
del method, url, json, timeout
return _GatedTwoTurnHarnessStream(
self._turns, self._call_index, self._started, self._release
)
async def post(
self,
url: str,
*,
json: dict[str, object],
timeout: float | None,
) -> httpx.Response:
"""Accept the runner's mid-turn injection forward (best-effort).
:param url: Harness endpoint path (ignored).
:param json: Forwarded message body (ignored — the buffered copy is
what drives the continuation; this fake never echoes an
``injection.consumed`` marker, so the buffer survives).
:param timeout: Request timeout (ignored).
:returns: A 202 so the forward is treated as accepted.
"""
del url, json, timeout
return httpx.Response(202, json={"queued": True})
@pytest.mark.asyncio
async def test_scaffold_subagent_defers_terminal_delivery_while_continuation_buffered() -> None:
"""A scaffold child running two turns delivers ONLY the final turn's text.
Reproduces the multi-turn delivery bug: ``_on_proxy_stream_end`` used to
mark a scaffold child's work entry terminal (``completed`` +
``_extract_last_assistant_text``) at EVERY successful turn end. A child
that runs a second turn without a fresh parent ``sys_session_send`` — here
a buffered continuation message drained by ``_check_and_start_next_turn``
— then had its FIRST turn's intermediate narration delivered as the
result, and its real final synthesis dropped by the already-terminal +
``delivered`` short-circuit in ``mark_subagent_work_terminal``.
Determinism: turn 1's harness stream blocks (``release``) after emitting
its intermediate delta and signals ``started`` so the test can post the
second message while turn 1 is provably the active turn (so it buffers a
continuation). Releasing turn 1 then lets the continuation run to its own
empty-buffer stream end. No sleeps, no polling of runner internals.
"""
from omnigent.runner import app as runner_app
parent_id = "conv_parent_multiturn_defer"
child_id = "conv_child_multiturn_defer"
started = asyncio.Event()
release = asyncio.Event()
turns = [_sse_text_turn("INTERMEDIATE_NARRATION"), _sse_text_turn("FINAL_SYNTHESIS")]
async def _spec_resolver(agent_id: str, session_id: str | None = None) -> AgentSpec:
"""Return a non-native scaffold spec so the success-delivery path runs.
The test harness name is unknown to ``_build_spawn_env_from_spec``
(no model needed) and is not ``claude-native`` / ``codex-native``, so
``_is_native_harness`` is False and the scaffold completion branch in
``_on_proxy_stream_end`` is exercised.
:param agent_id: Agent id requested by the runner (unused).
:param session_id: Session id (unused).
:returns: A minimal scaffold spec bound to the test harness.
"""
del agent_id, session_id
return AgentSpec(
spec_version=1,
name="scaffold-multiturn-agent",
executor=ExecutorSpec(type="omnigent", config={"harness": _TEST_HARNESS_NAME}),
)
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_FakeProcessManager(_GatedTwoTurnHarnessClient(turns, started, release)),
),
spec_resolver=_spec_resolver,
server_client=NullServerClient(), # type: ignore[arg-type]
)
# Register the child as tracked sub-agent work with a real parent inbox —
# the same module-level surface a parent ``sys_session_send`` uses. The
# inbox queue is the observable delivery surface.
parent_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
runner_app._session_inboxes_ref[parent_id] = parent_inbox
runner_app.register_subagent_work(
parent_session_id=parent_id,
child_session_id=child_id,
agent="worker",
title="multiturn",
)
try:
async with _runner_test_client(app) as http:
# Turn 1: starts a background turn that blocks before completing.
resp1 = await http.post(
f"/v1/sessions/{child_id}/events",
json={
"type": "message",
"role": "user",
"agent_id": "ag_scaffold",
"model": "x",
"content": [{"type": "input_text", "text": "do the work"}],
},
)
assert resp1.status_code == 202
# Sync gate: turn 1 is live (in _active_turns) and mid-stream.
await asyncio.wait_for(started.wait(), timeout=10.0)
# Concurrent action: a second message arrives while turn 1 is
# active, so it buffers a continuation (the fake never emits an
# injection.consumed marker, so the buffered copy survives).
resp2 = await http.post(
f"/v1/sessions/{child_id}/events",
json={
"type": "message",
"role": "user",
"agent_id": "ag_scaffold",
"model": "x",
"content": [{"type": "input_text", "text": "now refine"}],
},
)
# 202 with status "buffered" proves the message buffered against the
# active turn rather than starting its own turn — the precondition
# for the deferral path. If this were "accepted", turn 1 would not
# have a continuation pending and the bug wouldn't apply.
assert resp2.status_code == 202
assert resp2.json()["status"] == "buffered"
# End turn 1. With the fix, the non-empty buffer defers delivery;
# the continuation turn 2 then runs to its own empty-buffer end.
release.set()
# Wait for the child to fully finish: terminal work entry +
# exactly one delivered inbox item. The continuation turn is
# guaranteed to reach _on_proxy_stream_end again with an empty
# buffer, so this never hangs unless the result was stranded.
deadline = asyncio.get_running_loop().time() + 10.0
while asyncio.get_running_loop().time() < deadline:
entry = runner_app.get_subagent_work(child_id)
if entry is not None and entry.delivered:
break
await asyncio.sleep(0.02)
entry = runner_app.get_subagent_work(child_id)
delivered_items: list[dict[str, Any]] = []
while not parent_inbox.empty():
delivered_items.append(parent_inbox.get_nowait())
finally:
release.set()
runner_app.unregister_subagent_work(child_id)
runner_app._session_inboxes_ref.pop(parent_id, None)
# Exactly one terminal payload reached the parent inbox. Two items would
# mean the bug's double-delivery (intermediate + final); zero would mean
# the deferral stranded the result (a worse failure than the bug).
assert len(delivered_items) == 1, (
f"expected exactly one terminal delivery, got {len(delivered_items)}: "
f"{[i.get('output') for i in delivered_items]}"
)
item = delivered_items[0]
assert item["status"] == "completed" # success terminal status
# The delivered output is the FINAL turn's text, not turn 1's intermediate
# narration. Before the fix, turn 1's "INTERMEDIATE_NARRATION" was delivered
# as the terminal result and the final synthesis was dropped — so this
# asserts the exact content that proves the right turn won.
assert item["output"] == "FINAL_SYNTHESIS", (
f"parent must receive the final turn's synthesis, got {item['output']!r}; "
f"'INTERMEDIATE_NARRATION' here means turn 1 was delivered prematurely."
)
# The work entry settled terminal-and-delivered on the continuation turn.
assert entry is not None
assert entry.status == "completed"
assert entry.delivered is True
@pytest.mark.asyncio
@pytest.mark.parametrize(
("status", "policy_response", "expected_output", "blocked_output"),
[
pytest.param(
"completed",
{"result": "POLICY_ACTION_DENY", "reason": "secret output"},
"[Result suppressed by policy: secret output]",
"SECRET_MARKER",
id="completed-deny-suppresses",
),
pytest.param(
"completed",
{"result": "POLICY_ACTION_ALLOW", "data": "<REDACTED>"},
"<REDACTED>",
"SECRET_MARKER",
id="completed-allow-data-transforms",
),
pytest.param(
"failed",
{"result": "POLICY_ACTION_DENY", "reason": "failed secret output"},
"[Result suppressed by policy: failed secret output]",
"SECRET_MARKER",
id="failed-deny-suppresses",
),
],
)
async def test_sys_read_inbox_applies_subagent_tool_result_policy(
status: str,
policy_response: dict[str, Any],
expected_output: str,
blocked_output: str,
) -> None:
"""
``sys_read_inbox`` evaluates delayed sub-agent output as TOOL_RESULT.
``sys_session_send`` returns a launching handle immediately, so the
child output arrives after the original tool call. The delayed
output must still pass through Omnigent policy evaluation before the LLM
sees it in the inbox drain.
:param status: Terminal sub-agent status being drained.
:param policy_response: Fake Omnigent policy verdict body.
:param expected_output: Output expected in the drained inbox text.
:param blocked_output: Raw child output that policy must remove.
"""
from omnigent.runner.tool_dispatch import execute_tool
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
session_inbox.put_nowait(
{
"type": "sub_agent",
"task_id": "conv_child_policy",
"handle_id": "conv_child_policy",
"conversation_id": "conv_child_policy",
"tool_name": "worker",
"agent": "worker",
"title": "phase-policy",
"status": status,
"output": "SECRET_MARKER",
}
)
policy_requests: list[dict[str, Any]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Capture the Omnigent policy evaluation request."""
if (
request.method == "POST"
and request.url.path == "/v1/sessions/conv_parent_policy/policies/evaluate"
):
policy_requests.append(json.loads(request.content))
return httpx.Response(200, json=policy_response)
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
inbox_output = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
server_client=server_client,
conversation_id="conv_parent_policy",
session_inbox=session_inbox,
)
assert len(policy_requests) == 1, (
"Delayed sub-agent output must be policy-checked exactly once: "
"0 means raw output bypassed TOOL_RESULT policy; >1 means duplicate evaluation."
)
event = policy_requests[0]["event"]
assert event["type"] == "PHASE_TOOL_RESULT"
assert event["data"]["result"] == "SECRET_MARKER"
assert event["request_data"]["name"] == "sys_session_send"
assert event["request_data"]["args"] == {
"agent": "worker",
"title": "phase-policy",
"conversation_id": "conv_child_policy",
}
assert expected_output in inbox_output
assert blocked_output not in inbox_output
@pytest.mark.asyncio
async def test_sys_read_inbox_requeues_subagent_output_on_transient_policy_failure() -> None:
"""
Transient policy-evaluation failures must not destroy child output.
The first drain receives a non-JSON policy response, so
``sys_read_inbox`` must fail closed and hide the raw child output.
Because no real DENY/ASK/ALLOW verdict exists yet, the original
payload must remain retryable; otherwise the second drain could not
return the child result after the policy service recovers.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
parent_id = "conv_parent_policy_retry"
child_id = "conv_child_policy_retry"
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
runner_app._session_inboxes_ref[parent_id] = session_inbox
runner_app.register_subagent_work(
parent_session_id=parent_id,
child_session_id=child_id,
agent="worker",
title="retry-policy",
)
runner_app.mark_subagent_work_terminal(
child_id,
status="completed",
output="SECRET_RETRY_MARKER",
)
policy_attempts = 0
work_after_second_drain: object = "not-drained"
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Fail policy evaluation once, then allow the retry.
:param request: Omnigent policy-evaluation request.
:returns: Non-JSON response on first call, allow verdict later.
"""
nonlocal policy_attempts
if (
request.method == "POST"
and request.url.path == f"/v1/sessions/{parent_id}/policies/evaluate"
):
policy_attempts += 1
if policy_attempts == 1:
return httpx.Response(200, content=b"not-json")
return httpx.Response(200, json={"result": "POLICY_ACTION_UNSPECIFIED"})
return httpx.Response(404, json={"error": str(request.url)})
try:
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
first_drain = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
server_client=server_client,
conversation_id=parent_id,
session_inbox=session_inbox,
)
assert "[Result suppressed by policy: policy evaluation failed]" in first_drain
assert "SECRET_RETRY_MARKER" not in first_drain
assert runner_app.get_subagent_work(child_id) is not None, (
"A transient policy failure must not unregister completed "
"sub-agent work, or the real child output is lost permanently."
)
assert session_inbox.qsize() == 1, (
"The original payload must be requeued for a later policy "
"retry instead of being consumed by the failed drain."
)
second_drain = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
server_client=server_client,
conversation_id=parent_id,
session_inbox=session_inbox,
)
work_after_second_drain = runner_app.get_subagent_work(child_id)
finally:
runner_app.unregister_subagent_work(child_id)
runner_app._session_inboxes_ref.pop(parent_id, None)
assert policy_attempts == 2
assert "worker:retry-policy returned: SECRET_RETRY_MARKER" in second_drain
assert work_after_second_drain is None
def test_list_tasks_is_not_runner_local_builtin() -> None:
"""
``list_tasks`` is no longer a framework builtin.
User/local tools may still choose that name, so the runner must not
claim it as a local lifecycle tool or relay it to native harnesses as
a framework-owned builtin.
"""
from omnigent.runner.tool_dispatch import (
_NATIVE_RELAY_BUILTIN_TOOLS,
should_dispatch_locally,
)
assert should_dispatch_locally("list_tasks") is False
assert "list_tasks" not in _NATIVE_RELAY_BUILTIN_TOOLS
@pytest.mark.asyncio
async def test_sys_cancel_task_stops_subagent_and_dedupes_late_completion(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
``sys_cancel_task`` hard-stops a running claude-native child cleanly.
The created child carries the ``claude-code-native-ui`` wrapper label, so
the cancel routes to ``stop_session`` (the claude-native hard-stop). The
mock server marks the child cancelled when it receives the event, matching
the synchronous claude-native stop path (``_handle_claude_native_stop``
kills the pane and reclaims the work entry). A later completion attempt
must not enqueue a second completed inbox item.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
monkeypatch.setattr(runner_app, "get_session_agent_id", lambda _sid: "ag_parent")
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
stops: list[dict[str, Any]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""Serve the child create/message flow and cancellation event."""
if (
request.method == "GET"
and request.url.path == "/v1/sessions/conv_parent_cancel/child_sessions"
):
return httpx.Response(200, json={"data": []})
if request.method == "POST" and request.url.path == "/v1/sessions":
# claude-native sub-agent: the server stamps the wrapper label so
# the cancel routes to stop_session (the hard-stop path).
return httpx.Response(
201,
json={
"id": "conv_child_cancel",
"labels": {"omnigent.wrapper": "claude-code-native-ui"},
},
)
if (
request.method == "POST"
and request.url.path == "/v1/sessions/conv_child_cancel/events"
):
body = json.loads(request.content)
if body.get("type") == "stop_session":
stops.append(body)
runner_app.mark_subagent_work_terminal(
"conv_child_cancel",
status="cancelled",
output="[System: sub-agent stopped]",
)
return httpx.Response(202, json={"queued": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"agent": "runner",
"title": "phase-c",
"args": "run phase c",
}
),
server_client=server_client,
conversation_id="conv_parent_cancel",
agent_spec=SimpleNamespace(sub_agents=[SimpleNamespace(name="runner")]),
session_inbox=session_inbox,
)
cancel_output = json.loads(
await execute_tool(
tool_name="sys_cancel_task",
arguments=json.dumps({"task_id": "conv_child_cancel"}),
server_client=server_client,
conversation_id="conv_parent_cancel",
session_async_tasks={},
)
)
runner_app.mark_subagent_work_terminal(
"conv_child_cancel",
status="completed",
output="SHOULD_NOT_DELIVER",
)
inbox_output = await execute_tool(
tool_name="sys_read_inbox",
arguments="{}",
session_inbox=session_inbox,
)
finally:
runner_app.unregister_subagent_work("conv_child_cancel")
runner_app._session_inboxes_ref.pop("conv_parent_cancel", None)
# ``data`` rides along because SessionEventInput requires it on older
# servers — omitting it 422'd sub-agent cancellation in production.
assert stops == [{"type": "stop_session", "data": {}}]
assert cancel_output == {
"cancelled": True,
"task_id": "conv_child_cancel",
"status": "cancelled",
}
assert inbox_output == (
"[System: sub-agent task conv_child_cancel cancelled — runner:phase-c]"
), "Cancelled sub-agent work must produce exactly one cancelled inbox item."
@pytest.mark.asyncio
async def test_sys_cancel_task_reports_codex_native_cancel_as_best_effort() -> None:
"""
Unconfirmed codex-native cancel must not promise terminal inbox status.
Codex-native has no runner-side hard-stop path, so the cancel routes to
``interrupt`` (not ``stop_session``, which the runner 204 no-ops for every
non-claude-native harness). For codex-native that interrupt is itself a
best-effort no-op — the child can remain running after the POST returns —
so the tool result must say cancellation is best-effort instead of telling
the parent to wait forever for a terminal inbox item.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
parent_id = "conv_parent_codex_cancel"
child_id = "conv_child_codex_cancel"
runner_app.register_subagent_work(
parent_session_id=parent_id,
child_session_id=child_id,
agent="codex_impl",
title="native",
wrapper_label="codex-native-ui",
)
stops: list[dict[str, Any]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Accept the interrupt without marking the child terminal (codex no-op).
:param request: Request sent to the child session events route.
:returns: Accepted response.
"""
if request.method == "POST" and request.url.path == f"/v1/sessions/{child_id}/events":
stops.append(json.loads(request.content))
return httpx.Response(204)
return httpx.Response(404, json={"error": str(request.url)})
try:
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
cancel_output = json.loads(
await execute_tool(
tool_name="sys_cancel_task",
arguments=json.dumps({"task_id": child_id}),
server_client=server_client,
conversation_id=parent_id,
session_async_tasks={},
)
)
finally:
runner_app.unregister_subagent_work(child_id)
# codex-native routes to interrupt, not stop_session (which the runner
# 204 no-ops for non-claude-native harnesses). ``data`` rides along for
# SessionEventInput compatibility with older servers.
assert stops == [{"type": "interrupt", "data": {}}]
assert cancel_output == {
"cancel_requested": True,
"cancel_confirmed": False,
"best_effort": True,
"task_id": child_id,
"status": "launching",
"message": (
"Interrupt forwarded, but a runner-side hard-stop is not wired "
"for codex-native workers yet; the child may keep running and no "
"terminal inbox status is guaranteed."
),
}
@pytest.mark.asyncio
async def test_sys_cancel_task_interrupts_non_native_subagent() -> None:
"""
A non-native (in-process) sub-agent cancel must post ``interrupt``.
In-process harnesses (e.g. ``claude-sdk``) have no wrapper label. The
runner's ``stop_session`` handler 204 no-ops for them, so posting
``stop_session`` would silently leave the child running and the parent
work entry stuck ``running``. ``interrupt`` is the path they honor —
``_interrupted_sessions`` → ``_on_proxy_stream_end`` marks the turn
cancelled and wakes the parent. This guards against regressing to an
unconditional ``stop_session`` (which dropped in-process cancellation).
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
parent_id = "conv_parent_inproc_cancel"
child_id = "conv_child_inproc_cancel"
runner_app.register_subagent_work(
parent_session_id=parent_id,
child_session_id=child_id,
agent="researcher",
title="analysis",
wrapper_label=None, # in-process child has no native wrapper label
)
posts: list[dict[str, Any]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Accept the interrupt; the in-process turn is cancelled out-of-band.
:param request: Request sent to the child session events route.
:returns: Accepted response (deferred cancel, child still running).
"""
if request.method == "POST" and request.url.path == f"/v1/sessions/{child_id}/events":
posts.append(json.loads(request.content))
return httpx.Response(204)
return httpx.Response(404, json={"error": str(request.url)})
try:
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
cancel_output = json.loads(
await execute_tool(
tool_name="sys_cancel_task",
arguments=json.dumps({"task_id": child_id}),
server_client=server_client,
conversation_id=parent_id,
session_async_tasks={},
)
)
finally:
runner_app.unregister_subagent_work(child_id)
# The regression guard: a non-native child must route to interrupt, never
# the stop_session that the runner no-ops for non-claude-native harnesses.
# ``data`` rides along for SessionEventInput compatibility with older servers.
assert posts == [{"type": "interrupt", "data": {}}]
# Not codex → generic (non-best-effort) pending result; the terminal
# status will arrive on the inbox once the interrupted turn ends.
assert cancel_output == {
"cancel_requested": True,
"cancel_confirmed": False,
"task_id": child_id,
"status": "launching",
"message": (
"Cancel requested; cancellation has not been confirmed yet. "
"Use sys_read_inbox to observe terminal status."
),
}
def test_session_status_to_task_status_maps_known_values() -> None:
"""
``_session_status_to_task_status`` maps a session.status value to the
child-summary ``current_task_status`` (different vocabularies), and
returns None for unknown values so the caller omits the field.
"""
from omnigent.runner.app import _session_status_to_task_status
assert _session_status_to_task_status("launching") == "launching"
assert _session_status_to_task_status("running") == "in_progress"
assert _session_status_to_task_status("waiting") == "in_progress"
assert _session_status_to_task_status("idle") == "completed"
assert _session_status_to_task_status("failed") == "failed"
assert _session_status_to_task_status("bogus") is None
def test_truncate_child_preview_caps_with_ellipsis() -> None:
"""
``_truncate_child_preview`` returns short text unchanged and truncates
text past the cap to exactly the cap + a single ellipsis char (so the
child rail preview matches the server-side truncation).
"""
from omnigent.runner.app import _CHILD_PREVIEW_MAX_CHARS, _truncate_child_preview
assert _truncate_child_preview("hello world") == "hello world"
long_text = "x" * (_CHILD_PREVIEW_MAX_CHARS + 50)
out = _truncate_child_preview(long_text)
assert out.endswith("…")
assert len(out) == _CHILD_PREVIEW_MAX_CHARS + 1
def test_register_unregister_child_session_roundtrip() -> None:
"""
``register_child_session`` stores the parent fan-out metadata and
``unregister_child_session`` drops it (used to mirror a child's
status/preview deltas onto the parent stream).
"""
from omnigent.runner.app import (
_child_session_parents,
register_child_session,
unregister_child_session,
)
child_id = "conv_child_roundtrip_unique"
register_child_session(
child_id,
parent_session_id="conv_parent_roundtrip_unique",
title="researcher:auth",
tool="researcher",
session_name="auth",
)
meta = _child_session_parents.get(child_id)
assert meta is not None
assert meta.parent_id == "conv_parent_roundtrip_unique"
assert meta.title == "researcher:auth"
assert meta.last_busy is None
unregister_child_session(child_id)
assert _child_session_parents.get(child_id) is None
# ── sys_session_get_history / _list / _close runner dispatch ────────
#
# These verify the runner-local handler that makes get_history/list/close
# work for harness agents (claude-sdk/codex/openai-agents), whose
# Omnigent tool calls surface as action_required and route through
# the runner — NOT the in-process inner Session. Confirmed empirically:
# without this dispatch the runner returns "not in local dispatch
# table"; with it, a live harness agent reads a sibling's items. The
# handler calls the Omnigent server's existing REST endpoints, so tests use a
# real httpx.AsyncClient backed by MockTransport (not a MagicMock) — the
# code exercises the same request/response objects it sees in production.
def _session_query_client(
handler: Callable[[httpx.Request], httpx.Response],
) -> httpx.AsyncClient:
"""
Build an AsyncClient whose requests are answered by ``handler``.
:param handler: Maps an ``httpx.Request`` to a canned
``httpx.Response`` (routes by method + path).
:returns: An ``httpx.AsyncClient`` pointed at a fake Omnigent server.
"""
return httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="http://server",
)
@pytest.mark.asyncio
async def test_session_list_maps_children_and_skips_closed() -> None:
"""
``sys_session_list`` maps ``child_sessions`` rows to
``{agent, title, conversation_id}`` and drops closed and
colonless rows, matching ``SysSessionListTool``.
"""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
def handler(request: httpx.Request) -> httpx.Response:
# Parent-detection snapshot: this caller is top-level (no parent),
# so there is no main/sibling enrichment — only its own children.
if request.url.path == "/v1/sessions/conv_parent":
return httpx.Response(200, json={"id": "conv_parent", "parent_session_id": None})
assert request.url.path == "/v1/sessions/conv_parent/child_sessions"
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "c1",
"title": "researcher:auth",
"tool": "researcher",
"session_name": "auth",
},
{
"id": "c2",
"title": "ui:claude-native-ui:1",
"tool": "claude-native-ui",
"session_name": "1",
},
{
"id": "c3",
"title": "researcher:done",
"tool": "researcher",
"session_name": "done",
"labels": {CLOSED_LABEL_KEY: CLOSED_LABEL_VALUE},
},
{
"id": "c5",
"title": "researcher:legacy:closed:c5",
"tool": "researcher",
"session_name": "legacy",
},
{
"id": "c4",
"title": "legacy-untyped",
"tool": "legacy-untyped",
"session_name": None,
},
],
},
)
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_list", "{}", conversation_id="conv_parent", server_client=client
)
)
# c3 (explicitly closed), c5 (legacy title tombstone), and c4
# (no colon) dropped; the ui:-added child surfaces under its bound
# agent + label.
assert out["sub_agents"] == [
{"agent": "researcher", "title": "auth", "conversation_id": "c1"},
{"agent": "claude-native-ui", "title": "1", "conversation_id": "c2"},
]
@pytest.mark.asyncio
async def test_session_list_adds_main_and_siblings_for_child_caller() -> None:
"""
When the caller is itself a child (a user-added agent), sys_session_list
also surfaces ``main`` (its parent) and its siblings — so an added agent
with no children of its own can still discover the conversation_ids to
peek. The caller is excluded from its own sibling list.
"""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
def handler(request: httpx.Request) -> httpx.Response:
path = request.url.path
if path == "/v1/sessions/conv_added/child_sessions":
# The added agent has no children of its own.
return httpx.Response(200, json={"object": "list", "data": []})
if path == "/v1/sessions/conv_added":
# It IS a child of conv_main.
return httpx.Response(200, json={"id": "conv_added", "parent_session_id": "conv_main"})
if path == "/v1/sessions/conv_main/child_sessions":
# main's children = the added agent itself + one sibling.
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "conv_added",
"title": "ui:claude-native-ui:1",
"tool": "claude-native-ui",
"session_name": "1",
},
{
"id": "conv_sib",
"title": "researcher:auth",
"tool": "researcher",
"session_name": "auth",
},
],
},
)
raise AssertionError(f"unexpected path {path}")
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_list", "{}", conversation_id="conv_added", server_client=client
)
)
# No own children; gains main (its parent) + the sibling, with itself
# excluded from the sibling list.
assert out["sub_agents"] == [
{"agent": "main", "title": None, "conversation_id": "conv_main"},
{"agent": "researcher", "title": "auth", "conversation_id": "conv_sib"},
]
@pytest.mark.asyncio
async def test_session_peek_returns_chronological_projected_items() -> None:
"""
``sys_session_get_history`` reads ``GET /items`` (newest-first), reverses to
chronological, projects each item, and labels with the target's
parsed agent/title from its snapshot — matching ``SysSessionGetHistoryTool``.
"""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/v1/sessions/conv_target/items":
assert request.url.params["order"] == "desc"
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "i2",
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "found it"}],
},
{
"id": "i1",
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "where is the bug"}],
},
],
},
)
if request.url.path == "/v1/sessions/conv_target":
return httpx.Response(200, json={"id": "conv_target", "title": "researcher:auth"})
raise AssertionError(f"unexpected path {request.url.path}")
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_get_history",
json.dumps({"conversation_id": "conv_target", "tail_items": 5}),
conversation_id="conv_caller",
server_client=client,
)
)
assert out["conversation_id"] == "conv_target"
assert out["agent"] == "researcher"
assert out["title"] == "auth"
# Reversed to chronological (user ask first), and message text is
# extracted from the content blocks — proves the projection ran.
assert [(i["role"], i["text"]) for i in out["items"]] == [
("user", "where is the bug"),
("assistant", "found it"),
]
@pytest.mark.asyncio
async def test_session_peek_appends_pending_elicitation_from_snapshot() -> None:
"""
``sys_session_get_history`` appends the target's parked elicitations (read
off its snapshot) after the stored items.
A parked elicitation never lands in the conversation store, so the
``/items`` response ends at the last message. The snapshot's
``pending_elicitations`` block is the only place the prompt lives;
get_history must read it and project a ``pending_elicitation`` item so the
parent agent isn't blind to a sub-agent awaiting input.
"""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/v1/sessions/conv_target/items":
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "i1",
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "ask me 3 questions"}],
},
],
},
)
if request.url.path == "/v1/sessions/conv_target":
return httpx.Response(
200,
json={
"id": "conv_target",
"title": "researcher:auth",
"pending_elicitations": [
{
"type": "response.elicitation_request",
"elicitation_id": "elicit_bio",
"params": {
"mode": "form",
"message": "Answer 3 questions on human biology",
"requestedSchema": {"properties": {"q1": {}, "q2": {}}},
},
}
],
},
)
raise AssertionError(f"unexpected path {request.url.path}")
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_get_history",
json.dumps({"conversation_id": "conv_target", "tail_items": 5}),
conversation_id="conv_caller",
server_client=client,
)
)
items = out["items"]
# 2 = 1 stored message + 1 synthesized pending elicitation. If 1,
# the snapshot's pending_elicitations weren't read/appended and the
# parent stays blind to the prompt.
assert len(items) == 2
# Stored item first (chronological), elicitation appended last.
assert items[0]["type"] == "message"
elicit = items[-1]
assert elicit["type"] == "pending_elicitation"
assert elicit["elicitation_id"] == "elicit_bio"
# Prompt + fields prove the snapshot payload reached the projector.
assert elicit["prompt"] == "Answer 3 questions on human biology"
assert elicit["fields"] == ["q1", "q2"]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"status,expected_error",
[(404, "session_not_found"), (403, "session_out_of_tree")],
)
async def test_session_peek_maps_access_errors(status: int, expected_error: str) -> None:
"""A 404/403 from ``GET /items`` maps to the in-process tool's typed errors."""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(status, json={"error": "x"})
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_get_history",
json.dumps({"conversation_id": "conv_other"}),
conversation_id="conv_caller",
server_client=client,
)
)
assert out["error"] == expected_error
assert out["conversation_id"] == "conv_other"
@pytest.mark.asyncio
async def test_session_close_patches_tombstoned_title() -> None:
"""
``sys_session_close`` PATCHes a closed label and internal tombstone.
The title tombstone frees the DB unique slot so future
``sys_session_send`` of the same ``(agent, title)`` creates a
fresh child. The ``omnigent.closed=true`` label is the
behavioral marker that direct write paths and clients consume.
The caller (``conv_caller``) and target (``conv_target``) share the
same ``root_conversation_id`` and the target is a sub-agent, so the
tree-scope gate passes and the PATCH is issued.
"""
# _execute_session_query_tool is the runner's REST dispatch entry
# point for session-query tools — called directly here because these
# tests validate the REST path's tree-scoping (_session_close_via_rest)
# specifically, distinct from the in-process path covered in
# tests/tools/builtins/test_sys_session.py.
from omnigent.runner.tool_dispatch import _execute_session_query_tool
patched: dict[str, Any] = {}
def handler(request: httpx.Request) -> httpx.Response:
if request.method == "GET" and request.url.path == "/v1/sessions/conv_target":
return httpx.Response(
200,
json={
"id": "conv_target",
"title": "researcher:auth",
"root_conversation_id": "conv_root",
"parent_session_id": "conv_caller",
},
)
if request.method == "GET" and request.url.path == "/v1/sessions/conv_caller":
return httpx.Response(
200,
json={"id": "conv_caller", "root_conversation_id": "conv_root"},
)
if request.method == "PATCH" and request.url.path == "/v1/sessions/conv_target":
patched.update(json.loads(request.content))
return httpx.Response(200, json={"id": "conv_target"})
raise AssertionError(f"unexpected {request.method} {request.url.path}")
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_close",
json.dumps({"conversation_id": "conv_target"}),
conversation_id="conv_caller",
server_client=client,
)
)
# Tombstone embeds the conv id so repeated closes stay unique, and
# the explicit label makes the closed state observable without
# exposing the suffix as UI text.
assert patched["title"] == "researcher:auth:closed:conv_target"
assert patched["labels"] == {CLOSED_LABEL_KEY: CLOSED_LABEL_VALUE}
assert out == {
"closed": True,
"conversation_id": "conv_target",
"agent": "researcher",
"title": "auth",
}
@pytest.mark.asyncio
async def test_session_close_rejects_out_of_tree_target_without_patch() -> None:
"""
``sys_session_close`` refuses a target in a different spawn tree and
issues NO PATCH.
Close is a write: the REST path must enforce the same tree-scoping as
the in-process path. Here the target's ``root_conversation_id``
(``conv_other_root``) differs from the caller's (``conv_root``), so
the tool returns ``session_out_of_tree`` and the tombstone PATCH is
never sent — proving the target's title is left intact. Without the
gate, edit access alone would let an agent close a sub-agent in one
of its other, unrelated trees.
"""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
patched = False
def handler(request: httpx.Request) -> httpx.Response:
nonlocal patched
if request.method == "GET" and request.url.path == "/v1/sessions/conv_target":
return httpx.Response(
200,
json={
"id": "conv_target",
"title": "researcher:auth",
"root_conversation_id": "conv_other_root",
"parent_session_id": "conv_other_parent",
},
)
if request.method == "GET" and request.url.path == "/v1/sessions/conv_caller":
return httpx.Response(
200,
json={"id": "conv_caller", "root_conversation_id": "conv_root"},
)
if request.method == "PATCH":
patched = True
return httpx.Response(200, json={"id": "conv_target"})
raise AssertionError(f"unexpected {request.method} {request.url.path}")
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_close",
json.dumps({"conversation_id": "conv_target"}),
conversation_id="conv_caller",
server_client=client,
)
)
assert out == {"error": "session_out_of_tree", "conversation_id": "conv_target"}
# The tombstone write must never have been issued.
assert patched is False
@pytest.mark.asyncio
async def test_session_close_rejects_top_level_target() -> None:
"""
``sys_session_close`` refuses a top-level session (no parent) even
when it shares the caller's root, and issues no PATCH.
A top-level session in the caller's own tree (its root) has no
``parent_session_id``; close only operates on sub-agents, so the
tool returns ``session_not_a_sub_agent``.
"""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
patched = False
def handler(request: httpx.Request) -> httpx.Response:
nonlocal patched
if request.method == "GET" and request.url.path == "/v1/sessions/conv_root":
return httpx.Response(
200,
json={
"id": "conv_root",
"title": "some top-level title",
"root_conversation_id": "conv_root",
"parent_session_id": None,
},
)
if request.method == "PATCH":
patched = True
return httpx.Response(200, json={"id": "conv_root"})
raise AssertionError(f"unexpected {request.method} {request.url.path}")
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_close",
json.dumps({"conversation_id": "conv_root"}),
# Caller IS conv_root, so its self-snapshot is the same row.
conversation_id="conv_root",
server_client=client,
)
)
assert out == {"error": "session_not_a_sub_agent", "conversation_id": "conv_root"}
assert patched is False
def test_agent_tools_are_runner_local() -> None:
"""
``sys_agent_get`` / ``sys_agent_download`` dispatch locally in the
runner. If this regresses, native harnesses calling them fall
through to spec-callable resolution and the orchestrator can't
inspect or fork agents.
"""
from omnigent.runner.tool_dispatch import should_dispatch_locally
assert should_dispatch_locally("sys_agent_get") is True
assert should_dispatch_locally("sys_agent_download") is True
assert should_dispatch_locally("sys_agent_list") is True
@pytest.mark.asyncio
@pytest.mark.parametrize(
"tool_name",
[
pytest.param("sys_agent_get", id="get"),
pytest.param("sys_agent_download", id="download"),
],
)
async def test_agent_tools_map_404_to_agent_not_found(
tool_name: str,
tmp_path: Path,
) -> None:
"""
Both agent tools map a 404 to ``agent_not_found`` — the orchestrator
gets a typed reason instead of a raw status. If the mapping
regressed, it couldn't tell "no such agent/session" from a transport
error.
:param tool_name: The agent tool under test.
:param tmp_path: Workspace dir (only the download path needs it).
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(404, json={"error": "missing"})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name=tool_name,
arguments=json.dumps({"session_id": "conv_missing"}),
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=tmp_path,
)
info = json.loads(output)
assert info["error"] == "agent_not_found"
assert info["session_id"] == "conv_missing"
@pytest.mark.parametrize(
"opt_in, expected_writes",
[
pytest.param("none", set(), id="no-opt-in"),
pytest.param(
"agents",
{"sys_session_send", "sys_session_close"},
id="declared-agents",
),
pytest.param(
"spawn",
{"sys_session_send", "sys_session_close", "sys_session_create"},
id="spawn-flag",
),
],
)
def test_native_relay_builtin_set_matches_toolmanager_gating(
opt_in: str,
expected_writes: set[str],
) -> None:
"""
The native relay advertises exactly ``ToolManager``'s builtin schemas
intersected with ``_NATIVE_RELAY_BUILTIN_TOOLS``.
claude-native / codex-native ignore the harness ``tools`` list, so the
relay is their only tool surface; ``_ensure_comment_relay_started``
applies this same ``ToolManager(spec).get_tool_schemas()`` ∩
``_NATIVE_RELAY_BUILTIN_TOOLS`` filter. This locks two invariants:
- **Parity**: the always-on orchestrator/discovery surface (agent
reads, session reads, async inbox reads/cancels, comment tools)
reaches native harnesses.
- **Gating fidelity**: the spawn writes are relayed per the two
distinct grants, matching what non-native harnesses get via
``request.tools``. ``tools.agents`` permits only the declared
sub-agent list (send + close, NO create); ``spawn: true`` (set
by the native wrapper specs) additionally grants create —
launching arbitrary agents or custom bundles. A regressed gate
either strands an opted-in native agent or hands an un-opted
agent the spawn surface.
:param opt_in: Which opt-in arm the spec uses — ``"none"``,
``"agents"`` (declared ``tools.agents``), or ``"spawn"``
(top-level ``spawn: true``).
:param expected_writes: Exact spawn-write tool names expected in
the relayed set for this opt-in arm.
"""
from omnigent.runner.tool_dispatch import _NATIVE_RELAY_BUILTIN_TOOLS
from omnigent.spec.types import ToolsConfig
from omnigent.tools.manager import ToolManager
if opt_in == "agents":
spec = AgentSpec(
spec_version=1,
tools=ToolsConfig(agents=["researcher"]),
sub_agents=[AgentSpec(spec_version=1, name="researcher")],
)
elif opt_in == "spawn":
spec = AgentSpec(spec_version=1, spawn=True)
else:
spec = AgentSpec(spec_version=1)
schema_names = {s["function"]["name"] for s in ToolManager(spec).get_tool_schemas()}
relayed = schema_names & _NATIVE_RELAY_BUILTIN_TOOLS
# Always-on reads/discovery reach every native agent — if any is
# missing, the orchestrator running under claude-native can't list or
# inspect agents/sessions.
assert {"sys_agent_get", "sys_agent_download", "sys_agent_list"} <= relayed
assert {"sys_session_list", "sys_session_get_history", "sys_session_get_info"} <= relayed
assert {"sys_call_async", "sys_read_inbox", "sys_cancel_async", "sys_cancel_task"} <= relayed
assert {"list_comments", "update_comment"} <= relayed
# Exact-set check on the writes: an extra name means a grant leaked
# beyond its arm (e.g. create from tools.agents alone — letting a
# whitelisted-sub-agents spec launch arbitrary bundles); a missing
# name strands an opted-in agent.
spawn_writes = {"sys_session_send", "sys_session_close", "sys_session_create"}
assert relayed & spawn_writes == expected_writes
# Model awareness rides the dispatch grant: relayed iff send is.
assert ("sys_list_models" in relayed) == ("sys_session_send" in expected_writes)
# OS tools ride a separate unconditional relay path (overriding the
# bridge's static versions), so they must never be in the builtin set —
# otherwise they'd be double-advertised and bypass that override.
os_tools = {"sys_os_read", "sys_os_write", "sys_os_edit", "sys_os_shell"}
assert not (os_tools & _NATIVE_RELAY_BUILTIN_TOOLS)
@pytest.mark.parametrize(
"declares_terminals, expected_terminal_tools",
[
pytest.param(
True,
{
"sys_terminal_launch",
"sys_terminal_send",
"sys_terminal_read",
"sys_terminal_list",
"sys_terminal_close",
},
id="terminals-declared",
),
pytest.param(False, set(), id="no-terminals"),
],
)
def test_native_relay_advertises_terminal_tools_per_spec_gate(
declares_terminals: bool,
expected_terminal_tools: set[str],
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
The native relay advertises ``sys_terminal_*`` iff the spec declares
``terminals``.
claude-native / codex-native ignore the harness ``tools`` list, so
the relay (``ToolManager(spec).get_tool_schemas()`` ∩
``_NATIVE_RELAY_BUILTIN_TOOLS``, applied by
``_ensure_comment_relay_started``) is the ONLY way the five
terminal tools reach the real CLI. This locks both directions of
the gate:
- **Advertised when granted**: a spec with a ``terminals:`` block
must relay all five tools — a missing name strands a native
agent that was granted terminals.
- **Withheld when not granted**: a spec without ``terminals:``
must relay none — a leaked name hands tmux access to an
un-opted agent.
:param declares_terminals: Whether the spec carries a
``terminals:`` block.
:param expected_terminal_tools: Exact ``sys_terminal_*`` names
expected in the relayed set for this arm.
:param monkeypatch: Pytest monkeypatch fixture, used to install a
fresh :class:`TerminalRegistry` singleton so ToolManager's
terminal-tool registration (which looks it up via
``get_terminal_registry()``) works without runtime ``init()``.
"""
from omnigent.inner.datamodel import TerminalEnvSpec
from omnigent.runner.tool_dispatch import _NATIVE_RELAY_BUILTIN_TOOLS
from omnigent.runtime import _globals as rt_globals
from omnigent.terminals.registry import TerminalRegistry
from omnigent.tools.manager import ToolManager
monkeypatch.setattr(rt_globals, "_terminal_registry", TerminalRegistry())
terminals = {"bash": TerminalEnvSpec(command="bash")} if declares_terminals else None
spec = AgentSpec(spec_version=1, terminals=terminals)
schema_names = {s["function"]["name"] for s in ToolManager(spec).get_tool_schemas()}
relayed = schema_names & _NATIVE_RELAY_BUILTIN_TOOLS
all_terminal_tools = {
"sys_terminal_launch",
"sys_terminal_send",
"sys_terminal_read",
"sys_terminal_list",
"sys_terminal_close",
}
# Exact-set check on the terminal family: a missing name means the
# relay filter dropped a granted tool (set regression in
# _NATIVE_RELAY_BUILTIN_TOOLS); an extra name on the no-terminals
# arm means ToolManager registered terminal tools without the spec
# gate (registration regression).
assert relayed & all_terminal_tools == expected_terminal_tools
def test_session_create_is_runner_local() -> None:
"""
``sys_session_create`` dispatches locally in the runner. If it
regresses out of the local table, a native harness calling it falls
through to spec-callable resolution and the orchestrator can't spawn
child sessions.
"""
from omnigent.runner.tool_dispatch import should_dispatch_locally
assert should_dispatch_locally("sys_session_create") is True
@pytest.mark.asyncio
async def test_session_list_global_sessions_filter_and_connectivity() -> None:
"""
The global ``sessions`` view fetches GET /v1/sessions (forwarding the
``agent_name`` filter), projects each row, and annotates
``runner_online`` by checking each UNIQUE runner once. Proves: the
agent_name filter reaches the server; sessions are projected with
status + parentage; and connectivity is folded in without a
per-session status fan-out (two sessions share runner r1 → exactly
one /v1/runners/r1/status call).
"""
from omnigent.runner.tool_dispatch import _execute_session_query_tool
runner_status_calls: list[str] = []
sessions_params: dict[str, str] = {}
def handler(request: httpx.Request) -> httpx.Response:
path = request.url.path
if path == "/v1/sessions/conv_x/child_sessions":
return httpx.Response(200, json={"object": "list", "data": []})
if path == "/v1/sessions/conv_x":
return httpx.Response(200, json={"id": "conv_x", "parent_session_id": None})
if path == "/v1/sessions":
sessions_params.update(dict(request.url.params))
return httpx.Response(
200,
json={
"object": "list",
"data": [
{
"id": "s1",
"agent_name": "researcher",
"title": "auth",
"status": "running",
"runner_id": "r1",
"parent_session_id": None,
},
{
"id": "s2",
"agent_name": "researcher",
"title": "payments",
"status": "idle",
"runner_id": "r1",
"parent_session_id": None,
},
],
},
)
if path == "/v1/runners/r1/status":
runner_status_calls.append("r1")
return httpx.Response(200, json={"runner_id": "r1", "online": True})
raise AssertionError(f"unexpected path {path}")
async with _session_query_client(handler) as client:
out = json.loads(
await _execute_session_query_tool(
"sys_session_list",
json.dumps({"agent_name": "researcher"}),
conversation_id="conv_x",
server_client=client,
)
)
# agent_name forwarded to the server-side filter.
assert sessions_params.get("agent_name") == "researcher"
# Both sessions projected with status + connectivity from the single
# shared-runner status lookup.
assert out["sessions"] == [
{
"session_id": "s1",
"agent_name": "researcher",
"title": "auth",
"status": "running",
"runner_id": "r1",
"runner_online": True,
"parent_session_id": None,
},
{
"session_id": "s2",
"agent_name": "researcher",
"title": "payments",
"status": "idle",
"runner_id": "r1",
"runner_online": True,
"parent_session_id": None,
},
]
# Connectivity resolved once per UNIQUE runner — two sessions share
# r1, so exactly one status call (not one per session). A count of 2
# would mean the dedup regressed into a per-session fan-out.
assert runner_status_calls == ["r1"]
@pytest.mark.asyncio
async def test_sys_agent_download_rejects_path_in_dest_filename(tmp_path: Path) -> None:
"""
``sys_agent_download`` rejects a ``dest_filename`` containing a path
separator (a traversal attempt) and writes nothing. If the guard
regressed, a bundle could be written outside the working directory.
:param tmp_path: Pytest temp dir used as the runner workspace.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
content=b"data",
headers={"X-Agent-Name": "a", "X-Agent-Version": "1"},
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_agent_download",
arguments=json.dumps({"session_id": "conv_x", "dest_filename": "../escape.tar.gz"}),
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=tmp_path,
)
info = json.loads(output)
assert "error" in info
# Nothing was written anywhere under the workspace.
assert list(tmp_path.iterdir()) == []
@pytest.mark.asyncio
async def test_sys_agent_download_rejects_symlink_escape_from_cwd(tmp_path: Path) -> None:
"""
``sys_agent_download`` refuses to follow a symlink that redirects the
bundle write outside the os_env cwd. ``dest_filename`` is a bare name,
but if the cwd already holds a symlink of that name pointing elsewhere,
a naive ``write_bytes`` would clobber the symlink target outside the
sandbox. The realpath-containment guard must catch it and
write nothing to the outside target.
:param tmp_path: Pytest temp dir; holds both the workspace and an
outside directory the symlink points at.
"""
from omnigent.runner.tool_dispatch import execute_tool
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
outside_target = outside / "stolen.tar.gz"
# A symlink inside the workspace whose name matches the caller's
# dest_filename but whose target escapes the workspace.
(workspace / "escape.tar.gz").symlink_to(outside_target)
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
content=b"payload",
headers={"X-Agent-Name": "a", "X-Agent-Version": "1"},
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_agent_download",
arguments=json.dumps({"session_id": "conv_x", "dest_filename": "escape.tar.gz"}),
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=workspace,
)
info = json.loads(output)
assert "error" in info
# The outside target was never created — the guard blocked the write.
assert not outside_target.exists()
@pytest.mark.asyncio
async def test_sys_agent_download_writes_bundle_to_workspace(tmp_path: Path) -> None:
"""
``sys_agent_download`` writes the fetched ``.tar.gz`` bytes into the
agent's os_env cwd (here ``runner_workspace`` = tmp_path) and returns
the path. Proves the full path: fetch bytes, derive the default
filename from the X-Agent-* headers, and persist to the agent-visible
disk. If the write regressed, the file wouldn't exist or the bytes
wouldn't match.
:param tmp_path: Pytest temp dir used as the runner workspace, so the
resolved os_env cwd is a real local directory.
"""
from omnigent.runner.tool_dispatch import execute_tool
bundle_bytes = b"\x1f\x8b\x08fake-tar-gz-bytes"
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/v1/sessions/conv_x/agent/contents":
return httpx.Response(
200,
content=bundle_bytes,
headers={"X-Agent-Name": "my agent", "X-Agent-Version": "5"},
)
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_agent_download",
arguments=json.dumps({"session_id": "conv_x"}),
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=tmp_path,
)
info = json.loads(output)
# Default filename: agent name sanitized (space → "_") + version.
expected = tmp_path / "my_agent-v5.tar.gz"
assert info["path"] == str(expected)
assert info["bytes_written"] == len(bundle_bytes)
# The bundle actually landed on disk with the exact bytes — a
# mismatch means the write path or byte handling broke.
assert expected.read_bytes() == bundle_bytes
@pytest.mark.asyncio
async def test_sys_agent_get_projects_agent_metadata() -> None:
"""
``sys_agent_get`` projects ``GET /v1/sessions/{id}/agent`` into the
orchestrator-facing fields: agent_id, name, version, description,
harness, MCP server summaries, and policy summaries. If the
projection dropped a field or used the wrong key, the asserted
values would differ.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.method == "GET" and request.url.path == "/v1/sessions/conv_x/agent":
return httpx.Response(
200,
json={
"id": "ag_777",
"object": "agent",
"name": "researcher",
"version": 4,
"description": "Finds things",
"created_at": 1,
"harness": "claude-sdk",
"mcp_servers": [{"name": "fs", "transport": "stdio", "args": []}],
"policies": [{"name": "guard", "type": "label", "on": ["input"]}],
},
)
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_agent_get",
arguments=json.dumps({"session_id": "conv_x"}),
server_client=server_client,
conversation_id="conv_caller",
)
info = json.loads(output)
# agent_id comes from AgentObject.id (not a top-level "agent_id").
assert info["agent_id"] == "ag_777"
assert info["name"] == "researcher"
assert info["version"] == 4
assert info["harness"] == "claude-sdk"
# MCP/policy summaries pass through as-is so the orchestrator sees
# the agent's tool/guardrail surface.
assert info["mcp_servers"] == [{"name": "fs", "transport": "stdio", "args": []}]
assert info["policies"] == [{"name": "guard", "type": "label", "on": ["input"]}]
@pytest.mark.asyncio
async def test_sys_agent_list_degrades_when_sources_fail(tmp_path: Path) -> None:
"""
A failing source degrades to an empty section rather than failing the
whole call. Here the server 500s both list endpoints and no local
config dir exists, so all three sections come back empty — but the
tool still returns a well-formed result. If a source error
propagated, the tool would return an ``error`` instead.
:param tmp_path: Workspace dir with no agent-config subdir.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(500, json={"error": "boom"})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_agent_list",
arguments="{}",
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=tmp_path,
)
info = json.loads(output)
assert info == {"builtins": [], "session_agents": [], "local_configs": []}
@pytest.mark.asyncio
async def test_sys_agent_list_merges_three_sources(tmp_path: Path) -> None:
"""
``sys_agent_list`` merges built-ins (GET /v1/agents), session-bound
agents (GET /v1/sessions), and locally-authored config YAMLs (a scan
of the os_env cwd's agent-config subdir). Proves all three sources
are fetched and projected: a built-in's id/name, a session's
session_id+agent binding, and a local config's name/path from the
YAML on disk. If any source were dropped or mis-projected, the
corresponding section would be empty or wrong.
:param tmp_path: Pytest temp dir used as the runner workspace, so the
local-config scan reads a real directory.
"""
from omnigent.runner.tool_dispatch import _AGENT_CONFIG_SUBDIR, execute_tool
# Author a local config on disk so the scan has something to find.
configs_dir = tmp_path / _AGENT_CONFIG_SUBDIR
configs_dir.mkdir(parents=True)
(configs_dir / "my-agent.yaml").write_text(
"name: my-agent\ndescription: a local one\nprompt: hi\n", encoding="utf-8"
)
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/v1/agents":
return httpx.Response(
200,
json={"data": [{"id": "ag_b", "name": "claude-native-ui", "harness": "claude"}]},
)
if request.url.path == "/v1/sessions":
return httpx.Response(
200,
json={
"data": [
{
"id": "conv_1",
"agent_id": "ag_s",
"agent_name": "nessie",
"status": "idle",
}
]
},
)
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_agent_list",
arguments="{}",
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=tmp_path,
)
info = json.loads(output)
# Built-ins projected from GET /v1/agents (id → agent_id).
assert info["builtins"] == [
{"agent_id": "ag_b", "name": "claude-native-ui", "description": None, "harness": "claude"}
]
# Session-bound agents carry session_id so the caller can then
# sys_agent_get / sys_agent_download them.
assert info["session_agents"] == [
{"session_id": "conv_1", "agent_id": "ag_s", "agent_name": "nessie", "status": "idle"}
]
# Local config discovered by the on-disk scan, with its parsed name.
assert len(info["local_configs"]) == 1
assert info["local_configs"][0]["name"] == "my-agent"
assert info["local_configs"][0]["description"] == "a local one"
assert info["local_configs"][0]["path"].endswith("my-agent.yaml")
@pytest.mark.asyncio
async def test_sys_session_create_maps_agent_not_found() -> None:
"""
A 404 from the create maps to ``agent_not_found`` so the LLM gets a
typed reason rather than a raw status. If the mapping regressed, the
orchestrator couldn't tell a bad agent_id from a transport failure.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(404, json={"error": "no agent"})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_create",
arguments=json.dumps({"agent_id": "ag_missing"}),
server_client=server_client,
conversation_id="conv_caller",
)
info = json.loads(output)
assert info["error"] == "agent_not_found"
assert info["agent_id"] == "ag_missing"
@pytest.mark.asyncio
async def test_sys_session_create_spawns_child_under_caller() -> None:
"""
``sys_session_create`` POSTs a JSON create with
``parent_session_id`` forced to the caller (child-only), passes the
agent_id, title, and a queued initial message, and returns a handle
carrying the new child's id. If parent_session_id weren't forced to
the caller, an orchestrator could create top-level/sibling sessions —
so the asserted request body is the security-critical check.
"""
from omnigent.runner.tool_dispatch import execute_tool
captured: dict[str, Any] = {}
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.method == "POST" and request.url.path == "/v1/sessions":
captured.update(json.loads(request.content))
return httpx.Response(
201,
json={
"id": "conv_child",
"agent_id": "ag_x",
"agent_name": "researcher",
"status": "idle",
},
)
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_create",
arguments=json.dumps({"agent_id": "ag_x", "title": "auth", "message": "start"}),
server_client=server_client,
conversation_id="conv_caller",
)
# Child-only: parent forced to the caller; agent + title + queued
# message threaded through to the create body.
assert captured["parent_session_id"] == "conv_caller"
assert captured["agent_id"] == "ag_x"
assert captured["title"] == "auth"
assert captured["initial_items"][0]["data"]["content"][0]["text"] == "start"
handle = json.loads(output)
assert handle["conversation_id"] == "conv_child"
assert handle["agent_id"] == "ag_x"
assert handle["agent_name"] == "researcher"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"arguments",
[
# Both modes at once: the two create different agents, so the
# handler must refuse rather than silently pick one.
{"agent_id": "ag_x", "config_path": "helper.yaml"},
# Neither mode: nothing to launch.
{"title": "auth"},
],
)
async def test_sys_session_create_requires_exactly_one_mode(
arguments: dict[str, Any],
) -> None:
"""
``sys_session_create`` rejects both-or-neither of ``agent_id`` /
``config_path`` without touching the server.
If the mode split regressed to a silent preference, an orchestrator
passing both could launch the wrong agent with no signal.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
raise AssertionError(f"server must not be reached on invalid mode args: {request.url}")
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_create",
arguments=json.dumps(arguments),
server_client=server_client,
conversation_id="conv_caller",
)
info = json.loads(output)
assert "exactly one of 'agent_id'" in info["error"]
def _parse_multipart_create(request: httpx.Request) -> dict[str, Any]:
"""
Decode a captured multipart ``POST /v1/sessions`` request body.
Uses the stdlib email parser (multipart/form-data is MIME) rather
than hand-rolled boundary splitting.
:param request: The captured httpx request.
:returns: Dict with ``metadata`` (parsed JSON dict) and ``bundle``
(raw bytes of the uploaded file part).
"""
import email
import email.policy
header = f"Content-Type: {request.headers['content-type']}\r\n\r\n".encode()
message = email.message_from_bytes(header + request.content, policy=email.policy.HTTP)
parts: dict[str, Any] = {}
for part in message.iter_parts(): # type: ignore[attr-defined]
disposition = part.get("Content-Disposition", "")
payload = part.get_payload(decode=True)
if 'name="metadata"' in disposition:
parts["metadata"] = json.loads(payload.decode())
elif 'name="bundle"' in disposition:
parts["bundle"] = payload
return parts
@pytest.mark.asyncio
async def test_sys_session_create_bundle_mode_uploads_child_under_caller(
tmp_path: Path,
) -> None:
"""
Bundle mode bundles a local agent config, POSTs the multipart
create with ``parent_session_id`` forced to the caller, queues the
optional message as the child's first event, and returns a handle
built from the server's created-agent identifiers.
The asserted multipart metadata is the security-critical check
(child-only, mirroring agent_id mode); the tarball content check
proves the local config actually traversed bundling — an empty
bundle would create a session the server rejects at first turn.
"""
import io
import tarfile
from omnigent.runner.tool_dispatch import execute_tool
config_text = "name: helper\nprompt: do helpful things\n"
(tmp_path / "helper.yaml").write_text(config_text)
create_requests: list[httpx.Request] = []
event_bodies: list[dict[str, Any]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.method == "POST" and request.url.path == "/v1/sessions":
create_requests.append(request)
return httpx.Response(
201,
json={
"session_id": "conv_child",
"agent_id": "ag_new",
"agent_name": "helper",
},
)
if request.method == "POST" and request.url.path == "/v1/sessions/conv_child/events":
event_bodies.append(json.loads(request.content))
return httpx.Response(200, json={})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_create",
arguments=json.dumps(
{"config_path": "helper.yaml", "title": "auth", "message": "start"}
),
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=tmp_path,
)
# Exactly one multipart create; parent forced to the caller
# (child-only) and the title threaded into the metadata part.
assert len(create_requests) == 1, (
f"expected exactly one create POST, got {len(create_requests)}"
)
parts = _parse_multipart_create(create_requests[0])
assert parts["metadata"] == {"parent_session_id": "conv_caller", "title": "auth"}
# The uploaded bundle is a gzipped tar holding the authored config
# verbatim — proves the local file traversed materialize → tar.
with tarfile.open(fileobj=io.BytesIO(parts["bundle"]), mode="r:gz") as tf:
names = tf.getnames()
assert names == ["helper.yaml"]
member = tf.extractfile("helper.yaml")
assert member is not None
assert member.read().decode() == config_text
# The optional message was queued as the child's first user event —
# this is what starts the child's turn (same pattern as named send).
assert len(event_bodies) == 1
assert event_bodies[0]["data"]["content"][0]["text"] == "start"
handle = json.loads(output)
assert handle["conversation_id"] == "conv_child"
# agent_id/agent_name come from the server's CreatedSessionResponse,
# not the caller's args — the orchestrator needs the NEW agent's id.
assert handle["agent_id"] == "ag_new"
assert handle["agent_name"] == "helper"
@pytest.mark.asyncio
async def test_sys_session_create_config_path_escape_rejected(
tmp_path: Path,
) -> None:
"""
A ``config_path`` resolving outside the working directory is
refused before any disk read or server call.
This mirrors the sys_agent_download containment guard:
without it, an orchestrator could exfiltrate arbitrary host files
by bundling them into an uploaded agent.
"""
from omnigent.runner.tool_dispatch import execute_tool
workdir = tmp_path / "work"
workdir.mkdir()
(tmp_path / "outside.yaml").write_text("name: outside\n")
async def _server_handler(request: httpx.Request) -> httpx.Response:
raise AssertionError(f"server must not be reached on escape: {request.url}")
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_create",
arguments=json.dumps({"config_path": "../outside.yaml"}),
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=workdir,
)
info = json.loads(output)
assert "escapes the working directory" in info["error"]
@pytest.mark.asyncio
async def test_sys_session_create_config_not_found(tmp_path: Path) -> None:
"""
A missing ``config_path`` returns the typed ``config_not_found``
error so the LLM can distinguish a bad path from a transport
failure, without any server call.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
raise AssertionError(f"server must not be reached: {request.url}")
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_create",
arguments=json.dumps({"config_path": "nope.yaml"}),
server_client=server_client,
conversation_id="conv_caller",
runner_workspace=tmp_path,
)
info = json.loads(output)
assert info["error"] == "config_not_found"
assert info["config_path"] == "nope.yaml"
@pytest.mark.asyncio
async def test_sys_session_get_info_defaults_to_caller_session() -> None:
"""
Omitting ``session_id`` describes the caller's own session — the
runner targets ``GET /v1/sessions/{conversation_id}``. With no
runner bound, connectivity is unknown (``None``) and no
runner-status call is made. If the default-to-caller logic
regressed, the request path would be wrong and the GET would 404.
"""
from omnigent.runner.tool_dispatch import execute_tool
requested_paths: list[str] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
requested_paths.append(request.url.path)
if request.url.path == "/v1/sessions/conv_caller":
return httpx.Response(
200,
json={
"id": "conv_caller",
"agent_id": "ag_self",
"agent_name": "main",
"status": "idle",
"created_at": 1,
"runner_id": None,
"pending_elicitations": [],
},
)
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_get_info",
arguments="{}",
server_client=server_client,
conversation_id="conv_caller",
)
info = json.loads(output)
assert info["session_id"] == "conv_caller"
# No runner bound → connectivity unknown, and the status endpoint is
# never queried (a stray /v1/runners call would mean the None-runner
# short-circuit regressed).
assert info["runner_online"] is None
assert info["pending_elicitations"] == []
assert info["pending_elicitation_count"] == 0
assert not any(p.startswith("/v1/runners") for p in requested_paths)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"status_code,expected_error",
[
pytest.param(404, "session_not_found", id="not-found"),
pytest.param(403, "access_denied", id="forbidden"),
pytest.param(401, "access_denied", id="unauthorized"),
],
)
async def test_sys_session_get_info_maps_error_statuses(
status_code: int,
expected_error: str,
) -> None:
"""
A 404 maps to ``session_not_found``; 401/403 map to
``access_denied`` — so the LLM gets a typed reason instead of a raw
HTTP status. If the mapping regressed, the orchestrator couldn't
distinguish "no such session" from "you can't read it".
:param status_code: HTTP status the mocked Omnigent server returns.
:param expected_error: The typed error string the tool should emit.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(status_code, json={"error": "x"})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_get_info",
arguments=json.dumps({"session_id": "conv_missing"}),
server_client=server_client,
conversation_id="conv_caller",
)
info = json.loads(output)
assert info["error"] == expected_error
assert info["session_id"] == "conv_missing"
@pytest.mark.asyncio
async def test_sys_session_share_defaults_to_caller_and_puts_grant() -> None:
"""
Omitting ``session_id`` shares the caller's own session: the runner
PUTs to ``/v1/sessions/{conversation_id}/permissions`` with the
grantee and the numeric level mapped from the friendly name. If the
default-to-caller logic or the name->level mapping regressed, the
request path or body would be wrong (and an agent's "share this
session" would silently hit the wrong session or wrong level).
"""
from omnigent.runner.tool_dispatch import execute_tool
requests: list[tuple[str, str, dict[str, Any]]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
requests.append((request.method, request.url.path, json.loads(request.content)))
return httpx.Response(
200,
json={"user_id": "alice@example.com", "conversation_id": "conv_caller", "level": 2},
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_share",
arguments=json.dumps({"user_id": "alice@example.com", "level": "edit"}),
server_client=server_client,
conversation_id="conv_caller",
# Sharing a named user only needs the non-public tier enabled.
agent_spec=AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.NON_PUBLIC),
)
# Exactly one PUT to the caller's own permissions sub-resource, with
# level "edit" mapped to the server's numeric 2 (1=read/2=edit/3=manage).
assert requests == [
(
"PUT",
"/v1/sessions/conv_caller/permissions",
{"user_id": "alice@example.com", "level": 2},
)
]
result = json.loads(output)
assert result == {
"shared": True,
"session_id": "conv_caller",
"user_id": "alice@example.com",
"level": "edit",
}
@pytest.mark.asyncio
@pytest.mark.parametrize(
"status_code,expected_error",
[
pytest.param(404, "session_not_found", id="not-found"),
pytest.param(403, "access_denied", id="forbidden"),
pytest.param(401, "access_denied", id="unauthorized"),
],
)
async def test_sys_session_share_maps_error_statuses(
status_code: int,
expected_error: str,
) -> None:
"""
A 404 maps to ``session_not_found``; 401/403 map to ``access_denied``
— a typed reason instead of a raw status, matching the sibling
session tools so the LLM can distinguish "no such session" from
"you can't manage it".
:param status_code: HTTP status the mocked Omnigent server returns.
:param expected_error: The typed error string the tool should emit.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(status_code, json={"detail": "x"})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_share",
arguments=json.dumps({"user_id": "alice@example.com", "session_id": "conv_x"}),
server_client=server_client,
conversation_id="conv_caller",
agent_spec=AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.NON_PUBLIC),
)
result = json.loads(output)
assert result["error"] == expected_error
assert result["session_id"] == "conv_x"
@pytest.mark.asyncio
async def test_sys_session_share_rejects_bad_level_without_calling_server() -> None:
"""
An unknown ``level`` is rejected client-side before any PUT — so a
typo can't fall through to the server or silently skip the grant. A
request reaching the handler would mean the level validation
regressed.
"""
from omnigent.runner.tool_dispatch import execute_tool
called = False
async def _server_handler(request: httpx.Request) -> httpx.Response:
nonlocal called
called = True
return httpx.Response(200, json={})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_share",
arguments=json.dumps({"user_id": "alice@example.com", "level": "admin"}),
server_client=server_client,
conversation_id="conv_caller",
# Share enabled (non-public) so the call reaches level validation.
agent_spec=AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.NON_PUBLIC),
)
assert called is False # validation must short-circuit before the PUT
assert "level must be one of" in json.loads(output)["error"]
@pytest.mark.asyncio
async def test_sys_session_share_surfaces_server_message_on_4xx() -> None:
"""
A 4xx the typed branches don't claim (here the server's 400 for a
``__public__`` grant above read level) surfaces the server's own
``{"error": {"message": ...}}`` text rather than a bare "returned
400". If the detail-extraction regressed, the agent would see only
the status code and couldn't tell that public is read-only — the
exact actionable reason the server gave.
"""
from omnigent.runner.tool_dispatch import execute_tool
# Mirrors the OmnigentError envelope the server's exception handler
# emits (omnigent/server/app.py) for the public + level>read guard.
server_message = "Public access is limited to read-only (level 1)"
async def _server_handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
400, json={"error": {"code": "INVALID_INPUT", "message": server_message}}
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_share",
arguments=json.dumps(
{"user_id": "__public__", "level": "edit", "session_id": "conv_x"}
),
server_client=server_client,
conversation_id="conv_caller",
# agent_session_sharing: public lets __public__ pass the runner
# gate and reach the server, which rejects level>read for public.
agent_spec=AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.PUBLIC),
)
result = json.loads(output)
# The server's verbatim message is surfaced, not flattened to a status.
assert result["error"] == server_message
assert result["status_code"] == 400
assert result["session_id"] == "conv_x"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"share_policy",
[
pytest.param(None, id="no-spec"),
pytest.param(SharePolicy.NONE, id="share-none"),
],
)
async def test_sys_session_share_disabled_without_share_flag(
share_policy: SharePolicy | None,
) -> None:
"""
With no spec (``None``) or ``agent_session_sharing: none``, the
runner refuses the grant client-side and never PUTs — the
``agent_session_sharing`` flag is the real gate, not just tool
advertisement, so a prompt-injected call naming the tool can't
escalate. A PUT reaching the handler would mean the runner-side
policy gate regressed.
:param share_policy: The spec's ``agent_session_sharing`` policy
under test (or ``None`` for a missing spec).
"""
from omnigent.runner.tool_dispatch import execute_tool
called = False
async def _server_handler(request: httpx.Request) -> httpx.Response:
nonlocal called
called = True
return httpx.Response(200, json={})
spec = (
None
if share_policy is None
else AgentSpec(spec_version=1, agent_session_sharing=share_policy)
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_share",
arguments=json.dumps({"user_id": "alice@example.com", "session_id": "conv_x"}),
server_client=server_client,
conversation_id="conv_caller",
agent_spec=spec,
)
assert called is False # the gate must short-circuit before the PUT
assert "not enabled" in json.loads(output)["error"]
@pytest.mark.asyncio
async def test_sys_session_share_non_public_rejects_public_grant() -> None:
"""
Under ``agent_session_sharing: non-public`` a grant to a named user
is allowed, but a ``__public__`` grant is refused client-side before
any PUT — the
non-public tier must not be able to expose the transcript anonymously
even if the model (or an injection) asks for it. A PUT here would
mean the public sub-gate regressed.
"""
from omnigent.runner.tool_dispatch import execute_tool
called = False
async def _server_handler(request: httpx.Request) -> httpx.Response:
nonlocal called
called = True
return httpx.Response(200, json={})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_share",
arguments=json.dumps({"user_id": "__public__", "session_id": "conv_x"}),
server_client=server_client,
conversation_id="conv_caller",
agent_spec=AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.NON_PUBLIC),
)
assert called is False # public sub-gate must short-circuit before the PUT
assert "public" in json.loads(output)["error"]
@pytest.mark.asyncio
async def test_sys_session_share_public_allows_public_grant() -> None:
"""
Under ``agent_session_sharing: public`` a ``__public__`` read grant
passes the runner gate and PUTs to the permissions endpoint — the
positive case the
non-public/none gates exclude. If the gate wrongly blocked it, public
sharing would be impossible even when the spec explicitly opts in.
"""
from omnigent.runner.tool_dispatch import execute_tool
requests: list[tuple[str, str, dict[str, Any]]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
requests.append((request.method, request.url.path, json.loads(request.content)))
return httpx.Response(
200,
json={"user_id": "__public__", "conversation_id": "conv_caller", "level": 1},
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_share",
arguments=json.dumps({"user_id": "__public__"}),
server_client=server_client,
conversation_id="conv_caller",
agent_spec=AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.PUBLIC),
)
# __public__ reached the server as a level-1 (read) grant on the caller.
assert requests == [
("PUT", "/v1/sessions/conv_caller/permissions", {"user_id": "__public__", "level": 1})
]
result = json.loads(output)
assert result == {
"shared": True,
"session_id": "conv_caller",
"user_id": "__public__",
"level": "read",
}
@pytest.mark.asyncio
async def test_sys_session_get_info_projects_metadata_and_runner_connectivity() -> None:
"""
``sys_session_get_info`` projects ``GET /v1/sessions/{id}`` metadata
and folds in live runner connectivity from ``GET
/v1/runners/{id}/status``.
Proves the full runner-dispatch path: read the session snapshot,
derive the effective model (a per-session ``model_override`` wins
over the spec's ``llm_model``), count pending approval prompts, and
attach ``runner_online``. If the projection regressed (dropped a
field, skipped the runner-status call, or picked the wrong model),
the asserted values would differ. The transcript is intentionally
absent — get_info is metadata-only (``sys_session_get_history`` returns
items).
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.method == "GET" and request.url.path == "/v1/sessions/conv_target":
return httpx.Response(
200,
json={
"id": "conv_target",
"agent_id": "ag_xyz",
"agent_name": "researcher",
"status": "running",
"created_at": 1,
"title": "auth flow",
"runner_id": "runner_1",
"host_id": None,
"reasoning_effort": "high",
"parent_session_id": "conv_parent",
"sub_agent_name": "researcher",
"llm_model": "anthropic/claude-sonnet-4-6",
"model_override": "claude-opus-4-8",
"workspace": "/repo",
"git_branch": "feature/x",
"pending_elicitations": [{"id": "el_1"}, {"id": "el_2"}],
},
)
if request.method == "GET" and request.url.path == "/v1/runners/runner_1/status":
return httpx.Response(200, json={"runner_id": "runner_1", "online": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_get_info",
arguments=json.dumps({"session_id": "conv_target"}),
server_client=server_client,
conversation_id="conv_caller",
)
info = json.loads(output)
assert info["session_id"] == "conv_target"
assert info["status"] == "running"
assert info["title"] == "auth flow"
assert info["agent_id"] == "ag_xyz"
assert info["agent_name"] == "researcher"
assert info["runner_id"] == "runner_1"
# Live connectivity folded in from the runners status endpoint —
# None here would mean the best-effort status call was skipped.
assert info["runner_online"] is True
assert info["parent_session_id"] == "conv_parent"
# Effective model: the per-session override wins over the spec
# default. "anthropic/claude-sonnet-4-6" here would mean the
# override was ignored.
assert info["model"] == "claude-opus-4-8"
# Two outstanding approval prompts surfaced from the snapshot — both
# the prompts themselves and a count. If the projection dropped the
# prompts (count-only), the orchestrator couldn't tell what the
# blocked session is waiting on.
assert info["pending_elicitation_count"] == 2
assert info["pending_elicitations"] == [{"id": "el_1"}, {"id": "el_2"}]
# Metadata-only: the full transcript is never embedded.
assert "items" not in info
@pytest.mark.asyncio
async def test_sys_session_get_info_hides_native_ui_wrapper_agent_name() -> None:
"""A native-UI session describes itself with its clean public name.
Regression for the leak where ``sys_session_get_info`` returned the raw
bound ``agent_name`` ``"pi-native-ui"``, which the Pi agent then repeated to
the user ("I'm pi (agent name: pi-native-ui)"). The projection must map the
internal ``-native-ui`` wrapper name to its display name (``"Pi"``) so the
implementation detail never reaches the model.
"""
from omnigent.runner.tool_dispatch import execute_tool
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.method == "GET" and request.url.path == "/v1/sessions/conv_pi":
return httpx.Response(
200,
json={
"id": "conv_pi",
"agent_id": "ag_pi",
"agent_name": "pi-native-ui",
"status": "running",
"title": "hi what agent are you?",
"runner_id": "runner_1",
},
)
if request.method == "GET" and request.url.path == "/v1/runners/runner_1/status":
return httpx.Response(200, json={"runner_id": "runner_1", "online": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
output = await execute_tool(
tool_name="sys_session_get_info",
arguments=json.dumps({"session_id": "conv_pi"}),
server_client=server_client,
conversation_id="conv_pi",
)
info = json.loads(output)
# The internal wrapper name is rewritten to the public display name; the
# raw ``pi-native-ui`` must not appear anywhere in the tool output.
assert info["agent_name"] == "Pi"
assert "pi-native-ui" not in output
@pytest.mark.asyncio
async def test_sys_session_send_session_id_posts_to_direct_child(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
``sys_session_send`` in by-session-id mode verifies the target is a
direct child of the caller, posts the message, and returns a running
handle. Proves the unified session_id mode: the message reaches the
existing child and the handle carries its id with ``status:running``
(the result is delivered asynchronously via ``sys_read_inbox``).
:param monkeypatch: Stubs the runner-local child registration so the
dispatch runs without a live runner.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
monkeypatch.setattr(runner_app, "register_child_session", lambda *a, **k: None)
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
event_posts: list[dict[str, Any]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.method == "GET" and request.url.path == "/v1/sessions/conv_child":
# Target IS a direct child of the caller.
return httpx.Response(
200,
json={
"id": "conv_child",
"parent_session_id": "conv_caller",
"title": "researcher:auth",
},
)
if request.method == "POST" and request.url.path == "/v1/sessions/conv_child/events":
event_posts.append(json.loads(request.content))
return httpx.Response(200, json={"ok": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps({"session_id": "conv_child", "args": "continue please"}),
server_client=server_client,
conversation_id="conv_caller",
agent_spec=SimpleNamespace(sub_agents=[SimpleNamespace(name="researcher")]),
session_inbox=session_inbox,
)
finally:
runner_app.unregister_subagent_work("conv_child")
runner_app._session_inboxes_ref.pop("conv_caller", None)
# Message reached the existing child; handle carries its id + running status.
assert event_posts[0]["data"]["content"][0]["text"] == "continue please"
handle = json.loads(output)
assert handle["conversation_id"] == "conv_child"
assert handle["status"] == "launching"
@pytest.mark.asyncio
async def test_sys_session_send_session_id_rejects_non_child() -> None:
"""
By-session-id send refuses a target that is NOT a direct child of the
caller (``parent_session_id`` mismatch) — returning
``session_out_of_tree`` and posting NO message. This is the
child-only safety guarantee: an orchestrator can't drive a sibling or
an unrelated session it merely has read access to. If the parentage
check regressed, the message would be posted and the assertion on
``event_posts`` would fail.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
event_posts: list[dict[str, Any]] = []
async def _server_handler(request: httpx.Request) -> httpx.Response:
if request.method == "GET" and request.url.path == "/v1/sessions/conv_other":
# Target's parent is someone ELSE, not the caller.
return httpx.Response(
200,
json={
"id": "conv_other",
"parent_session_id": "conv_someone_else",
"title": "x:y",
},
)
if request.url.path == "/v1/sessions/conv_other/events":
event_posts.append(json.loads(request.content))
return httpx.Response(200, json={"ok": True})
return httpx.Response(404, json={"error": str(request.url)})
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps({"session_id": "conv_other", "args": "hi"}),
server_client=server_client,
conversation_id="conv_caller",
agent_spec=SimpleNamespace(sub_agents=[SimpleNamespace(name="researcher")]),
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_caller", None)
info = json.loads(output)
assert info["error"] == "session_out_of_tree"
# No message was posted to the non-child session.
assert event_posts == []
@pytest.mark.parametrize("empty_output", ["", " ", "\n\t "])
def test_format_async_task_item_empty_subagent_completion_reads_as_no_output(
empty_output: str,
) -> None:
"""
An empty sub-agent completion renders "produced no output", not "returned:".
A native child that idles with no assistant text is delivered as ``""`` —
the runner deliberately avoids fabricating output from stale runner
history (see
``test_external_status_idle_without_output_omits_stale_history_preview``).
The parent LLM must then see an explicit "produced no output" line rather
than a dangling ``"…returned: "`` that reads as a truncated/garbled handoff.
:param empty_output: An empty or whitespace-only child output.
"""
from omnigent.runner.tool_dispatch import _format_async_task_item
line = _format_async_task_item(
{
"type": "sub_agent",
"handle_id": "conv_child_empty",
"agent": "worker",
"title": "phase-a",
"status": "completed",
"output": empty_output,
}
)
# Reads as no-output; the dangling content-free "returned:" must not appear.
# With the old formatter, an empty output rendered "…returned: ]" and this
# would fail on the missing "produced no output" / present "returned:".
assert "produced no output" in line
assert "returned:" not in line
def test_format_async_task_item_nonempty_subagent_completion_shows_output() -> None:
"""
A non-empty sub-agent completion still renders its returned text.
Guards against the empty-output branch swallowing a real result.
"""
from omnigent.runner.tool_dispatch import _format_async_task_item
line = _format_async_task_item(
{
"type": "sub_agent",
"handle_id": "conv_child_real",
"agent": "worker",
"title": "phase-a",
"status": "completed",
"output": "review done: LGTM",
}
)
assert "returned: review done: LGTM" in line
assert "produced no output" not in line
@pytest.mark.asyncio
async def test_sys_session_send_rejects_both_session_id_and_named_target() -> None:
"""
Supplying both ``session_id`` and ``agent``/``title`` fails loud.
The by-session-id and named ``(agent, title)`` modes can point at different
children, so silently letting ``session_id`` win would misroute the message
with no signal to the caller. The dispatch must reject the ambiguity before
making any server call.
"""
from omnigent.runner import app as runner_app
from omnigent.runner.tool_dispatch import execute_tool
server_called = False
async def _server_handler(request: httpx.Request) -> httpx.Response:
"""
Fail the test if any server call is made.
:param request: Any Omnigent request — none should occur on the reject path.
:returns: A 404 (also records the unexpected call).
"""
nonlocal server_called
server_called = True
return httpx.Response(404, json={"error": str(request.url)})
session_inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async with httpx.AsyncClient(
transport=httpx.MockTransport(_server_handler),
base_url="http://server",
) as server_client:
try:
output = await execute_tool(
tool_name="sys_session_send",
arguments=json.dumps(
{
"session_id": "conv_direct_child",
"agent": "claude",
"title": "issue-1",
"args": "do the thing",
}
),
server_client=server_client,
conversation_id="conv_parent_ambiguous",
agent_spec=SimpleNamespace(sub_agents=[SimpleNamespace(name="claude")]),
session_inbox=session_inbox,
)
finally:
runner_app._session_inboxes_ref.pop("conv_parent_ambiguous", None)
# The ambiguity is rejected with a fail-loud error naming both modes.
# Without the guard, session_id silently wins and routing proceeds into
# _send_to_existing_session (which would hit the server handler below).
assert "both 'session_id' and 'agent'/'title'" in output
assert server_called is False
@pytest.mark.asyncio
async def test_create_session_reinit_preserves_existing_inbox() -> None:
"""
A reconnect re-POST of ``/v1/sessions`` must not wipe the session inbox.
The Omnigent server re-POSTs ``/v1/sessions`` for every bound conversation
on each runner WebSocket (re)connect — including in-process reconnects of a
still-alive runner after a transient blip. A sub-agent completion that lands
while the socket is down delivers its result into the parent's
``_session_inboxes`` queue and latches the work entry ``delivered``, which
makes redelivery short-circuit. If the re-init blindly replaced the queue,
that already-delivered payload would be orphaned and the parent would drain
an empty inbox and hang forever. This test drives the real
``POST /v1/sessions`` route twice for the same session id and asserts the
inbox (and a sentinel sitting in it) survives the second call.
:returns: None.
"""
from omnigent.runner import app as runner_app
session_id = "conv_reinit_inbox_guard"
agent_id = "ag_reinit_inbox_guard"
# Real stub manager: get_client must succeed so create_session reaches the
# inbox-init lines. ``_FakeHarnessClient([])`` is never streamed on this path.
app = create_runner_app(
process_manager=cast(
HarnessProcessManager,
_FakeProcessManager(_FakeHarnessClient([])),
),
server_client=NullServerClient(), # type: ignore[arg-type]
)
# Fresh-process hygiene: the inbox ref is a module-wide singleton, so clear
# any leftover from a prior test before exercising the route.
runner_app._session_inboxes_ref.pop(session_id, None)
sentinel = {"type": "subagent_result", "marker": "SURVIVE_REINIT"}
try:
async with _runner_test_client(app) as http:
first = await http.post(
"/v1/sessions",
json={"session_id": session_id, "agent_id": agent_id},
)
# 201 proves create_session ran end-to-end (past the inbox init)
# rather than short-circuiting (400 missing fields / 501 scaffold /
# 503 spawn failure), so the inbox we inspect next is route-created.
assert first.status_code == 201, (
f"First POST /v1/sessions should create the session (201); got "
f"{first.status_code} with body {first.text!r}."
)
assert first.json()["id"] == session_id
inbox_after_first = runner_app._session_inboxes_ref.get(session_id)
# The route must have installed a real inbox queue. If None, the
# init path didn't run and the rest of the test would be vacuous.
assert inbox_after_first is not None, (
"POST /v1/sessions did not create a session inbox; the "
"inbox-init line in create_session did not run."
)
# Stand in for a delivered sub-agent payload waiting to be drained.
inbox_after_first.put_nowait(sentinel)
second = await http.post(
"/v1/sessions",
json={"session_id": session_id, "agent_id": agent_id},
)
# The reconnect re-init must also succeed; a non-201 here would mean
# the simulated reconnect couldn't re-run the handler at all.
assert second.status_code == 201, (
f"Reconnect re-POST /v1/sessions should succeed (201); got "
f"{second.status_code} with body {second.text!r}."
)
inbox_after_second = runner_app._session_inboxes_ref.get(session_id)
# The load-bearing assertion: the inbox object is preserved across
# re-init. Without the ``if session_id not in _session_inboxes``
# guard, line ~4093 would assign a brand-new asyncio.Queue() here,
# so this would be a different object and the sentinel would vanish.
assert inbox_after_second is inbox_after_first, (
"Re-init replaced the session inbox with a new queue. The "
"guard in create_session must skip re-creating an existing "
"inbox or a delivered sub-agent payload is orphaned and the "
"parent hangs on an empty inbox."
)
# Content survives: the delivered payload is still drainable. A
# fresh empty queue would make this drain raise QueueEmpty.
assert inbox_after_second.get_nowait() == sentinel, (
"Sentinel payload did not survive the re-init; the parent's "
"delivered sub-agent result was lost when the inbox was wiped."
)
finally:
runner_app._session_inboxes_ref.pop(session_id, None)
# ── approval-event flattening (elicitation-approval hang regression) ──────
@pytest.mark.asyncio
async def test_approval_event_flattened_for_harness_scaffold() -> None:
"""A nested approval envelope is flattened to the scaffold's ApprovalEvent.
Regression for the elicitation-approval hang: the server forwards the
verdict as ``{"type": "approval", "data": {...}}``, but the harness
scaffold's ``ApprovalEvent`` requires ``elicitation_id`` / ``action`` /
``content`` at the TOP level. If the runner forwards the envelope verbatim
the harness 422s and the parked ``ctx.elicit`` Future never resolves (the
turn hangs after a human approves). The runner must translate the envelope
into the flat event the scaffold validates — for every scaffold harness.
"""
from omnigent.runtime.harnesses._scaffold import ApprovalEvent
captured: dict[str, Any] = {}
class _CapturingHarnessClient:
async def post(
self, url: str, *, json: dict[str, Any], timeout: float | None = None
) -> httpx.Response:
captured["url"] = url
captured["body"] = json
return httpx.Response(204)
mgr = _FakeProcessManager(harness_client=cast(Any, _CapturingHarnessClient()))
app = create_runner_app(
process_manager=cast(HarnessProcessManager, mgr),
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
resp = await http.post(
"/v1/sessions/conv_x/events",
json={
"type": "approval",
"data": {
"elicitation_id": "elicit_x",
"action": "accept",
"content": {"note": "ok"},
},
},
)
assert resp.status_code == 204
# Forwarded body is FLAT — no ``data`` envelope.
assert captured["body"] == {
"type": "approval",
"elicitation_id": "elicit_x",
"action": "accept",
"content": {"note": "ok"},
}
# And it validates as the scaffold's ApprovalEvent (i.e. no 422).
ApprovalEvent.model_validate(captured["body"])
@pytest.mark.asyncio
async def test_approval_event_without_content_flattened() -> None:
"""A decline verdict with no form content flattens without a ``content`` key."""
from omnigent.runtime.harnesses._scaffold import ApprovalEvent
captured: dict[str, Any] = {}
class _CapturingHarnessClient:
async def post(
self, url: str, *, json: dict[str, Any], timeout: float | None = None
) -> httpx.Response:
captured["body"] = json
return httpx.Response(204)
mgr = _FakeProcessManager(harness_client=cast(Any, _CapturingHarnessClient()))
app = create_runner_app(
process_manager=cast(HarnessProcessManager, mgr),
server_client=NullServerClient(), # type: ignore[arg-type]
)
async with _runner_test_client(app) as http:
resp = await http.post(
"/v1/sessions/conv_y/events",
json={"type": "approval", "data": {"elicitation_id": "e2", "action": "decline"}},
)
assert resp.status_code == 204
assert captured["body"] == {"type": "approval", "elicitation_id": "e2", "action": "decline"}
ApprovalEvent.model_validate(captured["body"])
@pytest.mark.asyncio
async def test_spawn_async_tool_cancels_losing_future_no_leak(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
``_spawn_async_tool`` races the tool coro against ``cancel_event.wait()``
via ``asyncio.wait(FIRST_COMPLETED)``. The losing future must be cancelled
in BOTH branches, otherwise every ``sys_call_async`` permanently leaks one
pending task (the orphaned ``cancel_event.wait()`` on success, or the
orphaned tool coro on cancel).
"""
from omnigent.runner import tool_dispatch
spawn_kw: dict[str, Any] = {
"server_client": None,
"terminal_registry": None,
"resource_registry": None,
"agent_spec": None,
"conversation_id": "conv_leak",
"task_id": None,
"agent_id": None,
"agent_name": None,
"runner_workspace": None,
"mcp_manager": None,
"filesystem_registry": None,
}
def _leaked(before: set[asyncio.Task[Any]]) -> list[str]:
cur = asyncio.current_task()
return [
t.get_name()
for t in asyncio.all_tasks()
if t not in before and t is not cur and not t.done()
]
# Success path: tool finishes first, so cancel_event.wait() is the loser.
async def _fast(**_kw: Any) -> str:
return "ok"
monkeypatch.setattr(tool_dispatch, "execute_tool", _fast)
inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
tasks: dict[str, tuple[asyncio.Task[str], asyncio.Event]] = {}
before = set(asyncio.all_tasks())
tool_dispatch._spawn_async_tool(
{"tool": "noop", "args": "{}"},
session_inbox=inbox,
session_async_tasks=tasks,
**spawn_kw,
)
bg_task, _evt = next(iter(tasks.values()))
await bg_task
await asyncio.sleep(0) # let the .cancel() propagate to the losing future
assert inbox.get_nowait()["status"] == "completed"
assert _leaked(before) == []
# Cancel path: cancel_event fires first, so the tool coro is the loser.
async def _slow(**_kw: Any) -> str:
await asyncio.sleep(30)
return "late"
monkeypatch.setattr(tool_dispatch, "execute_tool", _slow)
inbox = asyncio.Queue()
tasks = {}
before = set(asyncio.all_tasks())
tool_dispatch._spawn_async_tool(
{"tool": "slow", "args": "{}"},
session_inbox=inbox,
session_async_tasks=tasks,
**spawn_kw,
)
bg_task, evt = next(iter(tasks.values()))
await asyncio.sleep(0) # let _bg reach asyncio.wait
evt.set()
await bg_task
await asyncio.sleep(0)
assert inbox.get_nowait()["status"] == "cancelled"
assert _leaked(before) == []
Reference in New Issue View Git Blame Copy Permalink