"""Turn-completion push trigger in the orchestrator (fake agent + container)."""
from types import SimpleNamespace
import pytest
from navi.core.orchestrator import AgentSessionOrchestrator
from navi.core.events import StreamEnd
class FakePushService:
def __init__(self):
self.calls: list[tuple[str, str | None, str]] = []
async def notify_turn_complete(self, session_id, user_id, content):
self.calls.append((session_id, user_id, content))
class FakeSession:
def __init__(self, user_id=None):
self.user_id = user_id
self.session_metadata = {}
class FakeSessionStore:
def __init__(self, session):
self._session = session
async def get(self, session_id):
return self._session
async def save(self, session):
pass
class FakeAgent:
def __init__(self, content="Answer text"):
self._content = content
async def run_stream(self, *args, **kwargs):
yield StreamEnd(full_content=self._content)
@pytest.fixture
def make_orchestrator(monkeypatch):
def _make(push_service, auth_enabled=False):
container = SimpleNamespace(
push_service=push_service,
profile_registry=None,
tool_registry=None,
backend_registry=None,
workers=[],
memory_store=None,
cp_registry=None,
mcp_manager=None,
)
orch = AgentSessionOrchestrator(container)
# Stub the real Agent build; the trigger tests exercise only the
# orchestrator loop around StreamEnd.
monkeypatch.setattr(orch, "_build_agent", lambda store: FakeAgent())
# The orchestrator module reads `settings` by import — swap it for a
# light namespace carrying just the attributes the loop touches.
import navi.core.orchestrator as orch_mod
monkeypatch.setattr(orch_mod, "settings", SimpleNamespace(
navi_auth_enabled=auth_enabled,
ws_replay_buffer_size=500,
))
return orch
return _make
async def _run_turn(orch, session_id, store):
orch.create_run(session_id)
await orch.run_agent(
session_id,
user_content="hi",
raw_images=None,
display_content=None,
files=None,
session_store=store,
)
async def test_push_fires_when_no_websockets(make_orchestrator):
push = FakePushService()
orch = make_orchestrator(push)
store = FakeSessionStore(FakeSession(user_id="u1"))
await _run_turn(orch, "sess1", store)
assert push.calls == [("sess1", "u1", "Answer text")]
async def test_no_push_when_websocket_watching(make_orchestrator):
push = FakePushService()
orch = make_orchestrator(push)
store = FakeSessionStore(FakeSession(user_id="u1"))
session_id = "sess2"
orch.create_run(session_id)
state = orch._sessions[session_id]
state.websockets.append(object()) # a client tab is open on this session
await orch.run_agent(
session_id, user_content="hi", raw_images=None, display_content=None,
files=None, session_store=store,
)
assert push.calls == []
async def test_no_push_when_session_has_no_user(make_orchestrator):
push = FakePushService()
orch = make_orchestrator(push, auth_enabled=True)
store = FakeSessionStore(FakeSession(user_id=None))
await _run_turn(orch, "sess3", store)
assert push.calls == []
async def test_no_user_id_falls_back_to_anonymous_when_auth_off(make_orchestrator):
push = FakePushService()
orch = make_orchestrator(push, auth_enabled=False)
store = FakeSessionStore(FakeSession(user_id=None))
await _run_turn(orch, "sess4", store)
assert push.calls == [("sess4", "anonymous", "Answer text")]
async def test_push_failure_never_breaks_the_run(make_orchestrator):
class ExplodingPush:
async def notify_turn_complete(self, *a, **kw):
raise RuntimeError("push backend down")
orch = make_orchestrator(ExplodingPush())
store = FakeSessionStore(FakeSession(user_id="u1"))
# Must complete without raising.
await _run_turn(orch, "sess5", store)