solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import glob
5from pathlib import Path
6from textwrap import dedent
7from unittest.mock import patch
8
9import pytest
10
11
12def _write_note(
13 vault_dir: Path,
14 rel_path: str,
15 content: str,
16 mtime: float | None = None,
17) -> Path:
18 """Write a note file to the vault."""
19 path = vault_dir / rel_path
20 path.parent.mkdir(parents=True, exist_ok=True)
21 path.write_text(content, encoding="utf-8")
22 if mtime is not None:
23 import os
24
25 os.utime(path, (mtime, mtime))
26 return path
27
28
29SAMPLE_NOTE = dedent("""\
30 ---
31 tags: [project, alpha]
32 ---
33 # Alpha Project
34
35 This is a note about the [[Alpha Project]].
36 See also [[Bob Smith]] and [[Design Doc]].
37""")
38
39SAMPLE_NOTE_2 = dedent("""\
40 # Daily Note
41
42 Today I worked on [[Beta Launch]].
43""")
44
45UPDATED_NOTE = dedent("""\
46 ---
47 tags: [project, alpha]
48 ---
49 # Alpha Project
50
51 This is an updated note about the [[Alpha Project]].
52 See also [[Bob Smith]], [[Design Doc]], and [[Launch Plan]].
53""")
54
55
56def test_obsidian_sync_protocol_conformance():
57 """ObsidianSyncBackend satisfies SyncableBackend protocol."""
58 from solstone.think.importers.obsidian import ObsidianSyncBackend
59 from solstone.think.importers.sync import SyncableBackend
60
61 assert isinstance(ObsidianSyncBackend(), SyncableBackend)
62
63
64def test_obsidian_sync_registry_discovery():
65 """Registry discovery includes obsidian."""
66 from solstone.think.importers.sync import get_syncable_backends
67
68 backends = get_syncable_backends()
69 assert "obsidian" in [backend.name for backend in backends]
70
71
72def test_obsidian_sync_dry_run(tmp_path):
73 """Dry-run catalogs notes and saves state."""
74 from solstone.think.importers.obsidian import ObsidianSyncBackend
75 from solstone.think.importers.sync import load_sync_state
76
77 vault = tmp_path / "vault"
78 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
79 _write_note(vault, "Daily/2026-03-14.md", SAMPLE_NOTE_2, mtime=1_700_000_600)
80
81 result = ObsidianSyncBackend().sync(tmp_path, source_path=vault, dry_run=True)
82
83 assert result["total"] >= 2
84 assert result["available"] == 2
85 assert result["imported"] == 0
86 assert result["downloaded"] == 0
87
88 state = load_sync_state(tmp_path, "obsidian")
89 assert state is not None
90 assert state["files"]["Projects/Alpha.md"]["status"] == "available"
91 assert state["files"]["Daily/2026-03-14.md"]["status"] == "available"
92
93
94def test_obsidian_sync_import(tmp_path, monkeypatch):
95 """Import mode writes note segments and updates state."""
96 from solstone.think.importers.obsidian import ObsidianSyncBackend
97 from solstone.think.importers.sync import load_sync_state
98
99 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
100 vault = tmp_path / "vault"
101 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
102
103 result = ObsidianSyncBackend().sync(tmp_path, source_path=vault, dry_run=False)
104
105 assert result["downloaded"] >= 1
106 assert result["imported"] >= 1
107
108 segments = glob.glob(
109 str(
110 tmp_path
111 / "chronicle"
112 / "*"
113 / "import.obsidian"
114 / "*"
115 / "note_transcript.md"
116 )
117 )
118 assert len(segments) >= 1
119
120 state = load_sync_state(tmp_path, "obsidian")
121 assert state is not None
122 assert state["files"]["Projects/Alpha.md"]["status"] == "imported"
123
124
125def test_obsidian_sync_edit_creates_new_segments(tmp_path, monkeypatch):
126 """Editing a note creates new segments and preserves old ones."""
127 from solstone.think.importers.obsidian import ObsidianSyncBackend
128 from solstone.think.importers.sync import load_sync_state
129
130 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
131 vault = tmp_path / "vault"
132 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
133
134 backend = ObsidianSyncBackend()
135 first = backend.sync(tmp_path, source_path=vault, dry_run=False)
136 assert first["downloaded"] == 1
137 first_segments = sorted(
138 glob.glob(
139 str(
140 tmp_path
141 / "chronicle"
142 / "*"
143 / "import.obsidian"
144 / "*"
145 / "note_transcript.md"
146 )
147 )
148 )
149 assert len(first_segments) == 1
150
151 _write_note(vault, "Projects/Alpha.md", UPDATED_NOTE, mtime=1_700_000_900)
152 second = backend.sync(tmp_path, source_path=vault, dry_run=False)
153 assert second["downloaded"] == 1
154
155 all_segments = sorted(
156 glob.glob(
157 str(
158 tmp_path
159 / "chronicle"
160 / "*"
161 / "import.obsidian"
162 / "*"
163 / "note_transcript.md"
164 )
165 )
166 )
167 assert len(all_segments) == 2
168 assert first_segments[0] in all_segments
169
170 state = load_sync_state(tmp_path, "obsidian")
171 assert state is not None
172 assert state["files"]["Projects/Alpha.md"]["edit_count"] >= 2
173
174
175def test_obsidian_sync_unchanged_skip(tmp_path, monkeypatch):
176 """Mtime-only changes are skipped when content hash matches."""
177 from solstone.think.importers.obsidian import ObsidianSyncBackend
178
179 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
180 vault = tmp_path / "vault"
181 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
182
183 backend = ObsidianSyncBackend()
184 backend.sync(tmp_path, source_path=vault, dry_run=False)
185 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_300)
186
187 result = backend.sync(tmp_path, source_path=vault, dry_run=True)
188 assert result["available"] == 0
189
190
191def test_obsidian_sync_deleted_note(tmp_path, monkeypatch):
192 """Deleted notes are marked removed in state."""
193 from solstone.think.importers.obsidian import ObsidianSyncBackend
194 from solstone.think.importers.sync import load_sync_state
195
196 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
197 vault = tmp_path / "vault"
198 note = _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
199
200 backend = ObsidianSyncBackend()
201 backend.sync(tmp_path, source_path=vault, dry_run=False)
202 note.unlink()
203 backend.sync(tmp_path, source_path=vault, dry_run=True)
204
205 state = load_sync_state(tmp_path, "obsidian")
206 assert state is not None
207 assert state["files"]["Projects/Alpha.md"]["status"] == "removed"
208
209
210def test_obsidian_sync_force(tmp_path, monkeypatch):
211 """Force re-detects notes by clearing state."""
212 from solstone.think.importers.obsidian import ObsidianSyncBackend
213
214 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
215 vault = tmp_path / "vault"
216 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
217
218 backend = ObsidianSyncBackend()
219 backend.sync(tmp_path, source_path=vault, dry_run=False)
220 result = backend.sync(tmp_path, source_path=vault, dry_run=True, force=True)
221
222 assert result["available"] >= 1
223
224
225def test_obsidian_sync_vault_auto_detection(tmp_path, monkeypatch):
226 """Raises when no vault can be auto-detected."""
227 from solstone.think.importers.obsidian import ObsidianSyncBackend
228
229 home = tmp_path / "home"
230 home.mkdir()
231 monkeypatch.setattr("solstone.think.importers.obsidian.Path.home", lambda: home)
232
233 with pytest.raises(
234 ValueError,
235 match="No Obsidian vault found. Use --path to specify your vault location.",
236 ):
237 ObsidianSyncBackend().sync(tmp_path)
238
239
240def test_obsidian_sync_entity_seeding(tmp_path, monkeypatch):
241 """Wikilinks are converted into Topic entities on import."""
242 from solstone.think.importers.obsidian import ObsidianSyncBackend
243
244 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
245 vault = tmp_path / "vault"
246 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
247
248 captured: list[tuple[str, str, list[dict[str, str]]]] = []
249
250 def _fake_seed_entities(
251 facet: str,
252 day: str,
253 entities: list[dict[str, str]],
254 ) -> list[dict[str, str]]:
255 captured.append((facet, day, entities))
256 return entities
257
258 with patch(
259 "solstone.think.importers.obsidian.seed_entities",
260 side_effect=_fake_seed_entities,
261 ):
262 ObsidianSyncBackend().sync(tmp_path, source_path=vault, dry_run=False)
263
264 assert len(captured) == 1
265 facet, _day, entities = captured[0]
266 assert facet == "import.obsidian"
267 assert entities == [
268 {"name": "Alpha Project", "type": "Topic"},
269 {"name": "Bob Smith", "type": "Topic"},
270 {"name": "Design Doc", "type": "Topic"},
271 ]
272
273
274def test_obsidian_sync_incremental(tmp_path, monkeypatch):
275 """Incremental sync imports only newly added notes."""
276 from solstone.think.importers.obsidian import ObsidianSyncBackend
277
278 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
279 vault = tmp_path / "vault"
280 _write_note(vault, "Projects/Alpha.md", SAMPLE_NOTE, mtime=1_700_000_000)
281
282 backend = ObsidianSyncBackend()
283 first = backend.sync(tmp_path, source_path=vault, dry_run=False)
284 assert first["downloaded"] == 1
285
286 _write_note(vault, "Daily/2026-03-14.md", SAMPLE_NOTE_2, mtime=1_700_000_600)
287 second = backend.sync(tmp_path, source_path=vault, dry_run=False)
288
289 assert second["downloaded"] == 1
290 assert second["available"] == 0
291 assert second["imported"] >= 1
292
293
294def test_infer_entity_type_from_path():
295 """Folder path entity type inference."""
296 from solstone.think.importers.obsidian import infer_entity_type_from_path
297
298 # Direct folder match
299 assert infer_entity_type_from_path("People/Jane Smith.md") == "Person"
300 assert infer_entity_type_from_path("Contacts/John.md") == "Person"
301 assert infer_entity_type_from_path("Projects/Alpha.md") == "Project"
302 assert infer_entity_type_from_path("Companies/Acme.md") == "Organization"
303 assert infer_entity_type_from_path("Organizations/UN.md") == "Organization"
304 assert infer_entity_type_from_path("Places/Paris.md") == "Place"
305 assert infer_entity_type_from_path("Locations/HQ.md") == "Place"
306
307 # Case insensitive
308 assert infer_entity_type_from_path("people/jane.md") == "Person"
309 assert infer_entity_type_from_path("PEOPLE/Jane.md") == "Person"
310
311 # Any depth in path
312 assert infer_entity_type_from_path("00 knowledge/People/Jane Smith.md") == "Person"
313 assert infer_entity_type_from_path("Atlas/References/People/Jane.md") == "Person"
314
315 # Numeric prefix stripping
316 assert infer_entity_type_from_path("00 People/Jane.md") == "Person"
317 assert infer_entity_type_from_path("01 Projects/Alpha.md") == "Project"
318
319 # No match → None
320 assert infer_entity_type_from_path("Notes/random.md") is None
321 assert infer_entity_type_from_path("Daily/2026-03-14.md") is None
322 assert infer_entity_type_from_path("random.md") is None
323
324
325def test_clean_at_prefix():
326 """@ prefix stripping from entity names."""
327 from solstone.think.importers.obsidian import _clean_at_prefix
328
329 assert _clean_at_prefix("@JaneSmith") == ("JaneSmith", True)
330 assert _clean_at_prefix("@ Jane Smith") == ("Jane Smith", True)
331 assert _clean_at_prefix("@") == ("", True)
332 assert _clean_at_prefix("Jane Smith") == ("Jane Smith", False)
333 assert _clean_at_prefix("") == ("", False)
334
335
336def test_build_entity_dicts_precedence():
337 """Entity type inference precedence: @ > folder-path > Topic."""
338 from solstone.think.importers.obsidian import _build_entity_dicts
339
340 # Basic Topic fallback
341 result = _build_entity_dicts({"Design Doc"}, {})
342 assert result == [{"name": "Design Doc", "type": "Topic"}]
343
344 # Folder-path type
345 result = _build_entity_dicts({"Jane Smith"}, {"Jane Smith": "Person"})
346 assert result == [{"name": "Jane Smith", "type": "Person"}]
347
348 # @ prefix → Person
349 result = _build_entity_dicts({"@Jane Smith"}, {})
350 assert result == [{"name": "Jane Smith", "type": "Person"}]
351
352 # @ wins over folder-path
353 result = _build_entity_dicts(
354 {"@Jane Smith"},
355 {"Jane Smith": "Organization"},
356 )
357 assert result == [{"name": "Jane Smith", "type": "Person"}]
358
359 # @ filename added even without wikilink
360 result = _build_entity_dicts(set(), {}, at_filenames={"Jane Smith"})
361 assert result == [{"name": "Jane Smith", "type": "Person"}]
362
363 # Dedup: both @Jane and Jane as wikilinks — @ wins
364 result = _build_entity_dicts({"@Jane Smith", "Jane Smith"}, {})
365 assert result == [{"name": "Jane Smith", "type": "Person"}]
366
367
368def test_obsidian_sync_folder_path_entity_typing(tmp_path, monkeypatch):
369 """Notes in typed folders produce typed entities."""
370 from solstone.think.importers.obsidian import ObsidianSyncBackend
371
372 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
373 vault = tmp_path / "vault"
374
375 _write_note(
376 vault, "People/Jane Smith.md", "# Jane Smith\nA person.", mtime=1_700_000_000
377 )
378 _write_note(
379 vault,
380 "Notes/meeting.md",
381 "# Meeting\nMet with [[Jane Smith]] about [[Design Doc]].",
382 mtime=1_700_000_100,
383 )
384
385 captured: list[tuple[str, str, list[dict[str, str]]]] = []
386
387 def _fake_seed(facet, day, entities):
388 captured.append((facet, day, entities))
389 return entities
390
391 with patch(
392 "solstone.think.importers.obsidian.seed_entities", side_effect=_fake_seed
393 ):
394 ObsidianSyncBackend().sync(tmp_path, source_path=vault, dry_run=False)
395
396 all_entities = {}
397 for _, _, entities in captured:
398 for e in entities:
399 all_entities[e["name"]] = e["type"]
400
401 assert all_entities["Jane Smith"] == "Person"
402 assert all_entities["Design Doc"] == "Topic"
403
404
405def test_obsidian_sync_at_prefix_entity_typing(tmp_path, monkeypatch):
406 """Wikilinks with @ prefix produce Person entities."""
407 from solstone.think.importers.obsidian import ObsidianSyncBackend
408
409 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
410 vault = tmp_path / "vault"
411
412 _write_note(
413 vault,
414 "Notes/meeting.md",
415 "# Meeting\nMet with [[@Bob Jones]] and [[Design Doc]].",
416 mtime=1_700_000_000,
417 )
418
419 captured: list[tuple[str, str, list[dict[str, str]]]] = []
420
421 def _fake_seed(facet, day, entities):
422 captured.append((facet, day, entities))
423 return entities
424
425 with patch(
426 "solstone.think.importers.obsidian.seed_entities", side_effect=_fake_seed
427 ):
428 ObsidianSyncBackend().sync(tmp_path, source_path=vault, dry_run=False)
429
430 all_entities = {}
431 for _, _, entities in captured:
432 for e in entities:
433 all_entities[e["name"]] = e["type"]
434
435 assert all_entities["Bob Jones"] == "Person"
436 assert all_entities["Design Doc"] == "Topic"
437
438
439def test_obsidian_sync_numeric_prefix_folder(tmp_path, monkeypatch):
440 """Numeric-prefixed folder names are matched after stripping."""
441 from solstone.think.importers.obsidian import ObsidianSyncBackend
442
443 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
444 vault = tmp_path / "vault"
445
446 _write_note(vault, "00 People/Jane.md", "# Jane\nA person.", mtime=1_700_000_000)
447 _write_note(
448 vault,
449 "Notes/ref.md",
450 "# Ref\nSee [[Jane]].",
451 mtime=1_700_000_100,
452 )
453
454 captured: list[tuple[str, str, list[dict[str, str]]]] = []
455
456 def _fake_seed(facet, day, entities):
457 captured.append((facet, day, entities))
458 return entities
459
460 with patch(
461 "solstone.think.importers.obsidian.seed_entities", side_effect=_fake_seed
462 ):
463 ObsidianSyncBackend().sync(tmp_path, source_path=vault, dry_run=False)
464
465 all_entities = {}
466 for _, _, entities in captured:
467 for e in entities:
468 all_entities[e["name"]] = e["type"]
469
470 assert all_entities["Jane"] == "Person"
471
472
473def test_obsidian_walk_excludes_templates(tmp_path):
474 """templates/ and _templates/ folders are excluded from file walking."""
475 from solstone.think.importers.obsidian import _walk_md_files
476
477 vault = tmp_path / "vault"
478 _write_note(vault, "Notes/real.md", "# Real note", mtime=1_700_000_000)
479 _write_note(vault, "templates/daily.md", "# Template", mtime=1_700_000_000)
480 _write_note(vault, "_templates/weekly.md", "# Template", mtime=1_700_000_000)
481
482 files = _walk_md_files(vault)
483 rel_paths = [str(f.relative_to(vault)) for f in files]
484
485 assert "Notes/real.md" in rel_paths
486 assert "templates/daily.md" not in rel_paths
487 assert "_templates/weekly.md" not in rel_paths
488
489
490def test_obsidian_walk_excludes_templates_case_insensitive(tmp_path):
491 """Template folder exclusion is case-insensitive."""
492 from solstone.think.importers.obsidian import _walk_md_files
493
494 vault = tmp_path / "vault"
495 _write_note(vault, "Notes/real.md", "# Real note", mtime=1_700_000_000)
496 _write_note(vault, "Templates/daily.md", "# Template", mtime=1_700_000_000)
497 _write_note(vault, "TEMPLATES/weekly.md", "# Template", mtime=1_700_000_000)
498 _write_note(vault, "_Templates/meeting.md", "# Template", mtime=1_700_000_000)
499
500 files = _walk_md_files(vault)
501 rel_paths = [str(f.relative_to(vault)) for f in files]
502
503 assert "Notes/real.md" in rel_paths
504 assert "Templates/daily.md" not in rel_paths
505 assert "TEMPLATES/weekly.md" not in rel_paths
506 assert "_Templates/meeting.md" not in rel_paths
507
508
509def test_obsidian_walk_excludes_hidden_dirs(tmp_path):
510 """Hidden dirs (.obsidian, .trash) are excluded by _is_hidden."""
511 from solstone.think.importers.obsidian import _walk_md_files
512
513 vault = tmp_path / "vault"
514 _write_note(vault, "Notes/real.md", "# Real note", mtime=1_700_000_000)
515 _write_note(vault, ".obsidian/plugins/list.md", "# Plugin", mtime=1_700_000_000)
516 _write_note(vault, ".trash/deleted.md", "# Deleted", mtime=1_700_000_000)
517
518 files = _walk_md_files(vault)
519 rel_paths = [str(f.relative_to(vault)) for f in files]
520
521 assert "Notes/real.md" in rel_paths
522 assert ".obsidian/plugins/list.md" not in rel_paths
523 assert ".trash/deleted.md" not in rel_paths
524
525
526def test_obsidian_walk_excludes_nested_templates(tmp_path):
527 """Templates folders nested inside other folders are also excluded."""
528 from solstone.think.importers.obsidian import _walk_md_files
529
530 vault = tmp_path / "vault"
531 _write_note(vault, "Notes/real.md", "# Real note", mtime=1_700_000_000)
532 _write_note(
533 vault, "Areas/Work/templates/standup.md", "# Template", mtime=1_700_000_000
534 )
535
536 files = _walk_md_files(vault)
537 rel_paths = [str(f.relative_to(vault)) for f in files]
538
539 assert "Notes/real.md" in rel_paths
540 assert "Areas/Work/templates/standup.md" not in rel_paths
541
542
543def test_obsidian_sync_excludes_templates(tmp_path, monkeypatch):
544 """Incremental sync skips template folders."""
545 from solstone.think.importers.obsidian import ObsidianSyncBackend
546 from solstone.think.importers.sync import load_sync_state
547
548 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
549 vault = tmp_path / "vault"
550 _write_note(vault, "Notes/real.md", SAMPLE_NOTE, mtime=1_700_000_000)
551 _write_note(
552 vault, "templates/daily.md", "# Daily Template\n{{date}}", mtime=1_700_000_000
553 )
554 _write_note(
555 vault,
556 "Templates/weekly.md",
557 "# Weekly Template\n{{date}}",
558 mtime=1_700_000_000,
559 )
560
561 result = ObsidianSyncBackend().sync(tmp_path, source_path=vault, dry_run=True)
562
563 assert result["available"] == 1
564
565 state = load_sync_state(tmp_path, "obsidian")
566 assert "Notes/real.md" in state["files"]
567 assert "templates/daily.md" not in state["files"]
568 assert "Templates/weekly.md" not in state["files"]
569
570
571def test_obsidian_backends_cli_flag(capsys, monkeypatch, tmp_path):
572 """sol import --backends lists obsidian."""
573 import sys
574
575 from solstone.think.importers.cli import main
576
577 monkeypatch.setattr(sys, "argv", ["sol import", "--backends"])
578 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal"))
579
580 main()
581 captured = capsys.readouterr()
582 assert "obsidian" in captured.out