solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import pytest
7from flask import Flask
8
9from solstone.convey.sol_initiated.copy import KIND_SOL_CHAT_REQUEST
10from solstone.think.push import runtime
11from solstone.think.push.runtime import (
12 get_runtime_state,
13 start_push_runtime,
14 stop_all_push_runtime,
15 stop_push_runtime,
16)
17
18
19@pytest.fixture(autouse=True)
20def reset_runtime():
21 stop_all_push_runtime()
22 yield
23 stop_all_push_runtime()
24
25
26def test_start_push_runtime_attaches_state(monkeypatch):
27 calls: list[str] = []
28 monkeypatch.setattr(
29 "solstone.think.push.runtime.CallosumConnection.start",
30 lambda self, callback=None: calls.append("start"),
31 )
32 monkeypatch.setattr(
33 "solstone.think.push.runtime.CallosumConnection.stop",
34 lambda self: calls.append("stop"),
35 )
36 app = Flask(__name__)
37
38 start_push_runtime(app)
39 try:
40 runtime_state = get_runtime_state()
41 assert app.push_runtime_started is True
42 assert runtime_state is not None
43 assert runtime_state.loop is not None
44 assert runtime_state.thread is not None
45 assert calls == ["start"]
46 finally:
47 stop_push_runtime(app)
48
49
50def test_start_push_runtime_is_idempotent(monkeypatch):
51 monkeypatch.setattr(
52 "solstone.think.push.runtime.CallosumConnection.start",
53 lambda self, callback=None: None,
54 )
55 monkeypatch.setattr(
56 "solstone.think.push.runtime.CallosumConnection.stop", lambda self: None
57 )
58 app = Flask(__name__)
59
60 start_push_runtime(app)
61 runtime_state = get_runtime_state()
62 first_loop = runtime_state.loop if runtime_state else None
63 first_thread = runtime_state.thread if runtime_state else None
64 try:
65 start_push_runtime(app)
66 runtime_state = get_runtime_state()
67 assert runtime_state is not None
68 assert runtime_state.loop is first_loop
69 assert runtime_state.thread is first_thread
70 assert runtime_state.apps.count(app) == 1
71 finally:
72 stop_push_runtime(app)
73
74
75def test_stop_push_runtime_cleans_last_app(monkeypatch):
76 monkeypatch.setattr(
77 "solstone.think.push.runtime.CallosumConnection.start",
78 lambda self, callback=None: None,
79 )
80 monkeypatch.setattr(
81 "solstone.think.push.runtime.CallosumConnection.stop", lambda self: None
82 )
83 app = Flask(__name__)
84
85 start_push_runtime(app)
86 stop_push_runtime(app)
87
88 assert app.push_runtime_started is False
89 assert get_runtime_state() is None
90
91
92def test_stop_all_push_runtime_clears_runtime(monkeypatch):
93 monkeypatch.setattr(
94 "solstone.think.push.runtime.CallosumConnection.start",
95 lambda self, callback=None: None,
96 )
97 monkeypatch.setattr(
98 "solstone.think.push.runtime.CallosumConnection.stop", lambda self: None
99 )
100 app = Flask(__name__)
101
102 start_push_runtime(app)
103 stop_all_push_runtime()
104
105 assert app.push_runtime_started is False
106 assert get_runtime_state() is None
107
108
109def test_on_callosum_message_calls_all_handlers(monkeypatch):
110 calls: list[tuple[str, dict[str, str]]] = []
111 request_handler = "handle_sol_chat_" + "request"
112 monkeypatch.setattr(
113 runtime.triggers,
114 request_handler,
115 lambda message: calls.append(("request_handler", message)),
116 )
117 monkeypatch.setattr(
118 runtime.triggers,
119 "handle_chat_lifecycle",
120 lambda message: calls.append(("chat_lifecycle", message)),
121 )
122 monkeypatch.setattr(
123 runtime.triggers,
124 "handle_chat_fold",
125 lambda message: calls.append(("chat_fold", message)),
126 )
127 message = {"tract": "chat", "event": KIND_SOL_CHAT_REQUEST, "request_id": "req-1"}
128
129 runtime._on_callosum_message(message)
130
131 assert calls == [
132 ("request_handler", message),
133 ("chat_lifecycle", message),
134 ("chat_fold", message),
135 ]