Something went wrong. Try again.
A declarative, hermetic harness for Nix-defined agents.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184"""BackgroundRegistry tests.
These launch real short-lived `sh` subprocesses directly (no bwrap), so theyexercise the registry's tracking, log tailing, signalling, and async monitor inisolation from the jail. The jailed launch path is covered by test_jail.py."""
import asyncioimport osimport subprocessimport threading
import pytest
from tartarus.background import BackgroundError, BackgroundRegistryfrom tartarus.jail import BackgroundHandle
def _launch(args: list[str], log_path: str) -> BackgroundHandle: log_file = open(log_path, "wb") proc = subprocess.Popen( args, stdout=log_file, stderr=subprocess.STDOUT, start_new_session=True ) log_file.close() return BackgroundHandle(proc=proc, pgid=os.getpgid(proc.pid), log_path=log_path)
class MemorySink: def __init__(self): self.records = []
def record(self, event) -> None: self.records.append(event)
def test_register_assigns_sequential_ids(tmp_path): registry = BackgroundRegistry() one = _launch(["sh", "-c", "true"], str(tmp_path / "1.log")) two = _launch(["sh", "-c", "true"], str(tmp_path / "2.log"))
assert registry.register("cap", one) == "bg-1" assert registry.register("cap", two) == "bg-2"
def test_status_reflects_exit_without_a_loop(tmp_path): # No event loop/monitor attached: status must refresh liveness itself. registry = BackgroundRegistry() handle = _launch(["sh", "-c", "sleep 0.2; exit 0"], str(tmp_path / "s.log")) registry.register("cap", handle)
assert "running" in registry.status("bg-1") handle.proc.wait(timeout=5) assert "exited with code 0" in registry.status("bg-1")
def test_output_reads_log_from_offset(tmp_path): registry = BackgroundRegistry() handle = _launch(["sh", "-c", "printf abcdef"], str(tmp_path / "o.log")) handle.proc.wait(timeout=5) registry.register("cap", handle)
assert registry.output("bg-1") == "abcdef" assert registry.output("bg-1", 3) == "def"
def test_output_truncates_logs_larger_than_the_tail_window(tmp_path): # A log far larger than the budget must come back as only its final window # (with the truncation marker), exercising the seek-to-tail read path. registry = BackgroundRegistry(output_truncate=10) payload = "HEAD" + "x" * 200 + "TAIL" handle = _launch(["sh", "-c", f"printf '%s' {payload}"], str(tmp_path / "big.log")) handle.proc.wait(timeout=5) registry.register("cap", handle)
output = registry.output("bg-1")
assert output.startswith("...(truncated)\n") assert output.endswith("xxxxxxTAIL") # the last output_truncate characters assert "HEAD" not in output
def test_stop_kills_a_running_task(tmp_path): registry = BackgroundRegistry() handle = _launch(["sh", "-c", "sleep 30"], str(tmp_path / "k.log")) registry.register("cap", handle)
assert "SIGTERM" in registry.stop("bg-1") handle.proc.wait(timeout=5) assert "exited" in registry.status("bg-1")
def test_shutdown_all_reaps_running_tasks(tmp_path): registry = BackgroundRegistry() handle = _launch(["sh", "-c", "sleep 30"], str(tmp_path / "r.log")) registry.register("cap", handle)
registry.shutdown_all()
handle.proc.wait(timeout=5) assert handle.proc.returncode is not None
def test_unknown_task_raises(): registry = BackgroundRegistry() with pytest.raises(BackgroundError, match="unknown background task"): registry.status("bg-404")
def test_completion_is_audited(tmp_path): sink = MemorySink() registry = BackgroundRegistry(audit=sink) handle = _launch(["sh", "-c", "printf hi; exit 2"], str(tmp_path / "a.log")) handle.proc.wait(timeout=5) registry.register("cap", handle)
registry.status("bg-1") # refresh → finalize → one completion record
assert len(sink.records) == 1 assert sink.records[0].tool_name == "cap" assert sink.records[0].exec_result.code == 2 # Idempotent: refreshing again does not double-record. registry.status("bg-1") assert len(sink.records) == 1
def test_monitor_enqueues_notice_on_exit(tmp_path): async def run(): notices: asyncio.Queue = asyncio.Queue() registry = BackgroundRegistry(notices=notices, loop=asyncio.get_running_loop()) handle = _launch(["sh", "-c", "printf hello; exit 3"], str(tmp_path / "n.log")) bg_id = registry.register("cap", handle) notice = await asyncio.wait_for(notices.get(), timeout=5) return bg_id, notice
bg_id, notice = asyncio.run(run())
assert bg_id == "bg-1" assert notice.task_id == "bg-1" assert notice.capability == "cap" assert notice.exit_code == 3 assert "hello" in notice.output_tail
def test_register_is_thread_safe(tmp_path): registry = BackgroundRegistry() handles = [ _launch(["sh", "-c", "true"], str(tmp_path / f"{i}.log")) for i in range(50) ] ids: list[str] = [] errors: list[Exception] = []
def register_one(handle): try: ids.append(registry.register("cap", handle)) except Exception as exc: # pragma: no cover - failures should fail the test errors.append(exc)
threads = [threading.Thread(target=register_one, args=(h,)) for h in handles] for thread in threads: thread.start() for thread in threads: thread.join()
assert not errors assert len(ids) == 50 assert len(set(ids)) == 50 assert sorted(ids, key=lambda value: int(value.split("-", 1)[1])) == [ f"bg-{i}" for i in range(1, 51) ]
def test_finalize_is_idempotent_under_lock(tmp_path): sink = MemorySink() registry = BackgroundRegistry(audit=sink) handle = _launch(["sh", "-c", "printf hi; exit 2"], str(tmp_path / "a.log")) handle.proc.wait(timeout=5) registry.register("cap", handle) task = registry._get("bg-1")
# The first call finalizes; the second must not record again or stop proxy twice. assert registry._finalize(task, 2) is None assert registry._finalize(task, 2) is None assert len(sink.records) == 1