Files
ratatoskr/tests/test_tier3.py
T
vh 9fade55901 feat(local_agents): tier-3 index + picker merge (v0.8.0)
Worldtree's GET /agents doesn't return consumer-defined (tier-3)
agents — the public list excludes them by design. Confirmed live in
v0.7.0's smoke. Without server-side knowledge, ratatoskr's picker
couldn't show tier-3 agents the operator had defined; the workflow
was "remember the agent_id, pass --agent ratatoskr:<name>
explicitly." Friction grows with every tier-3 agent.

## Fix: client-side index, merged at picker time

New module `ratatoskr.local_agents` maintains a JSON-backed index at
$XDG_CONFIG_HOME/ratatoskr/local_agents.json (override via
$RATATOSKR_LOCAL_AGENTS). `tier3` CLI define / patch / delete update
the index as side-effects. `tui._resolve_then_run` loads the index
after `list_agents(client)` and appends entries not already in the
remote list (dedup by agent_id; remote wins on conflict).

Library-level `tier3.define_agent` / `patch_agent` / `delete_agent`
stay pure — local persistence lives in the CLI layer (`_run_define`
etc.), not in the library functions. Tests of the library don't
touch the filesystem.

## Public surface

  ratatoskr.local_agents:
    LocalAgentEntry (frozen dataclass)
    load_local_agents() -> list[LocalAgentEntry]
    add_local_agent(entry)
    update_local_agent(entry)  # same semantics as add (agent_id key)
    remove_local_agent(agent_id)
    make_description(system_prompt) -> str  # synthetic picker label

Failure modes are lenient: missing file → empty index; corrupt JSON
or schema mismatch → empty index (no crash). The picker continues
to show foundational agents either way; tier-3 surface degrades to
the pre-v0.8.0 workflow.

## Picker integration

Local entries convert to ratatoskr.sessions.AgentInfo with synthetic
fields:
  name        = agent_name (from LocalAgentEntry)
  description = "(tier 3) <first non-empty line of system prompt>"
  version, capabilities, supported_models, persona_traits, ui_hints
    = None / [] / [] / {} / {}

If Worldtree later starts returning tier-3 in GET /agents, this
module's role narrows to redundant local cache; can be removed
cleanly since the dedup-by-agent-id keeps remote-wins behavior.

## Tests

286/286 GREEN (was 265, +21: 20 local_agents + 1 picker-merge
integration). Ruff clean. Tests isolate the index via
$RATATOSKR_LOCAL_AGENTS pointed at pytest's tmp_path — no pollution
of operator's real ~/.config/ratatoskr/.

## Manual smoke

Sindra-like define against personal Worldtree:
  python -m ratatoskr.tier3 define --name foo --system-prompt "..." --model X
  cat ~/.config/ratatoskr/local_agents.json
  # ratatoskr --new picker now shows ratatoskr:foo alongside mimir et al.

Cross-machine: the file is per-host. Operator can sync via dotfiles
if needed; out of scope for this commit.

Minor bump (v0.7.1 → v0.8.0) — new public module + new picker
behavior (more agents shown). No caller-side breaking changes.
2026-05-24 21:13:30 -07:00

492 lines
19 KiB
Python

"""Tests for ratatoskr.tier3 per docs/contracts/issues/15.contract.md."""
from pathlib import Path
import httpx
import pytest
import respx
from ratatoskr.sessions import SessionApiFailed
from ratatoskr.tier3 import (
Tier3AgentInfo,
Tier3AgentNotFound,
Tier3FieldNotMutable,
Tier3LayerDeferred,
Tier3QuotaExceeded,
Tier3UserIdUnsupported,
define_agent,
delete_agent,
main,
patch_agent,
)
_FULL_AGENT_RESP = {
"agent_id": "ratatoskr:wizard",
"user_id": "ratatoskr",
"agent_name": "wizard",
"system_prompt": "You are a wizard.",
"model": "qwen3.6-35-a3b",
"created_at": "2026-05-25T03:20:09.703601+00:00",
"updated_at": "2026-05-25T03:20:09.703601+00:00",
}
class TestDefineAgent:
@respx.mock
async def test_happy_define(self) -> None:
"""happy_define [happy,tracer]: 201 → fully populated Tier3AgentInfo."""
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(201, json=_FULL_AGENT_RESP)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
info = await define_agent(
client,
agent_name="wizard",
system_prompt="You are a wizard.",
model="qwen3.6-35-a3b",
)
assert isinstance(info, Tier3AgentInfo)
assert info.agent_id == "ratatoskr:wizard"
assert info.user_id == "ratatoskr"
assert info.agent_name == "wizard"
assert info.model == "qwen3.6-35-a3b"
@respx.mock
async def test_request_body_shape(self) -> None:
"""request_body_shape [trace]: outbound JSON is exactly the three keys."""
import json as _json
route = respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(201, json=_FULL_AGENT_RESP)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await define_agent(
client,
agent_name="wizard",
system_prompt="You are a wizard.",
model="qwen3.6-35-a3b",
)
body = _json.loads(route.calls[0].request.content)
# INV-001: exactly these three keys — no layer fields, no metadata.
assert body == {
"agent_name": "wizard",
"system_prompt": "You are a wizard.",
"model": "qwen3.6-35-a3b",
}
@respx.mock
async def test_quota_exceeded(self) -> None:
"""quota_exceeded [error]: 429 + Retry-After → Tier3QuotaExceeded."""
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(
429,
headers={"Retry-After": "0"},
json={"detail": {"error_code": "agent_quota_exceeded"}},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(Tier3QuotaExceeded) as exc:
await define_agent(
client,
agent_name="overflow",
system_prompt="x",
model="m",
)
assert exc.value.retry_after == 0
@respx.mock
async def test_user_id_unsupported(self) -> None:
"""user_id_unsupported [error]: 403 + error_code → Tier3UserIdUnsupported."""
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(
403, json={"detail": {"error_code": "tier3_user_id_unsupported"}}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(Tier3UserIdUnsupported):
await define_agent(
client, agent_name="wizard", system_prompt="x", model="m"
)
@respx.mock
async def test_layer_deferred(self) -> None:
"""layer_deferred [error]: 422 + layer_deferred → Tier3LayerDeferred(field)."""
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(
422,
json={"detail": {"error_code": "layer_deferred", "field": "persona"}},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(Tier3LayerDeferred) as exc:
await define_agent(
client, agent_name="wizard", system_prompt="x", model="m"
)
assert exc.value.field == "persona"
@respx.mock
async def test_bad_slug_assert(self) -> None:
"""bad_slug_assert [adversarial]: agent_name with uppercase → AssertionError, no HTTP."""
route = respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(201, json=_FULL_AGENT_RESP)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await define_agent(
client, agent_name="Wizard", system_prompt="x", model="m"
)
assert route.call_count == 0
@respx.mock
async def test_short_slug_assert(self) -> None:
"""short_slug_assert [adversarial]: agent_name len < 3 → AssertionError."""
route = respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(201, json=_FULL_AGENT_RESP)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await define_agent(
client, agent_name="ab", system_prompt="x", model="m"
)
assert route.call_count == 0
@respx.mock
async def test_empty_prompt_assert(self) -> None:
"""empty_prompt_assert [adversarial]: empty system_prompt → AssertionError."""
route = respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(201, json=_FULL_AGENT_RESP)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await define_agent(
client, agent_name="wizard", system_prompt="", model="m"
)
assert route.call_count == 0
@respx.mock
async def test_other_5xx(self) -> None:
"""other_5xx [error]: 503 → SessionApiFailed(status=503)."""
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(503, content=b"upstream out")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await define_agent(
client, agent_name="wizard", system_prompt="x", model="m"
)
assert exc.value.status == 503
class TestPatchAgent:
@respx.mock
async def test_happy_patch_both_fields(self) -> None:
"""happy_patch_both_fields: both fields set → request body has both."""
import json as _json
updated = {
**_FULL_AGENT_RESP,
"system_prompt": "new prompt",
"model": "different-model",
}
route = respx.patch("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(200, json=updated)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
info = await patch_agent(
client,
"ratatoskr:wizard",
system_prompt="new prompt",
model="different-model",
)
body = _json.loads(route.calls[0].request.content)
assert body == {"system_prompt": "new prompt", "model": "different-model"}
assert info.system_prompt == "new prompt"
assert info.model == "different-model"
@respx.mock
async def test_happy_patch_single_field(self) -> None:
"""happy_patch_single_field: omit model → body has system_prompt only."""
import json as _json
updated = {**_FULL_AGENT_RESP, "system_prompt": "only this"}
route = respx.patch("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(200, json=updated)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await patch_agent(client, "ratatoskr:wizard", system_prompt="only this")
body = _json.loads(route.calls[0].request.content)
# INV-002: body omits the None-valued field entirely
assert body == {"system_prompt": "only this"}
@respx.mock
async def test_field_not_mutable(self) -> None:
"""field_not_mutable [error]: 422 + error_code → Tier3FieldNotMutable(field)."""
respx.patch("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(
422,
json={
"detail": {"error_code": "field_not_mutable", "field": "agent_name"}
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(Tier3FieldNotMutable) as exc:
await patch_agent(
client, "ratatoskr:wizard", system_prompt="x"
)
assert exc.value.field == "agent_name"
@respx.mock
async def test_404(self) -> None:
"""404 [error]: PATCH on non-existent agent → Tier3AgentNotFound."""
respx.patch("https://w.example/agents/ratatoskr:ghost").mock(
return_value=httpx.Response(404, content=b"")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(Tier3AgentNotFound) as exc:
await patch_agent(
client, "ratatoskr:ghost", system_prompt="x"
)
assert exc.value.agent_id == "ratatoskr:ghost"
@respx.mock
async def test_no_fields_assert(self) -> None:
"""no_fields_assert [adversarial]: both None → AssertionError, no HTTP."""
route = respx.patch("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(200, json=_FULL_AGENT_RESP)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await patch_agent(client, "ratatoskr:wizard")
assert route.call_count == 0
@respx.mock
async def test_non_tier3_id_assert(self) -> None:
"""non_tier3_id_assert [adversarial]: agent_id without `:` → AssertionError."""
route = respx.patch("https://w.example/agents/mimir").mock(
return_value=httpx.Response(200, json=_FULL_AGENT_RESP)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await patch_agent(client, "mimir", system_prompt="x")
assert route.call_count == 0
class TestDeleteAgent:
@respx.mock
async def test_happy_delete(self) -> None:
"""happy_delete [happy,tracer]: 204 → returns None."""
respx.delete("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(204)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
result = await delete_agent(client, "ratatoskr:wizard")
assert result is None
@respx.mock
async def test_404(self) -> None:
"""404 [error]: DELETE on non-existent agent → Tier3AgentNotFound."""
respx.delete("https://w.example/agents/ratatoskr:ghost").mock(
return_value=httpx.Response(404)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(Tier3AgentNotFound) as exc:
await delete_agent(client, "ratatoskr:ghost")
assert exc.value.agent_id == "ratatoskr:ghost"
@respx.mock
async def test_non_tier3_id_assert(self) -> None:
"""non_tier3_id_assert [adversarial]: agent_id without `:` → AssertionError."""
route = respx.delete("https://w.example/agents/mimir").mock(
return_value=httpx.Response(204)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AssertionError):
await delete_agent(client, "mimir")
assert route.call_count == 0
@respx.mock
async def test_other_5xx(self) -> None:
"""other_5xx [error]: 500 → SessionApiFailed."""
respx.delete("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(500, content=b"oops")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SessionApiFailed) as exc:
await delete_agent(client, "ratatoskr:wizard")
assert exc.value.status == 500
@pytest.fixture
def _isolated_local_agents(
tmp_path: "Path", monkeypatch: pytest.MonkeyPatch
) -> "Path":
"""Isolate the v0.8.0 local-tier-3 index from the operator's real file."""
path = tmp_path / "local_agents.json"
monkeypatch.setenv("RATATOSKR_LOCAL_AGENTS", str(path))
return path
class TestCli:
@respx.mock
def test_cli_define_happy(
self,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
_isolated_local_agents: "Path",
) -> None:
"""cli_define_happy [happy]: argv → 201 mock → stdout confirmation;
local index updated with the new entry (v0.8.0 hook).
"""
from ratatoskr.local_agents import load_local_agents
monkeypatch.setenv("WORLDTREE_API_URL", "https://w.example")
monkeypatch.setenv("WORLDTREE_API_KEY", "k")
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(201, json=_FULL_AGENT_RESP)
)
rc = main([
"define",
"--name", "wizard",
"--system-prompt", "You are a wizard.",
"--model", "qwen3.6-35-a3b",
])
out = capsys.readouterr()
assert rc == 0
assert out.out.strip() == "defined ratatoskr:wizard (qwen3.6-35-a3b)"
# v0.8.0: local index now has the new entry.
entries = load_local_agents()
assert len(entries) == 1
assert entries[0].agent_id == "ratatoskr:wizard"
assert entries[0].model == "qwen3.6-35-a3b"
@respx.mock
def test_cli_patch_happy(
self,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
_isolated_local_agents: "Path",
) -> None:
"""cli_patch_happy [happy]: argv → 200 mock → stdout confirmation;
local index refreshed with the post-patch state.
"""
from ratatoskr.local_agents import load_local_agents
monkeypatch.setenv("WORLDTREE_API_URL", "https://w.example")
monkeypatch.setenv("WORLDTREE_API_KEY", "k")
respx.patch("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(200, json=_FULL_AGENT_RESP)
)
rc = main(["patch", "ratatoskr:wizard", "--system-prompt", "new"])
out = capsys.readouterr()
assert rc == 0
assert out.out.strip() == "patched ratatoskr:wizard"
entries = load_local_agents()
assert len(entries) == 1
assert entries[0].agent_id == "ratatoskr:wizard"
@respx.mock
def test_cli_delete_happy(
self,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
_isolated_local_agents: "Path",
) -> None:
"""cli_delete_happy [happy]: argv → 204 mock → stdout confirmation;
local index entry removed (v0.8.0 hook).
"""
from ratatoskr.local_agents import (
LocalAgentEntry,
add_local_agent,
load_local_agents,
)
# Pre-populate so we can verify removal.
add_local_agent(LocalAgentEntry(
agent_id="ratatoskr:wizard",
agent_name="wizard",
model="m",
description="d",
defined_at="t",
))
monkeypatch.setenv("WORLDTREE_API_URL", "https://w.example")
monkeypatch.setenv("WORLDTREE_API_KEY", "k")
respx.delete("https://w.example/agents/ratatoskr:wizard").mock(
return_value=httpx.Response(204)
)
rc = main(["delete", "ratatoskr:wizard"])
out = capsys.readouterr()
assert rc == 0
assert out.out.strip() == "deleted ratatoskr:wizard"
assert load_local_agents() == []
def test_cli_missing_auth(
self, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
"""cli_missing_auth [error]: no api-key → stderr [auth_error] + exit 11."""
monkeypatch.delenv("WORLDTREE_API_KEY", raising=False)
rc = main([
"define",
"--name", "wizard",
"--system-prompt", "x",
"--model", "m",
])
err = capsys.readouterr().err
assert rc == 11
assert "[auth_error]" in err
@respx.mock
def test_cli_api_failed(
self, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
"""cli_api_failed [error]: 500 → stderr [api_failed] + exit 20."""
monkeypatch.setenv("WORLDTREE_API_URL", "https://w.example")
monkeypatch.setenv("WORLDTREE_API_KEY", "k")
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(500, content=b"upstream out")
)
rc = main([
"define",
"--name", "wizard",
"--system-prompt", "x",
"--model", "m",
])
err = capsys.readouterr().err
assert rc == 20
assert "[api_failed]" in err
@respx.mock
def test_cli_quota_exceeded(
self, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
"""cli_quota_exceeded [error]: 429 → stderr [quota_exceeded] + exit 20."""
monkeypatch.setenv("WORLDTREE_API_URL", "https://w.example")
monkeypatch.setenv("WORLDTREE_API_KEY", "k")
respx.post("https://w.example/agents/define").mock(
return_value=httpx.Response(
429,
headers={"Retry-After": "0"},
json={"detail": {"error_code": "agent_quota_exceeded"}},
)
)
rc = main([
"define",
"--name", "wizard",
"--system-prompt", "x",
"--model", "m",
])
err = capsys.readouterr().err
assert rc == 20
assert "[quota_exceeded]" in err
def test_cli_patch_no_fields(
self, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
"""cli_patch_no_fields [error]: patch with no flags → [usage_error] + exit 10."""
monkeypatch.setenv("WORLDTREE_API_URL", "https://w.example")
monkeypatch.setenv("WORLDTREE_API_KEY", "k")
rc = main(["patch", "ratatoskr:wizard"])
err = capsys.readouterr().err
assert rc == 10
assert "[usage_error]" in err