Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
from __future__ import annotations
import jsonimport loggingimport threadingimport timefrom datetime import datetime
import pytestfrom flask import Flask
from solstone.apps.chat.copy import CHAT_CLOSER_SUPPORT_SEND_FAILEDfrom solstone.convey.chat_stream import ( append_chat_event, read_chat_events, reduce_chat_state,)from tests.helpers.module_mocks import module_mock
def _reset_chat_state(chat_module) -> None: chat_module.stop_all_chat_runtime() with chat_module._state_lock: chat_module._current_chat_use_id = None chat_module._current_chat_state = None chat_module._queued_triggers.clear() chat_module._active_talents.clear() chat_module._reserved_use_ids.clear() chat_module._abandoned_raw_use_ids.clear() chat_module._raw_use_liveness.clear() for timer in chat_module._watchdog_timers.values(): timer.cancel() chat_module._watchdog_timers.clear() chat_module._last_use_id = 0
def _setup_journal(tmp_path, monkeypatch): journal = tmp_path / "journal" journal.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) return journal
def _ms(year: int, month: int, day: int, hour: int, minute: int, second: int) -> int: return int(datetime(year, month, day, hour, minute, second).timestamp() * 1000)
def _install_fake_timers(monkeypatch): import solstone.convey.chat as chat
timers: list[FakeTimer] = []
class FakeTimer: def __init__(self, interval, function, args=None, kwargs=None): self.interval = interval self.function = function self.args = tuple(args or ()) self.kwargs = dict(kwargs or {}) self.started = False self.cancelled = False self.daemon = False timers.append(self)
def start(self) -> None: self.started = True
def cancel(self) -> None: self.cancelled = True
def fire(self) -> None: if self.cancelled: return self.function(*self.args, **self.kwargs)
monkeypatch.setattr( chat, "threading", module_mock(chat.threading, Timer=FakeTimer), ) return timers
def _append_recoverable_talent_events( chat_use_id: str, talent_use_id: str, *, target: str = "exec", task: str = "research it",) -> None: now = datetime.now() start = _ms(now.year, now.month, now.day, 12, 0, 0) append_chat_event( "owner_message", ts=start, text="Help me with this", app="home", path="/app/home", facet="work", ) append_chat_event( "sol_message", ts=start + 1, use_id=chat_use_id, text="I am looking into that.", notes="need exec", requested_target=target, requested_task=task, ) append_chat_event( "talent_spawned", ts=start + 1_000, use_id=talent_use_id, name=target, task=task, started_at=start + 1_000, )
def test_chat_result_with_two_active_talents_queues_deferred_spawn( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
append_chat_event( "talent_spawned", use_id="1713620000001", name="exec", task="first task", started_at=1713620000001, ) append_chat_event( "talent_spawned", use_id="1713620000002", name="exec", task="second task", started_at=1713620000002, )
actions: list[dict] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None )
with chat._state_lock: chat._current_chat_use_id = "1713620000100" chat._current_chat_state = { "raw_use_id": "1713620000101", "raw_use_ids_seen": {"1713620000101"}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish( { "use_id": "1713620000101", "result": json.dumps( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": "research it", "context": json.dumps({"k": "v"}), }, } ), } )
assert actions == []
sol_messages = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "sol_message" ] assert sol_messages[-1]["requested_target"] == "exec" assert sol_messages[-1]["requested_task"] == "research it" queued = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "talent_queued" ] assert queued[-1]["name"] == "exec" assert queued[-1]["task"] == "research it" assert queued[-1]["chat_use_id"] == "1713620000100" assert queued[-1]["ask"] == "help" assert queued[-1]["context"] == {"k": "v"} assert reduce_chat_state(chat._today_day())["queued_talents"] == [ { "use_id": queued[-1]["use_id"], "name": "exec", "task": "research it", "queued_at": queued[-1]["queued_at"], } ]
def test_post_talent_finished_request_is_forced_terminal(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] finishes: list[tuple[str, str]] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda use_id, message: finishes.append((use_id, message)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None, )
with chat._state_lock: chat._current_chat_use_id = "1713621999999" chat._current_chat_state = { "raw_use_id": "1713622000000", "raw_use_ids_seen": {"1713622000000"}, "trigger": {"type": "talent_finished", "summary": "summary"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish( { "use_id": "1713622000000", "result": json.dumps( { "message": "Let me check logs. Found three relevant notes.", "notes": "blocked redispatch", "talent_request": { "target": "exec", "task": "one more pass", "context": json.dumps({}), }, } ), } )
assert actions == [] expected_text = ( "Here's what I have so far: Found three relevant notes. " "Want me to try a different angle?" ) events = read_chat_events(chat._today_day()) sol_messages = [event for event in events if event["kind"] == "sol_message"] assert len(sol_messages) == 1 assert sol_messages[-1]["text"] == expected_text assert sol_messages[-1]["requested_target"] is None assert sol_messages[-1]["requested_task"] is None assert [event for event in events if event["kind"] == "talent_spawned"] == [] assert finishes == [("1713621999999", expected_text)] with chat._state_lock: assert chat._current_chat_state is None assert chat._current_chat_use_id is None
def test_post_talent_errored_request_is_forced_terminal(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] finishes: list[tuple[str, str]] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda use_id, message: finishes.append((use_id, message)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None, )
with chat._state_lock: chat._current_chat_use_id = "1713622100000" chat._current_chat_state = { "raw_use_id": "1713622100001", "raw_use_ids_seen": {"1713622100001"}, "trigger": { "type": "talent_errored", "reason": "talent timed out waiting for provider response", }, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish( { "use_id": "1713622100001", "result": json.dumps( { "message": "I'll check again.", "notes": "blocked redispatch", "talent_request": { "target": "exec", "task": "one more pass", "context": json.dumps({}), }, } ), } )
expected_text = ( "I couldn't finish that lookup — talent timed out waiting for provider " "response. Want to try a different angle, or rephrase the question?" ) assert actions == [] events = read_chat_events(chat._today_day()) sol_messages = [event for event in events if event["kind"] == "sol_message"] assert len(sol_messages) == 1 assert sol_messages[-1]["text"] == expected_text assert sol_messages[-1]["requested_target"] is None assert sol_messages[-1]["requested_task"] is None assert [event for event in events if event["kind"] == "talent_spawned"] == [] assert finishes == [("1713622100000", expected_text)] with chat._state_lock: assert chat._current_chat_state is None assert chat._current_chat_use_id is None
def test_post_support_talent_errored_request_uses_send_failed_closer( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] finishes: list[tuple[str, str]] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda use_id, message: finishes.append((use_id, message)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None, )
with chat._state_lock: chat._current_chat_use_id = "1713622150000" chat._current_chat_state = { "raw_use_id": "1713622150001", "raw_use_ids_seen": {"1713622150001"}, "trigger": { "type": "talent_errored", "name": "support", "reason": "Traceback (most recent call last)", "reason_code": "wall_clock_exceeded", }, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish( { "use_id": "1713622150001", "result": json.dumps( { "message": "I drafted a ticket and will file it via live chat.", "notes": "blocked redispatch", "talent_request": { "target": "exec", "task": "one more pass", "context": json.dumps({}), }, } ), } )
assert actions == [] events = read_chat_events(chat._today_day()) sol_messages = [event for event in events if event["kind"] == "sol_message"] assert len(sol_messages) == 1 assert sol_messages[-1]["text"] == CHAT_CLOSER_SUPPORT_SEND_FAILED assert sol_messages[-1]["requested_target"] is None assert sol_messages[-1]["requested_task"] is None assert finishes == [("1713622150000", CHAT_CLOSER_SUPPORT_SEND_FAILED)] with chat._state_lock: assert chat._current_chat_state is None assert chat._current_chat_use_id is None
def test_owner_message_direct_reply_keeps_raw_model_text(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
finishes: list[tuple[str, str]] = [] monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda action: None) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda use_id, message: finishes.append((use_id, message)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None, )
with chat._state_lock: chat._current_chat_use_id = "1713622200000" chat._current_chat_state = { "raw_use_id": "1713622200001", "raw_use_ids_seen": {"1713622200001"}, "trigger": {"type": "owner_message", "message": "status?"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
raw_text = "Nothing new showed up in the digest." chat._on_cortex_finish( { "use_id": "1713622200001", "result": json.dumps( { "message": raw_text, "notes": "answered directly", "talent_request": None, } ), } )
sol_messages = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "sol_message" ] assert sol_messages[-1]["text"] == raw_text assert finishes == [("1713622200000", raw_text)]
def test_owner_message_request_still_spawns_talent(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None, )
with chat._state_lock: chat._current_chat_use_id = "1713622300000" chat._current_chat_state = { "raw_use_id": "1713622300001", "raw_use_ids_seen": {"1713622300001"}, "trigger": {"type": "owner_message", "message": "research this"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish( { "use_id": "1713622300001", "result": json.dumps( { "message": "I am looking into that.", "notes": "dispatch", "talent_request": { "target": "exec", "task": "research it", "context": json.dumps({}), }, } ), } )
assert actions[-1]["kind"] == "talent" events = read_chat_events(chat._today_day()) talent_spawns = [event for event in events if event["kind"] == "talent_spawned"] assert len(talent_spawns) == 1 assert talent_spawns[-1]["name"] == "exec"
@pytest.mark.parametrize( "trigger", [ None, {}, ],)def test_non_post_talent_triggers_allow_dispatch(tmp_path, monkeypatch, trigger): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None, )
with chat._state_lock: chat._current_chat_use_id = "1713622400000" chat._current_chat_state = { "raw_use_id": "1713622400001", "raw_use_ids_seen": {"1713622400001"}, "trigger": trigger, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish( { "use_id": "1713622400001", "result": json.dumps( { "message": "I am looking into that.", "notes": "dispatch", "talent_request": { "target": "exec", "task": "research it", "context": json.dumps({}), }, } ), } )
assert actions[-1]["kind"] == "talent" events = read_chat_events(chat._today_day()) talent_spawns = [event for event in events if event["kind"] == "talent_spawned"] assert len(talent_spawns) == 1 assert talent_spawns[-1]["name"] == "exec"
def test_cortex_finish_and_error_append_exec_terminal_events_by_use_id( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None )
with chat._state_lock: chat._current_chat_use_id = "1713623000000" chat._current_chat_state = { "raw_use_id": None, "raw_use_ids_seen": set(), "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, } chat._active_talents["1713623000001"] = { "chat_use_id": "1713623000000", "target": "exec", "task": "summarize", "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "ask": "help", }
chat._on_cortex_finish({"use_id": "1713623000001", "result": "done"}) finished_events = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "talent_finished" ] assert finished_events[-1]["use_id"] == "1713623000001" assert actions == [] with chat._state_lock: queued = chat._queued_triggers[-1] assert queued["trigger"]["type"] == "talent_finished" assert queued["trigger"]["origin"] == { "logical_use_id": "1713623000000", "ask": "help", }
_reset_chat_state(chat) actions.clear() with chat._state_lock: chat._current_chat_use_id = "1713624000000" chat._current_chat_state = { "raw_use_id": None, "raw_use_ids_seen": set(), "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, } chat._active_talents["1713624000001"] = { "chat_use_id": "1713624000000", "target": "exec", "task": "summarize", "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "ask": "help", }
chat._on_cortex_error( { "use_id": "1713624000001", "error": "boom", "reason_code": "wall_clock_exceeded", } ) errored_events = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "talent_errored" ] assert errored_events[-1]["use_id"] == "1713624000001" assert actions == [] with chat._state_lock: queued = chat._queued_triggers[-1] assert queued["trigger"]["type"] == "talent_errored" assert queued["trigger"]["reason"] == "boom" assert queued["trigger"]["reason_code"] == "wall_clock_exceeded" assert queued["trigger"]["origin"] == { "logical_use_id": "1713624000000", "ask": "help", }
def test_talent_errored_trigger_recovers_reason_code(): import solstone.convey.chat as chat
trigger = chat._trigger_from_stream_event( "20260420", { "kind": "talent_errored", "use_id": "1713624500001", "name": "support", "reason": "Traceback (most recent call last)", "reason_code": "wall_clock_exceeded", }, )
assert trigger["type"] == "talent_errored" assert trigger["name"] == "support" assert trigger["reason_code"] == "wall_clock_exceeded"
@pytest.mark.parametrize( ("terminal_kind", "result_field_name", "result_field_label", "result_value"), [ ("talent_finished", "summary", "result", "Found the answer."), ("talent_errored", "reason", "reason", "The lookup failed."), ],)def test_terminal_talent_reports_back_without_redispatch( tmp_path, monkeypatch, terminal_kind, result_field_name, result_field_label, result_value,): import solstone.convey.chat as chat from solstone.talent import chat_context
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) _install_fake_timers(monkeypatch)
monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None )
spawns: list[dict] = []
def fake_spawn_agent(prompt, name, config, use_id): spawns.append( { "prompt": prompt, "name": name, "config": dict(config), "use_id": str(use_id), } ) return use_id
monkeypatch.setattr("solstone.convey.utils.spawn_agent", fake_spawn_agent)
app = Flask(__name__) app.register_blueprint(chat.chat_bp) app.testing = True client = app.test_client()
response = client.post( "/api/chat", json={ "message": "Can you check this?", "app": "sol", "path": "/app/sol", "facet": "work", }, ) assert response.status_code == 200 assert response.get_json()["queued"] is False
first_chat_spawn = spawns[-1] assert first_chat_spawn["name"] == "chat" chat._on_cortex_finish( { "use_id": first_chat_spawn["use_id"], "result": json.dumps( { "message": "I am checking.", "notes": "dispatch exec", "talent_request": { "target": "exec", "task": "Check the thing", "context": json.dumps({"facet": "work"}), }, } ), } )
talent_spawn = spawns[-1] assert talent_spawn["name"] == "exec" if terminal_kind == "talent_finished": chat._on_cortex_finish( {"use_id": talent_spawn["use_id"], "result": result_value} ) else: chat._on_cortex_error({"use_id": talent_spawn["use_id"], "error": result_value})
report_back_spawn = spawns[-1] assert report_back_spawn["name"] == "chat" assert report_back_spawn["config"]["trigger"]["type"] == terminal_kind assert report_back_spawn["config"]["trigger"][result_field_name] == result_value
context_result = chat_context.pre_process(report_back_spawn["config"]) followup = context_result["messages"][-1]["content"] trigger_context = context_result["template_vars"]["trigger_context"] stop_and_report = ( "stop-and-report turn, not a dispatch turn. Do not retry this task " "or request another talent for it. Stop here and report to the owner " "directly using the" ) assert stop_and_report in followup assert stop_and_report in trigger_context assert f"{result_field_label.capitalize()}: {result_value}" in followup
raw_report_use_id = report_back_spawn["use_id"] chat._on_cortex_finish( { "use_id": raw_report_use_id, "result": json.dumps( { "message": "Here is the summary.", "notes": "reported terminal talent result", "talent_request": None, } ), } )
events = read_chat_events(chat._today_day()) sol_messages = [event for event in events if event["kind"] == "sol_message"] if terminal_kind == "talent_finished": expected_text = ( "Here's what I have so far: Here is the summary. " "Want me to try a different angle?" ) else: expected_text = ( "I couldn't finish that lookup — The lookup failed. " "Want to try a different angle, or rephrase the question?" ) assert sol_messages[-1]["text"] == expected_text assert sol_messages[-1]["requested_target"] is None talent_spawns = [event for event in events if event["kind"] == "talent_spawned"] assert len(talent_spawns) == 1 assert [spawn["name"] for spawn in spawns] == ["chat", "exec", "chat"]
def test_start_chat_runtime_recovers_exactly_one_unresponded_trigger( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
append_chat_event( "owner_message", text="recover me", app="sol", path="/app/sol", facet="work", )
starts: list[dict] = [] monkeypatch.setattr( "solstone.convey.chat.CallosumConnection.start", lambda self, callback=None: None, ) monkeypatch.setattr( "solstone.convey.chat.CallosumConnection.stop", lambda self: None ) monkeypatch.setattr( "solstone.convey.chat._spawn_chat_generate", lambda action: starts.append(action) or chat.ChatSpawnResult(ok=True), )
app = Flask(__name__) chat.start_chat_runtime(app) chat.start_chat_runtime(app)
assert len(starts) == 1
def test_start_chat_runtime_skips_debug_reloader_parent(tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
starts: list[object] = [] monkeypatch.setattr( "solstone.convey.chat.CallosumConnection.start", lambda self, callback=None: starts.append(callback), ) monkeypatch.setattr( "solstone.convey.chat.CallosumConnection.stop", lambda self: None )
app = Flask(__name__) app.debug = True
monkeypatch.delenv("WERKZEUG_RUN_MAIN", raising=False) with caplog.at_level("INFO"): chat.start_chat_runtime(app)
assert chat._runtime is None assert starts == [] assert app.chat_runtime_started is False assert "skipping chat runtime startup in Werkzeug reloader parent" in caplog.text
monkeypatch.setenv("WERKZEUG_RUN_MAIN", "true") chat.start_chat_runtime(app)
assert chat._runtime is not None assert len(starts) == 1 assert app.chat_runtime_started is True
def test_recover_active_talents_repopulates_from_chat_stream(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch) day = datetime.now().strftime("%Y%m%d") monkeypatch.setattr("solstone.convey.chat._today_day", lambda: day)
chat_use_id = "1713624500000" talent_use_id = "1713624500001" _append_recoverable_talent_events(chat_use_id, talent_use_id)
chat._recover_chat_if_needed()
with chat._state_lock: assert chat._active_talents[talent_use_id] == { "chat_use_id": chat_use_id, "target": "exec", "task": "research it", "trigger": "sol_message", "location": {"app": "home", "path": "/app/home", "facet": "work"}, "ask": "Help me with this", } assert talent_use_id in chat._watchdog_timers assert len(timers) == 1
def test_late_talent_finish_after_recovery_routes_to_chat_continuation( tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) _install_fake_timers(monkeypatch) day = datetime.now().strftime("%Y%m%d") monkeypatch.setattr("solstone.convey.chat._today_day", lambda: day)
chat_use_id = "1713624600000" talent_use_id = "1713624600001" _append_recoverable_talent_events(chat_use_id, talent_use_id) chat._recover_chat_if_needed()
actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None )
with chat._state_lock: chat._current_chat_use_id = chat_use_id chat._current_chat_state = { "raw_use_id": None, "raw_use_ids_seen": set(), "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "home", "path": "/app/home", "facet": "work"}, "retry_count": 0, }
with caplog.at_level("WARNING"): chat._on_cortex_finish({"use_id": talent_use_id, "result": "done"})
assert "unrouteable cortex event" not in caplog.text finished_events = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "talent_finished" ] assert finished_events[-1]["use_id"] == talent_use_id assert actions == [] with chat._state_lock: queued = chat._queued_triggers[-1] assert queued["trigger"]["type"] == "talent_finished" assert queued["trigger"]["origin"] == { "logical_use_id": chat_use_id, "ask": "Help me with this", } next_action = chat._clear_current_locked() assert next_action["logical_use_id"] == queued["use_id"] assert next_action["trigger"]["type"] == "talent_finished"
def test_recovery_is_idempotent_for_active_talents(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch) day = datetime.now().strftime("%Y%m%d") monkeypatch.setattr("solstone.convey.chat._today_day", lambda: day)
chat_use_id = "1713624700000" talent_use_id = "1713624700001" _append_recoverable_talent_events(chat_use_id, talent_use_id)
chat._recover_chat_if_needed() chat._recover_chat_if_needed()
with chat._state_lock: assert list(chat._active_talents) == [talent_use_id] assert talent_use_id in chat._watchdog_timers assert len(timers) == 1
def test_recovery_emits_info_log_per_reactivation(tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) _install_fake_timers(monkeypatch) day = datetime.now().strftime("%Y%m%d") monkeypatch.setattr("solstone.convey.chat._today_day", lambda: day)
chat_use_id = "1713624750000" talent_use_id = "1713624750001" _append_recoverable_talent_events(chat_use_id, talent_use_id)
with caplog.at_level(logging.INFO, logger="solstone.convey.chat"): chat._recover_chat_if_needed()
records = [ record for record in caplog.records if record.name == "solstone.convey.chat" and "reactivated" in record.getMessage() ] assert len(records) == 1 record = records[0] assert record.use_id == talent_use_id assert record.day == day assert record.trigger == "sol_message"
caplog.clear() empty_journal = tmp_path / "empty-journal" empty_journal.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(empty_journal)) _reset_chat_state(chat)
with caplog.at_level(logging.INFO, logger="solstone.convey.chat"): chat._recover_chat_if_needed()
assert [ record for record in caplog.records if record.name == "solstone.convey.chat" and record.levelno == logging.INFO ] == []
def test_chat_generate_schema_violation_retries_once_then_chat_errors( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] emitted_errors: list[tuple[str, str]] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **_kwargs: emitted_errors.append((use_id, reason)), )
with chat._state_lock: chat._current_chat_use_id = "1713625000000" chat._current_chat_state = { "raw_use_id": "1713625000001", "raw_use_ids_seen": {"1713625000001"}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish({"use_id": "1713625000001", "result": "not json"})
assert actions and actions[-1]["kind"] == "chat" assert actions[-1]["logical_use_id"] == "1713625000000" assert emitted_errors == []
with chat._state_lock: retry_use_id = chat._current_chat_state["raw_use_id"]
chat._on_cortex_finish({"use_id": retry_use_id, "result": "still not json"})
assert emitted_errors == [("1713625000000", "provider_response_invalid")] errors = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "chat_error" ] assert errors[-1]["use_id"] == "1713625000000"
def test_chat_generate_absorbs_prose_context_and_dispatches_talent( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] emitted_errors: list[tuple[str, str]] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **_kwargs: emitted_errors.append((use_id, reason)), )
with chat._state_lock: chat._current_chat_use_id = "1713625050000" chat._current_chat_state = { "raw_use_id": "1713625050001", "raw_use_ids_seen": {"1713625050001"}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
prose_context_result = json.dumps( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": "research it", "context": "not json{", }, } ) chat._on_cortex_finish({"use_id": "1713625050001", "result": prose_context_result})
assert actions and actions[-1]["kind"] == "talent" assert actions[-1]["logical_use_id"] == "1713625050000" assert actions[-1]["context"] == {"_raw": "not json{"} assert emitted_errors == [] spawned_events = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "talent_spawned" ] assert spawned_events[-1]["use_id"] == actions[-1]["use_id"]
def test_superseded_raw_finish_after_retry_is_dropped_without_warning( tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None )
raw_use_id = "1713625100001" with chat._state_lock: chat._current_chat_use_id = "1713625100000" chat._current_chat_state = { "raw_use_id": raw_use_id, "raw_use_ids_seen": {raw_use_id}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish({"use_id": raw_use_id, "result": "not json"})
with chat._state_lock: retry_use_id = str(chat._current_chat_state["raw_use_id"])
events_before = list(read_chat_events(chat._today_day())) with caplog.at_level("DEBUG"): chat._on_cortex_finish({"use_id": raw_use_id, "result": "still not json"})
assert ( "superseded raw cortex event use_id=1713625100001 event=finish reason=raw rotated" in caplog.text ) assert "unrouteable cortex event" not in caplog.text assert read_chat_events(chat._today_day()) == events_before
chat._on_cortex_finish( { "use_id": retry_use_id, "result": '{"message":"done","notes":"ok","talent_request":null}', } )
sol_messages = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "sol_message" ] assert sol_messages[-1]["text"] == "done"
def test_superseded_raw_error_after_followup_rotation_is_dropped_without_warning( tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None )
stale_raw_use_id = "1713625200001" with chat._state_lock: chat._current_chat_use_id = "1713625200000" chat._current_chat_state = { "raw_use_id": None, "raw_use_ids_seen": {stale_raw_use_id}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, } chat._active_talents["1713625200002"] = { "chat_use_id": "1713625200000", "target": "exec", "task": "summarize", "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, }
chat._on_cortex_finish({"use_id": "1713625200002", "result": "summary"})
with chat._state_lock: assert chat._current_chat_state["raw_use_id"] is None queued = chat._queued_triggers[-1] assert queued["trigger"]["type"] == "talent_finished"
events_before = list(read_chat_events(chat._today_day())) with caplog.at_level("DEBUG"): chat._on_cortex_error({"use_id": stale_raw_use_id, "error": "boom"})
assert ( "superseded raw cortex event use_id=1713625200001 event=error reason=raw rotated" in caplog.text ) assert "unrouteable cortex event" not in caplog.text assert read_chat_events(chat._today_day()) == events_before assert actions == []
def test_reserved_unknown_raw_use_id_still_warns(tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
use_id = "1713625300009" with chat._state_lock: chat._current_chat_use_id = "1713625300000" chat._current_chat_state = { "raw_use_id": "1713625300002", "raw_use_ids_seen": {"1713625300001", "1713625300002"}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, } chat._reserved_use_ids[use_id] = None
caplog.set_level(logging.WARNING, logger="solstone.convey.chat") chat._on_cortex_finish({"use_id": use_id, "result": "done"})
chat_records = [ record for record in caplog.records if record.name == "solstone.convey.chat" and record.levelno == logging.WARNING ] assert len(chat_records) == 1 assert ( "unrouteable cortex event use_id=1713625300009 event=finish " "reason=no matching active chat-generate or talent" ) == chat_records[0].getMessage()
def test_exec_dispatch_appends_sol_message_and_spawns_talent_real_path( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
spawn_calls: list[dict[str, object]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None )
def fake_spawn_agent(prompt, name, provider=None, config=None, use_id=None): spawn_calls.append( { "prompt": prompt, "name": name, "provider": provider, "config": config, "use_id": use_id, } ) return use_id
monkeypatch.setattr("solstone.convey.utils.spawn_agent", fake_spawn_agent)
with chat._state_lock: start_info = chat._activate_current_locked( "1713625500000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] chat._on_cortex_finish( { "use_id": raw_use_id, "result": json.dumps( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": "research it", "context": json.dumps({"k": "v"}), }, } ), } )
events = read_chat_events(chat._today_day()) sol_messages = [event for event in events if event["kind"] == "sol_message"] spawned_events = [event for event in events if event["kind"] == "talent_spawned"]
assert sol_messages[-1]["text"] == "I am looking into that." assert sol_messages[-1]["requested_target"] == "exec" assert sol_messages[-1]["requested_task"] == "research it" assert spawned_events[-1]["name"] == "exec" assert spawned_events[-1]["task"] == "research it" assert len(spawn_calls) == 1 spawn_call = spawn_calls[0] assert spawn_call["name"] == "exec" assert spawn_call["use_id"] == spawned_events[-1]["use_id"] assert spawn_call["config"] == { "app": "sol", "path": "/app/sol", "facet": "work", "chat_parent_use_id": "1713625500000", } assert "research it" in str(spawn_call["prompt"]) assert "Context hints:\n{'k': 'v'}" in str(spawn_call["prompt"]) assert len(timers) == 2 assert timers[0].cancelled is True with chat._state_lock: assert spawned_events[-1]["use_id"] in chat._active_talents
def test_support_dispatch_after_offer_spawns_app_talent_but_keeps_bare_event_name( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) _install_fake_timers(monkeypatch)
spawn_calls: list[dict[str, object]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None )
def fake_spawn_agent(prompt, name, provider=None, config=None, use_id=None): spawn_calls.append( { "prompt": prompt, "name": name, "provider": provider, "config": config, "use_id": use_id, } ) return use_id
monkeypatch.setattr("solstone.convey.utils.spawn_agent", fake_spawn_agent) append_chat_event( "sol_message", use_id="seed-offer", text="offer support", notes="offer", requested_target=None, requested_task=None, offer={"kind": "support"}, )
with chat._state_lock: start_info = chat._activate_current_locked( "1713625600000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] chat._on_cortex_finish( { "use_id": raw_use_id, "result": json.dumps( { "message": "I'll get support on that.", "notes": "need support", "talent_request": { "target": "support", "task": "File a bug report", "context": json.dumps({"severity": "medium"}), }, } ), } )
events = read_chat_events(chat._today_day()) sol_messages = [event for event in events if event["kind"] == "sol_message"] spawned_events = [event for event in events if event["kind"] == "talent_spawned"]
assert sol_messages[-1]["requested_target"] == "support" assert spawned_events[-1]["name"] == "support" assert spawned_events[-1]["task"] == "File a bug report" assert len(spawn_calls) == 1 spawn_call = spawn_calls[0] assert spawn_call["name"] == "support:support" assert spawn_call["use_id"] == spawned_events[-1]["use_id"] assert spawn_call["config"] == { "app": "sol", "path": "/app/sol", "facet": "work", "chat_parent_use_id": "1713625600000", } assert "File a bug report" in str(spawn_call["prompt"]) assert "Target: support" in str(spawn_call["prompt"]) with chat._state_lock: active = chat._active_talents[spawned_events[-1]["use_id"]] assert active["target"] == "support"
def test_watchdog_refreshed_by_progress_event(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_cortex: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), )
with chat._state_lock: start_info = chat._activate_current_locked( "1713627800000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] assert len(timers) == 1
chat._proxy_progress( { "tract": "cortex", "event": "thinking", "use_id": raw_use_id, } )
assert len(timers) == 2 assert timers[0].cancelled is True assert timers[-1].interval == chat._WATCHDOG_TIMEOUTS["chat"]
def test_chat_liveness_accumulates_across_watchdog_refreshes(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None )
with chat._state_lock: start_info = chat._activate_current_locked( "1713627801000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] for _index in range(2): chat._handle_callosum_message( { "tract": "cortex", "event": "progress", "use_id": raw_use_id, "phase": "generate", } )
with chat._state_lock: liveness = chat._raw_use_liveness[raw_use_id] assert liveness.observed_progress_count == 2 assert liveness.last_event_type == "progress"
chat._handle_callosum_message( { "tract": "cortex", "event": "start", "use_id": raw_use_id, "provider": "local", } )
with chat._state_lock: liveness = chat._raw_use_liveness[raw_use_id] assert liveness.observed_progress_count == 2 assert liveness.last_event_type == "start"
chat._handle_callosum_message( { "tract": "cortex", "event": "progress", "use_id": raw_use_id, "phase": "generate", } )
with chat._state_lock: liveness = chat._raw_use_liveness[raw_use_id] assert liveness.observed_progress_count == 3 assert liveness.last_event_type == "progress"
assert len(timers) == 5 assert all(timer.cancelled for timer in timers[:-1]) assert timers[-1].cancelled is False
def test_live_generate_progress_prevents_chat_watchdog_and_neutered_trips( tmp_path, monkeypatch): import solstone.convey.chat as chat from solstone.think import talents from solstone.think.utils import now_ms
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) monkeypatch.setattr(chat, "_WATCHDOG_TIMEOUTS", {"chat": 0.08, "talent": 0.3}) monkeypatch.setattr(talents, "_GENERATE_PROGRESS_INTERVAL_S", 0.02) monkeypatch.setattr(talents, "_GENERATE_PROGRESS_JOIN_TIMEOUT_S", 0.1)
emitted_errors: list[tuple[str, str, dict]] = [] emitted_cortex: list[tuple[str, dict]] = [] tripped = threading.Event() monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), )
def emit_error(use_id: str, reason: str, **kwargs) -> None: emitted_errors.append((use_id, reason, kwargs)) tripped.set()
monkeypatch.setattr("solstone.convey.chat._emit_error", emit_error) monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda action: None)
def activate(logical_use_id: str) -> str: with chat._state_lock: start_info = chat._activate_current_locked( logical_use_id, {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, ) return str(start_info["raw_use_id"])
def route_from_producer(raw_use_id: str, event: dict) -> None: body = dict(event) body.setdefault("ts", now_ms()) body.setdefault("use_id", raw_use_id) event_type = body.pop("event") chat._handle_callosum_message( { "tract": "cortex", "event": event_type, **body, } )
live_raw_use_id = activate("1713627810000") heartbeat = talents._GenerateProgressHeartbeat( lambda event: route_from_producer(live_raw_use_id, event), talent_name="chat", day=None, schedule=None, ) heartbeat.start() try: time.sleep(0.18) finally: heartbeat.stop_and_join()
assert not tripped.is_set() assert emitted_errors == [] assert any(event == "progress" for event, _fields in emitted_cortex)
_reset_chat_state(chat) emitted_errors.clear() emitted_cortex.clear() tripped.clear()
activate("1713627811000") deadline = time.monotonic() + 0.5 while not tripped.is_set() and time.monotonic() < deadline: time.sleep(0.01)
assert tripped.is_set() assert emitted_errors[-1][0:2] == ("1713627811000", "chat_timeout") assert [event for event, _fields in emitted_cortex if event == "cancel"] == [ "cancel" ]
def test_arm_watchdog_uses_per_kind_timeout(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
with chat._state_lock: chat._arm_watchdog_locked("u-chat-1", "chat", "u-chat-1") chat._arm_watchdog_locked("u-talent-1", "talent", "u-chat-1") chat._arm_watchdog_locked("u-unknown-1", "weirdkind", "u-chat-1")
assert timers[0].interval == 30 assert timers[1].interval == 180 assert timers[2].interval == 180
def test_chat_watchdog_timeout_constant_remains_30_seconds(): import solstone.convey.chat as chat
assert chat._WATCHDOG_TIMEOUTS["chat"] == 30
def test_proxy_progress_does_not_re_emit_request_event(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted.append((event, fields)), )
with chat._state_lock: start_info = chat._activate_current_locked( "1713627820000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
chat._proxy_progress( { "tract": "cortex", "event": "request", "use_id": start_info["raw_use_id"], } )
assert emitted == [] assert len(timers) == 1
@pytest.mark.parametrize("event_type", ["thinking", "progress", "start"])def test_proxy_progress_still_emits_other_event_types( tmp_path, monkeypatch, event_type): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) _install_fake_timers(monkeypatch)
emitted: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted.append((event, fields)), )
with chat._state_lock: start_info = chat._activate_current_locked( "1713627830000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
chat._proxy_progress( { "tract": "cortex", "event": event_type, "use_id": start_info["raw_use_id"], } )
assert emitted == [ ( event_type, { "use_id": "1713627830000", "chat_proxy": True, }, ) ]
def test_progress_event_is_not_thinking_and_does_not_fill_thinking_buffer( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) _install_fake_timers(monkeypatch)
emitted: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted.append((event, fields)), )
with chat._state_lock: start_info = chat._activate_current_locked( "1713627840000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
assert "progress" not in {"thinking", "start"} chat._handle_callosum_message( { "tract": "cortex", "event": "progress", "use_id": start_info["raw_use_id"], "phase": "generate", } )
assert emitted == [ ( "progress", { "phase": "generate", "use_id": "1713627840000", "chat_proxy": True, }, ) ] with chat._state_lock: assert chat._thinking_buffers == {}
def test_watchdog_refresh_is_no_op_when_no_timer_registered(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_cortex: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), )
with chat._state_lock: chat._current_chat_use_id = "1713627850000" chat._current_chat_state = { "raw_use_id": "1713627850001", "raw_use_ids_seen": {"1713627850001"}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._proxy_progress( { "tract": "cortex", "event": "thinking", "use_id": "1713627850002", } )
assert len(timers) == 0
def test_watchdog_refreshed_by_talent_progress(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None )
with chat._state_lock: chat._current_chat_use_id = "1713627900000" chat._current_chat_state = { "raw_use_id": None, "raw_use_ids_seen": set(), "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, } chat._active_talents["1713627900001"] = { "chat_use_id": "1713627900000", "target": "exec", "task": "summarize", "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, } chat._arm_watchdog_locked("1713627900001", "talent", "1713627900000")
assert len(timers) == 1
chat._proxy_progress( { "tract": "cortex", "event": "thinking", "use_id": "1713627900001", } )
assert len(timers) == 2 assert timers[0].cancelled is True assert timers[-1].interval == chat._WATCHDOG_TIMEOUTS["talent"]
def test_stalled_run_still_times_out_after_inactivity(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_errors: list[tuple[str, str]] = [] run_actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **_kwargs: emitted_errors.append((use_id, reason)), ) monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: run_actions.append(action), )
with chat._state_lock: start_info = chat._activate_current_locked( "1713627950000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] assert len(timers) == 1
for _ in range(3): chat._proxy_progress( { "tract": "cortex", "event": "thinking", "use_id": raw_use_id, } )
assert len(timers) == 4 timers[-1].fire()
errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "chat_error" ] assert emitted_errors == [("1713627950000", "chat_timeout")] assert run_actions == [None] assert errors[-1]["use_id"] == "1713627950000" assert errors[-1]["reason"] == "chat_timeout" with chat._state_lock: assert chat._current_chat_use_id is None assert chat._current_chat_state is None assert raw_use_id not in chat._watchdog_timers
def test_chat_watchdog_times_out_current_chat_generate(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_errors: list[tuple[str, str]] = [] emitted_cortex: list[tuple[str, dict]] = [] run_actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **_kwargs: emitted_errors.append((use_id, reason)), ) monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: run_actions.append(action), )
with chat._state_lock: start_info = chat._activate_current_locked( "1713628000000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] assert raw_use_id in chat._watchdog_timers
timers[-1].fire()
errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "chat_error" ] assert emitted_errors == [("1713628000000", "chat_timeout")] assert emitted_cortex == [ ( "cancel", { "use_id": raw_use_id, "reason_code": "chat_watchdog_cancelled", }, ) ] assert run_actions == [None] assert errors[-1]["use_id"] == "1713628000000" assert errors[-1]["reason"] == "chat_timeout" with chat._state_lock: assert chat._current_chat_use_id is None assert chat._current_chat_state is None assert raw_use_id not in chat._watchdog_timers
def test_chat_watchdog_cancel_emit_failure_does_not_escape(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_errors: list[tuple[str, str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **kwargs: emitted_errors.append( (use_id, reason, kwargs) ), )
def fail_cancel(event: str, **_fields) -> None: if event == "cancel": raise RuntimeError("callosum unavailable")
monkeypatch.setattr("solstone.convey.chat._emit_cortex_event", fail_cancel) monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda action: None)
with chat._state_lock: chat._activate_current_locked( "1713628020000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
timers[-1].fire()
errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "chat_error" ] assert emitted_errors[-1][0:2] == ("1713628020000", "chat_timeout") assert errors[-1]["reason"] == "chat_timeout"
def test_chat_watchdog_timeout_detail_reports_liveness(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_errors: list[tuple[str, str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **kwargs: emitted_errors.append( (use_id, reason, kwargs) ), ) monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None ) monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda action: None)
with chat._state_lock: start_info = chat._activate_current_locked( "1713628030000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] for event_type in ("start", "progress", "progress"): chat._handle_callosum_message( { "tract": "cortex", "event": event_type, "use_id": raw_use_id, "phase": "generate", } )
timers[-1].fire()
errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "chat_error" ] detail = errors[-1]["detail"] assert detail == emitted_errors[-1][2]["detail"] assert detail.startswith("silence ") assert "; last event progress; " in detail assert detail.endswith("liveness events 2") assert chat._normalize_chat_error_detail(detail) == detail
def test_late_chat_finish_after_watchdog_timeout_is_abandoned_raw( tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_errors: list[tuple[str, str, dict]] = [] emitted_finishes: list[tuple[str, str]] = [] emitted_cortex: list[tuple[str, dict]] = [] run_actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **kwargs: emitted_errors.append( (use_id, reason, kwargs) ), ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda use_id, message: emitted_finishes.append((use_id, message)), ) monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: run_actions.append(action), )
logical_use_id = "1713628050000" with chat._state_lock: start_info = chat._activate_current_locked( logical_use_id, {"type": "owner_message", "message": "first"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, ) queued_use_id, queued, start_action = chat._activate_or_enqueue_trigger_locked( {"type": "owner_message", "message": "second"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
assert queued is True assert start_action is None raw_use_id = start_info["raw_use_id"]
timers[-1].fire()
assert len(emitted_errors) == 1 assert emitted_errors[0][0:2] == (logical_use_id, "chat_timeout") assert emitted_errors[0][2]["detail"].startswith("silence ") assert emitted_cortex == [ ( "cancel", { "use_id": raw_use_id, "reason_code": "chat_watchdog_cancelled", }, ) ] assert run_actions and run_actions[-1]["logical_use_id"] == queued_use_id with chat._state_lock: assert chat._current_chat_use_id == queued_use_id assert chat._current_chat_state is not None assert chat._current_chat_state["raw_use_id"] != raw_use_id
events_before_late = read_chat_events(chat._today_day()) caplog.clear() caplog.set_level(logging.DEBUG, logger="solstone.convey.chat") chat._on_cortex_finish( { "use_id": raw_use_id, "result": json.dumps( {"message": "late answer", "notes": "ok", "talent_request": None} ), } )
assert read_chat_events(chat._today_day()) == events_before_late assert emitted_finishes == [] assert ( f"superseded raw cortex event use_id={raw_use_id} " "event=finish reason=raw rotated" ) in caplog.text warning_messages = [ record.getMessage() for record in caplog.records if record.name == "solstone.convey.chat" and record.levelno >= logging.WARNING ] assert all("unrouteable cortex event" not in msg for msg in warning_messages)
def test_chat_watchdog_cancel_precedes_queued_turn_spawn(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
order: list[tuple[str, str]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **_kwargs: order.append(("owner_error", use_id)), ) monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: order.append((event, str(fields.get("use_id") or ""))), ) monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: order.append(("run_next", str(action["logical_use_id"]))), )
with chat._state_lock: start_info = chat._activate_current_locked( "1713628060000", {"type": "owner_message", "message": "first"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, ) queued_use_id, queued, start_action = chat._activate_or_enqueue_trigger_locked( {"type": "owner_message", "message": "second"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
assert queued is True assert start_action is None timers[-1].fire()
assert order == [ ("owner_error", "1713628060000"), ("cancel", start_info["raw_use_id"]), ("run_next", queued_use_id), ]
def test_cancel_loopback_after_timeout_without_queue_is_inert(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_cortex: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None ) monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda action: None)
with chat._state_lock: start_info = chat._activate_current_locked( "1713628070000", {"type": "owner_message", "message": "help"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] timers[-1].fire() events_before_loopback = read_chat_events(chat._today_day()) emitted_cortex.clear()
chat._handle_callosum_message( {"tract": "cortex", "event": "cancel", "use_id": raw_use_id} )
assert emitted_cortex == [] assert read_chat_events(chat._today_day()) == events_before_loopback
def test_cancel_loopback_after_timeout_with_queue_is_superseded( tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_cortex: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None ) monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda action: None)
with chat._state_lock: start_info = chat._activate_current_locked( "1713628080000", {"type": "owner_message", "message": "first"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, ) chat._activate_or_enqueue_trigger_locked( {"type": "owner_message", "message": "second"}, {"app": "sol", "path": "/app/sol", "facet": "work"}, )
raw_use_id = start_info["raw_use_id"] timers[-1].fire() events_before_loopback = read_chat_events(chat._today_day()) emitted_cortex.clear() caplog.set_level(logging.DEBUG, logger="solstone.convey.chat")
chat._handle_callosum_message( {"tract": "cortex", "event": "cancel", "use_id": raw_use_id} )
assert emitted_cortex == [] assert read_chat_events(chat._today_day()) == events_before_loopback assert ( f"superseded raw cortex event use_id={raw_use_id} " "event=cancel reason=raw rotated" ) in caplog.text
def test_chat_watchdog_times_out_active_talent_and_queues_fold_when_chat_busy( tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
emitted_cortex: list[tuple[str, dict]] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda event, **fields: emitted_cortex.append((event, fields)), ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.utils.spawn_agent", lambda *args, **kwargs: kwargs["use_id"] )
with chat._state_lock: chat._current_chat_use_id = "1713629000000" chat._current_chat_state = { "raw_use_id": None, "raw_use_ids_seen": set(), "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, } chat._active_talents["1713629000001"] = { "chat_use_id": "1713629000000", "target": "exec", "task": "summarize", "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "ask": "help", }
chat._run_next_action( { "kind": "talent", "logical_use_id": "1713629000000", "target": "exec", "use_id": "1713629000001", "task": "summarize", "context": {}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, } )
assert "1713629000001" in chat._watchdog_timers timers[-1].fire()
chat_errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "chat_error" ] talent_errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "talent_errored" ] assert chat_errors == [] assert [event for event, _fields in emitted_cortex if event == "cancel"] == [] assert talent_errors[-1]["use_id"] == "1713629000001" assert talent_errors[-1]["reason"] == "talent took too long" with chat._state_lock: assert "1713629000001" not in chat._active_talents assert chat._current_chat_use_id == "1713629000000" assert chat._queued_triggers[-1]["trigger"]["type"] == "talent_errored" assert chat._queued_triggers[-1]["trigger"]["origin"] == { "logical_use_id": "1713629000000", "ask": "help", } assert "1713629000001" not in chat._watchdog_timers
def test_chat_watchdog_timeout_folds_talent_error_with_origin(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) timers = _install_fake_timers(monkeypatch)
actions: list[dict] = [] monkeypatch.setattr( "solstone.convey.chat._emit_cortex_event", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, )
with chat._state_lock: logical_use_id = chat._reserve_use_id_locked() talent_use_id = chat._reserve_use_id_locked()
append_chat_event( "talent_spawned", use_id=talent_use_id, name="exec", task="summarize", started_at=int(talent_use_id), )
with chat._state_lock: chat._active_talents[talent_use_id] = { "chat_use_id": logical_use_id, "target": "exec", "task": "summarize", "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "ask": "why did it hang?", } chat._arm_watchdog_locked(talent_use_id, "talent", logical_use_id)
timers[-1].fire()
chat_errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "chat_error" ] talent_errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "talent_errored" ] assert chat_errors == [] assert talent_errors[-1]["use_id"] == talent_use_id assert talent_errors[-1]["reason"] == "talent took too long" assert actions and actions[-1]["kind"] == "chat" assert actions[-1]["logical_use_id"] != logical_use_id assert actions[-1]["trigger"]["origin"] == { "logical_use_id": logical_use_id, "ask": "why did it hang?", }
chat._on_cortex_finish( { "use_id": actions[-1]["raw_use_id"], "result": {"message": "ignored", "notes": "ok", "talent_request": None}, } )
folded = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "sol_message" ][-1] assert folded["requested_target"] is None assert folded["origin"] == { "logical_use_id": logical_use_id, "ask": "why did it hang?", } assert folded["text"].startswith("I couldn't finish that lookup")
def test_cortex_finish_logs_warning_for_unrouteable_use_id( tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
with chat._state_lock: use_id = chat._reserve_use_id_locked()
caplog.set_level(logging.WARNING, logger="solstone.convey.chat") chat._on_cortex_finish({"use_id": use_id, "result": "done"})
chat_records = [ record for record in caplog.records if record.name == "solstone.convey.chat" and record.levelno == logging.WARNING ] assert len(chat_records) == 1 assert ( f"unrouteable cortex event use_id={use_id} event=finish " "reason=no matching active chat-generate or talent" ) == chat_records[0].getMessage()
@pytest.mark.parametrize( ("error_text", "expected_detail"), [ ("short msg", "short msg"), ( "something went very long with newlines\n\nand whitespace collapsed " + ("0123456789 " * 30), None, ), ],)def test_on_cortex_error_stores_normalized_detail( tmp_path, monkeypatch, error_text, expected_detail): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat) emitted_errors: list[tuple[str, str, dict]] = [] monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda action: None) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda use_id, reason, **kwargs: emitted_errors.append( (use_id, reason, kwargs) ), )
logical_use_id = "1713632500000" raw_use_id = "1713632500001" with chat._state_lock: chat._current_chat_use_id = logical_use_id chat._current_chat_state = { "raw_use_id": raw_use_id, "raw_use_ids_seen": {raw_use_id}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_error( { "use_id": raw_use_id, "reason_code": "unknown", "provider": "google", "error": error_text, } )
errors = [ event for event in read_chat_events(chat._today_day()) if event["kind"] == "chat_error" ] assert len(errors) == 1 assert errors[0]["reason"] == "unknown" assert errors[0]["provider"] == "google" detail = errors[0]["detail"] if expected_detail is not None: assert detail == expected_detail else: assert "\n" not in detail assert " " not in detail assert len(detail) <= 240 assert detail[-1] == "…" assert emitted_errors == [ ( logical_use_id, "unknown", {"provider": "google", "detail": detail}, ) ]
def test_cortex_error_logs_warning_for_unrouteable_use_id( tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
with chat._state_lock: use_id = chat._reserve_use_id_locked()
caplog.set_level(logging.WARNING, logger="solstone.convey.chat") chat._on_cortex_error({"use_id": use_id, "error": "boom"})
chat_records = [ record for record in caplog.records if record.name == "solstone.convey.chat" and record.levelno == logging.WARNING ] assert len(chat_records) == 1 assert ( f"unrouteable cortex event use_id={use_id} event=error " "reason=no matching active chat-generate or talent" ) == chat_records[0].getMessage()
def test_on_cortex_finish_unreserved_silent(tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
caplog.set_level(logging.WARNING, logger="solstone.convey.chat") chat._on_cortex_finish({"use_id": "1713632000000", "result": "done"})
chat_records = [ record for record in caplog.records if record.name == "solstone.convey.chat" and record.levelno == logging.WARNING ] assert chat_records == [] with chat._state_lock: assert chat._current_chat_use_id is None assert chat._current_chat_state is None assert chat._active_talents == {} assert chat._reserved_use_ids == {}
def test_on_cortex_error_unreserved_silent(tmp_path, monkeypatch, caplog): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
caplog.set_level(logging.WARNING, logger="solstone.convey.chat") chat._on_cortex_error({"use_id": "1713633000000", "error": "boom"})
chat_records = [ record for record in caplog.records if record.name == "solstone.convey.chat" and record.levelno == logging.WARNING ] assert chat_records == [] with chat._state_lock: assert chat._current_chat_use_id is None assert chat._current_chat_state is None assert chat._active_talents == {} assert chat._reserved_use_ids == {}
@pytest.mark.parametrize("target", ("read", "exec", "support"))def test_parse_chat_result_accepts_dispatch_targets(target): import solstone.convey.chat as chat
parsed = chat._parse_chat_result( { "message": "Let me think about that.", "notes": f"dispatch {target}", "talent_request": { "target": target, "task": "Handle the request", "context": {"facet": "work"}, }, } )
assert parsed["talent_request"] == { "target": target, "task": "Handle the request", "context": {"facet": "work"}, } # Exercises the scope-mandated defensive dict shim. assert parsed["talent_request"]["context"] == {"facet": "work"}
def test_parse_chat_result_decodes_json_string_context(): import solstone.convey.chat as chat
parsed = chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": "Research the last two weeks", "context": '{"window":"14d"}', }, } )
assert parsed["talent_request"]["context"] == {"window": "14d"}
@pytest.mark.parametrize( ("raw_context", "expected"), [ (None, {}), ("", {}), (" ", {}), ('{"foo": 1}', {"foo": 1}), ("123", {"_raw": "123"}), ("true", {"_raw": "true"}), ("[1, 2]", {"_raw": "[1, 2]"}), ('"hi"', {"_raw": '"hi"'}), ("not json at all{", {"_raw": "not json at all{"}), ({"foo": 1}, {"foo": 1}), (42, ValueError), ],)def test_parse_chat_result_context_absorbs_non_dict_shapes(raw_context, expected): import solstone.convey.chat as chat
payload = { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": "Research it", "context": raw_context, }, }
if expected is ValueError: with pytest.raises(ValueError): chat._parse_chat_result(payload) return
parsed = chat._parse_chat_result(payload) assert parsed["talent_request"]["context"] == expected
def test_parse_chat_result_accepts_null_talent_request(): import solstone.convey.chat as chat
parsed = chat._parse_chat_result( {"message": "hi", "notes": "n", "talent_request": None} )
assert parsed == { "message": "hi", "notes": "n", "talent_request": None, }
def test_parse_chat_result_rejects_unknown_target(): import solstone.convey.chat as chat
with pytest.raises(ValueError, match="unknown talent target: foo"): chat._parse_chat_result( { "message": "Let me think about that.", "notes": "dispatch unknown", "talent_request": {"target": "foo", "task": "Reflect on the week"}, } )
def test_parse_chat_result_rejects_missing_target(): import solstone.convey.chat as chat
with pytest.raises(ValueError, match="chat talent_request.target must be a string"): chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need exec", "talent_request": {"task": "Research it", "context": {"k": "v"}}, } )
def _target_alias_cases(): import solstone.convey.chat as chat
return list(chat.TARGET_ALIASES.items())
@pytest.mark.parametrize(("raw_target", "canonical_target"), _target_alias_cases())def test_parse_chat_result_coerces_target_case_variants(raw_target, canonical_target): import solstone.convey.chat as chat
parsed = chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need talent", "talent_request": { "target": raw_target, "task": "Research it", "context": None, }, } )
assert parsed["talent_request"]["target"] == canonical_target
@pytest.mark.parametrize("raw_target", ("reflection", "reflect"))def test_parse_chat_result_coerces_reflection_backcompat_to_read(raw_target): import solstone.convey.chat as chat
parsed = chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need read", "talent_request": { "target": raw_target, "task": "Reflect on the week", "context": None, }, } )
assert parsed["talent_request"]["target"] == "read"
def test_parse_chat_result_rejects_unknown_target_after_alias_lookup(): import solstone.convey.chat as chat
with pytest.raises(ValueError, match="unknown talent target: mystery"): chat._parse_chat_result( { "message": "Let me think about that.", "notes": "dispatch unknown", "talent_request": { "target": "mystery", "task": "Reflect on the week", "context": None, }, } )
def test_parse_chat_result_trims_task_whitespace(): import solstone.convey.chat as chat
parsed = chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": " do thing ", "context": None, }, } )
assert parsed["talent_request"]["task"] == "do thing"
def test_parse_chat_result_rejects_whitespace_only_task(): import solstone.convey.chat as chat
with pytest.raises( ValueError, match="chat talent_request.task must be a non-empty string" ): chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": " ", "context": None, }, } )
def test_parse_chat_result_emits_debug_on_target_coercion(caplog): import solstone.convey.chat as chat
with caplog.at_level(logging.DEBUG, logger="solstone.convey.chat"): chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "Exec", "task": "Research it", "context": None, }, }, use_id="test-use-id", )
assert any( "chat parser coerced target=Exec -> exec (use_id=test-use-id)" == record.getMessage() for record in caplog.records )
def test_parse_chat_result_emits_debug_on_task_coercion(caplog): import solstone.convey.chat as chat
with caplog.at_level(logging.DEBUG, logger="solstone.convey.chat"): chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need exec", "talent_request": { "target": "exec", "task": " run ", "context": None, }, }, use_id="test-use-id", )
assert any( "chat parser coerced task whitespace raw=' run ' -> 'run' " "(use_id=test-use-id)" == record.getMessage() for record in caplog.records )
def test_parse_chat_result_keeps_strict_on_non_dict_talent_request(): import solstone.convey.chat as chat
with pytest.raises( ValueError, match="chat talent_request must be an object or null" ): chat._parse_chat_result( { "message": "I am looking into that.", "notes": "need exec", "talent_request": [1, 2, 3], } )
def test_parse_chat_result_keeps_strict_on_non_string_message(): import solstone.convey.chat as chat
with pytest.raises( ValueError, match="chat result message must be a string or null" ): chat._parse_chat_result( { "message": 123, "notes": "need exec", "talent_request": None, } )
def test_parse_chat_result_keeps_strict_on_non_string_notes(): import solstone.convey.chat as chat
with pytest.raises(ValueError, match="chat result notes must be a string"): chat._parse_chat_result( { "message": "I am looking into that.", "notes": ["a", "b"], "talent_request": None, } )
def test_read_dispatch_spawns_read_talent(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None )
with chat._state_lock: chat._current_chat_use_id = "1713626000000" chat._current_chat_state = { "raw_use_id": "1713626000001", "raw_use_ids_seen": {"1713626000001"}, "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, }
chat._on_cortex_finish( { "use_id": "1713626000001", "result": json.dumps( { "message": "I want to sit with that.", "notes": "need read", "talent_request": { "target": "read", "task": "Reflect on the week", "context": json.dumps({"facet": "work"}), }, } ), } )
assert actions[-1]["kind"] == "talent" assert actions[-1]["target"] == "read"
events = read_chat_events(chat._today_day()) sol_message = next(event for event in events if event["kind"] == "sol_message") spawned = next(event for event in events if event["kind"] == "talent_spawned") assert sol_message["requested_target"] == "read" assert spawned["name"] == "read"
def test_read_finish_retriggers_chat_like_exec(tmp_path, monkeypatch): import solstone.convey.chat as chat
_setup_journal(tmp_path, monkeypatch) _reset_chat_state(chat)
actions: list[dict | None] = [] monkeypatch.setattr( "solstone.convey.chat._run_next_action", lambda action: actions.append(action) if action is not None else None, ) monkeypatch.setattr( "solstone.convey.chat._emit_finish", lambda *args, **kwargs: None ) monkeypatch.setattr( "solstone.convey.chat._emit_error", lambda *args, **kwargs: None )
with chat._state_lock: chat._current_chat_use_id = "1713627000000" chat._current_chat_state = { "raw_use_id": None, "raw_use_ids_seen": set(), "trigger": {"type": "owner_message", "message": "help"}, "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "retry_count": 0, } chat._active_talents["1713627000001"] = { "chat_use_id": "1713627000000", "target": "read", "task": "Reflect on the week", "location": {"app": "sol", "path": "/app/sol", "facet": "work"}, "ask": "help", }
chat._on_cortex_finish({"use_id": "1713627000001", "result": "A reflective note"})
finished_events = [ e for e in read_chat_events(chat._today_day()) if e["kind"] == "talent_finished" ] assert finished_events[-1]["name"] == "read" assert actions == [] with chat._state_lock: queued = chat._queued_triggers[-1] assert queued["trigger"]["type"] == "talent_finished" assert queued["trigger"]["name"] == "read" assert queued["trigger"]["origin"] == { "logical_use_id": "1713627000000", "ask": "help", }