Something went wrong. Try again.
A declarative, hermetic harness for Nix-defined agents.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230import asynciofrom typing import cast
import pytest
from tartarus.agent_loop import AgentLoopfrom tartarus.background import BackgroundRegistryfrom tartarus.broker import Brokerfrom tartarus.cli import ( SessionFlags, _bundle_manifest_source, _parse_agent_selector, _parse_session_flags, _persist_and_compact, _print_context_status, _run_one_shot,)from tartarus.config import ConfigErrorfrom tartarus.context import ContextLedger, ContextLimits, ContextManagerfrom tartarus.jail import JailBuilderfrom tartarus.policy import PolicyEnginefrom tartarus.provider.base import Providerfrom tartarus.session import SessionStorefrom tests.manifest_fixtures import echo_manifest
def test_parse_agent_selector_picks_named_agent(): agent_name, rest = _parse_agent_selector([".#research", "hello", "world"]) assert agent_name == "research" assert rest == ["hello", "world"]
def test_parse_agent_selector_accepts_uppercase_names(): agent_name, rest = _parse_agent_selector([".#readOnly"]) assert agent_name == "readOnly" assert rest == []
def test_parse_agent_selector_leaves_plain_prompt_unchanged(): agent_name, rest = _parse_agent_selector(["count", "lines", "in", "README.md"]) assert agent_name is None assert rest == ["count", "lines", "in", "README.md"]
def test_bundle_manifest_source_names_realized_bundle_path(): assert ( _bundle_manifest_source("/nix/store/abc-agent-bundle") == "/nix/store/abc-agent-bundle/manifest.json" )
def test_parse_agent_selector_rejects_empty_name(): with pytest.raises(ConfigError, match="missing a name"): _parse_agent_selector([".#"])
def test_parse_session_flags_extracts_continue_and_leaves_prompt(): flags, rest = _parse_session_flags(["--continue", "what", "did", "I", "ask"]) assert flags.continue_latest is True assert rest == ["what", "did", "I", "ask"]
def test_parse_session_flags_extracts_resume_id(): flags, rest = _parse_session_flags([".#default", "--resume", "20260627-1200", "go"]) assert flags.resume == "20260627-1200" # The selector and prompt survive untouched for downstream parsing. assert rest == [".#default", "go"]
def test_parse_session_flags_accepts_resume_equals_form(): flags, rest = _parse_session_flags(["--resume=abc123"]) assert flags.resume == "abc123" assert rest == []
def test_parse_session_flags_no_session_and_list(): flags, _ = _parse_session_flags(["--no-session"]) assert flags.disabled is True flags, _ = _parse_session_flags(["--list-sessions"]) assert flags.list_sessions is True
def test_parse_session_flags_context_commands(): flags, _ = _parse_session_flags(["--context-status"]) assert flags.context_status is True flags, _ = _parse_session_flags(["--compact-context"]) assert flags.compact_context is True
def test_parse_session_flags_resume_without_id_errors(): with pytest.raises(ConfigError, match="--resume requires"): _parse_session_flags(["--resume"])
def test_parse_session_flags_default_is_inert(): flags, rest = _parse_session_flags(["just", "a", "prompt"]) assert flags == SessionFlags() assert rest == ["just", "a", "prompt"]
def test_print_session_list_empty_dir(tmp_path, capsys): from tartarus.cli import _print_session_list
_print_session_list(str(tmp_path)) captured = capsys.readouterr() assert "no sessions" in captured.err assert captured.out == ""
def test_print_session_list_ignores_unreadable_session(tmp_path, capsys): from tartarus.cli import _print_session_list from tartarus.session import SessionStore
SessionStore(str(tmp_path), "good").append([{"role": "user", "content": "keep me"}]) bad = SessionStore(str(tmp_path), "bad") bad.append([{"role": "user", "content": "x"}]) bad_path = tmp_path / "bad.jsonl" bad_path.write_text("not valid json\n")
_print_session_list(str(tmp_path)) captured = capsys.readouterr() assert "good" in captured.out assert "keep me" in captured.out assert "warning: could not preview session bad" in captured.err
def test_print_session_list_reports_unreadable_dir(tmp_path, capsys): from tartarus.cli import _print_session_list
session_dir = tmp_path / "sessions" session_dir.write_text("not a directory") _print_session_list(str(session_dir)) captured = capsys.readouterr() assert "warning: could not list sessions" in captured.err
def test_print_context_status_uses_latest_session_without_api_key( tmp_path, monkeypatch, capsys): from tartarus.session import SessionStore
monkeypatch.setenv("TARTARUS_WORK_TREE", str(tmp_path)) session_dir = tmp_path / ".tartarus" / "sessions" SessionStore(str(session_dir), "s1").append( [ {"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}, ] )
result = _print_context_status(SessionFlags())
captured = capsys.readouterr() assert result == 0 assert "session: s1" in captured.out assert "messages: 2" in captured.out assert "ledger events: 0" in captured.out
def test_persist_and_compact_writes_summary_after_transcript_events(tmp_path): store = SessionStore(str(tmp_path / "sessions"), "s1") ledger = ContextLedger(str(tmp_path / "context"), "s1") manager = ContextManager( ledger, ContextLimits(max_chars=500, recent_turns=1), auto_compact=True, ) loop = AgentLoop( provider=cast(Provider, None), broker=Broker(echo_manifest(), cast(JailBuilder, None), PolicyEngine()), manifest=echo_manifest(), system_prompt="test", context_manager=manager, ) messages = [] for index in range(5): messages.append({"role": "user", "content": f"question {index} " + "x" * 200}) messages.append( {"role": "assistant", "content": f"answer {index} " + "y" * 200} )
_persist_and_compact(loop, store, ledger, messages)
events = ledger.load_events() assert events[-1]["type"] == "context_summary" assert [event["message_index"] for event in events[:-1]] == list(range(10)) assert events[-1]["covered"] == {"start": 0, "end": 8}
def test_run_one_shot_returns_one_when_background_reaction_fails(monkeypatch): """A provider-level failure while reacting to a completion yields exit code 1."""
state = {"running": True}
class FakeRegistry: @property def has_running(self) -> bool: return state["running"]
async def fake_send(_loop, _messages, _text): return True
async def fake_drain(_loop, _messages, _store, _ledger, _notices): state["running"] = False return False
monkeypatch.setattr("tartarus.cli._send", fake_send) monkeypatch.setattr("tartarus.cli._drain_notice", fake_drain)
loop = AgentLoop( provider=cast(Provider, None), broker=Broker(echo_manifest(), cast(JailBuilder, None), PolicyEngine()), manifest=echo_manifest(), system_prompt="test", )
result = asyncio.run( _run_one_shot( loop, "prompt", [], None, None, cast(BackgroundRegistry, FakeRegistry()), asyncio.Queue(), ) )
assert result == 1