Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""SQLite assertion helpers for content-stable table comparisons."""
from __future__ import annotations
import hashlibimport jsonimport sqlite3from typing import Any
EDGE_COLUMNS = [ "src", "dst", "kind", "directed", "src_name", "dst_name", "day", "facet", "source", "path", "anchor", "label", "ts", "weight",]
def table_content_hash( conn: sqlite3.Connection, table: str, columns: list[str],) -> str: """Hash explicit-column table contents without relying on rowid order.""" column_sql = ", ".join(columns) rows: list[list[Any]] = [ list(row) for row in conn.execute(f"SELECT {column_sql} FROM {table}") ] rows.sort(key=lambda row: json.dumps(row, ensure_ascii=True, separators=(",", ":"))) payload = json.dumps( rows, ensure_ascii=True, separators=(",", ":"), ) return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def edges_content_hash(conn: sqlite3.Connection) -> str: """Hash the edge table using its explicit L1 contract columns.""" return table_content_hash(conn, "edges", EDGE_COLUMNS)