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 json
7import logging
8import shutil
9from pathlib import Path
10from typing import Any
11
12import pytest
13
14from solstone.apps.entities import copy as entity_copy
15from solstone.apps.home.connections import _kind_words, build_connections_card
16from solstone.think.indexer import edges as edge_index
17from solstone.think.indexer.edges import insert_edges
18from solstone.think.indexer.journal import get_journal_index
19from tests.helpers.health_glance import healthy_backlog_source
20
21EDGE_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "edges_journal"
22CONTRACT_KIND_WORDS = {
23 "works-with": "works with",
24 "works-at": "works at",
25 "reports-to": "reports to",
26 "family-of": "family",
27 "knows": "knows",
28 "uses": "uses",
29 "created": "created",
30 "decided-with": "decided together",
31 "committed-to": "commitments",
32 "spoke-with": "spoke",
33 "mentioned": "mentions",
34 "messaged-with": "messaged",
35 "scheduled-with": "scheduled",
36 "party-of": "party to",
37 "other": "related",
38}
39
40
41@pytest.fixture
42def edges_journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
43 journal = tmp_path / "edges_journal"
44 shutil.copytree(EDGE_FIXTURE, journal)
45 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal.resolve()))
46 return journal
47
48
49def _insert(journal: Path, rows: list[dict[str, Any]]) -> None:
50 conn, _ = get_journal_index(str(journal))
51 insert_edges(conn, rows)
52 conn.commit()
53 conn.close()
54
55
56def _row(
57 src: str,
58 dst: str,
59 kind: str,
60 path: str,
61 *,
62 day: str = "20260701",
63 weight: int = 1,
64 label: str | None = None,
65 ts: int | None = 0,
66 src_name: str | None = "Alice Edge",
67 dst_name: str | None = None,
68) -> dict[str, Any]:
69 return {
70 "src": src,
71 "dst": dst,
72 "kind": kind,
73 "src_name": src_name,
74 "dst_name": dst_name,
75 "day": day,
76 "facet": "home-test",
77 "source": "home-test",
78 "path": path,
79 "anchor": None,
80 "label": label,
81 "ts": ts,
82 "weight": weight,
83 }
84
85
86def _write_entity(journal: Path, entity_id: str, *, principal: bool = False) -> None:
87 entity_dir = journal / "entities" / entity_id
88 entity_dir.mkdir(parents=True, exist_ok=True)
89 payload: dict[str, Any] = {
90 "id": entity_id,
91 "name": entity_id.replace("_", " ").title(),
92 "type": "Person",
93 "created_at": 0,
94 }
95 if principal:
96 payload["is_principal"] = True
97 (entity_dir / "entity.json").write_text(
98 json.dumps(payload, indent=2) + "\n",
99 encoding="utf-8",
100 )
101
102
103def _minimal_journal(
104 tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, principal: bool
105) -> Path:
106 journal = tmp_path / "journal"
107 (journal / "entities").mkdir(parents=True)
108 _write_entity(journal, "home_self", principal=principal)
109 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal.resolve()))
110 return journal
111
112
113def test_connections_card_populated_trims_loader_payload(edges_journal: Path):
114 _insert(
115 edges_journal,
116 [
117 _row(
118 "edge_alice",
119 "home_juliet",
120 "works-with",
121 "home/juliet-work.jsonl",
122 dst_name="Home Juliet",
123 label="Earlier work",
124 ts=1,
125 ),
126 _row(
127 "edge_alice",
128 "home_juliet",
129 "family-of",
130 "home/juliet-family.jsonl",
131 dst_name="Home Juliet",
132 label="Family tie",
133 ts=2,
134 ),
135 _row(
136 "edge_alice",
137 "home_juliet",
138 "mentioned",
139 "home/juliet-mentioned.jsonl",
140 weight=2,
141 dst_name="Home Juliet",
142 label="Latest mention",
143 ts=3,
144 ),
145 _row(
146 "edge_alice",
147 "home_friar",
148 "attended-with",
149 "home/friar-event.jsonl",
150 dst_name="Home Friar",
151 label="Shared event",
152 ts=4,
153 ),
154 ],
155 )
156
157 payload = build_connections_card()
158
159 assert payload["state"] == "ok"
160 assert payload["total"] == 2
161 assert payload["attendance_kinds"] == sorted(edge_index.ATTENDANCE_KINDS)
162 assert payload["kind_words"] == CONTRACT_KIND_WORDS
163 assert [neighbor["entity_id"] for neighbor in payload["neighbors"]] == [
164 "home_juliet",
165 "home_friar",
166 ]
167 juliet = payload["neighbors"][0]
168 assert set(juliet) == {
169 "entity_id",
170 "name",
171 "evidence_class",
172 "count",
173 "last_seen",
174 "kinds",
175 "latest_label",
176 "latest_kind",
177 "latest_day",
178 }
179 assert juliet["name"] == "Home Juliet"
180 assert juliet["evidence_class"] == "semantic"
181 assert juliet["count"] == 3
182 assert juliet["last_seen"] == "20260701"
183 assert [kind["kind"] for kind in juliet["kinds"]] == [
184 "mentioned",
185 "family-of",
186 "works-with",
187 ]
188 assert juliet["latest_label"] == "Latest mention"
189 assert juliet["latest_kind"] == "mentioned"
190 assert juliet["latest_day"] == "20260701"
191 assert "score" not in juliet
192 assert "directed" not in juliet
193 assert "evidence" not in juliet
194
195
196def test_connections_card_no_principal_is_empty(
197 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
198):
199 _minimal_journal(tmp_path, monkeypatch, principal=False)
200
201 assert build_connections_card() == {"state": "empty"}
202
203
204def test_connections_card_zero_neighbors_is_empty(
205 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
206):
207 journal = _minimal_journal(tmp_path, monkeypatch, principal=True)
208 conn, _ = get_journal_index(str(journal))
209 conn.commit()
210 conn.close()
211
212 assert build_connections_card() == {"state": "empty"}
213
214
215def test_connections_card_loader_error_is_unavailable(
216 tmp_path: Path,
217 monkeypatch: pytest.MonkeyPatch,
218 caplog: pytest.LogCaptureFixture,
219):
220 import solstone.apps.home.connections as connections
221
222 _minimal_journal(tmp_path, monkeypatch, principal=True)
223 monkeypatch.setattr(
224 connections,
225 "load_entity_network",
226 lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("boom")),
227 )
228 caplog.set_level(logging.WARNING, logger="solstone.apps.home.connections")
229
230 assert build_connections_card() == {"state": "unavailable"}
231 assert "home: failed to build connections card" in caplog.text
232
233
234def test_build_pulse_context_survives_missing_connections_index(
235 tmp_path: Path,
236 monkeypatch: pytest.MonkeyPatch,
237):
238 import solstone.apps.home.routes as home_routes
239
240 _minimal_journal(tmp_path, monkeypatch, principal=True)
241 monkeypatch.setattr(
242 home_routes,
243 "get_capture_health",
244 lambda: {"status": "active", "observers": []},
245 )
246 monkeypatch.setattr(home_routes, "get_cached_state", lambda: {})
247 monkeypatch.setattr(home_routes, "get_current", lambda: None)
248 monkeypatch.setattr(home_routes, "_resolve_attention", lambda awareness: None)
249 monkeypatch.setattr(home_routes, "_today", lambda: "20260713")
250 monkeypatch.setattr(home_routes, "_yesterday", lambda: "20260712")
251 monkeypatch.setattr(home_routes, "_count_journal_age_days", lambda today: 8)
252 monkeypatch.setattr(home_routes, "_load_stats", lambda today: {})
253 monkeypatch.setattr(home_routes, "_load_flow_md", lambda today: ("pulse", None))
254 monkeypatch.setattr(
255 home_routes,
256 "_load_pulse_narrative",
257 lambda today: ("pulse", None, []),
258 )
259 monkeypatch.setattr(
260 home_routes, "_collect_anticipated_activities", lambda today: []
261 )
262 monkeypatch.setattr(home_routes, "_collect_activities", lambda today: [])
263 monkeypatch.setattr(home_routes, "_load_latest_weekly_reflection", lambda: None)
264 monkeypatch.setattr(home_routes, "load_briefing", lambda today: None)
265 monkeypatch.setattr(home_routes, "read_steward_health", lambda: None)
266 monkeypatch.setattr(
267 home_routes, "read_steward_summary", lambda *args, **kwargs: None
268 )
269 monkeypatch.setattr(
270 home_routes,
271 "load_backlog_source",
272 lambda _journal_root: healthy_backlog_source(),
273 )
274 monkeypatch.setattr(
275 home_routes,
276 "build_brain_snapshot",
277 lambda *_a, **_k: {"state": "ready"},
278 )
279 monkeypatch.setattr(
280 home_routes,
281 "_summarize_yesterday_processing",
282 lambda yesterday, journal_age_days: None,
283 )
284
285 ctx = home_routes._build_pulse_context()
286
287 assert ctx["connections"] == {"state": "unavailable"}
288 assert ctx["narrative_content"] == "pulse"
289
290
291def test_connections_kind_vocabulary_matches_entities_copy_byte_for_byte():
292 entities_contract = {
293 **entity_copy.ENT_CONN_KIND_WORDS,
294 **entity_copy.ENT_CONN_KIND_CHIP_WORDS,
295 }
296
297 assert _kind_words() == CONTRACT_KIND_WORDS
298 assert {
299 key: entities_contract[key] for key in CONTRACT_KIND_WORDS
300 } == CONTRACT_KIND_WORDS
301 assert sorted(set(entities_contract) - set(CONTRACT_KIND_WORDS)) == [
302 "attended-with",
303 "co-present",
304 ]
305 assert sorted(set(CONTRACT_KIND_WORDS) - set(entities_contract)) == []