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
6from datetime import date
7
8import pytest
9from flask import Flask
10
11from solstone.convey.chat import ChatSpawnResult, chat_bp
12from solstone.convey.chat_stream import read_chat_events
13
14
15def _setup_journal(tmp_path, monkeypatch):
16 journal = tmp_path / "journal"
17 journal.mkdir()
18 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
19 return journal
20
21
22def _reset_chat_state(chat_module) -> None:
23 chat_module.stop_all_chat_runtime()
24 with chat_module._state_lock:
25 chat_module._current_chat_use_id = None
26 chat_module._current_chat_state = None
27 chat_module._queued_triggers.clear()
28 chat_module._active_talents.clear()
29 chat_module._reserved_use_ids.clear()
30 for timer in chat_module._watchdog_timers.values():
31 timer.cancel()
32 chat_module._watchdog_timers.clear()
33 chat_module._last_use_id = 0
34
35
36@pytest.fixture
37def chat_client(tmp_path, monkeypatch):
38 import solstone.convey.chat as chat
39
40 _setup_journal(tmp_path, monkeypatch)
41 _reset_chat_state(chat)
42 monkeypatch.setattr(
43 "solstone.think.identity.ensure_identity_directory",
44 lambda: None,
45 )
46 monkeypatch.setattr(
47 "solstone.convey.chat._spawn_chat_generate",
48 lambda action: ChatSpawnResult(ok=True),
49 )
50
51 app = Flask(__name__)
52 app.config["TESTING"] = True
53 app.register_blueprint(chat_bp)
54 return app.test_client()
55
56
57def _latest_owner_message() -> dict:
58 events = read_chat_events(date.today().strftime("%Y%m%d"))
59 owner_messages = [event for event in events if event["kind"] == "owner_message"]
60 assert owner_messages
61 return owner_messages[-1]
62
63
64def test_post_chat_persists_needs_you_source(chat_client):
65 source = {"kind": "needs_you", "item_text": "Review the launch checklist"}
66
67 response = chat_client.post(
68 "/api/chat",
69 json={
70 "message": "let's dig into Review the launch checklist",
71 "app": "home",
72 "path": "/app/home",
73 "facet": "work",
74 "source": source,
75 },
76 )
77
78 assert response.status_code == 200
79 assert _latest_owner_message()["source"] == source
80
81
82def test_post_chat_omits_source_when_absent(chat_client):
83 response = chat_client.post(
84 "/api/chat",
85 json={
86 "message": "hello there",
87 "app": "home",
88 "path": "/app/home",
89 "facet": "work",
90 },
91 )
92
93 assert response.status_code == 200
94 assert "source" not in _latest_owner_message()