# SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc from __future__ import annotations import json import logging import threading import time from datetime import datetime import pytest from flask import Flask from solstone.apps.chat.copy import CHAT_CLOSER_SUPPORT_SEND_FAILED from 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", }