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 os
8import shlex
9import socket
10import sqlite3
11import subprocess
12import sys
13from pathlib import Path
14
15import pytest
16
17from tests import verify_indexer_differential as harness
18
19try:
20 from tests._indexer_differential_fixtures import (
21 FULLTEXT_QUERY_CASES,
22 FULLTEXT_TOP10_JACCARD_MIN,
23 METADATA_FILTER_CASES,
24 )
25except ModuleNotFoundError:
26 from _indexer_differential_fixtures import (
27 FULLTEXT_QUERY_CASES,
28 FULLTEXT_TOP10_JACCARD_MIN,
29 METADATA_FILTER_CASES,
30 )
31
32FIXTURE_JOURNAL = Path("tests/fixtures/journal").resolve()
33FUNCTIONAL_PATHS = tuple(f"docs/p{i:02}.md" for i in range(10))
34
35
36def _quote_command(*parts: str | Path) -> str:
37 return " ".join(shlex.quote(str(part)) for part in parts)
38
39
40def _writer_script(tmp_path: Path) -> Path:
41 script = tmp_path / "write_index.py"
42 script.write_text(
43 """
44import os
45import sqlite3
46import sys
47from pathlib import Path
48
49mode = sys.argv[1]
50if mode == "fail":
51 sys.exit(7)
52if mode == "missing":
53 sys.exit(0)
54
55journal = Path(os.environ["SOLSTONE_JOURNAL"])
56db = journal / "indexer" / "journal.sqlite"
57db.parent.mkdir(parents=True, exist_ok=True)
58conn = sqlite3.connect(db)
59conn.execute("CREATE TABLE files(path TEXT PRIMARY KEY, mtime INTEGER)")
60conn.execute(\"\"\"
61CREATE VIRTUAL TABLE chunks USING fts5(
62 content,
63 path UNINDEXED,
64 day UNINDEXED,
65 facet UNINDEXED,
66 agent UNINDEXED,
67 stream UNINDEXED,
68 idx UNINDEXED,
69 time_bucket UNINDEXED
70)
71\"\"\")
72conn.execute("CREATE TABLE edge_files(path TEXT PRIMARY KEY, mtime INTEGER)")
73conn.execute(\"\"\"
74CREATE TABLE edges(
75 src TEXT NOT NULL,
76 dst TEXT NOT NULL,
77 kind TEXT NOT NULL,
78 directed INTEGER NOT NULL,
79 src_name TEXT,
80 dst_name TEXT,
81 day TEXT,
82 facet TEXT,
83 source TEXT NOT NULL,
84 path TEXT NOT NULL,
85 anchor TEXT,
86 label TEXT,
87 ts INTEGER,
88 weight INTEGER NOT NULL
89)
90\"\"\")
91if mode != "empty":
92 mtime = 20 if mode == "mtime2" else 10
93 content = "different token" if mode == "different" else "same token"
94 conn.execute(
95 "INSERT INTO files(path, mtime) VALUES (?, ?)",
96 ("entity_search:__mtime__", mtime),
97 )
98 conn.execute(
99 "INSERT INTO chunks(content, path, day, facet, agent, stream, idx, time_bucket) "
100 "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
101 (content, "source.md", "20260101", "test", "test", "", 0, ""),
102 )
103conn.commit()
104conn.close()
105""".lstrip(),
106 encoding="utf-8",
107 )
108 return script
109
110
111def _command(tmp_path: Path, mode: str) -> str:
112 return _quote_command(sys.executable, _writer_script(tmp_path), mode)
113
114
115def _journal_indexer_command(tmp_path: Path) -> str:
116 script = tmp_path / "run_journal_indexer.py"
117 script.write_text(
118 """
119import subprocess
120import sys
121from pathlib import Path
122
123journal_bin = Path(sys.executable).with_name("journal")
124completed = subprocess.run([str(journal_bin), "indexer", "--rescan-full"], check=False)
125raise SystemExit(completed.returncode)
126""".lstrip(),
127 encoding="utf-8",
128 )
129 return _quote_command(sys.executable, script)
130
131
132def _tree_inventory(root: Path) -> dict[str, tuple[int, int]]:
133 return {
134 path.relative_to(root).as_posix(): (stat.st_size, stat.st_mtime_ns)
135 for path in sorted(root.rglob("*"))
136 for stat in [path.lstat()]
137 }
138
139
140def _create_index_db(path: Path, *, content: str = "same token") -> None:
141 script = _writer_script(path.parent)
142 env = os.environ.copy()
143 journal = path.parent / f"{path.stem}_journal"
144 env["SOLSTONE_JOURNAL"] = str(journal)
145 subprocess.run(
146 [sys.executable, str(script), "same"],
147 env=env,
148 check=True,
149 capture_output=True,
150 text=True,
151 )
152 db = journal / "indexer" / "journal.sqlite"
153 path.parent.mkdir(parents=True, exist_ok=True)
154 path.write_bytes(db.read_bytes())
155 if content != "same token":
156 with sqlite3.connect(path) as conn:
157 conn.execute("DELETE FROM chunks")
158 conn.execute(
159 "INSERT INTO chunks(content, path, day, facet, agent, stream, idx, time_bucket) "
160 "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
161 (content, "source.md", "20260101", "test", "test", "", 0, ""),
162 )
163 conn.commit()
164
165
166def _functional_schema(conn: sqlite3.Connection) -> None:
167 conn.execute("CREATE TABLE files(path TEXT PRIMARY KEY, mtime INTEGER)")
168 conn.execute(
169 """
170 CREATE VIRTUAL TABLE chunks USING fts5(
171 content,
172 path UNINDEXED,
173 day UNINDEXED,
174 facet UNINDEXED,
175 agent UNINDEXED,
176 stream UNINDEXED,
177 idx UNINDEXED,
178 time_bucket UNINDEXED
179 )
180 """
181 )
182 conn.execute("CREATE TABLE edge_files(path TEXT PRIMARY KEY, mtime INTEGER)")
183 conn.execute(
184 """
185 CREATE TABLE edges(
186 src TEXT NOT NULL,
187 dst TEXT NOT NULL,
188 kind TEXT NOT NULL,
189 directed INTEGER NOT NULL,
190 src_name TEXT,
191 dst_name TEXT,
192 day TEXT,
193 facet TEXT,
194 source TEXT NOT NULL,
195 path TEXT NOT NULL,
196 anchor TEXT,
197 label TEXT,
198 ts INTEGER,
199 weight INTEGER NOT NULL
200 )
201 """
202 )
203
204
205def _functional_content(path: str, *, jwt: bool = True, suffix: str = "") -> str:
206 phrase = " JWT token" if jwt else ""
207 return f"authentication module FastAPI{phrase} common {path} {suffix}".strip()
208
209
210def _functional_rows(
211 paths: tuple[str, ...] = FUNCTIONAL_PATHS,
212 *,
213 idx_offset: int = 0,
214 suffix: str = "",
215 stream: str | None = "default",
216) -> list[dict[str, object]]:
217 return [
218 {
219 "content": _functional_content(path, suffix=suffix),
220 "path": path,
221 "day": "20240102",
222 "facet": "work",
223 "agent": "news",
224 "stream": stream,
225 "idx": idx + idx_offset,
226 "time_bucket": "morning",
227 }
228 for idx, path in enumerate(paths)
229 ]
230
231
232def _functional_edges() -> list[dict[str, object]]:
233 return [
234 {
235 "src": "alice",
236 "dst": "bob",
237 "kind": "works-with",
238 "directed": 0,
239 "src_name": "Alice",
240 "dst_name": "Bob",
241 "day": "20240102",
242 "facet": "work",
243 "source": "test",
244 "path": "edges.json",
245 "anchor": "a1",
246 "label": "Alice works with Bob",
247 "ts": 1,
248 "weight": 4,
249 },
250 {
251 "src": "alice",
252 "dst": "project",
253 "kind": "committed-to",
254 "directed": 1,
255 "src_name": "Alice",
256 "dst_name": "Project",
257 "day": "20240102",
258 "facet": "work",
259 "source": "test",
260 "path": "edges.json",
261 "anchor": "a2",
262 "label": "Alice committed to Project",
263 "ts": 2,
264 "weight": 5,
265 },
266 ]
267
268
269def _write_functional_db(
270 path: Path,
271 *,
272 rows: list[dict[str, object]] | None = None,
273 edges: list[dict[str, object]] | None = None,
274 file_paths: tuple[str, ...] = FUNCTIONAL_PATHS,
275) -> None:
276 path.parent.mkdir(parents=True, exist_ok=True)
277 with sqlite3.connect(path) as conn:
278 _functional_schema(conn)
279 for file_path in file_paths:
280 conn.execute(
281 "INSERT INTO files(path, mtime) VALUES (?, ?)",
282 (file_path, 10),
283 )
284 conn.executemany(
285 "INSERT INTO chunks(content, path, day, facet, agent, stream, idx, time_bucket) "
286 "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
287 [
288 (
289 row["content"],
290 row["path"],
291 row["day"],
292 row["facet"],
293 row["agent"],
294 row["stream"],
295 row["idx"],
296 row["time_bucket"],
297 )
298 for row in (rows if rows is not None else _functional_rows())
299 ],
300 )
301 conn.execute(
302 "INSERT INTO edge_files(path, mtime) VALUES (?, ?)",
303 ("edges.json", 10),
304 )
305 conn.executemany(
306 """
307 INSERT INTO edges(
308 src, dst, kind, directed, src_name, dst_name, day, facet,
309 source, path, anchor, label, ts, weight
310 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
311 """,
312 [
313 (
314 edge["src"],
315 edge["dst"],
316 edge["kind"],
317 edge["directed"],
318 edge["src_name"],
319 edge["dst_name"],
320 edge["day"],
321 edge["facet"],
322 edge["source"],
323 edge["path"],
324 edge["anchor"],
325 edge["label"],
326 edge["ts"],
327 edge["weight"],
328 )
329 for edge in (edges if edges is not None else _functional_edges())
330 ],
331 )
332 conn.commit()
333
334
335def _compare_functional_paths(tmp_path: Path, left: Path, right: Path) -> dict:
336 return harness.compare_functional(left, right, tmp_path / "functional")
337
338
339def _fixture_index(tmp_path: Path) -> Path:
340 journal = tmp_path / "indexed-journal"
341 harness.copytree_tracked(FIXTURE_JOURNAL, journal)
342 journal_bin = Path(sys.executable).with_name("journal")
343 env = os.environ.copy()
344 env["SOLSTONE_JOURNAL"] = str(journal)
345 env["SOL_SKIP_SUPERVISOR_CHECK"] = "1"
346 env["SOLSTONE_DISABLE_CONVEY_SIDE_RUNTIMES"] = "1"
347 subprocess.run(
348 [str(journal_bin), "indexer", "--rescan-full"],
349 cwd=harness.ROOT,
350 env=env,
351 capture_output=True,
352 text=True,
353 check=True,
354 )
355 return journal
356
357
358def test_stderr_classifier_allows_standalone_edge_skip_warning() -> None:
359 stderr = "\n".join(
360 [
361 f"{harness.EDGE_SKIP_PREFIX}20240102/default/234567_300/screen.jsonl"
362 ": invalid segment key 234567_300",
363 "unexpected diagnostic",
364 ]
365 )
366
367 classified = harness.classify_stderr(stderr)
368
369 assert classified["rules"][0]["count"] == 1
370 assert classified["unclassified"] == ["unexpected diagnostic"]
371
372
373def test_stderr_classifier_allows_markdown_sanitize_warning() -> None:
374 stderr = "\n".join(
375 [
376 "WARNING:solstone.think.markdown:Dropped 1 line(s) exceeding 2048 chars during markdown sanitization",
377 "WARNING:solstone.think.markdown:Dropped 5 line(s) exceeding 2048 chars during markdown sanitization",
378 ]
379 )
380 classified = harness.classify_stderr(stderr)
381 markdown_rule = next(
382 rule
383 for rule in classified["rules"]
384 if rule["name"] == harness.MARKDOWN_SANITIZE_RULE
385 )
386 assert markdown_rule["count"] == 2
387 assert markdown_rule["examples"] == stderr.splitlines()
388 assert classified["unclassified"] == []
389
390
391def test_stderr_classifier_allows_native_markdown_sanitize_warning() -> None:
392 stderr = "\n".join(
393 [
394 "warning: Dropped 1 line(s) exceeding 2048 chars during markdown sanitization",
395 "warning: Dropped 5 line(s) exceeding 2048 chars during markdown sanitization",
396 ]
397 )
398 classified = harness.classify_stderr(stderr)
399 native_markdown_rule = next(
400 rule
401 for rule in classified["rules"]
402 if rule["name"] == harness.NATIVE_MARKDOWN_SANITIZE_RULE
403 )
404 assert native_markdown_rule["count"] == 2
405 assert native_markdown_rule["examples"] == stderr.splitlines()
406 assert classified["unclassified"] == []
407
408
409def test_stderr_classifier_allows_native_edge_skip_warning() -> None:
410 stderr = "\n".join(
411 [
412 f"{harness.NATIVE_EDGE_SKIP_PREFIX}"
413 "20240102/default/234567_300/screen.jsonl"
414 ": invalid segment key 234567_300",
415 "unexpected diagnostic",
416 ]
417 )
418
419 classified = harness.classify_stderr(stderr)
420 native_edge_rule = next(
421 rule
422 for rule in classified["rules"]
423 if rule["name"] == harness.NATIVE_EDGE_SKIP_RULE
424 )
425
426 assert native_edge_rule["count"] == 1
427 assert classified["unclassified"] == ["unexpected diagnostic"]
428
429
430def test_stderr_classifier_rejects_markdown_near_misses() -> None:
431 stderr = "\n".join(
432 [
433 "ERROR:solstone.think.markdown:Dropped 1 line(s) exceeding 2048 chars during markdown sanitization",
434 "WARNING:solstone.think.other:Dropped 1 line(s) exceeding 2048 chars during markdown sanitization",
435 "WARNING:solstone.think.markdown:Some unrelated warning",
436 "warning: Dropped 1 line(s) exceeding 4096 chars during markdown sanitization",
437 "warning: dropped 1 line(s) exceeding 2048 chars during markdown sanitization",
438 "WARNING:some.other.module:generic warning",
439 ]
440 )
441 classified = harness.classify_stderr(stderr)
442 markdown_rule = next(
443 rule
444 for rule in classified["rules"]
445 if rule["name"] == harness.MARKDOWN_SANITIZE_RULE
446 )
447 assert markdown_rule["count"] == 0
448 assert classified["unclassified"] == stderr.splitlines()
449
450
451def test_stderr_classifier_mixed_run_fails_closed() -> None:
452 mixed = "\n".join(
453 [
454 "WARNING:solstone.think.markdown:Dropped 1 line(s) exceeding 2048 chars during markdown sanitization",
455 "unexpected diagnostic",
456 ]
457 )
458 classified = harness.classify_stderr(mixed)
459 markdown_rule = next(
460 rule
461 for rule in classified["rules"]
462 if rule["name"] == harness.MARKDOWN_SANITIZE_RULE
463 )
464 assert markdown_rule["count"] == 1
465 assert classified["unclassified"] == ["unexpected diagnostic"]
466
467 command = {
468 "id": "left",
469 "exit_code": 0,
470 "stderr_classification": classified,
471 "checks": {
472 "exit": {"status": "ok"},
473 "database": {"status": "ok"},
474 "stderr": {
475 "status": "unclassified" if classified["unclassified"] else "ok"
476 },
477 },
478 }
479 failure = harness._failure_for_commands([command])
480 assert failure == {
481 "class": "stderr_unclassified",
482 "command_id": "left",
483 "count": 1,
484 }
485
486
487def test_runner_prepares_tracked_clean_copies_with_equal_mtimes(tmp_path: Path) -> None:
488 copies = harness._prepare_working_copies(FIXTURE_JOURNAL, tmp_path / "work")
489
490 assert set(copies) == {"left", "right"}
491 assert not (copies["left"] / harness.DB_REL).exists()
492 assert not (copies["right"] / harness.DB_REL).exists()
493 assert harness._mtime_mismatches(copies["left"], copies["right"]) == []
494
495
496def test_runner_sets_journal_env_and_captures_exit_codes(tmp_path: Path) -> None:
497 report = harness.run_differential(
498 journal=FIXTURE_JOURNAL,
499 command_a=_command(tmp_path, "same"),
500 command_b=_command(tmp_path, "same"),
501 work_root=tmp_path / "work",
502 )
503
504 assert report["classification"] == "equal"
505 assert [command["exit_code"] for command in report["commands"]] == [0, 0]
506 assert report["commands"][0]["journal"] != report["commands"][1]["journal"]
507 assert all(
508 command["checks"]["database"]["status"] == "ok"
509 for command in report["commands"]
510 )
511
512
513def test_missing_database_is_failed_not_equal(tmp_path: Path) -> None:
514 report = harness.run_differential(
515 journal=FIXTURE_JOURNAL,
516 command_a=_command(tmp_path, "missing"),
517 command_b=_command(tmp_path, "missing"),
518 work_root=tmp_path / "work",
519 )
520
521 assert report["classification"] == "failed"
522 assert report["failure"]["class"] == "db_missing"
523
524
525def test_empty_database_is_failed_not_equal(tmp_path: Path) -> None:
526 report = harness.run_differential(
527 journal=FIXTURE_JOURNAL,
528 command_a=_command(tmp_path, "empty"),
529 command_b=_command(tmp_path, "empty"),
530 work_root=tmp_path / "work",
531 )
532
533 assert report["classification"] == "failed"
534 assert report["failure"]["class"] == "db_empty"
535
536
537def test_wal_representation_invariance_canonicalizes_equal(tmp_path: Path) -> None:
538 left = tmp_path / "left.sqlite"
539 right = tmp_path / "right.sqlite"
540 _create_index_db(right)
541
542 conn = sqlite3.connect(left)
543 try:
544 conn.execute("PRAGMA journal_mode=WAL")
545 conn.execute("CREATE TABLE files(path TEXT PRIMARY KEY, mtime INTEGER)")
546 conn.execute(
547 """
548 CREATE VIRTUAL TABLE chunks USING fts5(
549 content,
550 path UNINDEXED,
551 day UNINDEXED,
552 facet UNINDEXED,
553 agent UNINDEXED,
554 stream UNINDEXED,
555 idx UNINDEXED,
556 time_bucket UNINDEXED
557 )
558 """
559 )
560 conn.execute("CREATE TABLE edge_files(path TEXT PRIMARY KEY, mtime INTEGER)")
561 conn.execute(
562 """
563 CREATE TABLE edges(
564 src TEXT NOT NULL,
565 dst TEXT NOT NULL,
566 kind TEXT NOT NULL,
567 directed INTEGER NOT NULL,
568 src_name TEXT,
569 dst_name TEXT,
570 day TEXT,
571 facet TEXT,
572 source TEXT NOT NULL,
573 path TEXT NOT NULL,
574 anchor TEXT,
575 label TEXT,
576 ts INTEGER,
577 weight INTEGER NOT NULL
578 )
579 """
580 )
581 conn.execute(
582 "INSERT INTO files(path, mtime) VALUES (?, ?)",
583 ("entity_search:__mtime__", 10),
584 )
585 conn.execute(
586 "INSERT INTO chunks(content, path, day, facet, agent, stream, idx, time_bucket) "
587 "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
588 ("same token", "source.md", "20260101", "test", "test", "", 0, ""),
589 )
590 conn.commit()
591
592 assert left.read_bytes() != right.read_bytes()
593 _normalized, comparison = harness.canonicalize_pair(
594 left, right, tmp_path / "scratch"
595 )
596 finally:
597 conn.close()
598
599 assert comparison["classification"] == "equal"
600
601
602def test_shadow_table_only_change_still_equal(tmp_path: Path) -> None:
603 left = tmp_path / "left.sqlite"
604 right = tmp_path / "right.sqlite"
605 _create_index_db(left)
606 _create_index_db(right)
607 with sqlite3.connect(right) as conn:
608 conn.execute("INSERT INTO chunks(chunks, rank) VALUES('automerge', 2)")
609 conn.commit()
610
611 assert left.read_bytes() != right.read_bytes()
612 _normalized, comparison = harness.canonicalize_pair(
613 left, right, tmp_path / "scratch"
614 )
615
616 assert comparison["classification"] == "equal"
617
618
619def test_seeded_divergence_is_unexpected_differs_and_cli_nonzero(
620 tmp_path: Path,
621 capsys,
622) -> None:
623 exit_code = harness.main(
624 [
625 "--journal",
626 str(FIXTURE_JOURNAL),
627 "--a",
628 _command(tmp_path, "same"),
629 "--b",
630 _command(tmp_path, "different"),
631 "--work-dir",
632 str(tmp_path / "work"),
633 ]
634 )
635
636 report = json.loads(capsys.readouterr().out)
637 assert exit_code != 0
638 assert report["classification"] == "unexpected-differs"
639
640
641def test_mtime_only_divergence_is_functionally_equal(tmp_path: Path) -> None:
642 report = harness.run_differential(
643 journal=FIXTURE_JOURNAL,
644 command_a=_command(tmp_path, "same"),
645 command_b=_command(tmp_path, "mtime2"),
646 work_root=tmp_path / "work",
647 )
648
649 assert report["classification"] == "functionally-equal"
650 assert [rule["name"] for rule in report["normalization"]["rules_fired"]] == [
651 harness.ENTITY_SEARCH_MTIME_RULE
652 ]
653
654
655def test_command_failure_is_distinct_and_cli_nonzero(tmp_path: Path, capsys) -> None:
656 exit_code = harness.main(
657 [
658 "--journal",
659 str(FIXTURE_JOURNAL),
660 "--a",
661 _command(tmp_path, "same"),
662 "--b",
663 _command(tmp_path, "fail"),
664 "--work-dir",
665 str(tmp_path / "work"),
666 ]
667 )
668
669 report = json.loads(capsys.readouterr().out)
670 assert exit_code != 0
671 assert report["classification"] == "failed"
672 assert report["failure"]["class"] == "command_nonzero"
673 assert report["failure"]["command_id"] == "right"
674
675
676def test_fixture_corpus_reports_equal_with_visible_edge_skips(tmp_path: Path) -> None:
677 command = _journal_indexer_command(tmp_path)
678
679 report = harness.run_differential(
680 journal=FIXTURE_JOURNAL,
681 command_a=command,
682 command_b=command,
683 work_root=tmp_path / "work",
684 )
685
686 table_counts = {
687 table["name"]: table["row_counts"]["left"]
688 for table in report["canonical"]["tables"]
689 }
690 skip_counts = [
691 next(
692 rule["count"]
693 for rule in command_report["stderr_classification"]["rules"]
694 if rule["name"] == harness.NATIVE_EDGE_SKIP_RULE
695 )
696 for command_report in report["commands"]
697 ]
698 corpus = report["provenance"]["corpus"]
699
700 assert report["classification"] == "equal"
701 assert report["normalization"]["rules_fired"] == []
702 assert corpus["copy_route"] == "git-archive-head"
703 assert corpus["identity"]["repo_commit"]
704 assert table_counts == {
705 "files": 176,
706 "chunks": 590,
707 "edge_files": 101,
708 "edges": 30,
709 }
710 assert skip_counts == [1, 1]
711
712
713def test_functional_byte_different_but_equivalent(tmp_path: Path) -> None:
714 left = tmp_path / "left.sqlite"
715 right = tmp_path / "right.sqlite"
716 _write_functional_db(left, rows=_functional_rows())
717 _write_functional_db(
718 right,
719 rows=_functional_rows(idx_offset=100, suffix="right"),
720 )
721
722 assert left.read_bytes() != right.read_bytes()
723 comparison = _compare_functional_paths(tmp_path, left, right)
724
725 assert comparison["classification"] == "functionally-equal"
726 assert comparison["functional"]["failed_components"] == []
727
728
729def test_functional_missing_coverage_tuple_reports_missing(tmp_path: Path) -> None:
730 left = tmp_path / "left.sqlite"
731 right = tmp_path / "right.sqlite"
732 rows = [row for row in _functional_rows() if row["path"] != "docs/p09.md"]
733 _write_functional_db(left)
734 _write_functional_db(right, rows=rows)
735
736 comparison = _compare_functional_paths(tmp_path, left, right)
737 coverage = comparison["functional"]["chunk_coverage"]
738
739 assert comparison["classification"] == "unexpected-differs"
740 assert "chunk_coverage" in comparison["functional"]["failed_components"]
741 assert {
742 "path": "docs/p09.md",
743 "day": "20240102",
744 "facet": "work",
745 "agent": "news",
746 "stream": "default",
747 "time_bucket": "morning",
748 } in coverage["missing"]
749
750
751def test_functional_fulltext_overlap_failure_names_query(tmp_path: Path) -> None:
752 paths = tuple(f"docs/p{i:02}.md" for i in range(11))
753 left_rows = _functional_rows(paths)
754 right_rows = _functional_rows(paths)
755 for row in left_rows:
756 if row["path"] == "docs/p10.md":
757 row["content"] = _functional_content(str(row["path"]), jwt=False)
758 for row in right_rows:
759 if row["path"] == "docs/p09.md":
760 row["content"] = _functional_content(str(row["path"]), jwt=False)
761
762 left = tmp_path / "left.sqlite"
763 right = tmp_path / "right.sqlite"
764 _write_functional_db(left, rows=left_rows, file_paths=paths)
765 _write_functional_db(right, rows=right_rows, file_paths=paths)
766
767 comparison = _compare_functional_paths(tmp_path, left, right)
768 cases = comparison["functional"]["fulltext"]["cases"]
769 quoted = next(case for case in cases if case["name"] == "quoted_jwt_token")
770
771 assert comparison["classification"] == "unexpected-differs"
772 assert "fulltext" in comparison["functional"]["failed_components"]
773 assert quoted["passed"] is False
774 assert quoted["jaccard"] < FULLTEXT_TOP10_JACCARD_MIN
775 assert quoted["top3_subset_ok"] == {
776 "left_top3_in_right_top10": True,
777 "right_top3_in_left_top10": True,
778 "both": True,
779 }
780
781
782@pytest.mark.parametrize(
783 "edge_variant",
784 ("dropped", "relabeled", "new_kind"),
785)
786def test_functional_edge_differences_are_unexpected(
787 tmp_path: Path,
788 edge_variant: str,
789) -> None:
790 left_edges = _functional_edges()
791 right_edges = _functional_edges()
792 if edge_variant == "dropped":
793 right_edges = right_edges[1:]
794 elif edge_variant == "relabeled":
795 right_edges[0] = {**right_edges[0], "kind": "knows"}
796 elif edge_variant == "new_kind":
797 right_edges.append({**right_edges[0], "kind": "new-kind", "dst": "carol"})
798
799 left = tmp_path / "left.sqlite"
800 right = tmp_path / "right.sqlite"
801 _write_functional_db(left, edges=left_edges)
802 _write_functional_db(right, edges=right_edges)
803
804 comparison = _compare_functional_paths(tmp_path, left, right)
805
806 assert comparison["classification"] == "unexpected-differs"
807 assert "edges" in comparison["functional"]["failed_components"]
808
809
810def test_functional_metadata_filter_path_set_difference(tmp_path: Path) -> None:
811 left = tmp_path / "left.sqlite"
812 right = tmp_path / "right.sqlite"
813 right_rows = _functional_rows()
814 for row in right_rows:
815 if row["path"] == "docs/p09.md":
816 row["agent"] = "other"
817 _write_functional_db(left)
818 _write_functional_db(right, rows=right_rows)
819
820 comparison = _compare_functional_paths(tmp_path, left, right)
821 filters = comparison["functional"]["metadata_filters"]
822 work_news = next(
823 case for case in filters["cases"] if case["name"] == "work_news_all"
824 )
825
826 assert comparison["classification"] == "unexpected-differs"
827 assert "metadata_filters" in comparison["functional"]["failed_components"]
828 assert "docs/p09.md" in work_news["only_left"]
829
830
831def test_functional_null_empty_coverage_representation_is_equal(
832 tmp_path: Path,
833) -> None:
834 left = tmp_path / "left.sqlite"
835 right = tmp_path / "right.sqlite"
836 _write_functional_db(left, rows=_functional_rows(stream=None))
837 _write_functional_db(right, rows=_functional_rows(stream=""))
838
839 comparison = _compare_functional_paths(tmp_path, left, right)
840
841 assert comparison["classification"] == "functionally-equal"
842 assert comparison["functional"]["failed_components"] == []
843
844
845def test_functional_real_facet_difference_is_unexpected(tmp_path: Path) -> None:
846 left = tmp_path / "left.sqlite"
847 right = tmp_path / "right.sqlite"
848 right_rows = _functional_rows()
849 for row in right_rows:
850 if row["path"] == "docs/p09.md":
851 row["facet"] = "personal"
852 _write_functional_db(left)
853 _write_functional_db(right, rows=right_rows)
854
855 comparison = _compare_functional_paths(tmp_path, left, right)
856
857 assert comparison["classification"] == "unexpected-differs"
858
859
860def test_functional_fixture_cases_are_non_empty_on_reference_index(
861 tmp_path: Path,
862) -> None:
863 journal = _fixture_index(tmp_path)
864 from solstone.think.indexer.journal import search_journal
865
866 for case in (*FULLTEXT_QUERY_CASES, *METADATA_FILTER_CASES):
867 with harness._temporary_solstone_journal(journal):
868 total, _ = search_journal(
869 case["query"],
870 limit=0,
871 **case["filters"],
872 )
873 _, results = search_journal(
874 case["query"],
875 limit=total,
876 **case["filters"],
877 )
878 paths = {result["metadata"]["path"] for result in results}
879 assert total > 0, case["name"]
880 assert paths, case["name"]
881 assert total == case["reference_total"], case["name"]
882 assert len(paths) == case["reference_distinct_paths"], case["name"]
883
884
885def test_fixture_corpus_reports_functionally_equal(tmp_path: Path) -> None:
886 command = _journal_indexer_command(tmp_path)
887
888 report = harness.run_differential(
889 journal=FIXTURE_JOURNAL,
890 command_a=command,
891 command_b=command,
892 work_root=tmp_path / "work",
893 mode="functional",
894 )
895 functional = report["functional"]
896
897 assert report["mode"] == "functional"
898 assert report["classification"] == "functionally-equal"
899 assert functional["failed_components"] == []
900 assert functional["files"]["equal"] is True
901 assert functional["chunk_coverage"]["equal"] is True
902 assert functional["metadata_filters"]["passed"] is True
903 assert functional["fulltext"]["passed"] is True
904 assert functional["edges"]["passed"] is True
905
906
907def test_full_copy_mode_handles_non_git_journal_and_preserves_source_sqlite(
908 tmp_path: Path,
909) -> None:
910 source = tmp_path / "non-git-journal"
911 harness.copytree_tracked(FIXTURE_JOURNAL, source)
912 source_sqlite = source / "imports" / "health-dedupe.sqlite"
913 source_sqlite.parent.mkdir(parents=True, exist_ok=True)
914 source_sqlite.write_bytes(b"source sqlite content")
915
916 copies = harness._prepare_working_copies(
917 source,
918 tmp_path / "prep-work",
919 copy_mode="full",
920 )
921
922 assert not (copies["left"] / harness.DB_REL).exists()
923 assert not (copies["right"] / harness.DB_REL).exists()
924 assert harness._mtime_mismatches(copies["left"], copies["right"]) == []
925 assert (copies["left"] / "imports" / "health-dedupe.sqlite").read_bytes() == (
926 b"source sqlite content"
927 )
928
929 report = harness.run_differential(
930 journal=source,
931 command_a=_command(tmp_path, "same"),
932 command_b=_command(tmp_path, "same"),
933 work_root=tmp_path / "run-work",
934 copy_mode="full",
935 )
936
937 corpus = report["provenance"]["corpus"]
938 assert report["classification"] == "equal"
939 assert corpus["copy_route"] == "copytree-full"
940 assert corpus["copy_mode"] == "full"
941 assert corpus["copy_exclusions"] == list(harness.INDEX_DB_EXCLUSION_RELS)
942 assert (
943 tmp_path / "run-work" / "left" / "journal" / "imports" / "health-dedupe.sqlite"
944 ).read_bytes() == b"source sqlite content"
945
946
947def test_harness_does_not_use_network_or_write_outside_workdir(
948 tmp_path: Path,
949 monkeypatch,
950) -> None:
951 private_repo = tmp_path / "private-repo"
952 private_corpus = private_repo / "tests" / "fixtures" / "journal"
953 harness.copytree_tracked(FIXTURE_JOURNAL, private_corpus)
954 subprocess.run(
955 ["git", "init", "-q"],
956 cwd=private_repo,
957 capture_output=True,
958 text=True,
959 check=True,
960 )
961 subprocess.run(
962 ["git", "add", "."],
963 cwd=private_repo,
964 capture_output=True,
965 text=True,
966 check=True,
967 )
968 before_inventory = _tree_inventory(private_corpus)
969 connect_calls: list[tuple[object, object]] = []
970
971 def fail_connect(self: socket.socket, address: object) -> None:
972 connect_calls.append((self, address))
973 raise AssertionError("network call attempted")
974
975 # This catches in-process harness networking; subprocess networking is out of scope.
976 monkeypatch.setattr(socket.socket, "connect", fail_connect)
977
978 report = harness.run_differential(
979 journal=private_corpus,
980 command_a=_command(tmp_path, "same"),
981 command_b=_command(tmp_path, "same"),
982 work_root=tmp_path / "work",
983 )
984
985 after_inventory = _tree_inventory(private_corpus)
986
987 assert report["classification"] == "equal"
988 assert report["provenance"]["corpus"]["copy_route"] == "git-ls-files-live"
989 assert report["provenance"]["corpus"]["identity"] is None
990 assert connect_calls == []
991 assert before_inventory == after_inventory