solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for the derived edge index layer."""
5
6from __future__ import annotations
7
8import json
9import logging
10import os
11import shutil
12import sqlite3
13from pathlib import Path
14from typing import Any
15
16import pytest
17
18from solstone.think import edge_sources
19from solstone.think.activities import (
20 _relation_label,
21 extract_activity_edges,
22)
23from solstone.think.edge_sources import EdgeContext, get_edge_source
24from solstone.think.formatters import discover_files
25from solstone.think.indexer.edges import (
26 EDGES_SCHEMA_PATH,
27 _extract_file_edges,
28 discover_edge_files,
29 insert_edges,
30 rebuild_edges,
31)
32from solstone.think.indexer.journal import get_journal_index, index_file, scan_journal
33from tests._sqlite_assertions import (
34 CHUNK_COLUMNS,
35 EDGE_COLUMNS,
36 FILE_COLUMNS,
37 edges_content_hash,
38 table_content_hash,
39)
40
41EDGE_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "edges_journal"
42
43
44@pytest.fixture
45def edges_journal(tmp_path, monkeypatch):
46 journal = tmp_path / "edges_journal"
47 shutil.copytree(EDGE_FIXTURE, journal)
48 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal.resolve()))
49 return journal
50
51
52def synthetic_edge_extractor(entries: list[dict], ctx: EdgeContext) -> list[dict]:
53 return [
54 {
55 "src": "edge_ada",
56 "dst": "edge_byron",
57 "kind": "attended-with",
58 "src_name": None,
59 "dst_name": None,
60 "day": ctx.day or None,
61 "facet": ctx.facet,
62 "source": "participation",
63 "path": ctx.path,
64 "anchor": "synthetic",
65 "label": "Synthetic edge-only file",
66 "ts": 0,
67 "weight": 1,
68 }
69 ]
70
71
72def bad_kind_edge_extractor(entries: list[dict], ctx: EdgeContext) -> list[dict]:
73 return [
74 {
75 "src": "edge_ada",
76 "dst": "edge_byron",
77 "kind": "bad-kind",
78 "source": "participation",
79 "path": ctx.path,
80 "weight": 1,
81 }
82 ]
83
84
85def day_rooted_resolving_edge_extractor(
86 entries: list[dict],
87 ctx: EdgeContext,
88) -> list[dict]:
89 assert entries == [{"ok": True}]
90 src = ctx.resolve("Ada Edge")
91 assert src == "edge_ada"
92 ctx.drop()
93 return [
94 {
95 "src": src,
96 "dst": "edge_byron",
97 "kind": "attended-with",
98 "source": "participation",
99 "path": ctx.path,
100 "anchor": "day-rooted",
101 "label": "",
102 "day": ctx.day,
103 "facet": ctx.facet,
104 "ts": 0,
105 "weight": 1,
106 }
107 ]
108
109
110def _conn(journal: Path) -> sqlite3.Connection:
111 conn, _ = get_journal_index(str(journal))
112 conn.row_factory = sqlite3.Row
113 return conn
114
115
116def _edge_rows(conn: sqlite3.Connection, where: str = "") -> list[dict[str, Any]]:
117 sql = f"SELECT {', '.join(EDGE_COLUMNS)} FROM edges"
118 if where:
119 sql += f" WHERE {where}"
120 sql += " ORDER BY source, src, dst, anchor"
121 return [dict(row) for row in conn.execute(sql)]
122
123
124def _source_counts(conn: sqlite3.Connection) -> dict[str, int]:
125 return {
126 source: count
127 for source, count in conn.execute(
128 "SELECT source, count(*) FROM edges GROUP BY source"
129 )
130 }
131
132
133def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
134 path.write_text(
135 "".join(json.dumps(row, separators=(",", ":")) + "\n" for row in rows),
136 encoding="utf-8",
137 )
138
139
140def _write_json(path: Path, payload: dict[str, Any]) -> None:
141 path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
142
143
144def _read_jsonl(path: Path) -> list[dict[str, Any]]:
145 return [
146 json.loads(line)
147 for line in path.read_text(encoding="utf-8").splitlines()
148 if line.strip()
149 ]
150
151
152def _bump_mtime(path: Path) -> None:
153 bumped = path.stat().st_mtime + 2
154 os.utime(path, (bumped, bumped))
155
156
157def _edge_ctx(rel: str, *, facet: str = "edges-story") -> EdgeContext:
158 return EdgeContext(
159 path=rel,
160 day="20260430",
161 facet=facet,
162 resolve=lambda _name: None,
163 drop=lambda: None,
164 )
165
166
167def _valid_insert_edge_row(day: Any, *, path: str = "synthetic") -> dict[str, Any]:
168 return {
169 "src": "edge_ada",
170 "dst": "edge_byron",
171 "kind": "attended-with",
172 "src_name": None,
173 "dst_name": None,
174 "day": day,
175 "facet": "edges-activity",
176 "source": "participation",
177 "path": path,
178 "anchor": "ok",
179 "label": "valid row",
180 "ts": 0,
181 "weight": 1,
182 }
183
184
185def test_relation_label_formats_documented_forms():
186 assert _relation_label("Works together", None) == "Works together"
187 assert _relation_label("", "quoted text") == '"quoted text"'
188 assert (
189 _relation_label("Works together", "quoted text")
190 == 'Works together — "quoted text"'
191 )
192 assert _relation_label(" ", None) == ""
193
194
195def test_activity_story_generalization_preserves_commitment_and_closure_rows_byte_exact(
196 edges_journal,
197):
198 rel = "facets/edges-story/activities/20260430.jsonl"
199 rows = extract_activity_edges(_read_jsonl(edges_journal / rel), _edge_ctx(rel))
200
201 story_rows = [row for row in rows if row["source"] in {"commitment", "closure"}]
202 assert story_rows == [
203 {
204 "src": "edge_mina",
205 "dst": "edge_ravi",
206 "kind": "committed-to",
207 "src_name": None,
208 "dst_name": None,
209 "day": "20260430",
210 "facet": "edges-story",
211 "source": "commitment",
212 "path": rel,
213 "anchor": "story-commitments-1",
214 "label": "Send the proposal",
215 "ts": 1777554000000,
216 "weight": 1,
217 },
218 {
219 "src": "edge_tessa",
220 "dst": "edge_mina",
221 "kind": "committed-to",
222 "src_name": None,
223 "dst_name": None,
224 "day": "20260430",
225 "facet": "edges-story",
226 "source": "closure",
227 "path": rel,
228 "anchor": "story-commitments-1",
229 "label": "Confirm the handoff",
230 "ts": 1777554000000,
231 "weight": 1,
232 },
233 ]
234
235
236def test_activity_relations_and_decisions_emit_expected_rows(edges_journal):
237 rel = "facets/edges-story/activities/20260430.jsonl"
238 record = {
239 "id": "story-relations-1",
240 "title": "Relation session",
241 "created_at": 1777555000000,
242 "relations": [
243 {
244 "from": "Mina Edge",
245 "to": "Ravi Edge",
246 "from_entity_id": "edge_mina",
247 "to_entity_id": "edge_ravi",
248 "kind": "works-with",
249 "note": "Runs planning together",
250 "quote": "Let's pair on this",
251 },
252 {
253 "from": "Tessa Edge",
254 "to": "Tessa Edge",
255 "from_entity_id": "edge_tessa",
256 "to_entity_id": "edge_tessa",
257 "kind": "knows",
258 "note": "self",
259 "quote": None,
260 },
261 ],
262 "decisions": [
263 {
264 "owner": "Mina Edge",
265 "counterparty": "Ravi Edge",
266 "owner_entity_id": "edge_mina",
267 "counterparty_entity_id": "edge_ravi",
268 "action": "Use the stable plan",
269 },
270 {
271 "owner": "Mina Edge",
272 "counterparty": "Mina Edge",
273 "owner_entity_id": "edge_mina",
274 "counterparty_entity_id": "edge_mina",
275 "action": "Skip self",
276 },
277 ],
278 }
279
280 rows = extract_activity_edges([record], _edge_ctx(rel))
281
282 assert rows == [
283 {
284 "src": "edge_mina",
285 "dst": "edge_ravi",
286 "kind": "decided-with",
287 "src_name": None,
288 "dst_name": None,
289 "day": "20260430",
290 "facet": "edges-story",
291 "source": "decision",
292 "path": rel,
293 "anchor": "story-relations-1",
294 "label": "Use the stable plan",
295 "ts": 1777555000000,
296 "weight": 1,
297 },
298 {
299 "src": "edge_mina",
300 "dst": "edge_ravi",
301 "kind": "works-with",
302 "src_name": "Mina Edge",
303 "dst_name": "Ravi Edge",
304 "day": "20260430",
305 "facet": "edges-story",
306 "source": "relation",
307 "path": rel,
308 "anchor": "story-relations-1",
309 "label": 'Runs planning together — "Let\'s pair on this"',
310 "ts": 1777555000000,
311 "weight": 1,
312 },
313 ]
314
315
316def test_activity_unknown_relation_kind_raises_at_insert(edges_journal):
317 rel = "facets/edges-story/activities/20260430.jsonl"
318 rows = extract_activity_edges(
319 [
320 {
321 "id": "story-bad-relation-kind",
322 "created_at": 1777555000000,
323 "relations": [
324 {
325 "from": "Mina Edge",
326 "to": "Ravi Edge",
327 "from_entity_id": "edge_mina",
328 "to_entity_id": "edge_ravi",
329 "kind": "unknown-relation-kind",
330 "note": "Bad kind",
331 "quote": None,
332 }
333 ],
334 }
335 ],
336 _edge_ctx(rel),
337 )
338
339 conn = _conn(edges_journal)
340 with pytest.raises(ValueError, match="Unknown edge kind"):
341 insert_edges(conn, rows)
342 conn.close()
343
344
345def test_day_rooted_context_resolves_journal_entities_and_drop_hook_counts(
346 edges_journal,
347 monkeypatch,
348):
349 monkeypatch.setitem(
350 edge_sources.EDGE_SOURCES,
351 "*/*/*/synthetic.jsonl",
352 ("tests.test_index_edges", "day_rooted_resolving_edge_extractor"),
353 )
354 rel = "20260430/default/090000_300/synthetic.jsonl"
355 path = edges_journal / "chronicle" / rel
356 path.parent.mkdir(parents=True, exist_ok=True)
357 _write_jsonl(path, [{"ok": True}])
358
359 conn = _conn(edges_journal)
360 result = _extract_file_edges(conn, rel, str(path), {})
361
362 assert result.rows_inserted == 1
363 assert result.drops == 1
364 assert not result.failed
365 assert (
366 conn.execute("SELECT src FROM edges WHERE path=?", (rel,)).fetchone()[0]
367 == "edge_ada"
368 )
369 conn.close()
370
371
372def test_discover_edge_files_keeps_structural_sources_with_chronicle_root(
373 edges_journal,
374):
375 files = discover_edge_files(str(edges_journal))
376
377 assert "facets/edges-activity/activities/20260430.jsonl" in files
378 assert "facets/edges-story/activities/20260430.jsonl" in files
379 assert "facets/edges-copresence/entities/20260430.jsonl" in files
380 assert "facets/edges-events/events/20260430.jsonl" in files
381 assert "20260430/default/090000_300/screen.jsonl" in files
382 assert "20260430/default/090000_300/talents/documents.json" in files
383
384
385def test_scan_indexes_edges_and_second_scan_is_zero_delta(edges_journal):
386 assert scan_journal(str(edges_journal), full=True).changed is True
387
388 conn = _conn(edges_journal)
389 schema_version = conn.execute(
390 "SELECT mtime FROM edge_files WHERE path=?", (EDGES_SCHEMA_PATH,)
391 ).fetchone()[0]
392 assert schema_version == 1
393 assert conn.execute("SELECT count(*) FROM edges").fetchone()[0] == 24
394 assert _source_counts(conn) == {
395 "calendar": 3,
396 "co-presence": 2,
397 "closure": 1,
398 "commitment": 1,
399 "decision": 1,
400 "document": 3,
401 "event-legacy": 3,
402 "messaging": 4,
403 "observation": 1,
404 "participation": 3,
405 "relation": 2,
406 }
407
408 participation = _edge_rows(conn, "source='participation'")
409 assert len(participation) == 3
410 assert {row["anchor"] for row in participation} == {"activity-attendees-1"}
411 assert all(
412 row["src_name"] is None and row["dst_name"] is None for row in participation
413 )
414
415 commitment = _edge_rows(conn, "source='commitment'")
416 assert commitment == [
417 {
418 "src": "edge_mina",
419 "dst": "edge_ravi",
420 "kind": "committed-to",
421 "directed": 1,
422 "src_name": None,
423 "dst_name": None,
424 "day": "20260430",
425 "facet": "edges-story",
426 "source": "commitment",
427 "path": "facets/edges-story/activities/20260430.jsonl",
428 "anchor": "story-commitments-1",
429 "label": "Send the proposal",
430 "ts": 1777554000000,
431 "weight": 1,
432 }
433 ]
434 closure = _edge_rows(conn, "source='closure'")
435 assert closure[0]["src"] == "edge_tessa"
436 assert closure[0]["dst"] == "edge_mina"
437 assert closure[0]["directed"] == 1
438 assert closure[0]["src_name"] is None
439 assert closure[0]["dst_name"] is None
440
441 relation = {row["kind"]: row for row in _edge_rows(conn, "source='relation'")}
442 assert set(relation) == {"reports-to", "works-with"}
443 assert relation["works-with"]["label"] == (
444 'Runs planning together — "Let\'s pair on this"'
445 )
446 assert relation["works-with"]["directed"] == 0
447 assert relation["reports-to"]["directed"] == 1
448 decision = _edge_rows(conn, "source='decision'")
449 assert len(decision) == 1
450 assert decision[0]["kind"] == "decided-with"
451 assert decision[0]["label"] == "Use the stable plan together"
452
453 copresence = {
454 (row["src"], row["dst"]): row
455 for row in _edge_rows(conn, "source='co-presence'")
456 }
457 assert copresence[("edge_alice", "edge_bob")]["weight"] == 3
458 assert (
459 copresence[("edge_alice", "edge_bob")]["anchor"]
460 == "20260430/default/090000_300"
461 )
462 assert copresence[("edge_alice", "edge_bob")]["src_name"] == "Alice Edge"
463 assert copresence[("edge_alice", "edge_bob")]["dst_name"] == "Bob Edge"
464 assert copresence[("edge_alice", "edge_cora")]["weight"] == 1
465 assert copresence[("edge_alice", "edge_cora")]["src_name"] == "Alice Edge"
466
467 events = _edge_rows(conn, "source='event-legacy'")
468 assert len(events) == 3
469 assert all(row["src_name"] and row["dst_name"] for row in events)
470 assert {row["label"] for row in events} == {"Edge Legacy Meetup"}
471 assert all(row["ts"] > 0 for row in events)
472
473 assert conn.execute(
474 "SELECT 1 FROM edges WHERE src='edge_alice' OR dst='edge_alice' LIMIT 1"
475 ).fetchone()
476
477 first_hash = edges_content_hash(conn)
478 conn.close()
479
480 assert scan_journal(str(edges_journal), full=True).changed is False
481 conn = _conn(edges_journal)
482 assert edges_content_hash(conn) == first_hash
483 conn.close()
484
485
486def test_touching_one_edge_file_replaces_only_that_path(edges_journal):
487 scan_journal(str(edges_journal), full=True)
488 conn = _conn(edges_journal)
489 before = _source_counts(conn)
490 conn.close()
491
492 path = edges_journal / "facets" / "edges-activity" / "activities" / "20260430.jsonl"
493 records = _read_jsonl(path)
494 records[0]["participation"][4]["role"] = "attendee"
495 _write_jsonl(path, records)
496 _bump_mtime(path)
497
498 assert scan_journal(str(edges_journal), full=True).changed is True
499 conn = _conn(edges_journal)
500 after = _source_counts(conn)
501 assert after["participation"] == 6
502 assert {k: v for k, v in after.items() if k != "participation"} == {
503 k: v for k, v in before.items() if k != "participation"
504 }
505 assert conn.execute(
506 "SELECT 1 FROM edge_files WHERE path=?",
507 ("facets/edges-activity/activities/20260430.jsonl",),
508 ).fetchone()
509 conn.close()
510
511
512def test_deleted_edge_source_removes_rows_and_ledger(edges_journal):
513 scan_journal(str(edges_journal), full=True)
514 path = edges_journal / "facets" / "edges-events" / "events" / "20260430.jsonl"
515 path.unlink()
516
517 assert scan_journal(str(edges_journal), full=True).changed is True
518 conn = _conn(edges_journal)
519 assert (
520 conn.execute(
521 "SELECT count(*) FROM edges WHERE source='event-legacy'"
522 ).fetchone()[0]
523 == 0
524 )
525 assert (
526 conn.execute(
527 "SELECT 1 FROM edge_files WHERE path=?",
528 ("facets/edges-events/events/20260430.jsonl",),
529 ).fetchone()
530 is None
531 )
532 conn.close()
533
534
535def test_observation_relations_emit_edges_and_non_relations_still_index(
536 edges_journal,
537):
538 rel = "facets/edges-observations/entities/edge_mina/observations.jsonl"
539 extractor = get_edge_source("facets/work/entities/acme/observations.jsonl")
540 assert extractor is not None
541 assert extractor.__name__ == "extract_observation_edges"
542
543 conn = _conn(edges_journal)
544 result = _extract_file_edges(conn, rel, str(edges_journal / rel), {})
545 assert result.rows_inserted == 1
546 assert result.drops == 1
547 row = conn.execute(
548 """
549 SELECT src, dst, kind, day, facet, source, anchor, label
550 FROM edges
551 WHERE path=?
552 """,
553 (rel,),
554 ).fetchone()
555 assert dict(row) == {
556 "src": "edge_mina",
557 "dst": "edge_ravi",
558 "kind": "works-with",
559 "day": "20260430",
560 "facet": "edges-observations",
561 "source": "observation",
562 "anchor": "1777556000000",
563 "label": "Plans edge enrichment together",
564 }
565 conn.close()
566
567 assert index_file(str(edges_journal), rel) is True
568
569 conn = _conn(edges_journal)
570 assert (
571 conn.execute("SELECT count(*) FROM chunks WHERE path=?", (rel,)).fetchone()[0]
572 > 0
573 )
574 assert (
575 conn.execute("SELECT count(*) FROM edges WHERE path=?", (rel,)).fetchone()[0]
576 == 1
577 )
578 assert (
579 conn.execute(
580 "SELECT count(*) FROM chunks WHERE path=? AND chunks MATCH ?",
581 (rel, '"non relation observation"'),
582 ).fetchone()[0]
583 > 0
584 )
585 conn.close()
586
587
588def test_screen_sources_emit_messaging_calendar_and_event_day(edges_journal):
589 rel = "20260430/default/090000_300/screen.jsonl"
590 named_rel = "20260430/default/090000_300/left_screen.jsonl"
591
592 assert index_file(str(edges_journal), rel) is True
593 assert index_file(str(edges_journal), named_rel) is True
594
595 conn = _conn(edges_journal)
596 screen_rows = _edge_rows(conn, f"path='{rel}'")
597 named_rows = _edge_rows(conn, f"path='{named_rel}'")
598 conn.close()
599
600 messaging = [row for row in screen_rows if row["source"] == "messaging"]
601 calendar = [row for row in screen_rows if row["source"] == "calendar"]
602 assert len(messaging) == 3
603 assert {row["kind"] for row in messaging} == {"messaged-with"}
604 assert {row["label"] for row in messaging} == {"Edge Planning"}
605 assert {row["weight"] for row in messaging} == {2}
606
607 assert len(calendar) == 3
608 assert {row["kind"] for row in calendar} == {"scheduled-with"}
609 assert {row["day"] for row in calendar} == {"20260501"}
610 assert {row["label"] for row in calendar} == {"Future Edge Review"}
611
612 assert len(named_rows) == 1
613 assert named_rows[0]["kind"] == "messaged-with"
614 assert named_rows[0]["source"] == "messaging"
615 assert named_rows[0]["label"] == "Doc Review"
616 assert named_rows[0]["weight"] == 2
617
618
619def test_pretty_documents_json_edges_via_index_file_and_scan_journal(edges_journal):
620 rel = "20260430/default/090000_300/talents/documents.json"
621
622 conn = _conn(edges_journal)
623 result = _extract_file_edges(conn, rel, str(edges_journal / "chronicle" / rel), {})
624 assert result.rows_inserted == 3
625 assert result.drops == 1
626 conn.close()
627
628 assert index_file(str(edges_journal), rel) is True
629 conn = _conn(edges_journal)
630 assert (
631 conn.execute(
632 "SELECT count(*) FROM edges WHERE path=? AND source='document'",
633 (rel,),
634 ).fetchone()[0]
635 == 3
636 )
637 assert (
638 conn.execute("SELECT count(*) FROM chunks WHERE path=?", (rel,)).fetchone()[0]
639 > 0
640 )
641 conn.close()
642
643 assert scan_journal(str(edges_journal), full=True).changed is True
644 conn = _conn(edges_journal)
645 assert (
646 conn.execute(
647 "SELECT count(*) FROM edges WHERE path=? AND source='document'",
648 (rel,),
649 ).fetchone()[0]
650 == 3
651 )
652 conn.close()
653
654
655def _replace_new_source_fixture(shape: str, path: Path) -> None:
656 if shape == "observations":
657 _write_jsonl(
658 path,
659 [
660 {
661 "content": "Mina works with Ravi after replacement.",
662 "observed_at": 1777557000000,
663 "source_day": "20260430",
664 "relation": {
665 "kind": "works-with",
666 "target_entity_id": "edge_ravi",
667 "target_name": "Ravi Edge",
668 "note": "Replacement relation one",
669 },
670 },
671 {
672 "content": "Mina knows Tessa after replacement.",
673 "observed_at": 1777557100000,
674 "source_day": "20260430",
675 "relation": {
676 "kind": "knows",
677 "target_entity_id": "edge_tessa",
678 "target_name": "Tessa Edge",
679 "note": "Replacement relation two",
680 },
681 },
682 ],
683 )
684 return
685
686 if shape == "screen":
687 _write_jsonl(
688 path,
689 [
690 {"raw": "screen.png", "model": "fixture"},
691 {
692 "timestamp": 0,
693 "content": {
694 "messaging": {
695 "view": "conversation",
696 "app": "Signal",
697 "thread": "Replacement Thread",
698 "messages": [
699 {
700 "sender": "Alice Edge",
701 "timestamp": "2026-04-30T09:10:00Z",
702 "subject": "",
703 "text": "Replacement message",
704 },
705 {
706 "sender": "Bob Edge",
707 "timestamp": "2026-04-30T09:10:30Z",
708 "subject": "",
709 "text": "Replacement reply",
710 },
711 ],
712 }
713 },
714 },
715 ],
716 )
717 return
718
719 if shape == "named_screen":
720 _write_jsonl(
721 path,
722 [
723 {"raw": "left_screen.png", "model": "fixture"},
724 {
725 "timestamp": 0,
726 "content": {
727 "calendar": {
728 "view": "week",
729 "app": "Calendar",
730 "events": [
731 {
732 "title": "Replacement Calendar",
733 "start": "20260430T093000",
734 "end": "20260430T100000",
735 "calendar": "Work",
736 "guests": [
737 "Mina Edge",
738 "Ravi Edge",
739 "Tessa Edge",
740 ],
741 }
742 ],
743 }
744 },
745 },
746 ],
747 )
748 return
749
750 if shape == "documents":
751 _write_json(
752 path,
753 {
754 "overview": "Replacement document.",
755 "parties": [
756 {"name": "Mina Edge", "role": "author"},
757 {"name": "Ravi Edge", "role": "reviewer"},
758 ],
759 "key_provisions": [],
760 "assets": [],
761 "conditions": [],
762 "important_dates": [],
763 "summary": "Replacement parties.",
764 },
765 )
766 return
767
768 raise AssertionError(f"unknown source fixture shape: {shape}")
769
770
771@pytest.mark.parametrize(
772 ("shape", "rel", "initial_rows", "replacement_rows"),
773 [
774 (
775 "observations",
776 "facets/edges-observations/entities/edge_mina/observations.jsonl",
777 1,
778 2,
779 ),
780 ("screen", "20260430/default/090000_300/screen.jsonl", 6, 1),
781 ("named_screen", "20260430/default/090000_300/left_screen.jsonl", 1, 3),
782 (
783 "documents",
784 "20260430/default/090000_300/talents/documents.json",
785 3,
786 1,
787 ),
788 ],
789)
790def test_new_source_shapes_support_index_file_replacement_and_deletion(
791 edges_journal,
792 shape,
793 rel,
794 initial_rows,
795 replacement_rows,
796):
797 assert index_file(str(edges_journal), rel) is True
798 conn = _conn(edges_journal)
799 assert (
800 conn.execute("SELECT count(*) FROM edges WHERE path=?", (rel,)).fetchone()[0]
801 == initial_rows
802 )
803 conn.close()
804
805 path = edges_journal / ("chronicle" if rel.startswith("202") else "") / rel
806 _replace_new_source_fixture(shape, path)
807 _bump_mtime(path)
808
809 assert scan_journal(str(edges_journal), full=True).changed is True
810 conn = _conn(edges_journal)
811 assert (
812 conn.execute("SELECT count(*) FROM edges WHERE path=?", (rel,)).fetchone()[0]
813 == replacement_rows
814 )
815 conn.close()
816
817 path.unlink()
818 assert scan_journal(str(edges_journal), full=True).changed is True
819 conn = _conn(edges_journal)
820 assert (
821 conn.execute("SELECT count(*) FROM edges WHERE path=?", (rel,)).fetchone()[0]
822 == 0
823 )
824 assert (
825 conn.execute("SELECT 1 FROM edge_files WHERE path=?", (rel,)).fetchone() is None
826 )
827 conn.close()
828
829
830def test_malformed_new_source_fails_without_suppressing_sibling(
831 edges_journal,
832 caplog,
833):
834 bad_rel = "20260430/default/090000_300/talents/documents.json"
835 bad_path = edges_journal / "chronicle" / bad_rel
836 bad_path.write_text("{\n", encoding="utf-8")
837
838 caplog.set_level(logging.ERROR, logger="solstone.think.indexer.edges")
839 conn = _conn(edges_journal)
840 result = _extract_file_edges(conn, bad_rel, str(bad_path), {})
841 assert result.failed
842 assert result.rows_inserted == 0
843 conn.close()
844 assert f"Skipping edge extraction for {bad_rel}" in caplog.text
845
846 caplog.clear()
847 caplog.set_level(logging.ERROR, logger="solstone.think.indexer.edges")
848 assert scan_journal(str(edges_journal), full=True).changed is True
849 assert f"Skipping edge extraction for {bad_rel}" in caplog.text
850
851 conn = _conn(edges_journal)
852 assert (
853 conn.execute("SELECT count(*) FROM edges WHERE path=?", (bad_rel,)).fetchone()[
854 0
855 ]
856 == 0
857 )
858 assert (
859 conn.execute(
860 "SELECT count(*) FROM edges WHERE path=?",
861 ("20260430/default/090000_300/screen.jsonl",),
862 ).fetchone()[0]
863 == 6
864 )
865 assert (
866 conn.execute("SELECT 1 FROM edge_files WHERE path=?", (bad_rel,)).fetchone()
867 is not None
868 )
869 conn.close()
870
871
872def test_invalid_segment_dir_skips_at_warning_not_error(edges_journal, caplog):
873 rel = "20260430/default/234567_300/screen.jsonl"
874 path = edges_journal / "chronicle" / rel
875 path.parent.mkdir(parents=True, exist_ok=True)
876 _write_jsonl(path, [{"ok": True}])
877
878 caplog.set_level(logging.WARNING, logger="solstone.think.indexer.edges")
879 conn = _conn(edges_journal)
880 result = _extract_file_edges(conn, rel, str(path), {})
881 conn.close()
882
883 assert result.invalid_segment is True
884 assert result.failed is False
885 assert result.rows_inserted == 0
886 assert f"Skipping edge extraction for {rel}" in caplog.text
887 assert not [record for record in caplog.records if record.levelno >= logging.ERROR]
888
889
890def test_index_file_supports_edge_only_file(edges_journal, monkeypatch):
891 monkeypatch.setitem(
892 edge_sources.EDGE_SOURCES,
893 "edge-only/*.jsonl",
894 ("tests.test_index_edges", "synthetic_edge_extractor"),
895 )
896 rel = "edge-only/sample.jsonl"
897 path = edges_journal / rel
898 path.parent.mkdir(parents=True)
899 _write_jsonl(path, [{"ok": True}])
900
901 assert index_file(str(edges_journal), rel) is True
902
903 conn = _conn(edges_journal)
904 assert (
905 conn.execute("SELECT count(*) FROM edges WHERE path=?", (rel,)).fetchone()[0]
906 == 1
907 )
908 assert (
909 conn.execute("SELECT count(*) FROM chunks WHERE path=?", (rel,)).fetchone()[0]
910 == 0
911 )
912 assert (
913 conn.execute("SELECT count(*) FROM files WHERE path=?", (rel,)).fetchone()[0]
914 == 0
915 )
916 assert (
917 conn.execute("SELECT count(*) FROM edge_files WHERE path=?", (rel,)).fetchone()[
918 0
919 ]
920 == 1
921 )
922 conn.close()
923
924
925def test_edge_driver_catches_file_boundary_failures(
926 edges_journal,
927 monkeypatch,
928 caplog,
929):
930 monkeypatch.setitem(
931 edge_sources.EDGE_SOURCES,
932 "bad-edge/*.jsonl",
933 ("tests.test_index_edges", "bad_kind_edge_extractor"),
934 )
935 rel = "bad-edge/source.jsonl"
936 path = edges_journal / rel
937 path.parent.mkdir(parents=True)
938 _write_jsonl(path, [{"ok": True}])
939
940 caplog.set_level(logging.ERROR, logger="solstone.think.indexer.edges")
941 assert index_file(str(edges_journal), rel) is True
942
943 conn = _conn(edges_journal)
944 assert (
945 conn.execute("SELECT count(*) FROM edges WHERE path=?", (rel,)).fetchone()[0]
946 == 0
947 )
948 assert (
949 conn.execute("SELECT count(*) FROM edge_files WHERE path=?", (rel,)).fetchone()[
950 0
951 ]
952 == 1
953 )
954 conn.close()
955 assert f"Skipping edge extraction for {rel}" in caplog.text
956
957
958def test_rescan_file_indexes_activity_edges(edges_journal):
959 rel = "facets/edges-activity/activities/20260430.jsonl"
960 assert index_file(str(edges_journal), rel) is True
961
962 conn = _conn(edges_journal)
963 assert (
964 conn.execute(
965 "SELECT count(*) FROM edges WHERE path=? AND source='participation'",
966 (rel,),
967 ).fetchone()[0]
968 == 3
969 )
970 assert (
971 conn.execute("SELECT count(*) FROM chunks WHERE path=?", (rel,)).fetchone()[0]
972 > 0
973 )
974 conn.close()
975
976
977def test_non_date_activity_edge_source_fails_without_poisoning_sibling(
978 edges_journal,
979 caplog,
980):
981 bad_rel = "facets/edges-story/activities/notaday.jsonl"
982 bad_path = edges_journal / bad_rel
983 _write_jsonl(
984 bad_path,
985 [
986 {
987 "id": "invalid-day-activity",
988 "title": "Invalid day source",
989 "created_at": 1777555000000,
990 "participation": [
991 {"entity_id": "edge_ada", "role": "attendee"},
992 {"entity_id": "edge_byron", "role": "attendee"},
993 ],
994 }
995 ],
996 )
997
998 caplog.set_level(logging.ERROR, logger="solstone.think.indexer.edges")
999 conn = _conn(edges_journal)
1000 result = _extract_file_edges(conn, bad_rel, str(bad_path), {})
1001
1002 assert result.failed
1003 assert result.rows_inserted == 0
1004 assert result.drops == 0
1005 assert (
1006 conn.execute("SELECT count(*) FROM edges WHERE path=?", (bad_rel,)).fetchone()[
1007 0
1008 ]
1009 == 0
1010 )
1011 conn.close()
1012 assert f"Skipping edge extraction for {bad_rel}" in caplog.text
1013
1014 caplog.clear()
1015 assert scan_journal(str(edges_journal), full=True).changed is True
1016
1017 conn = _conn(edges_journal)
1018 assert (
1019 conn.execute("SELECT count(*) FROM edges WHERE path=?", (bad_rel,)).fetchone()[
1020 0
1021 ]
1022 == 0
1023 )
1024 assert (
1025 conn.execute(
1026 "SELECT count(*) FROM edges WHERE path=?",
1027 ("facets/edges-story/activities/20260430.jsonl",),
1028 ).fetchone()[0]
1029 > 0
1030 )
1031 conn.close()
1032 assert f"Skipping edge extraction for {bad_rel}" in caplog.text
1033
1034
1035def test_rebuild_edges_is_idempotent_and_preserves_chunks_files(edges_journal):
1036 scan_journal(str(edges_journal), full=True)
1037 conn = _conn(edges_journal)
1038 chunks_hash = table_content_hash(conn, "chunks", CHUNK_COLUMNS)
1039 files_hash = table_content_hash(conn, "files", FILE_COLUMNS)
1040 assert (
1041 conn.execute(
1042 "SELECT count(*) FROM chunks WHERE path=?",
1043 ("20260430/default/090000_300/talents/documents.json",),
1044 ).fetchone()[0]
1045 > 0
1046 )
1047 conn.close()
1048
1049 first = rebuild_edges(str(edges_journal))
1050 conn = _conn(edges_journal)
1051 first_edges_hash = edges_content_hash(conn)
1052 assert table_content_hash(conn, "chunks", CHUNK_COLUMNS) == chunks_hash
1053 assert table_content_hash(conn, "files", FILE_COLUMNS) == files_hash
1054 conn.close()
1055
1056 second = rebuild_edges(str(edges_journal))
1057 conn = _conn(edges_journal)
1058 assert edges_content_hash(conn) == first_edges_hash
1059 assert table_content_hash(conn, "chunks", CHUNK_COLUMNS) == chunks_hash
1060 assert table_content_hash(conn, "files", FILE_COLUMNS) == files_hash
1061 conn.close()
1062
1063 assert first["rows"] == 24
1064 assert second["rows"] == 24
1065 assert first["drops"] == 3
1066 assert second["drops"] == 3
1067
1068
1069def test_extract_file_edges_counts_only_resolution_drops(edges_journal):
1070 conn = _conn(edges_journal)
1071 rel = "facets/edges-copresence/entities/20260430.jsonl"
1072 result = _extract_file_edges(conn, rel, str(edges_journal / rel), {})
1073 assert result.rows_inserted == 2
1074 assert result.drops == 1
1075 assert not result.failed
1076 conn.close()
1077
1078
1079def test_insert_bad_kind_raises_directly(edges_journal):
1080 conn = _conn(edges_journal)
1081 with pytest.raises(ValueError, match="Unknown edge kind"):
1082 insert_edges(
1083 conn,
1084 [
1085 {
1086 "src": "a",
1087 "dst": "b",
1088 "kind": "not-a-kind",
1089 "source": "participation",
1090 "path": "synthetic",
1091 "weight": 1,
1092 }
1093 ],
1094 )
1095 conn.close()
1096
1097
1098@pytest.mark.parametrize("day", [None, "20260430"])
1099def test_insert_edges_accepts_null_and_valid_day_values(edges_journal, day):
1100 conn = _conn(edges_journal)
1101 path = f"synthetic/day-ok/{day}"
1102
1103 assert insert_edges(conn, [_valid_insert_edge_row(day, path=path)]) == 1
1104
1105 stored = conn.execute("SELECT day FROM edges WHERE path=?", (path,)).fetchone()[0]
1106 assert stored == day
1107 conn.close()
1108
1109
1110@pytest.mark.parametrize("day", ["", "notaday", "2026430", 20260430])
1111def test_insert_edges_rejects_invalid_day_values_atomically(edges_journal, day):
1112 conn = _conn(edges_journal)
1113 before = conn.execute("SELECT count(*) FROM edges").fetchone()[0]
1114
1115 with pytest.raises(ValueError, match="Invalid edge day"):
1116 insert_edges(
1117 conn,
1118 [
1119 _valid_insert_edge_row(
1120 "20260430",
1121 path="synthetic/day-validation/valid",
1122 ),
1123 _valid_insert_edge_row(
1124 day,
1125 path="synthetic/day-validation/invalid",
1126 ),
1127 ],
1128 )
1129
1130 assert conn.execute("SELECT count(*) FROM edges").fetchone()[0] == before
1131 assert (
1132 conn.execute(
1133 "SELECT count(*) FROM edges WHERE path LIKE 'synthetic/day-validation/%'"
1134 ).fetchone()[0]
1135 == 0
1136 )
1137 conn.close()
1138
1139
1140def test_insert_edges_validates_whole_batch_before_insert(edges_journal):
1141 conn = _conn(edges_journal)
1142 before = conn.execute("SELECT count(*) FROM edges").fetchone()[0]
1143 with pytest.raises(ValueError, match="Unknown edge kind"):
1144 insert_edges(
1145 conn,
1146 [
1147 {
1148 "src": "edge_ada",
1149 "dst": "edge_byron",
1150 "kind": "attended-with",
1151 "src_name": None,
1152 "dst_name": None,
1153 "day": "20260430",
1154 "facet": "edges-activity",
1155 "source": "participation",
1156 "path": "synthetic",
1157 "anchor": "ok",
1158 "label": "valid row",
1159 "ts": 0,
1160 "weight": 1,
1161 },
1162 {
1163 "src": "edge_ada",
1164 "dst": "edge_byron",
1165 "kind": "not-a-kind",
1166 "source": "participation",
1167 "path": "synthetic",
1168 "weight": 1,
1169 },
1170 ],
1171 )
1172 after = conn.execute("SELECT count(*) FROM edges").fetchone()[0]
1173 assert after == before
1174 conn.close()
1175
1176
1177def test_schema_version_migration_preserves_chunks_and_files(
1178 edges_journal,
1179 monkeypatch,
1180):
1181 import solstone.think.indexer.edges as edge_index
1182
1183 scan_journal(str(edges_journal), full=True)
1184 conn = _conn(edges_journal)
1185 chunks_hash = table_content_hash(conn, "chunks", CHUNK_COLUMNS)
1186 files_hash = table_content_hash(conn, "files", FILE_COLUMNS)
1187 assert conn.execute("SELECT count(*) FROM edges").fetchone()[0] == 24
1188 conn.close()
1189
1190 monkeypatch.setattr(edge_index, "EDGES_SCHEMA_VERSION", 2)
1191 conn = _conn(edges_journal)
1192 assert (
1193 conn.execute(
1194 "SELECT mtime FROM edge_files WHERE path=?", (EDGES_SCHEMA_PATH,)
1195 ).fetchone()[0]
1196 == 2
1197 )
1198 assert conn.execute("SELECT count(*) FROM edges").fetchone()[0] == 0
1199 assert table_content_hash(conn, "chunks", CHUNK_COLUMNS) == chunks_hash
1200 assert table_content_hash(conn, "files", FILE_COLUMNS) == files_hash
1201 conn.close()
1202
1203 assert scan_journal(str(edges_journal), full=True).changed is True
1204 conn = _conn(edges_journal)
1205 assert conn.execute("SELECT count(*) FROM edges").fetchone()[0] == 24
1206 conn.close()
1207
1208
1209def test_discover_files_day_rooted_paths_are_chronicle_free(tmp_path):
1210 journal = tmp_path / "journal"
1211 target = journal / "chronicle" / "20260430" / "synthetic" / "input.jsonl"
1212 target.parent.mkdir(parents=True)
1213 _write_jsonl(target, [{"ok": True}])
1214
1215 files = discover_files(str(journal), [], ["*/synthetic/*.jsonl"])
1216
1217 assert files == {"20260430/synthetic/input.jsonl": str(target)}