Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""Tests for the cortex_client module."""
import jsonimport threadingimport time
import pytest
from solstone.think.cortex_client import ( CortexNotClaimed, CortexSpawnUnavailable, cortex_request, cortex_uses, get_use_end_state, get_use_log_status, read_use_provider_model, read_use_provider_model_reason, wait_for_uses,)from solstone.think.models import GPT_5from solstone.think.utils import now_ms
@pytest.fixturedef sent_callosum_messages(tmp_path, monkeypatch): """Capture Cortex sends without exercising Callosum transport.""" messages = []
def capture(tract, event, **fields): messages.append({"tract": tract, "event": event, **fields}) return ""
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr( "solstone.think.cortex_client.callosum_send_classified", capture ) monkeypatch.setattr( "solstone.think.cortex_client.get_use_log_status", lambda _uid: "running" ) return messages
@pytest.fixturedef fake_callosum_listener(monkeypatch): """Capture wait_for_uses callbacks without opening a UNIX socket.""" callbacks = [] started = threading.Event()
class FakeCallosumConnection: def start(self, callback=None): callbacks.append(callback) started.set()
def stop(self): pass
monkeypatch.setattr( "solstone.think.cortex_client.CallosumConnection", FakeCallosumConnection ) return callbacks, started
def test_cortex_request_broadcasts_to_callosum(sent_callosum_messages): """Test that cortex_request broadcasts request to Callosum.""" messages = sent_callosum_messages
# Create a request use_id = cortex_request( prompt="Test prompt", name="chat", config={"facet": "work"}, )
# Verify broadcast was received assert len(messages) == 1 msg = messages[0] assert msg["tract"] == "cortex" assert msg["event"] == "request" assert msg["prompt"] == "Test prompt" assert msg["name"] == "chat" assert msg["facet"] == "work" assert "provider" not in msg assert "model" not in msg assert msg["use_id"] == use_id assert "ts" in msg
def test_cortex_request_returns_agent_id(sent_callosum_messages): """Test that cortex_request returns use_id string.""" use_id = cortex_request(prompt="Test", name="chat")
# Verify use_id is a string timestamp assert isinstance(use_id, str) assert use_id.isdigit() assert len(use_id) == 13 # Millisecond timestamp
def test_cortex_request_uses_explicit_use_id(sent_callosum_messages): messages = sent_callosum_messages
use_id = cortex_request( prompt="Test prompt", name="chat", use_id="1713629000000", )
assert use_id == "1713629000000" assert messages[-1]["use_id"] == "1713629000000"
def test_cortex_request_unique_agent_ids(sent_callosum_messages): """Test that cortex_request generates unique agent IDs.""" agent_ids = [] for i in range(3): use_id = cortex_request(prompt=f"Test {i}", name="chat") agent_ids.append(use_id)
# All agent IDs should be unique assert len(set(agent_ids)) == 3
def test_cortex_request_raises_when_callosum_unavailable(tmp_path, monkeypatch): """Test cortex_request classifies Callosum send failures.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr( "solstone.think.cortex_client.callosum_send_classified", lambda *a, **kw: "FileNotFoundError", )
with pytest.raises(CortexSpawnUnavailable) as excinfo: cortex_request(prompt="Test", name="chat")
assert excinfo.value.detail == "FileNotFoundError"
def test_cortex_request_empty_journal(tmp_path, monkeypatch): """Test cortex_request works with an empty journal directory.""" monkeypatch.setattr( "solstone.think.cortex_client.callosum_send_classified", lambda *a, **kw: "" ) monkeypatch.setattr( "solstone.think.cortex_client.get_use_log_status", lambda uid: "running" ) monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
use_id = cortex_request("test", "chat") assert use_id is not None assert len(use_id) > 0
def _install_fake_claim_clock(monkeypatch, poll_interval=0.01, windows=(0.02,) * 3): """Fake the claim loop's clock so schedules run without real sleeps.
Returns the cortex_client module and the mutable fake-monotonic clock, so tests can read elapsed time and timestamp individual sends. """ import solstone.think.cortex_client as cc
now = {"value": 0.0} monkeypatch.setattr(cc.time, "monotonic", lambda: now["value"])
def fake_sleep(seconds): now["value"] += seconds
monkeypatch.setattr(cc.time, "sleep", fake_sleep) monkeypatch.setattr(cc, "_CLAIM_POLL_INTERVAL_S", poll_interval) monkeypatch.setattr(cc, "_DEFAULT_CLAIM_WINDOWS", tuple(windows)) return cc, now
def test_cortex_request_returns_when_claim_appears_after_poll(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) cc, _ = _install_fake_claim_clock(monkeypatch) monkeypatch.setattr(cc, "callosum_send_classified", lambda *a, **kw: "") statuses = iter(["not_found", "running"])
def fake_status(use_id): return next(statuses, "running")
monkeypatch.setattr(cc, "get_use_log_status", fake_status)
use_id = cortex_request("test", "chat", use_id="1713629000001")
assert use_id == "1713629000001"
def test_cortex_request_rebroadcast_reuses_same_use_id(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) cc, _ = _install_fake_claim_clock(monkeypatch) send_calls = []
def fake_send(*args, **kwargs): send_calls.append(kwargs) return ""
monkeypatch.setattr(cc, "callosum_send_classified", fake_send) monkeypatch.setattr( cc, "get_use_log_status", lambda use_id: "running" if len(send_calls) >= 2 else "not_found", )
use_id = cortex_request("test", "chat", use_id="1713629000002")
assert use_id == "1713629000002" assert len(send_calls) >= 2 assert {call["use_id"] for call in send_calls} == {"1713629000002"}
def test_cortex_request_raises_when_not_claimed(tmp_path, monkeypatch): """Default schedule fast-fails: three sends, one window each, then raises.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) cc, clock = _install_fake_claim_clock(monkeypatch) send_calls = []
def fake_send(*args, **kwargs): send_calls.append(kwargs) return ""
monkeypatch.setattr(cc, "callosum_send_classified", fake_send) monkeypatch.setattr(cc, "get_use_log_status", lambda use_id: "not_found")
with pytest.raises(CortexNotClaimed) as excinfo: cortex_request("test", "chat", use_id="1713629000003")
assert excinfo.value.use_id == "1713629000003" assert len(send_calls) == 3 assert "after 3 broadcasts" in excinfo.value.detail assert clock["value"] == pytest.approx(sum(cc._DEFAULT_CLAIM_WINDOWS), abs=1e-6)
def test_cortex_request_patient_windows_are_non_decreasing(tmp_path, monkeypatch): """An explicit schedule drives send count, gaps, and total budget.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) cc, clock = _install_fake_claim_clock(monkeypatch, poll_interval=0.001) windows = (0.01, 0.02, 0.04, 0.08, 0.15) send_times = []
def fake_send(*args, **kwargs): send_times.append(clock["value"]) return ""
monkeypatch.setattr(cc, "callosum_send_classified", fake_send) monkeypatch.setattr(cc, "get_use_log_status", lambda use_id: "not_found")
with pytest.raises(CortexNotClaimed) as excinfo: cortex_request("test", "chat", use_id="1713629000005", claim_windows=windows)
assert len(send_times) == len(windows) assert "after 5 broadcasts" in excinfo.value.detail
gaps = [b - a for a, b in zip(send_times, send_times[1:])] assert gaps == pytest.approx(list(windows[:-1]), abs=1e-6) assert gaps == sorted(gaps), "rebroadcast windows must be non-decreasing" assert clock["value"] == pytest.approx(sum(windows), abs=1e-6)
def test_cortex_request_raises_spawn_unavailable_on_failed_rebroadcast( tmp_path, monkeypatch): """A send failure on a rebroadcast is CortexSpawnUnavailable, not NotClaimed.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) cc, _ = _install_fake_claim_clock(monkeypatch) send_calls = []
def fake_send(*args, **kwargs): send_calls.append(kwargs) return "FileNotFoundError" if len(send_calls) >= 2 else ""
monkeypatch.setattr(cc, "callosum_send_classified", fake_send) monkeypatch.setattr(cc, "get_use_log_status", lambda use_id: "not_found")
with pytest.raises(CortexSpawnUnavailable) as excinfo: cortex_request("test", "chat", use_id="1713629000006")
assert excinfo.value.detail == "FileNotFoundError" assert len(send_calls) == 2
def test_dispatch_cortex_request_patient_schedule_claims_after_default_would_fail( tmp_path, monkeypatch): """The orchestrator survives a claim that lands past the fast-fail budget.""" import solstone.think.thinking as think
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) default_windows = (0.01, 0.01, 0.01) # budget 0.03 patient_windows = (0.01, 0.02, 0.04, 0.08, 0.15) # budget 0.30 cc, clock = _install_fake_claim_clock( monkeypatch, poll_interval=0.001, windows=default_windows ) monkeypatch.setattr(think, "PATIENT_CLAIM_WINDOWS", patient_windows) monkeypatch.setattr(cc, "callosum_send_classified", lambda *a, **kw: "")
# Claim lands after the default budget expires but well inside the patient one. claim_at = 0.05 monkeypatch.setattr( cc, "get_use_log_status", lambda use_id: "running" if clock["value"] >= claim_at else "not_found", )
with pytest.raises(CortexNotClaimed): cortex_request("test", "chat", use_id="1713629000007")
clock["value"] = 0.0 result = think._dispatch_cortex_request( prompt="test", name="chat", use_id="1713629000008" )
assert result == "1713629000008"
def test_cortex_request_immediate_claim_does_not_sleep(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) import solstone.think.cortex_client as cc
monkeypatch.setattr(cc, "callosum_send_classified", lambda *a, **kw: "") monkeypatch.setattr(cc, "get_use_log_status", lambda use_id: "running")
def fail_sleep(seconds): raise AssertionError("sleep should not be called")
monkeypatch.setattr(cc.time, "sleep", fail_sleep)
use_id = cortex_request("test", "chat", use_id="1713629000004")
assert use_id == "1713629000004"
# Tests for cortex_uses remain mostly unchanged as they read from files
def test_cortex_agents_empty(tmp_path, monkeypatch): """Test cortex_uses with no agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
result = cortex_uses()
assert result["uses"] == [] assert result["pagination"]["total"] == 0 assert result["pagination"]["has_more"] is False assert result["live_count"] == 0 assert result["historical_count"] == 0
def test_cortex_agents_with_active(tmp_path, monkeypatch): """Test cortex_uses with active (running) agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir()
# Create active agent files ts1 = now_ms() ts2 = ts1 + 1000
unified_dir = talents_dir / "chat" tester_dir = talents_dir / "tester" unified_dir.mkdir() tester_dir.mkdir()
active_file1 = unified_dir / f"{ts1}_active.jsonl" with open(active_file1, "w") as f: json.dump( { "event": "request", "ts": ts1, "prompt": "Task 1", "name": "chat", "provider": "openai", }, f, ) f.write("\n")
active_file2 = tester_dir / f"{ts2}_active.jsonl" with open(active_file2, "w") as f: json.dump( { "event": "request", "ts": ts2, "prompt": "Task 2", "name": "tester", "provider": "google", }, f, ) f.write("\n")
result = cortex_uses()
assert len(result["uses"]) == 2 assert result["live_count"] == 2 assert result["historical_count"] == 0
def test_cortex_agents_with_completed(tmp_path, monkeypatch): """Test cortex_uses with completed (historical) agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir()
# Create completed agent files ts1 = now_ms() reviewer_dir = talents_dir / "reviewer" reviewer_dir.mkdir()
completed_file1 = reviewer_dir / f"{ts1}.jsonl" with open(completed_file1, "w") as f: json.dump( { "event": "request", "ts": ts1, "prompt": "Old task", "name": "reviewer", "provider": "anthropic", }, f, ) f.write("\n") json.dump({"event": "finish", "ts": ts1 + 100, "result": "Done"}, f) f.write("\n")
result = cortex_uses()
assert len(result["uses"]) == 1 assert result["live_count"] == 0 assert result["historical_count"] == 1 assert result["uses"][0]["status"] == "completed"
def test_cortex_agents_pagination(tmp_path, monkeypatch): """Test cortex_uses pagination.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir()
# Create multiple agents base_ts = now_ms() unified_dir = talents_dir / "chat" unified_dir.mkdir() for i in range(5): ts = base_ts + (i * 1000) file = unified_dir / f"{ts}.jsonl" with open(file, "w") as f: json.dump( { "event": "request", "ts": ts, "prompt": f"Task {i}", "name": "chat", }, f, ) f.write("\n")
# Test limit result = cortex_uses(limit=2) assert len(result["uses"]) == 2 assert result["pagination"]["limit"] == 2 assert result["pagination"]["total"] == 5 assert result["pagination"]["has_more"] is True
def test_cortex_agents_empty_journal(tmp_path, monkeypatch): """Test cortex_uses works with an empty journal directory.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
result = cortex_uses() assert "uses" in result assert "pagination" in result assert isinstance(result["uses"], list)
def test_get_agent_log_status_completed(tmp_path, monkeypatch): """Test get_use_log_status returns 'completed' for finished agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}.jsonl").write_text('{"event": "finish"}\n')
assert get_use_log_status(use_id) == "completed"
def test_get_agent_log_status_running(tmp_path, monkeypatch): """Test get_use_log_status returns 'running' for active agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}_active.jsonl").write_text('{"event": "start"}\n')
assert get_use_log_status(use_id) == "running"
def test_read_use_provider_model_reads_active_log(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" / "chat" talents_dir.mkdir(parents=True)
use_id = "1234567890123" (talents_dir / f"{use_id}_active.jsonl").write_text( json.dumps({"event": "request", "provider": "openai", "model": "wrong"}) + "\n" + json.dumps( { "event": "start", "provider": "anthropic", "model": "claude-opus-4-1", } ) + "\n" + json.dumps({"event": "error", "reason_code": "provider_key_missing"}) + "\n", encoding="utf-8", )
assert read_use_provider_model(use_id) == ("anthropic", "claude-opus-4-1") assert read_use_provider_model_reason(use_id) == ( "anthropic", "claude-opus-4-1", "provider_key_missing", )
def test_get_agent_log_status_not_found(tmp_path, monkeypatch): """Test get_use_log_status returns 'not_found' for missing agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) (tmp_path / "talents").mkdir()
assert get_use_log_status("nonexistent") == "not_found"
def test_get_agent_log_status_prefers_completed(tmp_path, monkeypatch): """Test get_use_log_status returns 'completed' when both files exist.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
# Edge case: both files exist (shouldn't happen, but check precedence) use_id = "1234567890123" (unified_dir / f"{use_id}.jsonl").write_text('{"event": "finish"}\n') (unified_dir / f"{use_id}_active.jsonl").write_text('{"event": "start"}\n')
assert get_use_log_status(use_id) == "completed"
def test_get_agent_end_state_finish(tmp_path, monkeypatch): """Test get_use_end_state returns 'finish' for successful agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}.jsonl").write_text( '{"event": "request", "prompt": "hello"}\n' '{"event": "finish", "result": "done"}\n' )
assert get_use_end_state(use_id) == "finish"
def test_get_agent_end_state_error(tmp_path, monkeypatch): """Test get_use_end_state returns 'error' for failed agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}.jsonl").write_text( '{"event": "request", "prompt": "hello"}\n' '{"event": "error", "error": "something went wrong"}\n' )
assert get_use_end_state(use_id) == "error"
def test_get_agent_end_state_no_output_maps_to_error(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}.jsonl").write_text( json.dumps({"event": "request", "prompt": "hello"}) + "\n" + json.dumps( { "event": "error", "error": "no_output: expects-final cogitate run finished without " "emitting a final result", "reason_code": "no_output", "terminal": True, } ) + "\n" )
assert get_use_end_state(use_id) == "error"
@pytest.mark.parametrize( "reason_code", ["token_budget_exceeded", "max_turns_exhausted", "agent_stuck"],)def test_force_finish_error_surfaces_budget_reason_code( tmp_path, monkeypatch, reason_code,): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}.jsonl").write_text( json.dumps({"event": "request", "prompt": "hello"}) + "\n" + json.dumps( { "event": "start", "provider": "openai", "model": GPT_5, } ) + "\n" + json.dumps( { "event": "error", "error": f"{reason_code}: force-finished", "reason_code": reason_code, "terminal": True, } ) + "\n" )
assert get_use_end_state(use_id) == "error" assert read_use_provider_model_reason(use_id) == ("openai", GPT_5, reason_code)
def test_get_agent_end_state_finish_after_nonterminal_error(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}.jsonl").write_text( json.dumps({"event": "request", "prompt": "hello"}) + "\n" + json.dumps( { "event": "start", "provider": "openai", "model": GPT_5, } ) + "\n" + json.dumps( { "event": "error", "error": "recoverable agent error", "terminal": False, } ) + "\n" + json.dumps( { "event": "finish", "result": "done", "usage": {"total_tokens": 10}, } ) + "\n" )
assert get_use_end_state(use_id) == "finish"
def test_get_agent_end_state_running(tmp_path, monkeypatch): """Test get_use_end_state returns 'running' for active agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir()
use_id = "1234567890123" (unified_dir / f"{use_id}_active.jsonl").write_text( '{"event": "request", "prompt": "hello"}\n' )
assert get_use_end_state(use_id) == "running"
def test_get_agent_end_state_unknown(tmp_path, monkeypatch): """Test get_use_end_state returns 'unknown' for missing agents.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) (tmp_path / "talents").mkdir()
assert get_use_end_state("nonexistent") == "unknown"
# Tests for wait_for_uses
def test_wait_for_agents_already_complete(tmp_path, monkeypatch): """Test wait_for_uses returns immediately if agents already completed.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
# Create completed agents agent_ids = ["1000", "2000"] for use_id in agent_ids: (unified_dir / f"{use_id}.jsonl").write_text('{"event": "finish"}\n')
completed, timed_out = wait_for_uses(agent_ids, timeout=1)
assert set(completed.keys()) == set(agent_ids) assert all(v == "finish" for v in completed.values()) assert timed_out == []
def test_wait_for_agents_event_completion( tmp_path, monkeypatch, fake_callosum_listener): """Test wait_for_uses completes when finish event is received.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" unified_dir = talents_dir / "chat" unified_dir.mkdir(parents=True) (tmp_path / "health").mkdir()
use_id = "1234567890123"
# Start wait in background thread result = {"completed": None, "timed_out": None}
def wait_thread(): result["completed"], result["timed_out"] = wait_for_uses([use_id], timeout=5)
waiter = threading.Thread(target=wait_thread) waiter.start()
callbacks, listener_started = fake_callosum_listener assert listener_started.wait(0.5)
# Create the completed file and emit finish event (unified_dir / f"{use_id}.jsonl").write_text('{"event": "finish"}\n')
callbacks[0]({"tract": "cortex", "event": "finish", "use_id": use_id})
waiter.join(timeout=3)
assert result["completed"] == {use_id: "finish"} assert result["timed_out"] == []
def test_wait_for_agents_error_event(tmp_path, monkeypatch, fake_callosum_listener): """Test wait_for_uses completes on error event too.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" unified_dir = talents_dir / "chat" unified_dir.mkdir(parents=True) (tmp_path / "health").mkdir()
use_id = "1234567890124"
result = {"completed": None, "timed_out": None}
def wait_thread(): result["completed"], result["timed_out"] = wait_for_uses([use_id], timeout=5)
waiter = threading.Thread(target=wait_thread) waiter.start() callbacks, listener_started = fake_callosum_listener assert listener_started.wait(0.5)
# Create completed file and emit error event (unified_dir / f"{use_id}.jsonl").write_text('{"event": "error"}\n')
callbacks[0]({"tract": "cortex", "event": "error", "use_id": use_id})
waiter.join(timeout=3)
assert result["completed"] == {use_id: "error"} assert result["timed_out"] == []
def test_wait_for_agents_initial_file_check(tmp_path, monkeypatch): """Test wait_for_uses finds already-completed agents via initial file check.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890125"
# Agent already completed before we start waiting (unified_dir / f"{use_id}.jsonl").write_text('{"event": "finish"}\n')
completed, timed_out = wait_for_uses([use_id], timeout=1)
# Should find via initial file check assert completed == {use_id: "finish"} assert timed_out == []
def test_wait_for_agents_timeout(tmp_path, monkeypatch, fake_callosum_listener): """Test wait_for_uses times out for agents that never complete.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890126" # Create active file (not completed) (unified_dir / f"{use_id}_active.jsonl").write_text('{"event": "start"}\n')
completed, timed_out = wait_for_uses([use_id], timeout=0)
assert completed == {} assert timed_out == [use_id]
def test_wait_for_agents_partial(tmp_path, monkeypatch, fake_callosum_listener): """Test wait_for_uses with some completing and some timing out.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" unified_dir = talents_dir / "chat" unified_dir.mkdir(parents=True) (tmp_path / "health").mkdir()
completing_agent = "1111" timeout_agent = "2222"
# Create active file for timeout agent (unified_dir / f"{timeout_agent}_active.jsonl").write_text('{"event": "start"}\n')
result = {"completed": None, "timed_out": None}
def wait_thread(): result["completed"], result["timed_out"] = wait_for_uses( [completing_agent, timeout_agent], timeout=0.1 )
waiter = threading.Thread(target=wait_thread) waiter.start() callbacks, listener_started = fake_callosum_listener assert listener_started.wait(0.5)
# Complete one agent (unified_dir / f"{completing_agent}.jsonl").write_text('{"event": "finish"}\n')
callbacks[0]({"tract": "cortex", "event": "finish", "use_id": completing_agent})
waiter.join(timeout=5)
assert result["completed"] == {completing_agent: "finish"} assert result["timed_out"] == [timeout_agent]
def test_wait_for_agents_missed_event_recovery(tmp_path, monkeypatch, caplog): """Test that missed events are recovered via final file check with INFO log.""" import logging
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890127"
# Start with active file (unified_dir / f"{use_id}_active.jsonl").write_text('{"event": "start"}\n')
result = {"completed": None, "timed_out": None}
def wait_and_complete(): # Wait a bit then "complete" the agent by renaming file time.sleep(0.3) (unified_dir / f"{use_id}.jsonl").write_text('{"event": "finish"}\n') (unified_dir / f"{use_id}_active.jsonl").unlink()
completer = threading.Thread(target=wait_and_complete) completer.start()
with caplog.at_level(logging.INFO): result["completed"], result["timed_out"] = wait_for_uses([use_id], timeout=1)
completer.join()
# Should recover via final file check assert result["completed"] == {use_id: "finish"} assert result["timed_out"] == []
# Should log about missed event assert any( "completion event not received but use completed" in record.message for record in caplog.records )
def test_wait_for_uses_renamed_terminal_via_polling(tmp_path, monkeypatch): """Test polling recovers a use renamed terminal without Callosum.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr("solstone.think.cortex_client._POLL_INTERVAL_S", 0.02) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890128" active_file = unified_dir / f"{use_id}_active.jsonl" final_file = unified_dir / f"{use_id}.jsonl" active_file.write_text('{"event": "start"}\n')
def complete_use(): time.sleep(0.05) final_file.write_text('{"event": "finish"}\n') active_file.unlink()
completer = threading.Thread(target=complete_use, daemon=True) completer.start()
started_at = time.monotonic() completed, timed_out = wait_for_uses([use_id], timeout=5) elapsed = time.monotonic() - started_at completer.join(timeout=5)
assert completed == {use_id: "finish"} assert timed_out == [] assert elapsed < 2.0
def test_wait_for_uses_terminal_before_rename_via_polling(tmp_path, monkeypatch): """Test polling recovers terminal events appended to the active log.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr("solstone.think.cortex_client._POLL_INTERVAL_S", 0.02) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890129" active_file = unified_dir / f"{use_id}_active.jsonl" active_file.write_text('{"event": "start"}\n')
result = {"completed": None, "timed_out": None}
def wait_thread(): result["completed"], result["timed_out"] = wait_for_uses([use_id], timeout=5)
waiter = threading.Thread(target=wait_thread, daemon=True) waiter.start() time.sleep(0.05)
with open(active_file, "a") as f: f.write('{"event": "finish"}\n')
waiter.join(timeout=5)
assert not waiter.is_alive() assert result["completed"] == {use_id: "finish"} assert result["timed_out"] == []
def test_wait_for_uses_timeout_none_returns_on_terminal(tmp_path, monkeypatch): """Test timeout=None returns when polling sees a terminal log.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr("solstone.think.cortex_client._POLL_INTERVAL_S", 0.02) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890130" active_file = unified_dir / f"{use_id}_active.jsonl" final_file = unified_dir / f"{use_id}.jsonl" active_file.write_text('{"event": "start"}\n')
result = {"completed": None, "timed_out": None}
def wait_thread(): result["completed"], result["timed_out"] = wait_for_uses([use_id], timeout=None)
waiter = threading.Thread(target=wait_thread, daemon=True) waiter.start() time.sleep(0.05)
final_file.write_text('{"event": "finish"}\n') active_file.unlink()
waiter.join(timeout=5)
assert not waiter.is_alive() assert result["completed"] == {use_id: "finish"} assert result["timed_out"] == []
def test_wait_for_uses_timeout_none_no_deadline_and_no_busy_spin( tmp_path, monkeypatch,): """Test timeout=None polls at the bounded interval until terminal.""" import solstone.think.cortex_client as cc
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr(cc, "_POLL_INTERVAL_S", 0.05) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890131" active_file = unified_dir / f"{use_id}_active.jsonl" final_file = unified_dir / f"{use_id}.jsonl" active_file.write_text('{"event": "start"}\n')
real = cc.get_use_end_state calls = []
def spy(uid): calls.append(uid) return real(uid)
monkeypatch.setattr(cc, "get_use_end_state", spy)
result = {"completed": None, "timed_out": None}
def wait_thread(): result["completed"], result["timed_out"] = wait_for_uses([use_id], timeout=None)
waiter = threading.Thread(target=wait_thread, daemon=True) waiter.start() time.sleep(0.3)
calls_before_completion = len(calls) assert calls_before_completion <= 20
final_file.write_text('{"event": "finish"}\n') active_file.unlink() waiter.join(timeout=5)
assert not waiter.is_alive() assert result["completed"] == {use_id: "finish"} assert result["timed_out"] == []
def test_wait_for_uses_unreadable_log_returns_unknown(tmp_path, monkeypatch): """Test get_use_end_state tolerates unreadable logs.""" import solstone.think.cortex_client as cc
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890132" (unified_dir / f"{use_id}_active.jsonl").write_text('{"event": "start"}\n')
def unreadable(_uid): raise PermissionError("boom")
monkeypatch.setattr(cc, "read_use_events", unreadable)
assert get_use_end_state(use_id) == "unknown"
def test_wait_for_uses_unreadable_sibling_does_not_block_terminal_sibling( tmp_path, monkeypatch,): """Test an unreadable pending use does not block a terminal sibling.""" import solstone.think.cortex_client as cc
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr(cc, "_POLL_INTERVAL_S", 0.02) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
terminal_use = "1234567890133" unreadable_use = "1234567890134" (unified_dir / f"{terminal_use}_active.jsonl").write_text('{"event": "start"}\n') (unified_dir / f"{unreadable_use}_active.jsonl").write_text('{"event": "start"}\n') (unified_dir / f"{terminal_use}.jsonl").write_text('{"event": "finish"}\n')
real = cc.read_use_events
def spy(uid): if uid == unreadable_use: raise OSError("unreadable") return real(uid)
monkeypatch.setattr(cc, "read_use_events", spy)
completed, timed_out = wait_for_uses([terminal_use, unreadable_use], timeout=0)
assert completed == {terminal_use: "finish"} assert timed_out == [unreadable_use]
def test_wait_for_uses_transient_read_error_then_recovery(tmp_path, monkeypatch): """Test a transient read error stays pending and recovers later.""" import solstone.think.cortex_client as cc
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr(cc, "_POLL_INTERVAL_S", 0.02) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890135" (unified_dir / f"{use_id}_active.jsonl").write_text('{"event": "start"}\n')
n = {"c": 0}
def spy(_uid): n["c"] += 1 if n["c"] == 1: raise OSError("race") return [{"event": "start"}, {"event": "finish"}]
monkeypatch.setattr(cc, "read_use_events", spy)
completed, timed_out = wait_for_uses([use_id], timeout=2)
assert completed == {use_id: "finish"} assert timed_out == []
def test_wait_for_uses_malformed_line_before_terminal(tmp_path, monkeypatch): """Test polling skips malformed JSON and still recovers terminal events.""" monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr("solstone.think.cortex_client._POLL_INTERVAL_S", 0.02) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890136" active_file = unified_dir / f"{use_id}_active.jsonl" active_file.write_text('{"event": "start"}\n')
result = {"completed": None, "timed_out": None}
def wait_thread(): result["completed"], result["timed_out"] = wait_for_uses([use_id], timeout=5)
waiter = threading.Thread(target=wait_thread, daemon=True) waiter.start() time.sleep(0.05)
with open(active_file, "a") as f: f.write("not valid json\n") f.write('{"event": "finish"}\n')
waiter.join(timeout=5)
assert not waiter.is_alive() assert result["completed"] == {use_id: "finish"} assert result["timed_out"] == []
def test_wait_for_uses_recovery_logs_exactly_once(tmp_path, monkeypatch, caplog): """Test missed-event recovery logs once, not once per poll.""" import logging
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) monkeypatch.setattr("solstone.think.cortex_client._POLL_INTERVAL_S", 0.05) talents_dir = tmp_path / "talents" talents_dir.mkdir() unified_dir = talents_dir / "chat" unified_dir.mkdir() (tmp_path / "health").mkdir()
use_id = "1234567890137" active_file = unified_dir / f"{use_id}_active.jsonl" final_file = unified_dir / f"{use_id}.jsonl" active_file.write_text('{"event": "start"}\n')
def complete_use(): time.sleep(0.2) final_file.write_text('{"event": "finish"}\n') active_file.unlink()
completer = threading.Thread(target=complete_use, daemon=True) completer.start()
with caplog.at_level(logging.INFO): completed, timed_out = wait_for_uses([use_id], timeout=2)
completer.join(timeout=5)
matching_records = [ record for record in caplog.records if "completion event not received but use completed" in record.message ] assert completed == {use_id: "finish"} assert timed_out == [] assert len(matching_records) == 1