solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""SQLite assertion helpers for content-stable table comparisons."""
5
6from __future__ import annotations
7
8import hashlib
9import json
10import sqlite3
11from collections.abc import Sequence
12from typing import Any
13
14CHUNK_COLUMNS = [
15 "content",
16 "path",
17 "day",
18 "facet",
19 "agent",
20 "stream",
21 "idx",
22 "time_bucket",
23]
24FILE_COLUMNS = ["path", "mtime"]
25EDGE_FILE_COLUMNS = ["path", "mtime"]
26EDGE_COLUMNS = [
27 "src",
28 "dst",
29 "kind",
30 "directed",
31 "src_name",
32 "dst_name",
33 "day",
34 "facet",
35 "source",
36 "path",
37 "anchor",
38 "label",
39 "ts",
40 "weight",
41]
42
43
44def rows_content_hash(rows: Sequence[Sequence[Any]]) -> str:
45 payload = json.dumps(
46 rows,
47 ensure_ascii=True,
48 separators=(",", ":"),
49 )
50 return hashlib.sha256(payload.encode("utf-8")).hexdigest()
51
52
53def table_content_hash(
54 conn: sqlite3.Connection,
55 table: str,
56 columns: list[str],
57) -> str:
58 """Hash explicit-column table contents without relying on rowid order."""
59 column_sql = ", ".join(columns)
60 rows: list[list[Any]] = [
61 list(row)
62 for row in conn.execute(
63 f"SELECT {column_sql} FROM {table} ORDER BY {column_sql}"
64 )
65 ]
66 return rows_content_hash(rows)
67
68
69def edges_content_hash(conn: sqlite3.Connection) -> str:
70 """Hash the edge table using its explicit L1 contract columns."""
71 return table_content_hash(conn, "edges", EDGE_COLUMNS)