solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import datetime as dt
5import json
6import zipfile
7
8from solstone.think.importers.chatgpt import ChatGPTImporter
9from solstone.think.importers.ics import ICSImporter
10from solstone.think.importers.shared import (
11 map_items_to_segments,
12 write_content_manifest,
13)
14from solstone.think.importers.utils import generate_content_manifest
15
16
17def test_write_content_manifest(tmp_path, monkeypatch):
18 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
19 entries = [
20 {
21 "id": "conv-0",
22 "title": "Conversation 1",
23 "date": "20260101",
24 "type": "conversation",
25 "preview": "hello",
26 "segments": [{"day": "20260101", "key": "100000_300"}],
27 }
28 ]
29
30 manifest_path = write_content_manifest("20260101_100000", entries)
31
32 assert (
33 manifest_path
34 == tmp_path / "imports" / "20260101_100000" / "content_manifest.jsonl"
35 )
36 lines = manifest_path.read_text(encoding="utf-8").strip().splitlines()
37 assert [json.loads(line) for line in lines] == entries
38
39
40def test_map_items_to_segments():
41 timestamps = [
42 dt.datetime(2026, 1, 1, 10, 0, 0).timestamp(),
43 dt.datetime(2026, 1, 1, 10, 1, 0).timestamp(),
44 dt.datetime(2026, 1, 1, 10, 10, 0).timestamp(),
45 ]
46
47 assert map_items_to_segments(timestamps, tz=None) == [
48 ("20260101", "100000_300"),
49 ("20260101", "100000_300"),
50 ("20260101", "101000_300"),
51 ]
52
53
54def test_generate_content_manifest_from_segments(tmp_path):
55 journal_root = tmp_path
56 import_dir = journal_root / "imports" / "20260101_090000"
57 segment_dir = journal_root / "chronicle" / "20260101" / "import.ics" / "090000_300"
58 segment_dir.mkdir(parents=True)
59 import_dir.mkdir(parents=True)
60
61 (segment_dir / "event_transcript.md").write_text(
62 "## Event One\n\nBody one.\n\n## Event Two\n\nBody two.\n",
63 encoding="utf-8",
64 )
65 (import_dir / "imported.json").write_text(
66 json.dumps(
67 {
68 "source_type": "ics",
69 "all_created_files": [
70 "20260101/import.ics/090000_300/event_transcript.md",
71 ],
72 }
73 ),
74 encoding="utf-8",
75 )
76
77 manifest_path = generate_content_manifest(journal_root, "20260101_090000")
78
79 assert manifest_path == import_dir / "content_manifest.jsonl"
80 entries = [
81 json.loads(line)
82 for line in manifest_path.read_text(encoding="utf-8").splitlines()
83 if line.strip()
84 ]
85 assert [entry["title"] for entry in entries] == ["Event One", "Event Two"]
86 assert all(entry["type"] == "event" for entry in entries)
87 assert all(
88 entry["segments"] == [{"day": "20260101", "key": "090000_300"}]
89 for entry in entries
90 )
91
92
93def test_chatgpt_importer_writes_content_manifest(tmp_path, monkeypatch):
94 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
95 archive = tmp_path / "chatgpt.zip"
96 conversations = [
97 {
98 "title": "Async help",
99 "create_time": dt.datetime(2026, 1, 1, 10, 0, 0).timestamp(),
100 "current_node": "assistant-node",
101 "mapping": {
102 "user-node": {
103 "parent": None,
104 "message": {
105 "author": {"role": "user"},
106 "content": {"parts": ["How do I debug asyncio cancellation?"]},
107 "create_time": dt.datetime(2026, 1, 1, 10, 0, 0).timestamp(),
108 },
109 },
110 "assistant-node": {
111 "parent": "user-node",
112 "message": {
113 "author": {"role": "assistant"},
114 "content": {"parts": ["Trace CancelledError propagation."]},
115 "create_time": dt.datetime(2026, 1, 1, 10, 1, 0).timestamp(),
116 "metadata": {"model_slug": "gpt-4o"},
117 },
118 },
119 },
120 }
121 ]
122 with zipfile.ZipFile(archive, "w") as zf:
123 zf.writestr("conversations.json", json.dumps(conversations))
124
125 result = ChatGPTImporter().process(
126 archive,
127 tmp_path,
128 import_id="20260101_100000",
129 )
130
131 assert result.entries_written == 2
132 manifest_path = tmp_path / "imports" / "20260101_100000" / "content_manifest.jsonl"
133 assert manifest_path.exists()
134 entries = [
135 json.loads(line)
136 for line in manifest_path.read_text(encoding="utf-8").splitlines()
137 if line.strip()
138 ]
139 assert entries[0]["title"] == "Async help"
140 assert entries[0]["segments"] == [{"day": "20260101", "key": "100000_300"}]
141
142
143def test_ics_importer_writes_content_manifest(tmp_path, monkeypatch):
144 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
145 ics_path = tmp_path / "calendar.ics"
146 ics_path.write_bytes(
147 b"""BEGIN:VCALENDAR
148VERSION:2.0
149BEGIN:VEVENT
150DTSTART:20260101T170000Z
151DTEND:20260101T173000Z
152SUMMARY:Design Review
153DESCRIPTION:Review the roadmap.
154CREATED:20260101T120000Z
155END:VEVENT
156END:VCALENDAR"""
157 )
158
159 result = ICSImporter().process(
160 ics_path,
161 tmp_path,
162 import_id="20260101_090000",
163 )
164
165 assert result.entries_written == 1
166 manifest_path = tmp_path / "imports" / "20260101_090000" / "content_manifest.jsonl"
167 assert manifest_path.exists()
168 entries = [
169 json.loads(line)
170 for line in manifest_path.read_text(encoding="utf-8").splitlines()
171 if line.strip()
172 ]
173 assert entries[0]["title"] == "Design Review"
174 assert entries[0]["type"] == "event"
175 assert entries[0]["segments"] == [{"day": "20260101", "key": "120000_300"}]