solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for journal merge command."""
5
6import json
7import shutil
8from pathlib import Path
9from types import SimpleNamespace
10
11import pytest
12from typer.testing import CliRunner
13
14from solstone.think.call import call_app
15from solstone.think.merge import DecisionLogWriteError, merge_journals
16
17runner = CliRunner()
18
19
20def _write_json(path: Path, payload: dict) -> None:
21 path.parent.mkdir(parents=True, exist_ok=True)
22 path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
23
24
25def _write_jsonl(path: Path, items: list[dict]) -> None:
26 path.parent.mkdir(parents=True, exist_ok=True)
27 path.write_text(
28 "".join(json.dumps(item) + "\n" for item in items),
29 encoding="utf-8",
30 )
31
32
33def _read_json(path: Path) -> dict:
34 return json.loads(path.read_text(encoding="utf-8"))
35
36
37def _read_jsonl(path: Path) -> list[dict]:
38 return [
39 json.loads(line)
40 for line in path.read_text(encoding="utf-8").splitlines()
41 if line.strip()
42 ]
43
44
45def _parse_cli_json(output: str) -> dict:
46 return json.loads(output.strip())
47
48
49def _find_merge_artifact_root(target: Path) -> Path:
50 merge_dir = target.parent / f"{target.name}.merge"
51 runs = sorted(path for path in merge_dir.iterdir() if path.is_dir())
52 assert len(runs) >= 1
53 return runs[-1]
54
55
56def _mock_indexer(monkeypatch):
57 import solstone.think.tools.call as call_module
58
59 calls = []
60
61 def _run(*args, **kwargs):
62 calls.append((args, kwargs))
63 return SimpleNamespace(returncode=0)
64
65 monkeypatch.setattr(call_module.subprocess, "run", _run)
66 return calls
67
68
69@pytest.fixture
70def merge_journals_fixture(tmp_path, monkeypatch):
71 target = tmp_path / "target"
72 source = tmp_path / "source"
73
74 (source / "20260101" / "143022_300").mkdir(parents=True)
75 (source / "20260101" / "143022_300" / "audio.jsonl").write_text(
76 '{"audio": "source-segment"}\n',
77 encoding="utf-8",
78 )
79 (source / "20260101" / "120000_60").mkdir(parents=True)
80 (source / "20260101" / "120000_60" / "audio.jsonl").write_text(
81 '{"audio": "source-existing-segment"}\n',
82 encoding="utf-8",
83 )
84
85 (target / "chronicle" / "20260101" / "120000_60").mkdir(parents=True)
86 (target / "chronicle" / "20260101" / "120000_60" / "audio.jsonl").write_text(
87 '{"audio": "target-existing-segment"}\n',
88 encoding="utf-8",
89 )
90
91 _write_json(
92 source / "entities" / "alice_johnson" / "entity.json",
93 {
94 "id": "alice_johnson",
95 "name": "Alice Johnson",
96 "type": "person",
97 "aka": ["Ali"],
98 "emails": ["alice@example.com"],
99 "is_principal": False,
100 "created_at": 1000,
101 },
102 )
103 _write_json(
104 target / "entities" / "alice_johnson" / "entity.json",
105 {
106 "id": "alice_johnson",
107 "name": "Alice Johnson",
108 "type": "person",
109 "aka": ["AJ"],
110 "emails": ["aj@work.com"],
111 "is_principal": False,
112 "created_at": 500,
113 },
114 )
115
116 _write_json(
117 source / "facets" / "work" / "facet.json",
118 {"title": "Work"},
119 )
120
121 (source / "identity").mkdir(parents=True)
122 (source / "identity" / "partner.md").write_text(
123 "source identity\n", encoding="utf-8"
124 )
125 (source / "config").mkdir(parents=True)
126 (source / "config" / "source-only.json").write_text("{}", encoding="utf-8")
127
128 (source / "imports" / "20260101_120000").mkdir(parents=True)
129 (source / "imports" / "20260101_120000" / "manifest.json").write_text(
130 '{"manifest": "source"}\n',
131 encoding="utf-8",
132 )
133
134 monkeypatch.setenv("SOLSTONE_JOURNAL", str(target))
135 import solstone.think.utils as think_utils
136
137 think_utils._journal_path_cache = None
138 yield {"target": target, "source": source}
139 think_utils._journal_path_cache = None
140
141
142def test_segment_copy(merge_journals_fixture, monkeypatch):
143 paths = merge_journals_fixture
144 _mock_indexer(monkeypatch)
145
146 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
147
148 assert result.exit_code == 0
149 assert (
150 paths["target"] / "chronicle" / "20260101" / "143022_300" / "audio.jsonl"
151 ).exists()
152
153
154def test_segment_skip(merge_journals_fixture, monkeypatch):
155 paths = merge_journals_fixture
156 _mock_indexer(monkeypatch)
157
158 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
159
160 assert result.exit_code == 0
161 assert (
162 paths["target"] / "chronicle" / "20260101" / "120000_60" / "audio.jsonl"
163 ).read_text(encoding="utf-8") == '{"audio": "target-existing-segment"}\n'
164
165
166def test_entity_create(merge_journals_fixture, monkeypatch):
167 paths = merge_journals_fixture
168 _mock_indexer(monkeypatch)
169 _write_json(
170 paths["source"] / "entities" / "bob_smith" / "entity.json",
171 {
172 "id": "bob_smith",
173 "name": "Bob Smith",
174 "type": "person",
175 "created_at": 2000,
176 },
177 )
178
179 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
180
181 assert result.exit_code == 0
182 merged = _read_json(paths["target"] / "entities" / "bob_smith" / "entity.json")
183 assert merged["name"] == "Bob Smith"
184
185
186def test_entity_merge_aka_emails(merge_journals_fixture, monkeypatch):
187 paths = merge_journals_fixture
188 _mock_indexer(monkeypatch)
189
190 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
191
192 assert result.exit_code == 0
193 merged = _read_json(paths["target"] / "entities" / "alice_johnson" / "entity.json")
194 assert merged["name"] == "Alice Johnson"
195 assert merged["type"] == "person"
196 assert merged["aka"] == ["AJ", "Ali"]
197 assert merged["emails"] == ["aj@work.com", "alice@example.com"]
198
199
200def test_entity_principal_dedup(merge_journals_fixture, monkeypatch):
201 paths = merge_journals_fixture
202 _mock_indexer(monkeypatch)
203 _write_json(
204 paths["target"] / "entities" / "jer" / "entity.json",
205 {
206 "id": "jer",
207 "name": "Jer",
208 "type": "person",
209 "is_principal": True,
210 "created_at": 1,
211 },
212 )
213 _write_json(
214 paths["source"] / "entities" / "principal_person" / "entity.json",
215 {
216 "id": "principal_person",
217 "name": "Principal Person",
218 "type": "person",
219 "is_principal": True,
220 "created_at": 2,
221 },
222 )
223
224 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
225
226 assert result.exit_code == 0
227 merged = _read_json(
228 paths["target"] / "entities" / "principal_person" / "entity.json"
229 )
230 assert merged["is_principal"] is False
231
232
233def test_entity_id_collision(merge_journals_fixture, monkeypatch):
234 paths = merge_journals_fixture
235 _mock_indexer(monkeypatch)
236 _write_json(
237 paths["source"] / "entities" / "alice_johnson" / "entity.json",
238 {
239 "id": "alice_johnson",
240 "name": "Alice Cooper",
241 "type": "person",
242 "created_at": 3000,
243 },
244 )
245
246 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
247
248 assert result.exit_code == 0
249 assert not (
250 paths["target"] / "entities" / "alice_johnson_2" / "entity.json"
251 ).exists()
252 artifact_root = _find_merge_artifact_root(paths["target"])
253 staged = _read_json(artifact_root / "staging" / "alice_johnson" / "entity.json")
254 assert staged["id"] == "alice_johnson"
255 assert staged["name"] == "Alice Cooper"
256 assert "staged" in result.output
257
258
259def test_ambiguous_entity_name_stages_instead_of_merging(
260 merge_journals_fixture, monkeypatch
261):
262 from solstone.think.entities import load_ambiguities
263
264 paths = merge_journals_fixture
265 _mock_indexer(monkeypatch)
266 _write_json(
267 paths["target"] / "entities" / "sarah_connor" / "entity.json",
268 {
269 "id": "sarah_connor",
270 "name": "Sarah Connor",
271 "type": "person",
272 "created_at": 1000,
273 },
274 )
275 _write_json(
276 paths["target"] / "entities" / "sarah_lee" / "entity.json",
277 {
278 "id": "sarah_lee",
279 "name": "Sarah Lee",
280 "type": "person",
281 "created_at": 1000,
282 },
283 )
284 _write_json(
285 paths["source"] / "entities" / "source_sarah" / "entity.json",
286 {
287 "id": "source_sarah",
288 "name": "Sarah",
289 "type": "person",
290 "aka": ["S"],
291 "created_at": 3000,
292 },
293 )
294 staging_path = paths["target"].parent / "manual-staging"
295
296 summary = merge_journals(
297 paths["source"],
298 paths["target"],
299 dry_run=False,
300 staging_path=staging_path,
301 )
302
303 assert summary.entities_staged == 1
304 assert (staging_path / "source_sarah" / "entity.json").exists()
305 assert not (paths["target"] / "entities" / "source_sarah").exists()
306 assert (
307 _read_json(paths["target"] / "entities" / "sarah_connor" / "entity.json")[
308 "name"
309 ]
310 == "Sarah Connor"
311 )
312 assert (
313 _read_json(paths["target"] / "entities" / "sarah_lee" / "entity.json")["name"]
314 == "Sarah Lee"
315 )
316 rows = load_ambiguities()
317 assert len(rows) == 1
318 assert rows[0]["normalized_query"] == "sarah"
319
320
321def test_facet_copy_new(merge_journals_fixture, monkeypatch):
322 paths = merge_journals_fixture
323 _mock_indexer(monkeypatch)
324 (paths["source"] / "facets" / "work" / "logs").mkdir(parents=True)
325 (paths["source"] / "facets" / "work" / "logs" / "20260101.jsonl").write_text(
326 '{"log": "copied"}\n',
327 encoding="utf-8",
328 )
329
330 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
331
332 assert result.exit_code == 0
333 assert (paths["target"] / "facets" / "work" / "facet.json").exists()
334 assert (paths["target"] / "facets" / "work" / "logs" / "20260101.jsonl").exists()
335
336
337def test_facet_merge_overlapping(merge_journals_fixture, monkeypatch):
338 paths = merge_journals_fixture
339 _mock_indexer(monkeypatch)
340
341 _write_json(paths["target"] / "facets" / "work" / "facet.json", {"title": "Work"})
342 _write_json(
343 paths["source"]
344 / "facets"
345 / "work"
346 / "entities"
347 / "alice_johnson"
348 / "entity.json",
349 {
350 "entity_id": "alice_johnson",
351 "description": "Source description",
352 "source_only": "keep-me",
353 },
354 )
355 _write_json(
356 paths["target"]
357 / "facets"
358 / "work"
359 / "entities"
360 / "alice_johnson"
361 / "entity.json",
362 {
363 "entity_id": "alice_johnson",
364 "description": "Target description",
365 "target_only": "wins",
366 },
367 )
368 _write_jsonl(
369 paths["source"]
370 / "facets"
371 / "work"
372 / "entities"
373 / "alice_johnson"
374 / "observations.jsonl",
375 [
376 {"content": "Shared fact", "observed_at": 100},
377 {"content": "Source fact", "observed_at": 200},
378 ],
379 )
380 _write_jsonl(
381 paths["target"]
382 / "facets"
383 / "work"
384 / "entities"
385 / "alice_johnson"
386 / "observations.jsonl",
387 [
388 {"content": "Shared fact", "observed_at": 100},
389 {"content": "Target fact", "observed_at": 300},
390 ],
391 )
392 (paths["source"] / "facets" / "work" / "news").mkdir(parents=True)
393 (paths["target"] / "facets" / "work" / "news").mkdir(parents=True)
394 (paths["source"] / "facets" / "work" / "news" / "20260101.md").write_text(
395 "source duplicate\n",
396 encoding="utf-8",
397 )
398 (paths["source"] / "facets" / "work" / "news" / "20260102.md").write_text(
399 "source new\n",
400 encoding="utf-8",
401 )
402 (paths["target"] / "facets" / "work" / "news" / "20260101.md").write_text(
403 "target duplicate\n",
404 encoding="utf-8",
405 )
406 _write_jsonl(
407 paths["source"] / "facets" / "work" / "activities" / "activities.jsonl",
408 [
409 {"id": "coding", "name": "Coding"},
410 {"id": "meeting", "name": "Meeting"},
411 ],
412 )
413 _write_jsonl(
414 paths["target"] / "facets" / "work" / "activities" / "activities.jsonl",
415 [
416 {"id": "coding", "name": "Coding"},
417 {"id": "email", "name": "Email"},
418 ],
419 )
420 _write_jsonl(
421 paths["source"] / "facets" / "work" / "activities" / "20260101.jsonl",
422 [
423 {"id": "coding_100000_300", "activity": "coding"},
424 {"id": "meeting_110000_300", "activity": "meeting"},
425 ],
426 )
427 _write_jsonl(
428 paths["target"] / "facets" / "work" / "activities" / "20260101.jsonl",
429 [
430 {"id": "coding_100000_300", "activity": "coding"},
431 ],
432 )
433 source_output = (
434 paths["source"]
435 / "facets"
436 / "work"
437 / "activities"
438 / "20260101"
439 / "coding_100000_300"
440 )
441 source_output.mkdir(parents=True)
442 (source_output / "session_review.md").write_text(
443 "source review\n",
444 encoding="utf-8",
445 )
446 _write_jsonl(
447 paths["source"] / "facets" / "work" / "logs" / "20260101.jsonl",
448 [{"action": "test_action", "ts": 1000}],
449 )
450 _write_jsonl(
451 paths["source"] / "facets" / "work" / "entities" / "20260101.jsonl",
452 [
453 {"id": "alice_johnson", "type": "Person", "name": "Alice Johnson"},
454 {"id": "bob_smith", "type": "Person", "name": "Bob Smith"},
455 ],
456 )
457 _write_jsonl(
458 paths["target"] / "facets" / "work" / "entities" / "20260101.jsonl",
459 [
460 {"id": "alice_johnson", "type": "Person", "name": "Alice Johnson"},
461 ],
462 )
463 (paths["source"] / "facets" / "work" / "entities.jsonl").write_text(
464 '{"skip": true}\n',
465 encoding="utf-8",
466 )
467
468 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
469
470 assert result.exit_code == 0
471 relationship = _read_json(
472 paths["target"]
473 / "facets"
474 / "work"
475 / "entities"
476 / "alice_johnson"
477 / "entity.json"
478 )
479 assert relationship["description"] == "Target description"
480 assert relationship["target_only"] == "wins"
481 assert relationship["source_only"] == "keep-me"
482
483 observations = _read_jsonl(
484 paths["target"]
485 / "facets"
486 / "work"
487 / "entities"
488 / "alice_johnson"
489 / "observations.jsonl"
490 )
491 assert len(observations) == 3
492 assert {item["content"] for item in observations} == {
493 "Shared fact",
494 "Source fact",
495 "Target fact",
496 }
497
498 assert (paths["target"] / "facets" / "work" / "news" / "20260102.md").read_text(
499 encoding="utf-8"
500 ) == "source new\n"
501 assert (paths["target"] / "facets" / "work" / "news" / "20260101.md").read_text(
502 encoding="utf-8"
503 ) == "target duplicate\n"
504
505 activities_config = _read_jsonl(
506 paths["target"] / "facets" / "work" / "activities" / "activities.jsonl"
507 )
508 config_ids = {item["id"] for item in activities_config}
509 assert config_ids == {"coding", "email", "meeting"}
510
511 activity_records = _read_jsonl(
512 paths["target"] / "facets" / "work" / "activities" / "20260101.jsonl"
513 )
514 record_ids = {item["id"] for item in activity_records}
515 assert record_ids == {"coding_100000_300", "meeting_110000_300"}
516
517 assert (
518 paths["target"]
519 / "facets"
520 / "work"
521 / "activities"
522 / "20260101"
523 / "coding_100000_300"
524 / "session_review.md"
525 ).exists()
526
527 logs = _read_jsonl(paths["target"] / "facets" / "work" / "logs" / "20260101.jsonl")
528 assert any(item.get("action") == "test_action" for item in logs)
529
530 detected = _read_jsonl(
531 paths["target"] / "facets" / "work" / "entities" / "20260101.jsonl"
532 )
533 detected_ids = {item["id"] for item in detected}
534 assert detected_ids == {"alice_johnson", "bob_smith"}
535
536 assert not (paths["target"] / "facets" / "work" / "entities.jsonl").exists()
537
538
539def test_import_copy(merge_journals_fixture, monkeypatch):
540 paths = merge_journals_fixture
541 _mock_indexer(monkeypatch)
542
543 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
544
545 assert result.exit_code == 0
546 assert (paths["target"] / "imports" / "20260101_120000" / "manifest.json").exists()
547
548
549def test_import_skip(merge_journals_fixture, monkeypatch):
550 paths = merge_journals_fixture
551 _mock_indexer(monkeypatch)
552 (paths["target"] / "imports" / "20260101_120000").mkdir(parents=True)
553 (paths["target"] / "imports" / "20260101_120000" / "manifest.json").write_text(
554 '{"manifest": "target"}\n',
555 encoding="utf-8",
556 )
557
558 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
559
560 assert result.exit_code == 0
561 assert (
562 paths["target"] / "imports" / "20260101_120000" / "manifest.json"
563 ).read_text(encoding="utf-8") == '{"manifest": "target"}\n'
564
565
566def test_source_identity_skipped(merge_journals_fixture, monkeypatch):
567 paths = merge_journals_fixture
568 _mock_indexer(monkeypatch)
569
570 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
571
572 assert result.exit_code == 0
573 assert not (paths["target"] / "identity" / "partner.md").exists()
574
575
576def test_source_config_skipped(merge_journals_fixture, monkeypatch):
577 paths = merge_journals_fixture
578 _mock_indexer(monkeypatch)
579
580 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
581
582 assert result.exit_code == 0
583 assert not (paths["target"] / "config" / "source-only.json").exists()
584
585
586def test_dry_run(merge_journals_fixture):
587 paths = merge_journals_fixture
588 _write_json(
589 paths["source"] / "entities" / "bob_smith" / "entity.json",
590 {
591 "id": "bob_smith",
592 "name": "Bob Smith",
593 "type": "person",
594 "created_at": 2000,
595 },
596 )
597
598 result = runner.invoke(
599 call_app,
600 ["journal", "merge", str(paths["source"]), "--dry-run"],
601 )
602
603 assert result.exit_code == 0
604 assert "Would merge:" in result.output
605 assert not (paths["target"] / "chronicle" / "20260101" / "143022_300").exists()
606 assert not (paths["target"] / "entities" / "bob_smith").exists()
607 assert not (paths["target"] / "facets" / "work").exists()
608 assert not (paths["target"] / "imports" / "20260101_120000").exists()
609
610
611def test_invalid_source(tmp_path):
612 result = runner.invoke(call_app, ["journal", "merge", str(tmp_path / "missing")])
613
614 assert result.exit_code == 1
615 assert "is not a directory" in result.output
616
617
618def test_invalid_source_no_days(tmp_path):
619 source = tmp_path / "source"
620 source.mkdir()
621
622 result = runner.invoke(call_app, ["journal", "merge", str(source)])
623
624 assert result.exit_code == 1
625 assert "does not appear to be a journal" in result.output
626
627
628def test_accepts_unpacked_export_root(merge_journals_fixture, monkeypatch):
629 paths = merge_journals_fixture
630 _mock_indexer(monkeypatch)
631
632 chronicle_dir = paths["source"] / "chronicle"
633 chronicle_dir.mkdir()
634 shutil.move(
635 str(paths["source"] / "20260101"),
636 str(chronicle_dir / "20260101"),
637 )
638
639 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
640
641 assert result.exit_code == 0
642 assert (
643 paths["target"] / "chronicle" / "20260101" / "143022_300" / "audio.jsonl"
644 ).exists()
645
646
647def test_error_resilience(merge_journals_fixture, monkeypatch):
648 paths = merge_journals_fixture
649 _mock_indexer(monkeypatch)
650 _write_json(
651 paths["source"] / "entities" / "bob_smith" / "entity.json",
652 {
653 "id": "bob_smith",
654 "name": "Bob Smith",
655 "type": "person",
656 "created_at": 2000,
657 },
658 )
659 (paths["source"] / "20260101" / "150000_60").mkdir(parents=True)
660 (paths["source"] / "20260101" / "150000_60" / "audio.jsonl").write_text(
661 '{"audio": "bad"}\n',
662 encoding="utf-8",
663 )
664
665 import solstone.think.merge as journal_merge_module
666
667 real_copytree = shutil.copytree
668 bad_segment = paths["source"] / "20260101" / "150000_60"
669
670 def failing_copytree(src, dst, *args, **kwargs):
671 if Path(src) == bad_segment:
672 raise OSError("boom")
673 return real_copytree(src, dst, *args, **kwargs)
674
675 monkeypatch.setattr(journal_merge_module.shutil, "copytree", failing_copytree)
676
677 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
678
679 assert result.exit_code == 0
680 assert (
681 paths["target"] / "chronicle" / "20260101" / "143022_300" / "audio.jsonl"
682 ).exists()
683 assert (paths["target"] / "entities" / "bob_smith" / "entity.json").exists()
684 assert "1 errors:" in result.output
685
686
687def test_segment_copy_mid_copy_failure_leaves_no_target_or_temp(
688 merge_journals_fixture, monkeypatch
689):
690 paths = merge_journals_fixture
691
692 import solstone.think.merge as journal_merge_module
693
694 real_copytree = shutil.copytree
695 bad_segment = paths["source"] / "20260101" / "143022_300"
696 target_path = paths["target"] / "chronicle" / "20260101" / "143022_300"
697
698 def failing_copytree(src, dst, *args, **kwargs):
699 if Path(src) == bad_segment:
700 Path(dst).mkdir(parents=True)
701 (Path(dst) / "partial.jsonl").write_text("partial\n", encoding="utf-8")
702 raise OSError("boom")
703 return real_copytree(src, dst, *args, **kwargs)
704
705 monkeypatch.setattr(journal_merge_module.shutil, "copytree", failing_copytree)
706
707 result = merge_journals(paths["source"], paths["target"])
708
709 assert result.segments_errored == 1
710 assert not target_path.exists()
711 assert not any(entry.name.startswith(".") for entry in target_path.parent.iterdir())
712
713 monkeypatch.setattr(journal_merge_module.shutil, "copytree", real_copytree)
714
715 retry = merge_journals(paths["source"], paths["target"])
716
717 assert retry.segments_errored == 0
718 assert (target_path / "audio.jsonl").read_text(encoding="utf-8") == (
719 '{"audio": "source-segment"}\n'
720 )
721
722
723def test_decision_log_written(merge_journals_fixture, monkeypatch):
724 paths = merge_journals_fixture
725 _mock_indexer(monkeypatch)
726
727 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
728
729 assert result.exit_code == 0
730 artifact_root = _find_merge_artifact_root(paths["target"])
731 decision_log = artifact_root / "decisions.jsonl"
732 assert decision_log.exists()
733 entries = _read_jsonl(decision_log)
734 assert entries
735 for entry in entries:
736 assert {"ts", "action", "item_type", "item_id", "reason"} <= set(entry)
737
738
739def test_decision_log_entity_merge_snapshots(merge_journals_fixture, monkeypatch):
740 paths = merge_journals_fixture
741 _mock_indexer(monkeypatch)
742
743 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
744
745 assert result.exit_code == 0
746 artifact_root = _find_merge_artifact_root(paths["target"])
747 entries = _read_jsonl(artifact_root / "decisions.jsonl")
748 entity_merged = next(
749 entry for entry in entries if entry["action"] == "entity_merged"
750 )
751 assert "source" in entity_merged
752 assert "target" in entity_merged
753 assert "fields_changed" in entity_merged
754
755
756def test_entity_staged_count_in_output(merge_journals_fixture, monkeypatch):
757 paths = merge_journals_fixture
758 _mock_indexer(monkeypatch)
759 _write_json(
760 paths["source"] / "entities" / "alice_johnson" / "entity.json",
761 {
762 "id": "alice_johnson",
763 "name": "Alice Cooper",
764 "type": "person",
765 "created_at": 3000,
766 },
767 )
768
769 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
770
771 assert result.exit_code == 0
772 assert "staged" in result.output
773
774
775def test_merge_json_success_envelope(merge_journals_fixture, monkeypatch):
776 paths = merge_journals_fixture
777 _mock_indexer(monkeypatch)
778
779 result = runner.invoke(
780 call_app, ["journal", "merge", str(paths["source"]), "--json"]
781 )
782
783 assert result.exit_code == 0
784 payload = _parse_cli_json(result.output)
785 assert payload["ok"] is True
786 assert payload["code"] == "ok"
787 assert payload["source"] == str(paths["source"].resolve())
788 assert payload["dry_run"] is False
789 assert payload["summary"]["segments_copied"] >= 1
790 assert payload["decision_log"].endswith("decisions.jsonl")
791 assert payload["staging_dir"] is None
792 assert payload["indexer_returncode"] == 0
793 assert result.stderr == ""
794 assert len(result.output.strip().splitlines()) == 1
795
796
797def test_merge_json_validation_error_envelope():
798 result = runner.invoke(
799 call_app, ["journal", "merge", "/tmp/missing-source", "--json"]
800 )
801
802 assert result.exit_code == 1
803 payload = _parse_cli_json(result.stderr)
804 assert payload == {
805 "ok": False,
806 "code": "source-not-a-directory",
807 "message": "Source path is not a directory.",
808 "source": str(Path("/tmp/missing-source").resolve()),
809 "dry_run": False,
810 "details": {},
811 }
812 assert result.stdout == ""
813 assert len(result.stderr.strip().splitlines()) == 1
814
815
816def test_merge_json_partial_errors_still_succeed(merge_journals_fixture, monkeypatch):
817 paths = merge_journals_fixture
818 _mock_indexer(monkeypatch)
819 (paths["source"] / "20260101" / "150000_60").mkdir(parents=True)
820 (paths["source"] / "20260101" / "150000_60" / "audio.jsonl").write_text(
821 '{"audio": "bad"}\n',
822 encoding="utf-8",
823 )
824
825 import solstone.think.merge as journal_merge_module
826
827 real_copytree = shutil.copytree
828 bad_segment = paths["source"] / "20260101" / "150000_60"
829
830 def failing_copytree(src, dst, *args, **kwargs):
831 if Path(src) == bad_segment:
832 raise OSError("boom")
833 return real_copytree(src, dst, *args, **kwargs)
834
835 monkeypatch.setattr(journal_merge_module.shutil, "copytree", failing_copytree)
836
837 result = runner.invoke(
838 call_app, ["journal", "merge", str(paths["source"]), "--json"]
839 )
840
841 assert result.exit_code == 0
842 payload = _parse_cli_json(result.output)
843 assert payload["ok"] is True
844 assert payload["summary"]["errors"] == ["segment 20260101/_default/150000_60: boom"]
845
846
847def test_merge_json_decision_log_failure(merge_journals_fixture, monkeypatch):
848 paths = merge_journals_fixture
849 _mock_indexer(monkeypatch)
850
851 import solstone.think.merge as journal_merge_module
852
853 def failing_log(*args, **kwargs):
854 raise DecisionLogWriteError("boom")
855
856 monkeypatch.setattr(journal_merge_module, "_log_decision", failing_log)
857
858 result = runner.invoke(
859 call_app, ["journal", "merge", str(paths["source"]), "--json"]
860 )
861
862 assert result.exit_code == 1
863 payload = _parse_cli_json(result.stderr)
864 assert payload["code"] == "decision-log-write-failed"
865 assert payload["details"]["exception_type"] == "DecisionLogWriteError"
866
867
868def test_merge_json_merge_engine_error(merge_journals_fixture, monkeypatch):
869 paths = merge_journals_fixture
870 _mock_indexer(monkeypatch)
871
872 import solstone.think.merge as journal_merge_module
873
874 def boom(*args, **kwargs):
875 raise RuntimeError("merge boom")
876
877 monkeypatch.setattr(journal_merge_module, "merge_journals", boom)
878
879 result = runner.invoke(
880 call_app, ["journal", "merge", str(paths["source"]), "--json"]
881 )
882
883 assert result.exit_code == 1
884 payload = _parse_cli_json(result.stderr)
885 assert payload["code"] == "merge-engine-error"
886 assert payload["details"]["exception_type"] == "RuntimeError"
887
888
889def test_merge_progress_callback_reports_all_phases(merge_journals_fixture):
890 paths = merge_journals_fixture
891 events = []
892
893 def progress(phase, completed, total, item_name):
894 events.append((phase, completed, total, item_name))
895
896 summary = merge_journals(
897 paths["source"],
898 paths["target"],
899 dry_run=True,
900 log_path=paths["target"].parent / "dry-run.log",
901 staging_path=paths["target"].parent / "staging",
902 progress=progress,
903 )
904
905 assert summary.segments_copied >= 1
906 assert ("segments", 0, 2, None) in events
907 assert ("segments", 2, 2, None) in events
908 assert ("entities", 0, 1, None) in events
909 assert ("entities", 1, 1, "alice_johnson") in events
910 assert ("facets", 1, 1, "work") in events
911 assert ("imports", 1, 1, "20260101_120000") in events
912
913
914def test_merge_progress_callback_exception_propagates(merge_journals_fixture):
915 paths = merge_journals_fixture
916
917 def progress(phase, completed, total, item_name):
918 if phase == "segments" and completed == 1:
919 raise RuntimeError("stop here")
920
921 with pytest.raises(RuntimeError, match="stop here"):
922 merge_journals(
923 paths["source"],
924 paths["target"],
925 dry_run=True,
926 log_path=paths["target"].parent / "dry-run.log",
927 staging_path=paths["target"].parent / "staging",
928 progress=progress,
929 )
930
931
932def test_merge_no_flag_snapshot_unchanged(merge_journals_fixture, monkeypatch):
933 paths = merge_journals_fixture
934 _mock_indexer(monkeypatch)
935
936 result = runner.invoke(call_app, ["journal", "merge", str(paths["source"])])
937
938 assert result.exit_code == 0
939 assert "\nMerged:\n" in result.output
940 assert "Segments: 1 copied, 1 skipped, 0 errored" in result.output
941 assert "Entities: 0 created, 1 merged, 0 staged, 0 skipped" in result.output
942 assert "Facets: 1 created, 0 merged" in result.output
943 assert "Imports: 1 copied, 0 skipped" in result.output
944 assert "Decision log:" in result.output
945 assert "Index rebuild started." in result.output