wehub-resource-sync
876 lines
34 KiB
Python
876 lines
34 KiB
Python
"""Tests for ``POST /v1/sessions/{id}/switch-agent``.
|
|
|
|
Exercises the in-place agent-switch endpoint: validation (404 missing
|
|
session, 400 sub-agent / no binding / unloadable bundle, 404 non-bindable
|
|
target, 409 while busy) and the happy-path wiring (it clones the target,
|
|
computes the same-family model/history/label deltas, and forwards them to
|
|
``switch_conversation_agent``). Real-type store stubs — no MagicMock.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.testclient import TestClient
|
|
|
|
from omnigent.entities import Agent, Conversation, ConversationItem, MessageData, PagedList
|
|
from omnigent.errors import OmnigentError
|
|
from omnigent.server.routes import sessions as sessions_mod
|
|
from omnigent.server.routes.sessions import create_sessions_router
|
|
|
|
# ── Stubs ────────────────────────────────────────────────────────
|
|
|
|
|
|
class _AgentStore:
|
|
"""Agent store stub: get + list for the switch route.
|
|
|
|
:param agents: Pre-populated map of agent_id → Agent.
|
|
"""
|
|
|
|
def __init__(self, agents: dict[str, Agent]) -> None:
|
|
self._agents = dict(agents)
|
|
|
|
def get(self, agent_id: str) -> Agent | None:
|
|
""":returns: The agent if present, else None."""
|
|
return self._agents.get(agent_id)
|
|
|
|
def list(
|
|
self,
|
|
limit: int = 20,
|
|
after: str | None = None,
|
|
before: str | None = None,
|
|
order: str = "desc",
|
|
) -> PagedList[Agent]:
|
|
"""Return the built-in (session_id is None) agents.
|
|
|
|
:param limit: Max agents (ignored — stubs are small).
|
|
:returns: A PagedList of the template agents.
|
|
"""
|
|
del after, before, order
|
|
builtins = [a for a in self._agents.values() if a.session_id is None][:limit]
|
|
return PagedList(data=builtins, first_id=None, last_id=None, has_more=False)
|
|
|
|
|
|
class _ConversationStore:
|
|
"""Conversation store stub for the switch route.
|
|
|
|
:param conversations: Map of id → Conversation.
|
|
:param items_by_conv: Map of conv_id → items.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
conversations: dict[str, Conversation],
|
|
items_by_conv: dict[str, list[ConversationItem]] | None = None,
|
|
) -> None:
|
|
self._convs = conversations
|
|
self._items = items_by_conv or {}
|
|
self.switch_calls: list[dict[str, Any]] = []
|
|
|
|
def get_conversation(self, conversation_id: str) -> Conversation | None:
|
|
""":returns: The conversation if present, else None."""
|
|
return self._convs.get(conversation_id)
|
|
|
|
def switch_conversation_agent(
|
|
self,
|
|
conversation_id: str,
|
|
*,
|
|
new_agent_id: str,
|
|
new_agent_name: str,
|
|
new_agent_bundle_location: str,
|
|
new_agent_description: str | None,
|
|
copy_model_settings: bool,
|
|
carry_history_into_native: bool,
|
|
presentation_labels: dict[str, str],
|
|
previous_builtin_id: str | None,
|
|
) -> Conversation:
|
|
"""Record the call and return the updated conversation.
|
|
|
|
:param conversation_id: Session being switched.
|
|
:param new_agent_id: Cloned agent id the route generated.
|
|
:param new_agent_name: Cloned agent name.
|
|
:param new_agent_bundle_location: Target bundle to clone.
|
|
:param new_agent_description: Target description.
|
|
:param copy_model_settings: Same-family flag from the route.
|
|
:param carry_history_into_native: Native-rebuild flag.
|
|
:param presentation_labels: Target-harness ui/wrapper labels.
|
|
:param previous_builtin_id: Built-in switched away from.
|
|
:returns: The conversation rebound to ``new_agent_id``.
|
|
:raises LookupError: If the conversation is unknown.
|
|
"""
|
|
self.switch_calls.append(
|
|
{
|
|
"conversation_id": conversation_id,
|
|
"new_agent_id": new_agent_id,
|
|
"new_agent_name": new_agent_name,
|
|
"new_agent_bundle_location": new_agent_bundle_location,
|
|
"new_agent_description": new_agent_description,
|
|
"copy_model_settings": copy_model_settings,
|
|
"carry_history_into_native": carry_history_into_native,
|
|
"presentation_labels": presentation_labels,
|
|
"previous_builtin_id": previous_builtin_id,
|
|
}
|
|
)
|
|
src = self._convs.get(conversation_id)
|
|
if src is None:
|
|
raise LookupError(conversation_id)
|
|
return Conversation(
|
|
id=conversation_id,
|
|
created_at=src.created_at,
|
|
updated_at=200,
|
|
root_conversation_id=conversation_id,
|
|
agent_id=new_agent_id,
|
|
title=src.title,
|
|
)
|
|
|
|
def list_items(
|
|
self,
|
|
conversation_id: str,
|
|
limit: int = 100,
|
|
after: str | None = None,
|
|
before: str | None = None,
|
|
order: str = "asc",
|
|
type: str | None = None,
|
|
) -> PagedList[ConversationItem]:
|
|
""":returns: A PagedList of the conversation's items."""
|
|
del limit, after, before, order, type
|
|
items = self._items.get(conversation_id, [])
|
|
return PagedList(
|
|
data=items,
|
|
first_id=items[0].id if items else None,
|
|
last_id=items[-1].id if items else None,
|
|
has_more=False,
|
|
)
|
|
|
|
|
|
class _AgentCacheStub:
|
|
"""Stub for ``get_agent_cache()`` — controls the bundle precheck.
|
|
|
|
:param raise_on_load: When True, ``load`` raises to simulate an
|
|
unloadable target bundle (the route maps this to 400).
|
|
"""
|
|
|
|
def __init__(self, raise_on_load: bool = False) -> None:
|
|
self._raise = raise_on_load
|
|
|
|
def load(
|
|
self,
|
|
agent_id: str,
|
|
bundle_location: str,
|
|
*,
|
|
expand_env: bool = False,
|
|
) -> object:
|
|
"""Pretend to load a bundle.
|
|
|
|
:param agent_id: Agent id (unused).
|
|
:param bundle_location: Bundle key (unused).
|
|
:param expand_env: Ignored — accepted to match the real
|
|
``AgentCache.load`` signature. Without it the
|
|
route's keyword call would raise ``TypeError``.
|
|
:returns: A sentinel object (the route ignores the value).
|
|
:raises RuntimeError: When configured to fail.
|
|
"""
|
|
del agent_id, bundle_location, expand_env
|
|
if self._raise:
|
|
raise RuntimeError("bundle missing")
|
|
return object()
|
|
|
|
|
|
class _LoadedAgentStub:
|
|
"""Loaded-agent stub exposing ``spec.executor.harness_kind``."""
|
|
|
|
def __init__(self, harness_kind: str) -> None:
|
|
class _Executor:
|
|
def __init__(self, hk: str) -> None:
|
|
self.harness_kind = hk
|
|
|
|
class _Spec:
|
|
def __init__(self, hk: str) -> None:
|
|
self.executor = _Executor(hk)
|
|
|
|
self.spec = _Spec(harness_kind)
|
|
|
|
|
|
class _HarnessAgentCacheStub:
|
|
"""Agent cache stub mapping agent id to harness kind."""
|
|
|
|
def __init__(self, harness_by_id: dict[str, str]) -> None:
|
|
self._harness_by_id = harness_by_id
|
|
|
|
def load(
|
|
self,
|
|
agent_id: str,
|
|
bundle_location: str,
|
|
*,
|
|
expand_env: bool = False,
|
|
) -> _LoadedAgentStub:
|
|
del bundle_location, expand_env
|
|
return _LoadedAgentStub(self._harness_by_id[agent_id])
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────
|
|
|
|
|
|
def _conv(
|
|
conv_id: str = "conv_src",
|
|
agent_id: str | None = "ag_session_scoped",
|
|
kind: str = "default",
|
|
) -> Conversation:
|
|
"""Build a Conversation entity.
|
|
|
|
:param conv_id: Conversation id.
|
|
:param agent_id: Bound agent id, or None.
|
|
:param kind: ``"default"`` or ``"sub_agent"``.
|
|
:returns: A Conversation.
|
|
"""
|
|
return Conversation(
|
|
id=conv_id,
|
|
created_at=1,
|
|
updated_at=1,
|
|
root_conversation_id=conv_id,
|
|
agent_id=agent_id,
|
|
title="Source",
|
|
kind=kind,
|
|
)
|
|
|
|
|
|
def _agent(agent_id: str, name: str, bundle: str, session_id: str | None) -> Agent:
|
|
"""Build an Agent entity.
|
|
|
|
:param agent_id: Agent id.
|
|
:param name: Agent name.
|
|
:param bundle: Bundle location.
|
|
:param session_id: Owning session id (None for a built-in).
|
|
:returns: An Agent.
|
|
"""
|
|
return Agent(
|
|
id=agent_id,
|
|
created_at=1,
|
|
name=name,
|
|
bundle_location=bundle,
|
|
version=1,
|
|
session_id=session_id,
|
|
)
|
|
|
|
|
|
def _build_app(conv_store: _ConversationStore, agent_store: _AgentStore) -> FastAPI:
|
|
"""Build a FastAPI app mounting the sessions router + error handler.
|
|
|
|
:param conv_store: Conversation store stub.
|
|
:param agent_store: Agent store stub.
|
|
:returns: A configured FastAPI app.
|
|
"""
|
|
router = create_sessions_router(
|
|
conversation_store=conv_store, # type: ignore[arg-type]
|
|
agent_store=agent_store, # type: ignore[arg-type]
|
|
)
|
|
app = FastAPI()
|
|
|
|
@app.exception_handler(OmnigentError)
|
|
async def _handle(request: Request, exc: OmnigentError) -> JSONResponse:
|
|
del request
|
|
return JSONResponse(
|
|
status_code=exc.http_status,
|
|
content={"error": {"code": exc.code, "message": exc.message}},
|
|
)
|
|
|
|
app.include_router(router, prefix="/v1")
|
|
return app
|
|
|
|
|
|
def _patch_family_helpers(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
*,
|
|
same_family: bool,
|
|
native: bool,
|
|
labels: dict[str, str],
|
|
raise_on_load: bool = False,
|
|
) -> None:
|
|
"""Stub the bundle-loading helpers so the route runs without a real
|
|
bundle, with controlled family/native/label outcomes.
|
|
|
|
:param monkeypatch: Pytest monkeypatch fixture.
|
|
:param same_family: Value ``_same_provider_family`` returns.
|
|
:param native: Value ``_agent_is_native`` /
|
|
``_agent_carries_native_fork_history`` return (these switch tests
|
|
target claude/codex native, which both classify and carry history).
|
|
:param labels: Value ``_presentation_labels_for_agent`` returns.
|
|
:param raise_on_load: Whether the bundle precheck should fail.
|
|
"""
|
|
monkeypatch.setattr(
|
|
sessions_mod, "get_agent_cache", lambda: _AgentCacheStub(raise_on_load=raise_on_load)
|
|
)
|
|
monkeypatch.setattr(sessions_mod, "_same_provider_family", lambda a, b: same_family)
|
|
monkeypatch.setattr(sessions_mod, "_agent_is_native", lambda a: native)
|
|
monkeypatch.setattr(sessions_mod, "_agent_carries_native_fork_history", lambda a: native)
|
|
monkeypatch.setattr(sessions_mod, "_presentation_labels_for_agent", lambda a: labels)
|
|
|
|
|
|
# The session's currently-bound agent is a session-scoped clone of the
|
|
# claude-sdk built-in (shares its bundle_location). ``_BUILTIN_ORIGIN`` is
|
|
# that built-in — what "switch back" resolves to, and the no-op target the
|
|
# route now rejects (same bundle as the current agent). The switch *targets*
|
|
# are built-ins with a DIFFERENT bundle.
|
|
_CURRENT = _agent("ag_session_scoped", "claude (switch src)", "bundle/claude-sdk", "conv_src")
|
|
_BUILTIN_ORIGIN = _agent("ag_builtin_origin", "claude", "bundle/claude-sdk", None)
|
|
_BUILTIN_CLAUDE = _agent("ag_builtin_claude", "claude-native-ui", "bundle/claude-native", None)
|
|
_BUILTIN_CODEX = _agent("ag_builtin_codex", "codex-native-ui", "bundle/codex", None)
|
|
_BUILTIN_CURSOR = _agent("ag_builtin_cursor", "cursor-native-ui", "bundle/cursor", None)
|
|
_BUILTIN_PI = _agent("ag_builtin_pi", "pi-native-ui", "bundle/pi", None)
|
|
_BUILTIN_QWEN = _agent("ag_builtin_qwen", "qwen-native-ui", "bundle/qwen", None)
|
|
|
|
|
|
# ── Tests ────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_same_family_native_carries_history(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A same-family native target keeps model settings, marks the native
|
|
rebuild, applies target labels, and resolves the previous built-in.
|
|
"""
|
|
conv_store = _ConversationStore(
|
|
conversations={"conv_src": _conv()},
|
|
items_by_conv={
|
|
"conv_src": [
|
|
ConversationItem(
|
|
id="msg_1",
|
|
type="message",
|
|
status="completed",
|
|
response_id="r1",
|
|
created_at=1,
|
|
data=MessageData(role="user", content=[{"type": "input_text", "text": "hi"}]),
|
|
)
|
|
]
|
|
},
|
|
)
|
|
agent_store = _AgentStore(
|
|
{
|
|
"ag_session_scoped": _CURRENT,
|
|
"ag_builtin_claude": _BUILTIN_CLAUDE,
|
|
"ag_builtin_origin": _BUILTIN_ORIGIN,
|
|
}
|
|
)
|
|
labels = {"omnigent.ui": "terminal", "omnigent.wrapper": "claude-code-native-ui"}
|
|
_patch_family_helpers(monkeypatch, same_family=True, native=True, labels=labels)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["status"] == "idle"
|
|
# Bound to a freshly cloned session-scoped agent, not the built-in itself.
|
|
assert body["agent_id"].startswith("ag_") and body["agent_id"] != "ag_builtin_claude"
|
|
# 1 item returned — the in-place transcript is preserved (not copied/empty).
|
|
assert len(body["items"]) == 1
|
|
|
|
assert len(conv_store.switch_calls) == 1, "route must call switch exactly once"
|
|
call = conv_store.switch_calls[0]
|
|
assert call["conversation_id"] == "conv_src"
|
|
assert call["new_agent_bundle_location"] == "bundle/claude-native"
|
|
# Same family → keep model settings AND carry native history (target native).
|
|
assert call["copy_model_settings"] is True
|
|
assert call["carry_history_into_native"] is True
|
|
assert call["presentation_labels"] == labels
|
|
# The origin built-in shares the current agent's bundle → that's the
|
|
# built-in to offer for "Switch back" (NOT the switched-to target).
|
|
assert call["previous_builtin_id"] == "ag_builtin_origin"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_cross_family_resets_model_but_carries_history(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A cross-family switch resets model settings but still carries history
|
|
into a native target.
|
|
|
|
The model id is provider-bound so it must reset; history is NOT — the
|
|
switch clears ``external_session_id`` and the runner rebuilds the
|
|
native transcript from this session's own Omnigent items, a conversion
|
|
that doesn't depend on the source harness.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore({"ag_session_scoped": _CURRENT, "ag_builtin_codex": _BUILTIN_CODEX})
|
|
# native=True with same_family=False → model resets, history still carries.
|
|
_patch_family_helpers(monkeypatch, same_family=False, native=True, labels={})
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post("/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_codex"})
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
call = conv_store.switch_calls[0]
|
|
# Cross-family → reset model settings (a model id is provider-bound).
|
|
assert call["copy_model_settings"] is False
|
|
# Native target carries history regardless of family: the runner
|
|
# rebuilds the transcript from Omnigent items. False here would mean
|
|
# the cross-family gate regressed and the session resumes blank.
|
|
assert call["carry_history_into_native"] is True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"target_agent,target_harness,expected_labels,expected_carry",
|
|
[
|
|
# cursor-native has no resumable session file to rebuild → no carry.
|
|
(
|
|
_BUILTIN_CURSOR,
|
|
"cursor-native",
|
|
{"omnigent.ui": "terminal", "omnigent.wrapper": "cursor-native-ui"},
|
|
False,
|
|
),
|
|
# pi-native rebuilds its JSONL session file from the copied items →
|
|
# carry history (parity with claude/codex native).
|
|
(
|
|
_BUILTIN_PI,
|
|
"pi-native",
|
|
{"omnigent.ui": "terminal", "omnigent.wrapper": "pi-native-ui"},
|
|
True,
|
|
),
|
|
# qwen-native rebuilds qwen's on-disk chat recording from the copied
|
|
# items → carry history (parity with claude/codex/pi native).
|
|
(
|
|
_BUILTIN_QWEN,
|
|
"qwen-native",
|
|
{"omnigent.ui": "terminal", "omnigent.wrapper": "qwen-native-ui"},
|
|
True,
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_switch_cursor_pi_native_targets_carry_history_gating(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
target_agent: Agent,
|
|
target_harness: str,
|
|
expected_labels: dict[str, str],
|
|
expected_carry: bool,
|
|
) -> None:
|
|
"""Switching into cursor/pi native keeps terminal UI; carry gates per harness.
|
|
|
|
Cross-family from a claude SDK source so ``copy_model_settings`` is always
|
|
False. ``carry_history_into_native`` is True only for the harness with a
|
|
rebuildable session file (pi-native), False for cursor-native.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore(
|
|
{
|
|
"ag_session_scoped": _CURRENT,
|
|
target_agent.id: target_agent,
|
|
"ag_builtin_origin": _BUILTIN_ORIGIN,
|
|
}
|
|
)
|
|
monkeypatch.setattr(
|
|
sessions_mod,
|
|
"get_agent_cache",
|
|
lambda: _HarnessAgentCacheStub(
|
|
{"ag_session_scoped": "claude_sdk", target_agent.id: target_harness}
|
|
),
|
|
)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post("/v1/sessions/conv_src/switch-agent", json={"agent_id": target_agent.id})
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
call = conv_store.switch_calls[0]
|
|
assert call["copy_model_settings"] is False
|
|
assert call["carry_history_into_native"] is expected_carry, (
|
|
f"{target_harness} carry_history_into_native should be {expected_carry}."
|
|
)
|
|
assert call["presentation_labels"] == expected_labels
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_400_noop_same_bundle() -> None:
|
|
"""Switching to the built-in the session already runs (same bundle) is a
|
|
no-op and rejected with 400 — no store mutation.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
# _BUILTIN_ORIGIN shares the current clone's bundle_location.
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_origin": _BUILTIN_ORIGIN}
|
|
)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_origin"}
|
|
)
|
|
|
|
assert resp.status_code == 400, resp.text
|
|
# No-op rejected before any mutation.
|
|
assert conv_store.switch_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_publishes_agent_changed_event(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A successful switch publishes ``session.agent_changed`` on the session
|
|
stream, carrying the cloned agent's id and the clean target-agent
|
|
display name, so connected clients
|
|
re-derive their binding-dependent state (the chat store's
|
|
``isNativeTerminalSession`` gate) without a navigation. No event → a
|
|
bound web client keeps treating the session as the old harness and
|
|
drops the first post-switch optimistic bubble on idle churn.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_claude": _BUILTIN_CLAUDE}
|
|
)
|
|
_patch_family_helpers(monkeypatch, same_family=True, native=True, labels={})
|
|
|
|
# Capture session-stream publishes by rebinding the module's
|
|
# ``session_stream`` reference to a recorder (not patching ``publish``
|
|
# through the shared module singleton) — omnigent-testing rule 14.
|
|
published: list[dict[str, object]] = []
|
|
|
|
class _RecordingStream:
|
|
@staticmethod
|
|
def publish(conversation_id: str, event: dict[str, object]) -> None:
|
|
published.append({"_conversation_id": conversation_id, **event})
|
|
|
|
monkeypatch.setattr(sessions_mod, "session_stream", _RecordingStream)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
changed = [e for e in published if e.get("type") == "session.agent_changed"]
|
|
# Exactly one broadcast per switch. Zero means clients never learn about
|
|
# the in-place rebind; more than one would double the snapshot refetch.
|
|
assert len(changed) == 1, f"expected exactly one agent_changed event, got {published}"
|
|
event = changed[0]
|
|
# Published on the switched session's channel — a wrong id would deliver
|
|
# the refresh signal to some other session's subscribers.
|
|
assert event["_conversation_id"] == "conv_src"
|
|
assert event["conversation_id"] == "conv_src"
|
|
# agent_id must be the agent the store actually bound (the fresh
|
|
# clone) — that's the durable reference clients re-bind to.
|
|
call = conv_store.switch_calls[0]
|
|
assert event["agent_id"] == call["new_agent_id"]
|
|
# agent_name must be the clean target name, NOT the clone row's
|
|
# "<name> (switch ag_…)" disambiguation name — clients display it
|
|
# verbatim. The clone name leaking here is the web's
|
|
# flash-of-ugly-name bug.
|
|
assert event["agent_name"] == "claude-native-ui"
|
|
assert event["agent_name"] != call["new_agent_name"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_rejected_publishes_no_event(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A rejected switch (no-op same-bundle target, 400) publishes nothing —
|
|
clients must not refetch a snapshot for a binding that didn't change.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
# _BUILTIN_ORIGIN shares the current clone's bundle_location → 400.
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_origin": _BUILTIN_ORIGIN}
|
|
)
|
|
published: list[dict[str, object]] = []
|
|
|
|
class _RecordingStream:
|
|
@staticmethod
|
|
def publish(conversation_id: str, event: dict[str, object]) -> None:
|
|
published.append({"_conversation_id": conversation_id, **event})
|
|
|
|
monkeypatch.setattr(sessions_mod, "session_stream", _RecordingStream)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_origin"}
|
|
)
|
|
|
|
assert resp.status_code == 400, resp.text
|
|
# No mutation happened, so no broadcast may go out.
|
|
assert published == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_schedules_runner_resource_reset(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A successful switch schedules a runner-side resource reset so the new
|
|
agent's os_env/sandbox governs the web filesystem/shell endpoints (the
|
|
cached primary OSEnv is dropped) and no lingering native terminal shadows
|
|
the next harness's transcript rebuild.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_claude": _BUILTIN_CLAUDE}
|
|
)
|
|
_patch_family_helpers(monkeypatch, same_family=True, native=True, labels={})
|
|
reset_calls: list[str] = []
|
|
|
|
async def _record_reset(session_id: str) -> None:
|
|
reset_calls.append(session_id)
|
|
|
|
monkeypatch.setattr(sessions_mod, "_reset_runner_resources_after_switch", _record_reset)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
# The reset must be scheduled (and run, since TestClient drains background
|
|
# tasks after the response) for exactly this session. An empty list means
|
|
# the route stopped wiring the reset — the new agent's sandbox would never
|
|
# take effect on the cached primary env and a stale terminal could shadow
|
|
# the rebuild.
|
|
assert reset_calls == ["conv_src"]
|
|
|
|
|
|
class _RunnerClientStub:
|
|
"""Async runner-HTTP-client stub recording reset-state POSTs.
|
|
|
|
:param fail: When True, ``post`` raises ``httpx.ConnectError`` after
|
|
recording the call, simulating a transport-level runner hiccup.
|
|
:param status_code: HTTP status of the stubbed response, e.g. ``500``
|
|
for a runner-side reset failure — httpx does NOT raise on it, so
|
|
production must check it explicitly via ``raise_for_status``.
|
|
"""
|
|
|
|
def __init__(self, fail: bool = False, status_code: int = 200) -> None:
|
|
self._fail = fail
|
|
self._status_code = status_code
|
|
self.posts: list[str] = []
|
|
|
|
async def post(self, url: str, timeout: float) -> httpx.Response:
|
|
"""Record the POST and return a real response of the stubbed status.
|
|
|
|
:param url: Runner-relative URL, e.g.
|
|
``"/v1/sessions/conv_src/reset-state"``.
|
|
:param timeout: Per-request timeout in seconds (unused).
|
|
:returns: An ``httpx.Response`` so production's
|
|
``raise_for_status`` behaves exactly as on a live client.
|
|
"""
|
|
del timeout
|
|
self.posts.append(url)
|
|
if self._fail:
|
|
raise httpx.ConnectError("runner unreachable")
|
|
return httpx.Response(
|
|
self._status_code,
|
|
request=httpx.Request("POST", f"http://runner.test{url}"),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_reset_publishes_changed_files_invalidated_after_reset(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""After the post-switch runner reset succeeds, the route's background
|
|
task publishes ``session.changed_files.invalidated`` so web clients
|
|
refetch filesystem state — including environment availability, which
|
|
flips the Files tab when the switch crosses an os_env boundary. The
|
|
event must come AFTER the reset POST (and after ``agent_changed``):
|
|
publishing before the reset would have clients refetch the OLD agent's
|
|
still-cached env.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_claude": _BUILTIN_CLAUDE}
|
|
)
|
|
_patch_family_helpers(monkeypatch, same_family=True, native=True, labels={})
|
|
runner = _RunnerClientStub()
|
|
|
|
async def _get_runner(session_id: str) -> _RunnerClientStub:
|
|
del session_id
|
|
return runner
|
|
|
|
monkeypatch.setattr(sessions_mod, "_get_runner_client_for_resource_access", _get_runner)
|
|
published: list[dict[str, object]] = []
|
|
|
|
class _RecordingStream:
|
|
@staticmethod
|
|
def publish(conversation_id: str, event: dict[str, object]) -> None:
|
|
published.append({"_conversation_id": conversation_id, **event})
|
|
|
|
monkeypatch.setattr(sessions_mod, "session_stream", _RecordingStream)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
# The real reset ran (TestClient drains background tasks) and hit the
|
|
# runner's dedicated endpoint — a different URL means the old env was
|
|
# never closed and the event below would advertise a stale refetch.
|
|
assert runner.posts == ["/v1/sessions/conv_src/reset-state"]
|
|
types = [e["type"] for e in published]
|
|
# Exactly one invalidation per switch; zero means the Files tab never
|
|
# learns availability changed until the 60 s staleTime + a refetch
|
|
# trigger.
|
|
assert types.count("session.changed_files.invalidated") == 1, published
|
|
# Ordering: agent_changed (at store-commit time) strictly before the
|
|
# invalidation (after the reset). Reversed order would mean the
|
|
# invalidation was published while the old env was still cached.
|
|
assert types.index("session.agent_changed") < types.index("session.changed_files.invalidated")
|
|
event = published[types.index("session.changed_files.invalidated")]
|
|
assert event["_conversation_id"] == "conv_src"
|
|
assert event["session_id"] == "conv_src"
|
|
# The web client keys filesystem queries by the default environment.
|
|
assert event["environment_id"] == "default"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"getter_kind",
|
|
[
|
|
# Reset POST raises mid-flight (transport-level runner hiccup).
|
|
"post_fails",
|
|
# Reset POST returns non-2xx — httpx does NOT raise on this; the
|
|
# route must check the status itself or it would publish against
|
|
# a runner that never closed the old env.
|
|
"post_500",
|
|
# No runner client at all (session not runner-bound / unit setup).
|
|
"no_client",
|
|
],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_switch_reset_failure_publishes_no_changed_files_event(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
getter_kind: str,
|
|
) -> None:
|
|
"""When the post-switch reset doesn't complete, NO
|
|
``session.changed_files.invalidated`` goes out: the runner's env cache
|
|
is still the OLD agent's, so a triggered refetch would re-serve stale
|
|
availability. (``agent_changed`` still fires — the store commit
|
|
succeeded.)
|
|
|
|
:param getter_kind: Which failure shape to simulate —
|
|
``"post_fails"`` (reset POST raises), ``"post_500"`` (reset POST
|
|
returns HTTP 500), or ``"no_client"`` (no runner client resolved).
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_claude": _BUILTIN_CLAUDE}
|
|
)
|
|
_patch_family_helpers(monkeypatch, same_family=True, native=True, labels={})
|
|
|
|
async def _get_runner(session_id: str) -> _RunnerClientStub | None:
|
|
del session_id
|
|
if getter_kind == "post_fails":
|
|
return _RunnerClientStub(fail=True)
|
|
if getter_kind == "post_500":
|
|
return _RunnerClientStub(status_code=500)
|
|
return None
|
|
|
|
monkeypatch.setattr(sessions_mod, "_get_runner_client_for_resource_access", _get_runner)
|
|
published: list[dict[str, object]] = []
|
|
|
|
class _RecordingStream:
|
|
@staticmethod
|
|
def publish(conversation_id: str, event: dict[str, object]) -> None:
|
|
published.append({"_conversation_id": conversation_id, **event})
|
|
|
|
monkeypatch.setattr(sessions_mod, "session_stream", _RecordingStream)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
types = [e["type"] for e in published]
|
|
# The switch itself committed, so the binding broadcast still goes out.
|
|
assert types.count("session.agent_changed") == 1
|
|
# But no filesystem invalidation: the old env wasn't closed, so a
|
|
# refetch now would re-serve it; recovery is the client-side stale-mark
|
|
# (focus/remount refetch) or the runner relaunching with fresh caches.
|
|
assert types.count("session.changed_files.invalidated") == 0, published
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_404_missing_session() -> None:
|
|
"""404 when the session does not exist (before any mutation)."""
|
|
conv_store = _ConversationStore(conversations={})
|
|
agent_store = _AgentStore({})
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post("/v1/sessions/conv_missing/switch-agent", json={"agent_id": "ag_x"})
|
|
|
|
assert resp.status_code == 404, resp.text
|
|
assert conv_store.switch_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_400_sub_agent() -> None:
|
|
"""400 when the session is a sub-agent (only top-level can switch)."""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv(kind="sub_agent")})
|
|
agent_store = _AgentStore({"ag_session_scoped": _CURRENT})
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 400, resp.text
|
|
assert conv_store.switch_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_404_target_not_bindable() -> None:
|
|
"""404 when the target is a session-scoped agent (not a built-in)."""
|
|
other_session_agent = _agent("ag_other", "other", "bundle/x", "conv_other")
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore({"ag_session_scoped": _CURRENT, "ag_other": other_session_agent})
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post("/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_other"})
|
|
|
|
# Session-scoped target → not bindable, mapped to 404, nothing mutated.
|
|
assert resp.status_code == 404, resp.text
|
|
assert conv_store.switch_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_409_when_busy(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""409 when a turn is running — switching mid-turn is rejected."""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_claude": _BUILTIN_CLAUDE}
|
|
)
|
|
# Mark the session as running in the relay status cache.
|
|
monkeypatch.setitem(sessions_mod._session_status_cache, "conv_src", "running")
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 409, resp.text
|
|
# The 409 fires before the irreversible mutation.
|
|
assert conv_store.switch_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_400_unloadable_target_bundle(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""400 when the target bundle can't load — fails before deleting the old
|
|
agent, so the session is left untouched.
|
|
"""
|
|
conv_store = _ConversationStore(conversations={"conv_src": _conv()})
|
|
agent_store = _AgentStore(
|
|
{"ag_session_scoped": _CURRENT, "ag_builtin_claude": _BUILTIN_CLAUDE}
|
|
)
|
|
_patch_family_helpers(
|
|
monkeypatch, same_family=True, native=False, labels={}, raise_on_load=True
|
|
)
|
|
client = TestClient(_build_app(conv_store, agent_store))
|
|
|
|
resp = client.post(
|
|
"/v1/sessions/conv_src/switch-agent", json={"agent_id": "ag_builtin_claude"}
|
|
)
|
|
|
|
assert resp.status_code == 400, resp.text
|
|
# Pre-commit failure → no switch attempted (old agent intact).
|
|
assert conv_store.switch_calls == []
|