mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-25 17:18:11 +00:00
3960315af7
Two compression-tip hydration tests simulated legacy state by emptying the parent AFTER end_session(compression) — exactly the durable write the new closed-parent guard refuses. Reordered: empty first, close second. The tests' actual contract (old id hydrates from the live tip) is unchanged and still pinned.
1200 lines
48 KiB
Python
1200 lines
48 KiB
Python
"""Focused tests for API server session-control endpoints."""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from aiohttp import web
|
|
from aiohttp.test_utils import TestClient, TestServer
|
|
|
|
from gateway.config import PlatformConfig
|
|
from gateway.platforms.api_server import APIServerAdapter
|
|
from hermes_state import SessionDB
|
|
|
|
|
|
@pytest.fixture
|
|
def session_db(tmp_path):
|
|
db = SessionDB(tmp_path / "state.db")
|
|
try:
|
|
yield db
|
|
finally:
|
|
close = getattr(db, "close", None)
|
|
if callable(close):
|
|
close()
|
|
|
|
|
|
@pytest.fixture
|
|
def adapter(session_db):
|
|
adapter = APIServerAdapter(PlatformConfig(enabled=True))
|
|
adapter._session_db = session_db
|
|
return adapter
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_adapter(session_db):
|
|
adapter = APIServerAdapter(PlatformConfig(enabled=True, extra={"key": "sk-test"}))
|
|
adapter._session_db = session_db
|
|
return adapter
|
|
|
|
|
|
def _create_session_app(adapter: APIServerAdapter) -> web.Application:
|
|
app = web.Application()
|
|
app.router.add_get("/v1/capabilities", adapter._handle_capabilities)
|
|
app.router.add_get("/api/sessions", adapter._handle_list_sessions)
|
|
app.router.add_post("/api/sessions", adapter._handle_create_session)
|
|
app.router.add_get("/api/sessions/{session_id}", adapter._handle_get_session)
|
|
app.router.add_patch("/api/sessions/{session_id}", adapter._handle_patch_session)
|
|
app.router.add_delete("/api/sessions/{session_id}", adapter._handle_delete_session)
|
|
app.router.add_get("/api/sessions/{session_id}/messages", adapter._handle_session_messages)
|
|
app.router.add_post("/api/sessions/{session_id}/fork", adapter._handle_fork_session)
|
|
app.router.add_post("/api/sessions/{session_id}/chat", adapter._handle_session_chat)
|
|
app.router.add_post("/api/sessions/{session_id}/chat/stream", adapter._handle_session_chat_stream)
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capabilities_advertises_session_control_surface(adapter):
|
|
app = _create_session_app(adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.get("/v1/capabilities")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
|
|
features = data["features"]
|
|
assert features["session_resources"] is True
|
|
assert features["session_chat"] is True
|
|
assert features["session_chat_streaming"] is True
|
|
assert features["session_fork"] is True
|
|
assert features["admin_config_rw"] is False
|
|
assert features["memory_write_api"] is False
|
|
assert features["skills_api"] is True
|
|
assert features["realtime_voice"] is False
|
|
assert data["endpoints"]["sessions"] == {"method": "GET", "path": "/api/sessions"}
|
|
assert data["endpoints"]["session_chat_stream"] == {
|
|
"method": "POST",
|
|
"path": "/api/sessions/{session_id}/chat/stream",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_binds_api_session_context_for_tool_env(adapter, monkeypatch):
|
|
"""API-server request sessions should reach tools and terminal subprocess env."""
|
|
monkeypatch.setenv("HERMES_SESSION_ID", "stale-session")
|
|
observed = {}
|
|
|
|
class FakeAgent:
|
|
session_prompt_tokens = 0
|
|
session_completion_tokens = 0
|
|
session_total_tokens = 0
|
|
|
|
def __init__(self, session_id: str):
|
|
self.session_id = session_id
|
|
|
|
def run_conversation(self, user_message, conversation_history, task_id):
|
|
from gateway.session_context import get_session_env
|
|
from tools.environments.local import _make_run_env
|
|
|
|
observed["task_id"] = task_id
|
|
observed["context_session_id"] = get_session_env("HERMES_SESSION_ID")
|
|
observed["context_platform"] = get_session_env("HERMES_SESSION_PLATFORM")
|
|
observed["context_session_key"] = get_session_env("HERMES_SESSION_KEY")
|
|
observed["child_session_id"] = _make_run_env({}).get("HERMES_SESSION_ID")
|
|
return {"final_response": "ok"}
|
|
|
|
def fake_create_agent(**kwargs):
|
|
return FakeAgent(kwargs["session_id"])
|
|
|
|
monkeypatch.setattr(adapter, "_create_agent", fake_create_agent)
|
|
|
|
result, usage = await adapter._run_agent(
|
|
user_message="hello",
|
|
conversation_history=[],
|
|
session_id="request-session",
|
|
gateway_session_key="request-key",
|
|
)
|
|
|
|
assert result["session_id"] == "request-session"
|
|
assert usage["input_tokens"] == 0
|
|
assert usage["output_tokens"] == 0
|
|
assert usage["total_tokens"] == 0
|
|
assert "runtime" not in usage
|
|
assert observed == {
|
|
"task_id": "request-session",
|
|
"context_session_id": "request-session",
|
|
"context_platform": "api_server",
|
|
"context_session_key": "request-key",
|
|
"child_session_id": "request-session",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_crud_and_message_history(adapter, session_db):
|
|
app = _create_session_app(adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
create_resp = await cli.post("/api/sessions", json={"title": "Mobile chat", "model": "test-model"})
|
|
assert create_resp.status == 201
|
|
created = await create_resp.json()
|
|
session_id = created["session"]["id"]
|
|
assert created["object"] == "hermes.session"
|
|
assert created["session"]["title"] == "Mobile chat"
|
|
|
|
session_db.append_message(session_id, "user", "hello from phone")
|
|
session_db.append_message(session_id, "assistant", "hello from hermes")
|
|
|
|
list_resp = await cli.get("/api/sessions?limit=10&offset=0")
|
|
assert list_resp.status == 200
|
|
listed = await list_resp.json()
|
|
assert listed["object"] == "list"
|
|
assert [s["id"] for s in listed["data"]] == [session_id]
|
|
assert listed["data"][0]["message_count"] == 2
|
|
|
|
get_resp = await cli.get(f"/api/sessions/{session_id}")
|
|
assert get_resp.status == 200
|
|
got = await get_resp.json()
|
|
assert got["session"]["id"] == session_id
|
|
assert got["session"]["message_count"] == 2
|
|
|
|
messages_resp = await cli.get(f"/api/sessions/{session_id}/messages")
|
|
assert messages_resp.status == 200
|
|
messages = await messages_resp.json()
|
|
assert messages["object"] == "list"
|
|
assert [m["role"] for m in messages["data"]] == ["user", "assistant"]
|
|
assert messages["data"][0]["content"] == "hello from phone"
|
|
|
|
patch_resp = await cli.patch(f"/api/sessions/{session_id}", json={"title": "Renamed"})
|
|
assert patch_resp.status == 200
|
|
patched = await patch_resp.json()
|
|
assert patched["session"]["title"] == "Renamed"
|
|
|
|
delete_resp = await cli.delete(f"/api/sessions/{session_id}")
|
|
assert delete_resp.status == 200
|
|
deleted = await delete_resp.json()
|
|
assert deleted == {"object": "hermes.session.deleted", "id": session_id, "deleted": True}
|
|
assert session_db.get_session(session_id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_messages_follow_compression_tip(adapter, session_db):
|
|
source_id = session_db.create_session("source-session", "api_server")
|
|
session_db.append_message(source_id, "user", "before compression")
|
|
# Empty the parent BEFORE closing it: the closed-parent write guard
|
|
# (CompressionSessionClosedError) refuses durable writes to a session
|
|
# ended by compression, so the legacy-state simulation must run first.
|
|
session_db.replace_messages(source_id, [])
|
|
session_db.end_session(source_id, "compression")
|
|
session_db.create_session("tip-session", "api_server", parent_session_id=source_id)
|
|
session_db.append_message("tip-session", "user", "after compression")
|
|
|
|
app = _create_session_app(adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
messages_resp = await cli.get(f"/api/sessions/{source_id}/messages")
|
|
assert messages_resp.status == 200
|
|
messages = await messages_resp.json()
|
|
|
|
assert messages["object"] == "list"
|
|
assert messages["session_id"] == "tip-session"
|
|
assert [m["content"] for m in messages["data"]] == ["after compression"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_fork_uses_current_sessiondb_branch_primitives(adapter, session_db):
|
|
source_id = session_db.create_session("source-session", "api_server", model="test-model")
|
|
session_db.set_session_title(source_id, "Original")
|
|
session_db.append_message(source_id, "user", "first path")
|
|
session_db.append_message(source_id, "assistant", "answer")
|
|
|
|
app = _create_session_app(adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(f"/api/sessions/{source_id}/fork", json={"title": "Alternative"})
|
|
assert resp.status == 201
|
|
payload = await resp.json()
|
|
|
|
fork = payload["session"]
|
|
assert payload["object"] == "hermes.session"
|
|
assert fork["id"] != source_id
|
|
assert fork["parent_session_id"] == source_id
|
|
assert fork["title"] == "Alternative"
|
|
assert [m["content"] for m in session_db.get_messages(fork["id"])] == ["first path", "answer"]
|
|
assert session_db.get_session(source_id)["end_reason"] == "branched"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_loads_history_and_preserves_session_headers(auth_adapter, session_db):
|
|
session_id = session_db.create_session("chat-session", "api_server")
|
|
session_db.set_session_title(session_id, "Chat")
|
|
session_db.append_message(session_id, "user", "earlier")
|
|
session_db.append_message(session_id, "assistant", "prior answer")
|
|
|
|
mock_run = AsyncMock(return_value=({"final_response": "fresh answer", "session_id": session_id}, {"total_tokens": 3}))
|
|
app = _create_session_app(auth_adapter)
|
|
with patch.object(auth_adapter, "_run_agent", mock_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={"message": "next", "system_message": "stay focused"},
|
|
headers={"Authorization": "Bearer sk-test", "X-Hermes-Session-Key": "client-42"},
|
|
)
|
|
assert resp.status == 200
|
|
payload = await resp.json()
|
|
|
|
assert resp.headers["X-Hermes-Session-Id"] == session_id
|
|
assert resp.headers["X-Hermes-Session-Key"] == "client-42"
|
|
assert payload["object"] == "hermes.session.chat.completion"
|
|
assert payload["session_id"] == session_id
|
|
assert payload["message"]["role"] == "assistant"
|
|
assert payload["message"]["content"] == "fresh answer"
|
|
mock_run.assert_awaited_once()
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs["session_id"] == session_id
|
|
assert kwargs["gateway_session_key"] == "client-42"
|
|
assert kwargs["ephemeral_system_prompt"] == "stay focused"
|
|
history = kwargs["conversation_history"]
|
|
assert len(history) == 2
|
|
assert isinstance(history[0].pop("timestamp"), (int, float))
|
|
assert isinstance(history[1].pop("timestamp"), (int, float))
|
|
assert history == [
|
|
{"role": "user", "content": "earlier"},
|
|
{"role": "assistant", "content": "prior answer"},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_accepts_multimodal_message(auth_adapter, session_db):
|
|
session_id = session_db.create_session("image-session", "api_server")
|
|
image_payload = [
|
|
{"type": "input_text", "text": "What's in this image?"},
|
|
{"type": "input_image", "image_url": "data:image/png;base64,AAAA"},
|
|
]
|
|
expected_user_message = [
|
|
{"type": "text", "text": "What's in this image?"},
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
|
|
]
|
|
|
|
mock_run = AsyncMock(return_value=({"final_response": "A cat.", "session_id": session_id}, {"total_tokens": 4}))
|
|
app = _create_session_app(auth_adapter)
|
|
with patch.object(auth_adapter, "_run_agent", mock_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={"message": image_payload},
|
|
headers={"Authorization": "Bearer sk-test"},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs["user_message"] == expected_user_message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_stream_accepts_multimodal_message(adapter, session_db):
|
|
session_id = session_db.create_session("image-stream-session", "api_server")
|
|
image_payload = [
|
|
{"type": "input_text", "text": "What's in this image?"},
|
|
{"type": "input_image", "image_url": "data:image/png;base64,AAAA"},
|
|
]
|
|
expected_user_message = [
|
|
{"type": "text", "text": "What's in this image?"},
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
|
|
]
|
|
captured_kwargs = {}
|
|
|
|
async def fake_run(**kwargs):
|
|
captured_kwargs.update(kwargs)
|
|
kwargs["stream_delta_callback"]("A cat.")
|
|
return {"final_response": "A cat.", "session_id": session_id}, {"total_tokens": 4}
|
|
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_run_agent", side_effect=fake_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat/stream",
|
|
json={"message": image_payload},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
assert resp.headers["Content-Type"].startswith("text/event-stream")
|
|
body = await resp.text()
|
|
|
|
assert "event: assistant.completed" in body
|
|
assert captured_kwargs["user_message"] == expected_user_message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_stream_emits_lifecycle_events_and_keepalive_safe_shape(adapter, session_db):
|
|
session_id = session_db.create_session("stream-session", "api_server")
|
|
session_db.set_session_title(session_id, "Stream")
|
|
|
|
async def fake_run(**kwargs):
|
|
kwargs["stream_delta_callback"]("Hello")
|
|
kwargs["stream_delta_callback"](" world")
|
|
kwargs["tool_progress_callback"]("reasoning.available", tool_name="_thinking", preview="thinking")
|
|
return {"final_response": "Hello world", "session_id": session_id}, {"total_tokens": 2}
|
|
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_run_agent", side_effect=fake_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(f"/api/sessions/{session_id}/chat/stream", json={"message": "stream please"})
|
|
assert resp.status == 200
|
|
assert resp.headers["Content-Type"].startswith("text/event-stream")
|
|
body = await resp.text()
|
|
|
|
assert "event: run.started" in body
|
|
assert "event: message.started" in body
|
|
assert "event: assistant.delta" in body
|
|
assert "Hello world" in body
|
|
assert "event: tool.progress" in body
|
|
assert "event: assistant.completed" in body
|
|
assert "event: run.completed" in body
|
|
assert "event: done" in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_stream_run_completed_carries_turn_transcript(adapter, session_db):
|
|
"""run.completed must include the full interleaved turn transcript so a
|
|
client that lost intermediate (pre-tool-call) assistant text from the live
|
|
delta stream can reconcile without a separate /messages fetch. Refs #34703.
|
|
"""
|
|
import json as _json
|
|
|
|
session_id = session_db.create_session("transcript-session", "api_server")
|
|
|
|
async def fake_run(**kwargs):
|
|
# Stream the intermediate planning text the way a real turn would.
|
|
kwargs["stream_delta_callback"]("Let me search for that:")
|
|
kwargs["stream_delta_callback"]("Here is the summary.")
|
|
result = {
|
|
"final_response": "Here is the summary.",
|
|
"session_id": session_id,
|
|
"messages": [
|
|
{"role": "user", "content": "search then summarize"},
|
|
{
|
|
"role": "assistant",
|
|
"content": "Let me search for that:",
|
|
"tool_calls": [
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "web_search", "arguments": "{}"},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "content": "results", "tool_call_id": "call_1", "tool_name": "web_search"},
|
|
{"role": "assistant", "content": "Here is the summary."},
|
|
],
|
|
}
|
|
return result, {"total_tokens": 6}
|
|
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_run_agent", side_effect=fake_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat/stream",
|
|
json={"message": "search then summarize"},
|
|
)
|
|
assert resp.status == 200
|
|
body = await resp.text()
|
|
|
|
# Pull the run.completed event payload out of the SSE body.
|
|
run_completed_payload = None
|
|
for block in body.split("\n\n"):
|
|
if "event: run.completed" in block:
|
|
for line in block.splitlines():
|
|
if line.startswith("data: "):
|
|
run_completed_payload = _json.loads(line[len("data: "):])
|
|
break
|
|
assert run_completed_payload is not None, body
|
|
messages = run_completed_payload.get("messages")
|
|
assert isinstance(messages, list) and messages, run_completed_payload
|
|
|
|
# The colon-ended intermediate text that preceded the tool call must be present.
|
|
contents = [m.get("content") for m in messages]
|
|
assert "Let me search for that:" in contents
|
|
assert "Here is the summary." in contents
|
|
# No prior-turn user message should leak into the per-turn slice.
|
|
assert all(m.get("role") in ("assistant", "tool") for m in messages)
|
|
# The tool call is preserved alongside the intermediate text.
|
|
assert any(m.get("tool_calls") for m in messages)
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_endpoints_require_auth_when_key_configured(auth_adapter):
|
|
app = _create_session_app(auth_adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.get("/api/sessions")
|
|
assert resp.status == 401
|
|
body = await resp.json()
|
|
assert body["error"]["code"] == "gateway_auth_failed"
|
|
|
|
ok = await cli.get("/api/sessions", headers={"Authorization": "Bearer sk-test"})
|
|
assert ok.status == 200
|
|
data = await ok.json()
|
|
assert data["object"] == "list"
|
|
assert data["data"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_header_rejected_without_api_key(adapter, session_db):
|
|
session_id = session_db.create_session("unsafe-session", "api_server")
|
|
app = _create_session_app(adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={"message": "hello"},
|
|
headers={"X-Hermes-Session-Key": "client-42"},
|
|
)
|
|
assert resp.status == 403
|
|
data = await resp.json()
|
|
assert "X-Hermes-Session-Key requires API key" in data["error"]["message"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session-persisted model threading + provider-auth failure surfacing
|
|
# (salvaged from PR #57947 by @FvanW and PR #59941 by @kaishi00)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_threads_session_model_to_run_agent(auth_adapter, session_db):
|
|
"""POST /api/sessions persists a per-session model, but the chat handler
|
|
previously fetched the session record and threw it away — the session's
|
|
chosen model silently had no effect on any chat turn."""
|
|
session_id = session_db.create_session("model-pinned-session", "api_server", model="claude-sonnet-4-6")
|
|
|
|
mock_run = AsyncMock(return_value=({"final_response": "ok", "session_id": session_id}, {"total_tokens": 1}))
|
|
app = _create_session_app(auth_adapter)
|
|
with patch.object(auth_adapter, "_run_agent", mock_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={"message": "hi"},
|
|
headers={"Authorization": "Bearer sk-test"},
|
|
)
|
|
assert resp.status == 200
|
|
|
|
mock_run.assert_awaited_once()
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs["session_model"] == "claude-sonnet-4-6"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_stream_threads_session_model_to_run_agent(adapter, session_db):
|
|
"""Streaming twin of the session-model threading test above."""
|
|
session_id = session_db.create_session("model-pinned-stream-session", "api_server", model="gpt-5.5")
|
|
|
|
mock_run = AsyncMock(return_value=({"final_response": "ok", "session_id": session_id}, {"total_tokens": 1}))
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_run_agent", mock_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat/stream",
|
|
json={"message": "hi"},
|
|
)
|
|
assert resp.status == 200
|
|
await resp.read()
|
|
|
|
mock_run.assert_awaited_once()
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs["session_model"] == "gpt-5.5"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_resolves_stored_model_route_alias(session_db, monkeypatch):
|
|
"""A session-persisted model that matches a model_routes alias must go
|
|
through the route path (so route provider/credentials apply) and NOT be
|
|
passed as a raw session_model (idea from PR #59941 by @kaishi00)."""
|
|
adapter = APIServerAdapter(
|
|
PlatformConfig(
|
|
enabled=True,
|
|
extra={"model_routes": {"alias": {"model": "route/model", "provider": "openrouter"}}},
|
|
)
|
|
)
|
|
adapter._session_db = session_db
|
|
session_id = session_db.create_session("route-pinned-session", "api_server", model="alias")
|
|
|
|
mock_run = AsyncMock(return_value=({"final_response": "ok", "session_id": session_id}, {"total_tokens": 1}))
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_run_agent", mock_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={"message": "hi"},
|
|
)
|
|
assert resp.status == 200
|
|
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs["route"] == {"model": "route/model", "provider": "openrouter"}
|
|
assert kwargs["session_model"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_returns_controlled_response_on_provider_auth_failure(adapter, monkeypatch):
|
|
"""_resolve_runtime_agent_kwargs() (inside _create_agent()) raises
|
|
RuntimeError on provider auth/credential failure. Previously this
|
|
propagated unhandled out of _run_agent(): /v1/chat/completions caught it
|
|
as a generic 500, and /api/sessions/{id}/chat didn't catch it at all
|
|
(raw aiohttp 500, no JSON body). Must now return run.py's controlled
|
|
response shape instead of raising. Exercises the REAL boundary
|
|
(gateway.run._resolve_runtime_agent_kwargs, the sole raiser)."""
|
|
monkeypatch.setattr(
|
|
"gateway.run._resolve_runtime_agent_kwargs",
|
|
lambda: (_ for _ in ()).throw(
|
|
RuntimeError("No credentials found for provider 'nous' — run `hermes auth add nous`")
|
|
),
|
|
)
|
|
|
|
result, usage = await adapter._run_agent(
|
|
user_message="hello",
|
|
conversation_history=[],
|
|
session_id="request-session",
|
|
)
|
|
|
|
assert result == {
|
|
"final_response": "⚠️ Provider authentication failed: No credentials found for provider 'nous' — run `hermes auth add nous`",
|
|
"messages": [],
|
|
"api_calls": 0,
|
|
"tools": [],
|
|
}
|
|
assert usage == {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_does_not_swallow_unrelated_exceptions(adapter, monkeypatch):
|
|
"""The _ProviderAuthResolutionError catch must stay narrow — a TypeError
|
|
elsewhere in _create_agent()/run_conversation() must still propagate."""
|
|
def fake_create_agent(**kwargs):
|
|
raise TypeError("unrelated bug: unexpected keyword argument")
|
|
|
|
monkeypatch.setattr(adapter, "_create_agent", fake_create_agent)
|
|
|
|
with pytest.raises(TypeError, match="unrelated bug"):
|
|
await adapter._run_agent(
|
|
user_message="hello",
|
|
conversation_history=[],
|
|
session_id="request-session",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_does_not_swallow_unrelated_runtime_error_from_run_conversation(adapter, monkeypatch):
|
|
"""agent.run_conversation() can legitimately raise a RuntimeError
|
|
unrelated to provider auth (e.g. run_agent.py's "Failed to recreate
|
|
closed OpenAI client"). A bare `except RuntimeError` around the whole
|
|
_create_agent()+run_conversation() span would mislabel it as
|
|
"Provider authentication failed". Only _ProviderAuthResolutionError —
|
|
raised exclusively inside _create_agent() at the
|
|
_resolve_runtime_agent_kwargs() call site — may trigger the controlled
|
|
response; this unrelated RuntimeError must propagate unhandled."""
|
|
class _FakeAgent:
|
|
def run_conversation(self, **kwargs):
|
|
raise RuntimeError("Failed to recreate closed OpenAI client")
|
|
|
|
monkeypatch.setattr(adapter, "_create_agent", lambda **kwargs: _FakeAgent())
|
|
|
|
with pytest.raises(RuntimeError, match="Failed to recreate closed OpenAI client"):
|
|
await adapter._run_agent(
|
|
user_message="hello",
|
|
conversation_history=[],
|
|
session_id="request-session",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_surfaces_controlled_response_on_provider_auth_failure(auth_adapter, session_db, monkeypatch):
|
|
"""End-to-end: POST /api/sessions/{id}/chat previously had zero wrapping
|
|
around _run_agent() — an unhandled RuntimeError produced a raw aiohttp
|
|
500 with no JSON body. Must now return 200 with the controlled error
|
|
message as the assistant content. Exercises the real
|
|
gateway.run._resolve_runtime_agent_kwargs() boundary, not a mocked
|
|
_create_agent()."""
|
|
session_id = session_db.create_session("auth-fail-session", "api_server")
|
|
|
|
monkeypatch.setattr(
|
|
"gateway.run._resolve_runtime_agent_kwargs",
|
|
lambda: (_ for _ in ()).throw(RuntimeError("Auth failed: token expired")),
|
|
)
|
|
|
|
app = _create_session_app(auth_adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={"message": "hi"},
|
|
headers={"Authorization": "Bearer sk-test"},
|
|
)
|
|
assert resp.status == 200
|
|
payload = await resp.json()
|
|
|
|
assert payload["message"]["content"] == "⚠️ Provider authentication failed: Auth failed: token expired"
|
|
def _register_session_model_route(app, adapter):
|
|
app.router.add_post("/api/sessions/{session_id}/model", adapter._handle_session_model_lock)
|
|
|
|
|
|
def _patch_api_server_runtime(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"gateway.run._resolve_runtime_agent_kwargs",
|
|
lambda: {
|
|
"provider": "openrouter",
|
|
"api_key": "sk-global",
|
|
"base_url": "https://openrouter.example/v1",
|
|
"api_mode": "chat_completions",
|
|
},
|
|
)
|
|
monkeypatch.setattr("gateway.run._resolve_gateway_model", lambda: "global/model")
|
|
monkeypatch.setattr("gateway.run._load_gateway_config", lambda: {})
|
|
monkeypatch.setattr(
|
|
"gateway.run.GatewayRunner._load_reasoning_config",
|
|
staticmethod(lambda model="": {}),
|
|
)
|
|
monkeypatch.setattr(
|
|
"gateway.run.GatewayRunner._load_fallback_model",
|
|
staticmethod(lambda: None),
|
|
)
|
|
monkeypatch.setattr("gateway.run._current_max_iterations", lambda: 90)
|
|
monkeypatch.setattr("hermes_cli.tools_config._get_platform_tools", lambda *_: set())
|
|
monkeypatch.setattr(
|
|
"gateway.run._resolve_runtime_agent_kwargs_for_provider",
|
|
lambda provider: {
|
|
"provider": provider,
|
|
"api_key": f"sk-{provider}",
|
|
"base_url": f"https://{provider}.example/v1",
|
|
"api_mode": "chat_completions",
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_builds_raw_provider_model_route_when_alias_missing(adapter, session_db):
|
|
session_id = session_db.create_session("route-session", "api_server")
|
|
mock_run = AsyncMock(
|
|
return_value=(
|
|
{
|
|
"final_response": "ok",
|
|
"session_id": session_id,
|
|
"runtime": {"provider": "nous", "model": "x-ai/grok-4.5", "route_source": "raw_request"},
|
|
},
|
|
{"total_tokens": 2, "runtime": {"provider": "nous", "model": "x-ai/grok-4.5"}},
|
|
)
|
|
)
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", mock_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={
|
|
"message": "hello",
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"require_model_lock": True,
|
|
},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
payload = await resp.json()
|
|
|
|
kwargs = mock_run.call_args.kwargs
|
|
assert kwargs["route"] == {"provider": "nous", "model": "x-ai/grok-4.5"}
|
|
assert payload["runtime"]["provider"] == "nous"
|
|
assert payload["runtime"]["model"] == "x-ai/grok-4.5"
|
|
assert payload["runtime"]["requested"]["model"] == "x-ai/grok-4.5"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_passes_runtime_options_to_run_agent(adapter, session_db):
|
|
session_id = session_db.create_session("options-session", "api_server")
|
|
mock_run = AsyncMock(return_value=({"final_response": "ok", "session_id": session_id}, {}))
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", mock_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={
|
|
"message": "hello",
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"model_options": {
|
|
"reasoning": {"enabled": True, "effort": "xhigh"},
|
|
"service_tier": "priority",
|
|
"fast": True,
|
|
},
|
|
},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
|
|
kwargs = mock_run.call_args.kwargs
|
|
# In the merged design model_options travel raw to _create_agent, which
|
|
# parses reasoning/service-tier itself (see _request_reasoning_config /
|
|
# _request_service_tier) — there is no separate runtime_options kwarg.
|
|
assert kwargs["model_options"] == {
|
|
"reasoning": {"enabled": True, "effort": "xhigh"},
|
|
"service_tier": "priority",
|
|
"fast": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_chat_stream_uses_same_runtime_lock(adapter, session_db):
|
|
session_id = session_db.create_session("stream-lock-session", "api_server")
|
|
captured = {}
|
|
|
|
async def fake_run(**kwargs):
|
|
captured.update(kwargs)
|
|
kwargs["stream_delta_callback"]("hi")
|
|
return (
|
|
{
|
|
"final_response": "hi",
|
|
"session_id": session_id,
|
|
"runtime": {
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
"route_source": "raw_request",
|
|
},
|
|
},
|
|
{
|
|
"total_tokens": 1,
|
|
"runtime": {
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
},
|
|
},
|
|
)
|
|
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", side_effect=fake_run):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat/stream",
|
|
json={
|
|
"message": "stream",
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"model_options": {"reasoning": {"enabled": False}},
|
|
"require_model_lock": True,
|
|
},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
body = await resp.text()
|
|
|
|
assert captured["route"] == {"provider": "nous", "model": "x-ai/grok-4.5"}
|
|
assert captured["model_options"] == {"reasoning": {"enabled": False}}
|
|
assert captured["confirmed_runtime_lock"] is True
|
|
assert "x-ai/grok-4.5" in body
|
|
assert "run.started" in body or "event: run.started" in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_session_respects_browser_source_and_model_lock(adapter, session_db):
|
|
app = _create_session_app(adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
"/api/sessions",
|
|
json={
|
|
"id": "browser-lock-session",
|
|
"source": "hermes_browser",
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"require_model_lock": True,
|
|
"title": "Browser lock",
|
|
"system_prompt": "browser prompt",
|
|
},
|
|
)
|
|
assert resp.status == 201, await resp.text()
|
|
payload = await resp.json()
|
|
|
|
assert payload["session"]["source"] == "hermes_browser"
|
|
assert payload["session"]["model"] == "x-ai/grok-4.5"
|
|
row = session_db.get_session("browser-lock-session")
|
|
assert row["source"] == "hermes_browser"
|
|
assert row["model"] == "x-ai/grok-4.5"
|
|
import json as _json
|
|
model_config = row.get("model_config")
|
|
if isinstance(model_config, str):
|
|
model_config = _json.loads(model_config)
|
|
assert model_config["browser_model_lock"]["provider"] == "nous"
|
|
assert model_config["browser_model_lock"]["model"] == "x-ai/grok-4.5"
|
|
assert model_config["browser_model_lock"]["confirmed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_model_lock_endpoint_persists_and_invalidates_prompt(adapter, session_db):
|
|
session_id = session_db.create_session(
|
|
"lock-endpoint-session",
|
|
"api_server",
|
|
model="gpt-5.5",
|
|
model_config={"_branched_from": "parent-session"},
|
|
system_prompt="Conversation started:\nModel: gpt-5.5\nProvider: openai-codex\n",
|
|
)
|
|
app = _create_session_app(adapter)
|
|
_register_session_model_route(app, adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/model",
|
|
json={
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"model_options": {"reasoning": {"enabled": True, "effort": "high"}},
|
|
"require_model_lock": True,
|
|
},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
payload = await resp.json()
|
|
|
|
assert payload["object"] == "hermes.session.model_lock"
|
|
assert payload["runtime"]["requested"]["provider"] == "nous"
|
|
assert payload["runtime"]["model"] == "x-ai/grok-4.5"
|
|
assert payload["runtime"]["model_lock"] in {"accepted", "confirmed"}
|
|
row = session_db.get_session(session_id)
|
|
assert row["model"] == "x-ai/grok-4.5"
|
|
assert row["system_prompt"] is None
|
|
import json as _json
|
|
model_config = row.get("model_config")
|
|
if isinstance(model_config, str):
|
|
model_config = _json.loads(model_config)
|
|
assert model_config["_branched_from"] == "parent-session"
|
|
assert model_config["browser_model_lock"]["provider"] == "nous"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_model_lock_endpoint_then_chat_reuses_persisted_lock_and_provider_credentials(
|
|
adapter,
|
|
session_db,
|
|
monkeypatch,
|
|
):
|
|
session_id = session_db.create_session(
|
|
"endpoint-lock-chat",
|
|
"api_server",
|
|
model="gpt-5.5",
|
|
system_prompt="Conversation started:\nModel: gpt-5.5\nProvider: openai-codex\n",
|
|
)
|
|
captured = {}
|
|
|
|
class FakeAgent:
|
|
session_prompt_tokens = 0
|
|
session_completion_tokens = 0
|
|
session_total_tokens = 0
|
|
|
|
def __init__(self, **kwargs):
|
|
captured.update(kwargs)
|
|
self.session_id = kwargs["session_id"]
|
|
self.provider = kwargs.get("provider") or ""
|
|
self.model = kwargs.get("model") or ""
|
|
|
|
def run_conversation(self, user_message, conversation_history, task_id):
|
|
return {"final_response": "locked", "session_id": self.session_id}
|
|
|
|
_patch_api_server_runtime(monkeypatch)
|
|
monkeypatch.setattr("run_agent.AIAgent", FakeAgent)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"_session_model_override_for",
|
|
lambda *_: {
|
|
"model": "session/override-model",
|
|
"provider": "openai-codex",
|
|
"api_key": "sk-session-override",
|
|
"base_url": "https://override.example/v1",
|
|
"api_mode": "codex_responses",
|
|
},
|
|
)
|
|
|
|
app = _create_session_app(adapter)
|
|
_register_session_model_route(app, adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
lock_resp = await cli.post(
|
|
f"/api/sessions/{session_id}/model",
|
|
json={
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"require_model_lock": True,
|
|
},
|
|
)
|
|
assert lock_resp.status == 200, await lock_resp.text()
|
|
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={"message": "use the stored lock"},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
payload = await resp.json()
|
|
|
|
assert captured["provider"] == "nous"
|
|
assert captured["model"] == "x-ai/grok-4.5"
|
|
assert captured["api_key"] == "sk-nous"
|
|
assert captured["base_url"] == "https://nous.example/v1"
|
|
assert payload["runtime"]["provider"] == "nous"
|
|
assert payload["runtime"]["model"] == "x-ai/grok-4.5"
|
|
assert payload["runtime"]["requested"] == {
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
}
|
|
assert payload["runtime"]["route_source"] == "session_model_lock"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_model_lock_endpoint_then_chat_stream_reuses_persisted_lock(
|
|
adapter,
|
|
session_db,
|
|
):
|
|
session_id = session_db.create_session("endpoint-lock-stream", "api_server")
|
|
captured = {}
|
|
|
|
async def fake_run(**kwargs):
|
|
captured.update(kwargs)
|
|
kwargs["stream_delta_callback"]("hi")
|
|
return (
|
|
{
|
|
"final_response": "hi",
|
|
"session_id": session_id,
|
|
"runtime": {
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
"route_source": "session_model_lock",
|
|
},
|
|
},
|
|
{
|
|
"total_tokens": 1,
|
|
"runtime": {
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"requested": {"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
"route_source": "session_model_lock",
|
|
},
|
|
},
|
|
)
|
|
|
|
app = _create_session_app(adapter)
|
|
_register_session_model_route(app, adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None), patch.object(
|
|
adapter,
|
|
"_run_agent",
|
|
side_effect=fake_run,
|
|
):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
lock_resp = await cli.post(
|
|
f"/api/sessions/{session_id}/model",
|
|
json={
|
|
"provider": "nous",
|
|
"model": "x-ai/grok-4.5",
|
|
"require_model_lock": True,
|
|
},
|
|
)
|
|
assert lock_resp.status == 200, await lock_resp.text()
|
|
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat/stream",
|
|
json={"message": "stream with stored lock"},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
body = await resp.text()
|
|
|
|
assert captured["route"] == {"provider": "nous", "model": "x-ai/grok-4.5"}
|
|
assert captured["requested_runtime"]["provider"] == "nous"
|
|
assert captured["requested_runtime"]["model"] == "x-ai/grok-4.5"
|
|
assert captured["route_source"] == "session_model_lock"
|
|
assert "x-ai/grok-4.5" in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_reports_actual_agent_runtime_not_requested_metadata(adapter, monkeypatch):
|
|
class FakeAgent:
|
|
session_prompt_tokens = 0
|
|
session_completion_tokens = 0
|
|
session_total_tokens = 0
|
|
|
|
def __init__(self):
|
|
self.session_id = "runtime-session"
|
|
self.provider = "actual-provider"
|
|
self.model = "actual-model"
|
|
self._hermes_api_runtime = {
|
|
"provider": "requested-provider",
|
|
"model": "requested-model",
|
|
"route_source": "raw_request",
|
|
}
|
|
|
|
def run_conversation(self, user_message, conversation_history, task_id):
|
|
return {"final_response": "ok", "session_id": self.session_id}
|
|
|
|
monkeypatch.setattr(adapter, "_create_agent", lambda **kwargs: FakeAgent())
|
|
|
|
result, usage = await adapter._run_agent(
|
|
user_message="hello",
|
|
conversation_history=[],
|
|
session_id="runtime-session",
|
|
route={"provider": "requested-provider", "model": "requested-model"},
|
|
requested_runtime={
|
|
"provider": "requested-provider",
|
|
"model": "requested-model",
|
|
},
|
|
route_source="session_model_lock",
|
|
)
|
|
|
|
assert result["runtime"]["provider"] == "actual-provider"
|
|
assert result["runtime"]["model"] == "actual-model"
|
|
assert result["runtime"]["requested"] == {
|
|
"provider": "requested-provider",
|
|
"model": "requested-model",
|
|
}
|
|
assert usage["runtime"]["provider"] == "actual-provider"
|
|
assert usage["runtime"]["model"] == "actual-model"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirmed_runtime_lock_rejects_actual_runtime_mismatch(adapter, monkeypatch):
|
|
class FakeAgent:
|
|
session_prompt_tokens = 0
|
|
session_completion_tokens = 0
|
|
session_total_tokens = 0
|
|
session_id = "mismatch-session"
|
|
provider = "fallback-provider"
|
|
model = "fallback-model"
|
|
|
|
def run_conversation(self, user_message, conversation_history, task_id):
|
|
return {"final_response": "wrong runtime", "session_id": self.session_id}
|
|
|
|
monkeypatch.setattr(adapter, "_create_agent", lambda **kwargs: FakeAgent())
|
|
|
|
with pytest.raises(RuntimeError, match="confirmed model lock runtime mismatch"):
|
|
await adapter._run_agent(
|
|
user_message="hello",
|
|
conversation_history=[],
|
|
session_id="mismatch-session",
|
|
route={"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
requested_runtime={"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
route_source="session_model_lock",
|
|
confirmed_runtime_lock=True,
|
|
)
|
|
|
|
|
|
def test_confirmed_runtime_lock_fails_closed_on_provider_resolution_error(adapter, monkeypatch):
|
|
_patch_api_server_runtime(monkeypatch)
|
|
# Break BOTH resolution paths (primary picker-based resolver + the
|
|
# gateway fallback) — a confirmed lock must propagate the failure
|
|
# instead of constructing an agent on the previous global credentials.
|
|
monkeypatch.setattr(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("provider unavailable")),
|
|
)
|
|
monkeypatch.setattr(
|
|
"gateway.run._resolve_runtime_agent_kwargs_for_provider",
|
|
lambda provider: (_ for _ in ()).throw(RuntimeError("provider unavailable")),
|
|
)
|
|
agent_ctor = patch("run_agent.AIAgent")
|
|
with agent_ctor as mocked_agent:
|
|
with pytest.raises(RuntimeError, match="provider unavailable"):
|
|
adapter._create_agent(
|
|
session_id="locked-session",
|
|
route={"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
confirmed_runtime_lock=True,
|
|
)
|
|
mocked_agent.assert_not_called()
|
|
|
|
|
|
def test_confirmed_runtime_lock_disables_global_fallback_model(adapter, monkeypatch):
|
|
_patch_api_server_runtime(monkeypatch)
|
|
monkeypatch.setattr(
|
|
"gateway.run.GatewayRunner._load_fallback_model",
|
|
staticmethod(lambda: "openrouter/fallback-model"),
|
|
)
|
|
captured = {}
|
|
|
|
class FakeAgent:
|
|
provider = "nous"
|
|
model = "x-ai/grok-4.5"
|
|
|
|
def __init__(self, **kwargs):
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr("run_agent.AIAgent", FakeAgent)
|
|
|
|
adapter._create_agent(
|
|
session_id="locked-session",
|
|
route={"provider": "nous", "model": "x-ai/grok-4.5"},
|
|
confirmed_runtime_lock=True,
|
|
)
|
|
|
|
assert captured["fallback_model"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unconfirmed_request_does_not_replace_confirmed_session_lock(adapter, session_db):
|
|
session_id = session_db.create_session("one-off-override", "api_server")
|
|
session_db.update_session_runtime_lock(
|
|
session_id,
|
|
provider="nous",
|
|
model="x-ai/grok-4.5",
|
|
route_source="raw_request",
|
|
confirmed=True,
|
|
)
|
|
mock_run = AsyncMock(
|
|
return_value=(
|
|
{
|
|
"final_response": "ok",
|
|
"session_id": session_id,
|
|
"runtime": {"provider": "openrouter", "model": "anthropic/claude-sonnet"},
|
|
},
|
|
{"total_tokens": 1},
|
|
)
|
|
)
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None), patch.object(
|
|
adapter,
|
|
"_run_agent",
|
|
mock_run,
|
|
):
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={
|
|
"message": "one turn only",
|
|
"provider": "openrouter",
|
|
"model": "anthropic/claude-sonnet",
|
|
},
|
|
)
|
|
assert resp.status == 200, await resp.text()
|
|
|
|
import json as _json
|
|
|
|
row = session_db.get_session(session_id)
|
|
config = row["model_config"]
|
|
if isinstance(config, str):
|
|
config = _json.loads(config)
|
|
assert config["browser_model_lock"]["provider"] == "nous"
|
|
assert config["browser_model_lock"]["model"] == "x-ai/grok-4.5"
|
|
assert config["browser_model_lock"]["confirmed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_require_model_lock_hard_fails_when_global_default_would_be_used(adapter, session_db, monkeypatch):
|
|
session_id = session_db.create_session("lock-fail-session", "api_server")
|
|
monkeypatch.setattr(adapter, "_model_name", "gpt-5.5")
|
|
app = _create_session_app(adapter)
|
|
with patch.object(adapter, "_resolve_route", return_value=None), patch.object(adapter, "_run_agent", new_callable=AsyncMock) as mock_run:
|
|
async with TestClient(TestServer(app)) as cli:
|
|
# empty model + require_model_lock must not silently fall through
|
|
resp = await cli.post(
|
|
f"/api/sessions/{session_id}/chat",
|
|
json={
|
|
"message": "hello",
|
|
"provider": "nous",
|
|
"model": "",
|
|
"require_model_lock": True,
|
|
},
|
|
)
|
|
assert resp.status in (400, 409), await resp.text()
|
|
body = await resp.json()
|
|
assert body["error"]["code"] in {"model_lock_unavailable", "invalid_model_lock", "missing_model"}
|
|
mock_run.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capabilities_advertises_session_model_lock(adapter):
|
|
app = _create_session_app(adapter)
|
|
async with TestClient(TestServer(app)) as cli:
|
|
resp = await cli.get("/v1/capabilities")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["features"]["session_model_lock"] is True
|
|
assert data["endpoints"]["session_model_lock"] == {
|
|
"method": "POST",
|
|
"path": "/api/sessions/{session_id}/model",
|
|
}
|