vh
fc256bbaa4
The slice-3 heid-bug-hunt panel (3/3) caught a real regression the cutover introduced, plus a chokepoint-invariant gap: - ConnectFailed escaped both rewired CLI probes. When --set-persona-pad and --seed-first-message moved off raw httpx onto the wt adapter, transport failures changed class: the SDK normalizes any pre-response transport error to worldtree_sdk.ConnectFailed (request.py), a WorldtreeError (not ApiError), so it passed the adapter unmapped AND the probes' httpx-only except tuples → an uncaught traceback instead of the graceful [network_error] exit 21. _amain (slice-2) already handled it; the probes lagged. Fix: add ConnectFailed to both probe except tuples (mirrors _amain). Live-verified at a refused host → [network_error] exit 21. - Finite-PAD enforced only at the CLI, not the adapter chokepoint. wt.set_persona_state delegated finiteness to the caller (documented), so a direct/non-CLI caller passing nan/inf got a raw SDK ConfigurationError. Fix: assert finiteness in the adapter precondition (consistent with its other precondition asserts) so the invariant holds at the chokepoint in ratatoskr's own terms; the CLI pre-check stays for the friendly usage error. Triaged-and-declined (all correct per the panel + Heid's source-check): the deleted sessions.py exports (intended no-shim cutover, zero un-migrated importers), the session["session_id"] index (accept-known-risk, matches --new), and Regin's "web indefinite block" (refuted — the seed is asyncio.wait_for-bounded). The concurrent heid-code-review panel returned zero drift, no code change. TDD: 3 RED tests (both probes' ConnectFailed → exit 21; adapter nan/inf/-inf → AssertionError, never reaches the SDK) → GREEN. Suite 469; ruff clean; mypy no new errors.
531 lines
23 KiB
Python
531 lines
23 KiB
Python
"""Unit tests for the worldtree-sdk adapter (`ratatoskr.wt`) — slice-1 foundation.
|
|
|
|
Covers the two foundation surfaces (issue #20 cutover contract, slice 1):
|
|
* `build_client` — construction wiring + injected-transport ownership (INV-CUT-1:
|
|
the SDK must never close ratatoskr's transport).
|
|
* `translate_error` — the § Error map DEFAULT (`ApiError` → `SessionApiFailed`)
|
|
plus discriminated-`WorldtreeError` passthrough (INV-CUT-2).
|
|
|
|
No ratatoskr surface (CLI / web / TUI) is exercised here — that wiring lands in
|
|
slice 2. These tests hit no network (WorldtreeClient does no I/O at construction).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, cast
|
|
|
|
import httpx
|
|
import pytest
|
|
import worldtree_sdk as wtsdk
|
|
from worldtree_sdk import ApiError, CancelResult, PadState, WorldtreeClient
|
|
|
|
from ratatoskr.sessions import (
|
|
AgentNotFound,
|
|
AuthoredHistoryUnavailable,
|
|
BifrostBinding,
|
|
BifrostConsumerKeyMissing,
|
|
BifrostHandshakeFailed,
|
|
InvalidCursor,
|
|
)
|
|
from ratatoskr.sse_client import (
|
|
AgentNotAvailable,
|
|
CancelAlreadyCompleted,
|
|
CancelFailed,
|
|
CancelTurnNotFound,
|
|
MalformedSseData,
|
|
MalformedSseId,
|
|
SseConnectFailed,
|
|
SseConnectionDropped,
|
|
TurnIdFlip,
|
|
TurnLaunchUnavailable,
|
|
)
|
|
from ratatoskr.wt import (
|
|
SessionApiFailed,
|
|
build_client,
|
|
cancel_turn,
|
|
create_session,
|
|
get_session_messages,
|
|
get_session_tools,
|
|
list_sessions,
|
|
set_persona_state,
|
|
stream_turn,
|
|
translate_error,
|
|
write_authored_history,
|
|
)
|
|
|
|
|
|
class _FakeSessions:
|
|
"""A stand-in for `WorldtreeClient.sessions` — records the last call and
|
|
returns a canned result or raises a canned error. Lets the adapter's
|
|
body-building + error-mapping be unit-tested without any SDK HTTP."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
result: Any = None,
|
|
error: BaseException | None = None,
|
|
events: list[Any] | None = None,
|
|
stream_error: BaseException | None = None,
|
|
) -> None:
|
|
self._result = result
|
|
self._error = error
|
|
self._events = events or []
|
|
self._stream_error = stream_error
|
|
self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = []
|
|
|
|
async def _dispatch(self, name: str, *args: Any, **kwargs: Any) -> Any:
|
|
self.calls.append((name, args, kwargs))
|
|
if self._error is not None:
|
|
raise self._error
|
|
return self._result
|
|
|
|
async def create(self, *args: Any, **kwargs: Any) -> Any:
|
|
return await self._dispatch("create", *args, **kwargs)
|
|
|
|
async def list(self, *args: Any, **kwargs: Any) -> Any:
|
|
return await self._dispatch("list", *args, **kwargs)
|
|
|
|
async def messages(self, *args: Any, **kwargs: Any) -> Any:
|
|
return await self._dispatch("messages", *args, **kwargs)
|
|
|
|
async def tools(self, *args: Any, **kwargs: Any) -> Any:
|
|
return await self._dispatch("tools", *args, **kwargs)
|
|
|
|
def stream_turn(self, *args: Any, **kwargs: Any) -> Any:
|
|
self.calls.append(("stream_turn", args, kwargs))
|
|
return self._astream()
|
|
|
|
async def _astream(self) -> Any:
|
|
for event in self._events:
|
|
yield event
|
|
if self._stream_error is not None:
|
|
raise self._stream_error
|
|
|
|
async def cancel_turn(self, *args: Any, **kwargs: Any) -> Any:
|
|
return await self._dispatch("cancel_turn", *args, **kwargs)
|
|
|
|
async def set_persona_state(self, *args: Any, **kwargs: Any) -> Any:
|
|
return await self._dispatch("set_persona_state", *args, **kwargs)
|
|
|
|
async def write_history(self, *args: Any, **kwargs: Any) -> Any:
|
|
return await self._dispatch("write_history", *args, **kwargs)
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, sessions: _FakeSessions) -> None:
|
|
self.sessions = sessions
|
|
|
|
|
|
def _wt(sessions: _FakeSessions) -> WorldtreeClient:
|
|
"""Cast the structural fake to the nominal client type the adapter is typed
|
|
against — the route functions only touch `client.sessions.*`, which the fake
|
|
provides. (No network; construction does no I/O.)"""
|
|
return cast(WorldtreeClient, _FakeClient(sessions))
|
|
|
|
|
|
class TestBuildClient:
|
|
async def test_constructs_worldtree_client(self) -> None:
|
|
transport = httpx.AsyncClient()
|
|
try:
|
|
client = build_client(
|
|
"https://wt.example:8081", api_key="ck-test", transport=transport
|
|
)
|
|
assert isinstance(client, WorldtreeClient)
|
|
assert client.base_url == "https://wt.example:8081"
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
async def test_injected_transport_is_ratatoskr_owned(self) -> None:
|
|
# INV-CUT-1 [hard]: aclose() on the SDK client must NOT close ratatoskr's
|
|
# transport — ratatoskr owns the lifecycle exactly as it does today.
|
|
transport = httpx.AsyncClient()
|
|
client = build_client("https://wt.example", api_key="ck", transport=transport)
|
|
await client.aclose()
|
|
assert client.closed is True
|
|
assert transport.is_closed is False
|
|
await transport.aclose()
|
|
|
|
async def test_admin_key_optional(self) -> None:
|
|
transport = httpx.AsyncClient()
|
|
try:
|
|
# Absent admin_key → admin_auth=None; still constructs.
|
|
without_admin = build_client(
|
|
"https://wt.example", api_key="ck", transport=transport
|
|
)
|
|
assert isinstance(without_admin, WorldtreeClient)
|
|
# Present admin_key → constructs (admin surface available in later slices).
|
|
with_admin = build_client(
|
|
"https://wt.example", api_key="ck", admin_key="ak", transport=transport
|
|
)
|
|
assert isinstance(with_admin, WorldtreeClient)
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
class TestTranslateError:
|
|
def test_apierror_maps_to_session_api_failed_default(self) -> None:
|
|
exc = ApiError("some_code", "boom", status=500, body="raw-body")
|
|
mapped = translate_error(exc)
|
|
assert isinstance(mapped, SessionApiFailed)
|
|
assert mapped.status == 500
|
|
assert mapped.error_code == "some_code"
|
|
assert mapped.body == "raw-body"
|
|
|
|
def test_apierror_with_no_body_maps_cleanly(self) -> None:
|
|
exc = ApiError("nope", "no body", status=404)
|
|
mapped = translate_error(exc)
|
|
assert isinstance(mapped, SessionApiFailed)
|
|
assert mapped.status == 404
|
|
assert mapped.error_code == "nope"
|
|
assert mapped.body is None
|
|
|
|
def test_discriminated_subclass_passes_through_unchanged(self) -> None:
|
|
# Discriminated WorldtreeError subclasses are already the right semantic
|
|
# type at the REST layer — translate_error passes them through by identity
|
|
# (the stream routes re-wrap them; that is stream_turn's job, not this one).
|
|
exc = wtsdk.AgentNotAvailable("agent_not_available", "gone", status=409)
|
|
assert translate_error(exc) is exc
|
|
|
|
def test_non_worldtree_error_passes_through_unchanged(self) -> None:
|
|
exc = ValueError("unrelated")
|
|
assert translate_error(exc) is exc
|
|
|
|
|
|
class TestCreateSession:
|
|
async def test_happy_returns_sdk_dict_and_builds_body(self) -> None:
|
|
info = {"session_id": "s-1", "agent_id": "mimir", "created_at": "t", "last_active": "t"}
|
|
fake = _FakeSessions(result=info)
|
|
client = _wt(fake)
|
|
out = await create_session(client, "mimir", end_user_id="u-9")
|
|
assert out is info # open-world passthrough — no re-shaping
|
|
name, args, kwargs = fake.calls[-1]
|
|
assert name == "create"
|
|
assert args[0] == {"agent_id": "mimir", "end_user_id": "u-9"}
|
|
assert kwargs["consumer_key"] is None
|
|
|
|
async def test_config_passthrough(self) -> None:
|
|
fake = _FakeSessions(result={"session_id": "s"})
|
|
await create_session(
|
|
_wt(fake), "echo", config={"system_prompt": "be terse"}
|
|
)
|
|
assert fake.calls[-1][1][0] == {
|
|
"agent_id": "echo",
|
|
"config": {"system_prompt": "be terse"},
|
|
}
|
|
|
|
async def test_bifrost_bound_body_and_consumer_key(self) -> None:
|
|
fake = _FakeSessions(result={"session_id": "s"})
|
|
binding = BifrostBinding(endpoint_url="http://h:8391", scope=None)
|
|
await create_session(
|
|
_wt(fake), "sindra", bifrost=binding, consumer_key="ck-real"
|
|
)
|
|
_name, args, kwargs = fake.calls[-1]
|
|
assert args[0] == {
|
|
"agent_id": "sindra",
|
|
"bifrost": {"endpoint_url": "http://h:8391", "scope": None},
|
|
}
|
|
# INV-CUT: the consumer key rides the SDK's per-request auth, NOT a header.
|
|
assert kwargs["consumer_key"] == "ck-real"
|
|
|
|
async def test_unbound_create_drops_consumer_key(self) -> None:
|
|
# A consumer_key must NOT reach the SDK on an UNBOUND create — the SDK's
|
|
# credential precedence would otherwise auth as the Bifrost consumer instead
|
|
# of the default bearer (heid-bug-hunt Gróa#4 / Regin#4).
|
|
fake = _FakeSessions(result={"session_id": "s"})
|
|
await create_session(_wt(fake), "mimir", consumer_key="ck-should-be-dropped")
|
|
assert fake.calls[-1][2]["consumer_key"] is None
|
|
|
|
async def test_bifrost_without_consumer_key_rejected_pre_http(self) -> None:
|
|
fake = _FakeSessions(result={"session_id": "s"})
|
|
binding = BifrostBinding(endpoint_url="http://h:8391", scope=None)
|
|
with pytest.raises(BifrostConsumerKeyMissing):
|
|
await create_session(_wt(fake), "sindra", bifrost=binding)
|
|
assert fake.calls == [] # never reached the SDK
|
|
|
|
async def test_404_maps_to_agent_not_found(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("agent_not_found", "no", status=404))
|
|
with pytest.raises(AgentNotFound) as ei:
|
|
await create_session(_wt(fake), "ghost")
|
|
assert ei.value.agent_id == "ghost"
|
|
|
|
async def test_bound_502_maps_to_bifrost_handshake_failed(self) -> None:
|
|
body = '{"detail": {"bifrost_error": "bifrost.auth_rejected"}}'
|
|
fake = _FakeSessions(
|
|
error=ApiError("bifrost_handshake_failed", "boom", status=502, body=body)
|
|
)
|
|
binding = BifrostBinding(endpoint_url="http://h:8391", scope=None)
|
|
with pytest.raises(BifrostHandshakeFailed) as ei:
|
|
await create_session(
|
|
_wt(fake), "sindra", bifrost=binding, consumer_key="ck"
|
|
)
|
|
assert ei.value.bifrost_error == "bifrost.auth_rejected"
|
|
|
|
async def test_unbound_502_stays_session_api_failed(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("upstream", "boom", status=502, body="x"))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await create_session(_wt(fake), "mimir")
|
|
assert ei.value.status == 502
|
|
|
|
async def test_default_error_maps_to_session_api_failed(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("weird", "boom", status=418, body="teapot"))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await create_session(_wt(fake), "mimir")
|
|
assert ei.value.status == 418
|
|
assert ei.value.error_code == "weird"
|
|
|
|
|
|
class TestListSessions:
|
|
async def test_passes_params_and_returns_dict(self) -> None:
|
|
page: dict[str, Any] = {"items": [], "next_cursor": None}
|
|
fake = _FakeSessions(result=page)
|
|
out = await list_sessions(_wt(fake), limit=10, cursor="c1", include_archived=True)
|
|
assert out is page
|
|
kwargs = fake.calls[-1][2]
|
|
assert kwargs["limit"] == 10
|
|
assert kwargs["cursor"] == "c1"
|
|
assert kwargs["include_archived"] is True
|
|
|
|
async def test_422_cursor_invalid_maps_to_invalid_cursor(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("cursor_invalid", "bad", status=422))
|
|
with pytest.raises(InvalidCursor) as ei:
|
|
await list_sessions(_wt(fake), cursor="bogus")
|
|
assert ei.value.raw == "bogus"
|
|
|
|
async def test_other_422_stays_session_api_failed(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("validation_failed", "x", status=422))
|
|
with pytest.raises(SessionApiFailed):
|
|
await list_sessions(_wt(fake))
|
|
|
|
|
|
class TestReadPassthroughs:
|
|
async def test_messages_returns_dict(self) -> None:
|
|
data = {"session_id": "s", "items": []}
|
|
fake = _FakeSessions(result=data)
|
|
assert await get_session_messages(_wt(fake), "s") is data
|
|
assert fake.calls[-1][0] == "messages"
|
|
|
|
async def test_tools_returns_dict(self) -> None:
|
|
data = {"agent_id": "mimir", "builtin_tools": []}
|
|
fake = _FakeSessions(result=data)
|
|
assert await get_session_tools(_wt(fake), "s") is data
|
|
assert fake.calls[-1][0] == "tools"
|
|
|
|
async def test_messages_error_maps_to_session_api_failed(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("auth_revoked", "no", status=401))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await get_session_messages(_wt(fake), "s")
|
|
assert ei.value.status == 401
|
|
|
|
async def test_tools_error_maps_to_session_api_failed(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("auth_revoked", "no", status=401))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await get_session_tools(_wt(fake), "s")
|
|
assert ei.value.status == 401
|
|
|
|
|
|
async def _drain(aiter: Any) -> list[Any]:
|
|
out: list[Any] = []
|
|
async for ev in aiter:
|
|
out.append(ev)
|
|
return out
|
|
|
|
|
|
class TestStreamTurn:
|
|
async def test_yields_events_verbatim(self) -> None:
|
|
e1, e2 = object(), object()
|
|
fake = _FakeSessions(events=[e1, e2])
|
|
got = await _drain(stream_turn(_wt(fake), "s-1", "hello"))
|
|
assert got == [e1, e2]
|
|
assert fake.calls[-1] == ("stream_turn", ("s-1", "hello"), {})
|
|
|
|
async def test_session_retired_maps_to_session_api_failed(self) -> None:
|
|
fake = _FakeSessions(
|
|
stream_error=wtsdk.SessionRetired("session_retired", "gone", status=410)
|
|
)
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.status == 410
|
|
assert ei.value.error_code == "session_retired"
|
|
|
|
async def test_agent_not_available_rewraps_to_ratatoskr(self) -> None:
|
|
fake = _FakeSessions(
|
|
stream_error=wtsdk.AgentNotAvailable("agent_not_available", "no agent", status=409)
|
|
)
|
|
with pytest.raises(AgentNotAvailable) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.error_code == "agent_not_available"
|
|
assert ei.value.status == 409
|
|
|
|
async def test_turn_launch_unavailable_rewraps_to_ratatoskr(self) -> None:
|
|
fake = _FakeSessions(
|
|
stream_error=wtsdk.TurnLaunchUnavailable("not_ready", "busy", status=503)
|
|
)
|
|
with pytest.raises(TurnLaunchUnavailable) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.retryable is True
|
|
|
|
async def test_generic_connect_failed_maps_to_sse_connect_failed(self) -> None:
|
|
fake = _FakeSessions(stream_error=wtsdk.ConnectFailed("connect_failed", "boom", status=500))
|
|
with pytest.raises(SseConnectFailed) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.status == 500
|
|
|
|
async def test_connection_dropped_maps_and_carries_cursor(self) -> None:
|
|
fake = _FakeSessions(stream_error=wtsdk.ConnectionDropped("12:3"))
|
|
with pytest.raises(SseConnectionDropped) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.last_seen_sse_id == "12:3"
|
|
|
|
async def test_resume_error_maps_to_sse_connect_failed(self) -> None:
|
|
fake = _FakeSessions(stream_error=wtsdk.ResumeError("resume_failed", "boom", status=412))
|
|
with pytest.raises(SseConnectFailed) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.status == 412
|
|
|
|
async def test_malformed_sse_id_passes_through_as_ratatoskr(self) -> None:
|
|
fake = _FakeSessions(stream_error=wtsdk.MalformedSseId("bad-id"))
|
|
with pytest.raises(MalformedSseId) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.raw == "bad-id"
|
|
|
|
async def test_malformed_sse_data_passes_through_as_ratatoskr(self) -> None:
|
|
fake = _FakeSessions(stream_error=wtsdk.MalformedSseData("not json"))
|
|
with pytest.raises(MalformedSseData):
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
|
|
async def test_turn_id_flip_carries_established_and_got(self) -> None:
|
|
fake = _FakeSessions(stream_error=wtsdk.TurnIdFlip(5, 7))
|
|
with pytest.raises(TurnIdFlip) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert (ei.value.established, ei.value.got) == (5, 7)
|
|
|
|
async def test_undiscriminated_api_error_maps_to_session_api_failed(self) -> None:
|
|
# INV-CUT-2 default: an undiscriminated ApiError surfacing from the stream
|
|
# (not a discriminated stream error) → SessionApiFailed.
|
|
fake = _FakeSessions(stream_error=ApiError("weird", "boom", status=500))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await _drain(stream_turn(_wt(fake), "s", "hi"))
|
|
assert ei.value.status == 500
|
|
|
|
|
|
class TestCancelTurn:
|
|
async def test_happy_returns_cancel_result(self) -> None:
|
|
res = CancelResult(turn_id=42, cancelled=True, reason=None, partial_message_id=None)
|
|
fake = _FakeSessions(result=res)
|
|
out = await cancel_turn(_wt(fake), "s-1", 42, persist_partial=True)
|
|
assert out is res
|
|
assert fake.calls[-1] == ("cancel_turn", ("s-1", 42), {"persist_partial": True})
|
|
|
|
async def test_late_cancel_race_is_a_result_not_an_error(self) -> None:
|
|
# B-CAN-3: a 200 with cancelled=False is the benign late-cancel no-op.
|
|
res = CancelResult(turn_id=42, cancelled=False, reason=None, partial_message_id=None)
|
|
out = await cancel_turn(_wt(_FakeSessions(result=res)), "s", 42)
|
|
assert out.cancelled is False
|
|
|
|
async def test_turn_not_found_maps_to_ratatoskr(self) -> None:
|
|
fake = _FakeSessions(
|
|
error=wtsdk.CancelTurnNotFound(42, error_code="turn_not_found", message="gone")
|
|
)
|
|
with pytest.raises(CancelTurnNotFound) as ei:
|
|
await cancel_turn(_wt(fake), "s", 42)
|
|
assert ei.value.turn_id == 42
|
|
|
|
async def test_turn_finished_maps_to_ratatoskr(self) -> None:
|
|
fake = _FakeSessions(
|
|
error=wtsdk.CancelAlreadyCompleted(42, error_code="turn_finished", message="done")
|
|
)
|
|
with pytest.raises(CancelAlreadyCompleted):
|
|
await cancel_turn(_wt(fake), "s", 42)
|
|
|
|
async def test_other_cancel_failure_maps_to_cancel_failed(self) -> None:
|
|
fake = _FakeSessions(error=wtsdk.CancelFailed(42, error_code="boom", message="failed"))
|
|
with pytest.raises(CancelFailed):
|
|
await cancel_turn(_wt(fake), "s", 42)
|
|
|
|
async def test_undiscriminated_api_error_maps_to_session_api_failed(self) -> None:
|
|
# INV-CUT-2 default: an undiscriminated ApiError on the cancel route (not a
|
|
# typed Cancel* race) → SessionApiFailed, never leaked as a bare ApiError.
|
|
fake = _FakeSessions(error=ApiError("weird", "boom", status=500))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await cancel_turn(_wt(fake), "s", 42)
|
|
assert ei.value.status == 500
|
|
|
|
|
|
class TestSetPersonaState:
|
|
"""slice-3: set_persona_state → SDK sessions.set_persona_state(PadState). The
|
|
adapter builds the canonical PadState (the SDK owns the {"pad": {...}} wire
|
|
shape); no error row beyond the § Error map default (SessionApiFailed)."""
|
|
|
|
async def test_happy_builds_padstate_and_returns_none(self) -> None:
|
|
fake = _FakeSessions(result=None) # SDK resolves the 204 to None
|
|
out = await set_persona_state(
|
|
_wt(fake), "s-1", pleasure=0.4, arousal=0.1, dominance=-0.2
|
|
)
|
|
assert out is None
|
|
name, args, _kwargs = fake.calls[-1]
|
|
assert name == "set_persona_state"
|
|
assert args[0] == "s-1"
|
|
pad = args[1]
|
|
assert isinstance(pad, PadState)
|
|
assert (pad.pleasure, pad.arousal, pad.dominance) == (0.4, 0.1, -0.2)
|
|
|
|
async def test_falsy_zero_pad_preserved(self) -> None:
|
|
# A 0.0 axis must survive verbatim (not be dropped as falsy).
|
|
fake = _FakeSessions(result=None)
|
|
await set_persona_state(_wt(fake), "s", pleasure=0.0, arousal=0.0, dominance=0.0)
|
|
pad = fake.calls[-1][1][1]
|
|
assert (pad.pleasure, pad.arousal, pad.dominance) == (0.0, 0.0, 0.0)
|
|
|
|
async def test_error_maps_to_session_api_failed(self) -> None:
|
|
fake = _FakeSessions(error=ApiError("upstream", "boom", status=500, body="x"))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await set_persona_state(_wt(fake), "s", pleasure=0.0, arousal=0.0, dominance=0.0)
|
|
assert ei.value.status == 500
|
|
|
|
async def test_non_finite_pad_rejected_at_the_chokepoint(self) -> None:
|
|
# The finite-PAD invariant is enforced at the adapter (not only the CLI): a
|
|
# non-finite axis would serialize to null and corrupt the injection, so a
|
|
# direct caller is rejected pre-SDK — never a leaked SDK ConfigurationError.
|
|
fake = _FakeSessions(result=None)
|
|
for bad in (float("nan"), float("inf"), float("-inf")):
|
|
with pytest.raises(AssertionError):
|
|
await set_persona_state(_wt(fake), "s", pleasure=bad, arousal=0.0, dominance=0.0)
|
|
assert fake.calls == [] # never reached the SDK
|
|
|
|
|
|
class TestWriteAuthoredHistory:
|
|
"""slice-3: write_authored_history → SDK sessions.write_history. Builds the
|
|
v1 authored-write entry (author="assistant", the only accepted author);
|
|
404 → AuthoredHistoryUnavailable (hide-existence); else the default."""
|
|
|
|
async def test_happy_builds_entry_and_returns_dict(self) -> None:
|
|
ack = {"seq": 0, "phase": "seeded", "turn_id": "t1", "content_chars": 3}
|
|
fake = _FakeSessions(result=ack)
|
|
out = await write_authored_history(
|
|
_wt(fake), "s-1", content="hi!", idempotency_key="k1"
|
|
)
|
|
assert out is ack # open-world passthrough
|
|
name, args, _kwargs = fake.calls[-1]
|
|
assert name == "write_history"
|
|
assert args[0] == "s-1"
|
|
assert args[1] == {
|
|
"author": "assistant",
|
|
"content": "hi!",
|
|
"idempotency_key": "k1",
|
|
}
|
|
|
|
async def test_404_maps_to_authored_history_unavailable(self) -> None:
|
|
# Hide-existence: the ROUTE is the discriminator (never the body) — any 404
|
|
# on write_history → AuthoredHistoryUnavailable, no capability-probe.
|
|
fake = _FakeSessions(error=ApiError("session_not_found", "no", status=404))
|
|
with pytest.raises(AuthoredHistoryUnavailable) as ei:
|
|
await write_authored_history(_wt(fake), "s-1", content="hi", idempotency_key="k")
|
|
assert ei.value.session_id == "s-1"
|
|
|
|
async def test_other_error_maps_to_session_api_failed(self) -> None:
|
|
# 409 generation_active (retryable) is NOT a hide-existence 404 → default.
|
|
fake = _FakeSessions(error=ApiError("generation_active", "busy", status=409))
|
|
with pytest.raises(SessionApiFailed) as ei:
|
|
await write_authored_history(_wt(fake), "s", content="hi", idempotency_key="k")
|
|
assert ei.value.status == 409
|