From fc398184d17f66450a24708bbb33f3dc3016bad5 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 08:53:41 -0500 Subject: [PATCH 1/3] fix: Publish complete L0 profile and style generations atomically --- docs/resource-budgets.md | 44 ++ tests/test-source-invalidation-748.py | 701 ++++++++++++++++++++++++++ truememory/personality.py | 124 +++-- truememory/personality_style_vec.py | 138 +++-- 4 files changed, 923 insertions(+), 84 deletions(-) create mode 100644 tests/test-source-invalidation-748.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index 7ca0cbc5..9fe5c949 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -540,3 +540,47 @@ in a separate failure-status transaction. Successful output and successful provenance must commit atomically. A cluster run must retain validated runtime model ownership through that outer commit; an inner builder savepoint release alone does not supply that guarantee. + +## Atomic L0 full generations + +Full personality-profile and character-style builders now read a coherent source +snapshot, compute and serialize their complete output, then validate the exact +relevant source values and source schema under SQLite writer ownership. A changed +source raises a categorical `sqlite3.OperationalError`; neither builder retries +internally. Publication replaces only its fully owned table. Vanished senders are +removed, and successful empty input clears the previous output. + +Ordinary messages contribute to both full builders, matching the incremental +add path's directive exclusion. Existing schemas without a `directive` column +continue to treat all messages as ordinary. No maintenance tables are required +for these public builders. Profile source order remains timestamp order; style +source order remains sender and timestamp order. The existing formulas, +thresholds, Python lowercase identity keys and 256-dimensional style algorithm +remain unchanged. Generated update timestamps are incidental to the generation. + +When the builder owns its transaction, its read transaction ends before CPU work +and its write transaction begins only for validation and replacement. A caller's +existing transaction is retained instead, with savepoints around each phase. The +builder never commits or restarts unrelated caller work. Any writer lock already +held by the caller remains held during computation; the builder cannot release +that lock on the caller's behalf. A stale WAL reader fails at write upgrade. + +Write, cancellation and terminal COMMIT/RELEASE failures roll back the attempted +replacement. The previous complete generation survives a later caller commit +after successful rollback. If SQLite also rejects rollback, cleanup propagates +that failure without releasing partial output; the connection owner must recover +or close the transaction. This is not a guarantee that a caller can ignore a +failed rollback and safely commit. + +These are full recomputations. They retain source rows and computed output, and +style computation still retains one entity's per-message 256-element vectors. +Validation streams the source again while holding the writer; it is linear in +the read source size, not constant-time. No memory or latency improvement is +claimed without measurement. They perform no embedding-model work. + +This is the first #748 foundation stage. It does not register L0 maintenance +checkpoints, suppress stale cache reads, change incremental updaters, schedule +repair after source mutations or alter opt-in entity-profile summary sheets. +Those integration steps remain necessary to resolve the complete invalidation +issue. An A10 runner can wrap a builder in its own source transaction and commit +the successful output/checkpoint together once that shared contract is wired. diff --git a/tests/test-source-invalidation-748.py b/tests/test-source-invalidation-748.py new file mode 100644 index 00000000..2bbe7ce2 --- /dev/null +++ b/tests/test-source-invalidation-748.py @@ -0,0 +1,701 @@ +"""A05 stage 1: real stdlib L0 code and disposable SQLite, no package startup. + +The two frozen references come from b97e586. Directive rows are removed only +from the reference input: all other formulas, ordering and returned values +must match exactly. No native dependencies, models or external data are read. +""" +from __future__ import annotations + +import ast +import random +import sqlite3 +import tempfile +import types +import unittest +import warnings +from pathlib import Path +from unittest.mock import patch + +SOURCE = Path(__file__).resolve().parents[1] / "truememory" + + +def _load_modules() -> tuple[types.ModuleType, types.ModuleType]: + style = types.ModuleType("a05_style") + exec(compile((SOURCE / "personality_style_vec.py").read_text(), "personality_style_vec.py", "exec"), style.__dict__) + tree = ast.parse((SOURCE / "personality.py").read_text()) + tree.body = [node for node in tree.body if not ( + isinstance(node, ast.ImportFrom) and (node.module or "").startswith("truememory.") + )] + personality = types.ModuleType("a05_personality") + for name in ("_l0_directive_filter", "_l0_source_schema", "_l0_transaction", "_l0_validate_source"): + if hasattr(style, name): + personality.__dict__[name] = getattr(style, name) + exec(compile(tree, "personality.py", "exec"), personality.__dict__) + return personality, style + + +SCHEMA = """ +CREATE TABLE messages ( + id INTEGER PRIMARY KEY, content TEXT, sender TEXT, recipient TEXT, + timestamp TEXT, category TEXT, modality TEXT, directive INTEGER +); +CREATE TABLE entity_profiles ( + entity TEXT PRIMARY KEY, message_count INTEGER, traits TEXT, + communication_style TEXT, topics TEXT, relationships TEXT, updated_at TEXT +); +CREATE TABLE entity_style_vectors ( + entity TEXT PRIMARY KEY, vector TEXT, message_count INTEGER, updated_at TEXT +); +CREATE TABLE sentinel (value TEXT); +""" + + +def _db(path: str = ":memory:", *, directives: bool = True) -> sqlite3.Connection: + conn = sqlite3.connect(path, timeout=0, cached_statements=0) + schema = SCHEMA if directives else SCHEMA.replace(", directive INTEGER", "") + conn.executescript(schema) + return conn + + +def _seed(conn: sqlite3.Connection) -> None: + conn.executemany( + "INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + [ + (9, "hey I write code and deploy the database", "Alice", "Bob", "2026-01-03", "work", "chat", 0), + (3, "hello I enjoy coffee every morning", "ALICE", "Bob", "2026-01-01", "routine", "chat", None), + (7, "I am worried about the server and customer pricing", "Bob", "Alice", "2026-01-03", "work", "chat", 0), + (11, "directive-only sentinel should never become a profile", "Directive", "Alice", "2026-01-01", "control", "chat", 1), + (12, "", "", "Bob", "2026-01-02", "", "chat", 0), + ], + ) + conn.commit() + + +def _allow(*_args: object) -> int: + return sqlite3.SQLITE_OK + + +LEGACY_PROFILES = '''def _get_all_messages(conn: sqlite3.Connection) -> list[dict]: + """Fetch all messages ordered by timestamp.""" + rows = conn.execute( + "SELECT id, content, sender, recipient, timestamp, category, modality " + "FROM messages ORDER BY timestamp" + ).fetchall() + return [ + { + "id": r[0], "content": r[1], "sender": r[2], + "recipient": r[3], "timestamp": r[4], + "category": r[5], "modality": r[6], + } + for r in rows + ] + +def build_entity_profiles(conn: sqlite3.Connection) -> dict: + """ + Analyze all messages and build personality profiles for key entities. + + For each entity (sender), extracts: + + - **message_count**: total messages sent. + - **topics**: frequent themes (startup, health, food, etc.). + - **communication_style**: average message length, emoji usage, + formality level, and typical greeting. + - **relationships**: who they message most and approximate topic focus + per recipient. + - **traits**: personality descriptors inferred from content analysis. + + Results are stored in the ``entity_profiles`` table and also returned + as a dict keyed by entity name. + + Args: + conn: Open database connection (from :func:`truememory.storage.create_db`). + + Returns: + ``{entity: profile_dict}`` for every sender who has at least one + message in the database. + """ + all_msgs = _get_all_messages(conn) + + # Group messages by sender (normalized to lowercase for case-insensitive matching) + by_sender: dict[str, list[dict]] = defaultdict(list) + for msg in all_msgs: + if msg["sender"]: + by_sender[msg["sender"].lower()].append(msg) + + profiles: dict[str, dict] = {} + + for sender, messages in by_sender.items(): + # Communication style + lengths = [len(m["content"]) for m in messages] + avg_length = sum(lengths) / len(lengths) if lengths else 0.0 + uses_emoji = any(_detect_emoji(m["content"]) for m in messages) + formality = _assess_formality(messages) + greeting = _find_typical_greeting(messages) + + comm_style = { + "avg_length": round(avg_length, 1), + "uses_emoji": uses_emoji, + "formality": formality, + "typical_greeting": greeting, + } + + # Relationships: who they talk to and rough topic per recipient + recipient_counts: dict[str, int] = defaultdict(int) + recipient_topics: dict[str, list[str]] = defaultdict(list) + for msg in messages: + recip = msg["recipient"] + if recip: + recipient_counts[recip] += 1 + # Quick topic tag for each message + lower = msg["content"].lower() + if any(w in lower for w in ("code", "deploy", "api", + "database", "bug", "server")): + recipient_topics[recip].append("technical") + elif any(w in lower for w in ("worried", "anxious", + "scared", "stress")): + recipient_topics[recip].append("emotional") + elif any(w in lower for w in ("dinner", "drinks", + "hang out", "watch")): + recipient_topics[recip].append("social") + elif any(w in lower for w in ("revenue", "investor", + "pricing", "customer")): + recipient_topics[recip].append("business") + + relationships = {} + for recip, count in sorted(recipient_counts.items(), + key=lambda x: x[1], reverse=True): + topic_freq: dict[str, int] = defaultdict(int) + for t in recipient_topics.get(recip, []): + topic_freq[t] += 1 + top_topic = max(topic_freq, key=topic_freq.get) if topic_freq else "general" + relationships[recip] = { + "message_count": count, + "primary_topic": top_topic, + } + + with _warnings_ctx(): + topics = _extract_topics(messages) + traits = _extract_traits(messages) + + profile = { + "message_count": len(messages), + "topics": topics, + "communication_style": comm_style, + "relationships": relationships, + "traits": traits, + } + profiles[sender] = profile + + # Store in database + now = datetime.now(timezone.utc).isoformat() + conn.execute( + """INSERT OR REPLACE INTO entity_profiles + (entity, message_count, traits, communication_style, + topics, relationships, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + ( + sender, + len(messages), + json.dumps(traits), + json.dumps(comm_style), + json.dumps(topics), + json.dumps(relationships), + now, + ), + ) + + conn.commit() + return profiles''' + +LEGACY_STYLE = '''def build_entity_style_vectors(conn: sqlite3.Connection) -> dict[str, list[float]]: + """Batch-build style vectors for every entity (sender) in the database. + + For each sender: + 1. Compute ``compute_style_vector(msg.content)`` for each message. + 2. Mean-pool all per-message vectors via ``mean_pool_vectors``. + 3. Store the result in the ``entity_style_vectors`` table. + + Args: + conn: Open database connection (from :func:`truememory.storage.create_db`). + + Returns: + ``{entity: vector}`` for every sender. + """ + conn.execute( + """CREATE TABLE IF NOT EXISTS entity_style_vectors ( + entity TEXT PRIMARY KEY, + vector TEXT, + message_count INTEGER DEFAULT 0, + updated_at TEXT + )""" + ) + + rows = conn.execute( + "SELECT sender, content FROM messages WHERE sender != '' ORDER BY sender, timestamp" + ).fetchall() + + from collections import defaultdict + by_sender: dict[str, list[str]] = defaultdict(list) + for sender, content in rows: + by_sender[sender.lower()].append(content) + + result: dict[str, list[float]] = {} + now = datetime.now(timezone.utc).isoformat() + + for sender, contents in by_sender.items(): + vecs = [compute_style_vector(c) for c in contents] + mean_vec = mean_pool_vectors(vecs) + result[sender] = mean_vec + + conn.execute( + """INSERT OR REPLACE INTO entity_style_vectors + (entity, vector, message_count, updated_at) + VALUES (?, ?, ?, ?)""", + (sender, json.dumps(mean_vec), len(contents), now), + ) + + conn.commit() + return result''' + +class TestSourceInvalidation748(unittest.TestCase): + def setUp(self) -> None: + self.personality, self.style = _load_modules() + self.temp = tempfile.TemporaryDirectory(prefix="a05-748-") + self.addCleanup(self.temp.cleanup) + self.connections: list[sqlite3.Connection] = [] + self.addCleanup(self._close) + + def _close(self) -> None: + for conn in self.connections: + conn.set_authorizer(_allow) + conn.close() + + def db(self, *, file: bool = False, directives: bool = True) -> sqlite3.Connection: + path = str(Path(self.temp.name) / f"source-{len(self.connections)}.db") if file else ":memory:" + conn = _db(path, directives=directives) + self.connections.append(conn) + if directives: + _seed(conn) + return conn + + def builders(self) -> list[tuple[types.ModuleType, str, str]]: + return [ + (self.personality, "build_entity_profiles", "entity_profiles"), + (self.style, "build_entity_style_vectors", "entity_style_vectors"), + ] + + def generation(self, conn: sqlite3.Connection, table: str, *, stamps: bool = True) -> list[tuple]: + columns = ( + "entity,message_count,traits,communication_style,topics,relationships" + if table == "entity_profiles" else "entity,vector,message_count" + ) + if stamps: + columns += ",updated_at" + return [tuple(row) for row in conn.execute(f"SELECT {columns} FROM {table} ORDER BY entity")] + + def reference(self, conn: sqlite3.Connection, kind: str) -> tuple[dict, list[tuple]]: + ref = _db() + try: + rows = conn.execute("SELECT * FROM messages WHERE directive = 0 OR directive IS NULL").fetchall() + ref.executemany("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [tuple(r) for r in rows]) + ref.commit() + module = self.personality if kind == "entity_profiles" else self.style + namespace = dict(module.__dict__) + exec(LEGACY_PROFILES if kind == "entity_profiles" else LEGACY_STYLE, namespace) + name = "build_entity_profiles" if kind == "entity_profiles" else "build_entity_style_vectors" + result = namespace[name](ref) + return result, self.generation(ref, kind, stamps=False) + finally: + ref.close() + + def assert_reference(self, conn: sqlite3.Connection, module: types.ModuleType, name: str, table: str) -> None: + expected, stored = self.reference(conn, table) + self.assertEqual(getattr(module, name)(conn), expected) + self.assertEqual(self.generation(conn, table, stamps=False), stored) + + def test_unchanged_formulas_and_tied_order_match_frozen_reference(self) -> None: + rng = random.Random(748) + phrases = ["hello friend coffee morning", "please deploy the code", "worried about pricing", "", "a", "gym every day"] + for size in (0, 1, 5, 51, 103): + with self.subTest(size=size): + conn = self.db() + conn.execute("DELETE FROM messages") + for index in range(size): + conn.execute("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ( + index * 3 - 5, rng.choice(phrases), rng.choice(["Alice", "ALICE", "Bob", "Änne", ""]), + rng.choice(["Alice", "Bob", "BOB", ""]), rng.choice(["", "2026-01-01", "2026-01-03"]), + "fixture", "chat", rng.choice([0, 0, 1, None]), + )) + conn.commit() + for module, name, table in self.builders(): + self.assert_reference(conn, module, name, table) + self.assertFalse(conn.in_transaction) + + def test_edits_deletes_replacement_and_empty_match_fresh_reference(self) -> None: + changes = [ + "UPDATE messages SET content='hello new ordinary content', recipient='Carol' WHERE id=9", + "UPDATE messages SET sender='Carol', timestamp='2025-12-01' WHERE id=3", + "DELETE FROM messages WHERE sender='Bob'", + "INSERT OR REPLACE INTO messages VALUES (9,'replaced content','Dana','Carol','2025-01-01','','chat',0)", + "DELETE FROM messages", + ] + for module, name, table in self.builders(): + conn = self.db() + getattr(module, name)(conn) + for sql in changes: + with self.subTest(builder=name, change=sql.split()[0]): + conn.execute(sql) + conn.commit() + self.assert_reference(conn, module, name, table) + self.assertEqual(self.generation(conn, table), []) + + def test_directive_index_does_not_reorder_tied_ordinary_rows(self) -> None: + conn = self.db() + conn.execute("DELETE FROM messages") + conn.executescript( + "CREATE INDEX idx_messages_sender ON messages(sender);" + "CREATE INDEX idx_messages_timestamp ON messages(timestamp);" + "CREATE INDEX idx_messages_directive ON messages(directive);" + ) + rows = [ + (1, "hello one code", "Alice", "Bob", "2026-01-01", "", "", None), + (2, "hey two coffee", "Alice", "Carol", "2026-01-01", "", "", 0), + ] + [(100 + i, "ignored", "Control", "", "2026-01-01", "", "", 1) for i in range(100)] + conn.executemany("INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, ?, ?)", rows) + conn.commit() + for module, name, table in self.builders(): + self.assert_reference(conn, module, name, table) + + def test_directive_toggle_and_directive_only_generation(self) -> None: + for module, name, table in self.builders(): + conn = self.db() + self.assert_reference(conn, module, name, table) + self.assertNotIn("directive", dict((r[0], r) for r in self.generation(conn, table))) + for flag in (1, 0, None): + conn.execute("UPDATE messages SET directive=? WHERE sender != ''", (flag,)) + conn.commit() + self.assert_reference(conn, module, name, table) + if flag == 1: + self.assertEqual(self.generation(conn, table), []) + + def test_standalone_schema_without_maintenance_or_directive(self) -> None: + for module, name, table in self.builders(): + conn = self.db(directives=False) + conn.execute("INSERT INTO messages VALUES (1,'hello code','Alice','Bob','2026-01-01','','chat')") + conn.commit() + result = getattr(module, name)(conn) + self.assertEqual(set(result), {"alice"}) + conn.execute("DELETE FROM messages") + conn.commit() + self.assertEqual(getattr(module, name)(conn), {}) + self.assertEqual(self.generation(conn, table), []) + + def test_row_factory_and_autocommit_compatibility(self) -> None: + for module, name, table in self.builders(): + conn = self.db() + conn.row_factory = sqlite3.Row + conn.isolation_level = None + self.assert_reference(conn, module, name, table) + self.assertIsNone(conn.isolation_level) + self.assertFalse(conn.in_transaction) + + def test_style_builder_accepts_its_original_minimal_source_schema(self) -> None: + conn = sqlite3.connect(":memory:") + self.connections.append(conn) + conn.execute("CREATE TABLE messages (sender TEXT, content TEXT, timestamp TEXT)") + conn.execute("INSERT INTO messages VALUES ('Alice','hello ordinary content','2026-01-01')") + conn.commit() + result = self.style.build_entity_style_vectors(conn) + self.assertEqual(result, {"alice": self.style.compute_style_vector("hello ordinary content")}) + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], 1) + + def test_failed_source_read_does_not_leave_owned_transaction(self) -> None: + for module, name, _table in self.builders(): + conn = sqlite3.connect(":memory:") + self.connections.append(conn) + with self.assertRaises(sqlite3.OperationalError): + getattr(module, name)(conn) + self.assertFalse(conn.in_transaction) + + def test_style_table_creation_is_after_compute_and_atomic(self) -> None: + conn = self.db() + conn.execute("DROP TABLE entity_style_vectors") + original = self.style.compute_style_vector + observed = [] + def compute(text: str) -> list[float]: + observed.append(conn.execute("SELECT count(*) FROM sqlite_master WHERE name='entity_style_vectors'").fetchone()[0]) + self.assertFalse(conn.in_transaction) + return original(text) + with patch.object(self.style, "compute_style_vector", compute): + self.style.build_entity_style_vectors(conn) + self.assertTrue(observed) + self.assertEqual(set(observed), {0}) + self.assertTrue(self.generation(conn, "entity_style_vectors")) + + def test_publication_preserves_caller_transaction_and_rollback(self) -> None: + for module, name, table in self.builders(): + conn = self.db() + getattr(module, name)(conn) + before = self.generation(conn, table) + conn.execute("INSERT INTO sentinel VALUES ('uncommitted')") + conn.execute("UPDATE messages SET sender='Carol' WHERE sender='Bob'") + result = getattr(module, name)(conn) + self.assertIn("carol", result) + self.assertTrue(conn.in_transaction) + conn.rollback() + self.assertEqual(self.generation(conn, table), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 0) + self.assertEqual(conn.execute("SELECT sender FROM messages WHERE id=7").fetchone()[0], "Bob") + + def test_publication_retains_caller_writes_until_explicit_commit(self) -> None: + for module, name, table in self.builders(): + conn = self.db(file=True) + path = conn.execute("PRAGMA database_list").fetchone()[2] + other = sqlite3.connect(path, timeout=0) + self.connections.append(other) + conn.execute("INSERT INTO sentinel VALUES ('uncommitted')") + getattr(module, name)(conn) + self.assertEqual(other.execute("SELECT count(*) FROM sentinel").fetchone()[0], 0) + self.assertEqual(other.execute(f"SELECT count(*) FROM {table}").fetchone()[0], 0) + conn.commit() + self.assertEqual(other.execute("SELECT count(*) FROM sentinel").fetchone()[0], 1) + self.assertGreater(other.execute(f"SELECT count(*) FROM {table}").fetchone()[0], 0) + + def test_late_compute_failure_and_cancellation_leave_previous_generation(self) -> None: + for module, name, table in self.builders(): + for error_type in (RuntimeError, KeyboardInterrupt): + with self.subTest(builder=name, error=error_type.__name__): + conn = self.db() + getattr(module, name)(conn) + before = self.generation(conn, table) + conn.execute("UPDATE messages SET content=content || ' corrected'") + conn.commit() + conn.execute("INSERT INTO sentinel VALUES ('pending')") + function = "_extract_topics" if table == "entity_profiles" else "compute_style_vector" + original = getattr(module, function) + calls = 0 + error = error_type("synthetic computation failure") + def compute(*args: object, **kwargs: object) -> object: + nonlocal calls + calls += 1 + if calls == 3: + raise error + return original(*args, **kwargs) + # Profile fixture has two ordinary senders; add a third to fail late. + conn.execute("INSERT INTO messages VALUES (20,'new content','Carol','','2026-01-04','','chat',0)") + with patch.object(module, function, compute): + with self.assertRaises(error_type) as raised: + getattr(module, name)(conn) + self.assertIs(raised.exception, error) + self.assertGreaterEqual(calls, 3) + conn.commit() + self.assertEqual(self.generation(conn, table), before) + + def test_partial_insert_failure_rolls_back_replacement_not_caller(self) -> None: + for module, name, table in self.builders(): + for caller in (False, True): + with self.subTest(builder=name, caller=caller): + conn = self.db() + getattr(module, name)(conn) + before = self.generation(conn, table) + conn.execute("UPDATE messages SET content=content || ' corrected'") + conn.commit() + conn.execute(f"CREATE TRIGGER fail_l0 BEFORE INSERT ON {table} WHEN new.entity='bob' BEGIN SELECT RAISE(ABORT,'synthetic write failure'); END") + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + with self.assertRaisesRegex(sqlite3.IntegrityError, "synthetic write failure"): + getattr(module, name)(conn) + self.assertEqual(conn.in_transaction, caller) + conn.commit() + self.assertEqual(self.generation(conn, table), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_publication_commit_or_release_failure_rolls_back(self) -> None: + for module, name, table in self.builders(): + for caller in (False, True): + with self.subTest(builder=name, caller=caller): + conn = self.db() + getattr(module, name)(conn) + before = self.generation(conn, table) + conn.execute("UPDATE messages SET content=content || ' corrected'") + conn.commit() + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + writing = False + denied = False + def authorize(action: int, one: str | None, two: str | None, *_rest: object) -> int: + nonlocal writing, denied + if action == sqlite3.SQLITE_DELETE and one == table: + writing = True + commit = action == sqlite3.SQLITE_TRANSACTION and one == "COMMIT" + release = action == sqlite3.SQLITE_SAVEPOINT and one == "RELEASE" + if writing and not denied and (release if caller else commit): + denied = True + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + conn.set_authorizer(authorize) + try: + with self.assertRaises(sqlite3.DatabaseError): + getattr(module, name)(conn) + finally: + conn.set_authorizer(_allow) + self.assertTrue(denied) + self.assertEqual(conn.in_transaction, caller) + conn.commit() + self.assertEqual(self.generation(conn, table), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_rollback_failure_never_releases_partial_publication(self) -> None: + for module, name, table in self.builders(): + conn = self.db() + getattr(module, name)(conn) + before = self.generation(conn, table) + conn.execute(f"CREATE TRIGGER fail_l0 BEFORE INSERT ON {table} WHEN new.entity='bob' BEGIN SELECT RAISE(ABORT,'synthetic write failure'); END") + conn.execute("INSERT INTO sentinel VALUES ('pending')") + blocked = False + releases_after_failure = [] + def authorize(action: int, one: str | None, two: str | None, *_rest: object) -> int: + nonlocal blocked + if action == sqlite3.SQLITE_SAVEPOINT and one == "ROLLBACK": + blocked = True + return sqlite3.SQLITE_DENY + if blocked and action == sqlite3.SQLITE_SAVEPOINT and one == "RELEASE": + releases_after_failure.append(two) + return sqlite3.SQLITE_OK + conn.set_authorizer(authorize) + try: + with self.assertRaises(sqlite3.DatabaseError): + getattr(module, name)(conn) + finally: + conn.set_authorizer(_allow) + self.assertTrue(blocked) + self.assertTrue(conn.in_transaction) + self.assertEqual(releases_after_failure, []) + conn.rollback() + self.assertEqual(self.generation(conn, table), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 0) + + def test_foreground_source_changes_commit_during_compute_and_fence_publish(self) -> None: + changes = [ + "UPDATE messages SET content='concurrent new text' WHERE id=9", + "UPDATE messages SET sender='Carol' WHERE id=9", + "UPDATE messages SET timestamp='2025-01-01' WHERE id=9", + "UPDATE messages SET directive=1 WHERE id=9", + "DELETE FROM messages WHERE id=9", + "INSERT INTO messages VALUES (2,'backdated ordinary text','Alice','Bob','2024-01-01','','chat',0)", + "DELETE FROM messages", + "ALTER TABLE messages ADD COLUMN extra TEXT", + ] + for module, name, table in self.builders(): + for sql in changes: + with self.subTest(builder=name, mutation=sql.split()[0]): + conn = self.db(file=True) + getattr(module, name)(conn) + before = self.generation(conn, table) + path = conn.execute("PRAGMA database_list").fetchone()[2] + other = sqlite3.connect(path, timeout=0) + self.connections.append(other) + function = "_extract_topics" if table == "entity_profiles" else "compute_style_vector" + original = getattr(module, function) + calls = 0 + def compute(*args: object, **kwargs: object) -> object: + nonlocal calls + calls += 1 + self.assertFalse(conn.in_transaction) + if calls == 1: + other.execute(sql) + other.commit() + return original(*args, **kwargs) + with patch.object(module, function, compute): + with self.assertRaisesRegex(sqlite3.OperationalError, "L0 source changed"): + getattr(module, name)(conn) + self.assertGreater(calls, 0) + self.assertFalse(conn.in_transaction) + conn.commit() + self.assertEqual(self.generation(conn, table), before) + + def test_stale_caller_wal_snapshot_cannot_publish_or_restart(self) -> None: + for module, name, table in self.builders(): + conn = self.db(file=True) + conn.execute("PRAGMA journal_mode=WAL") + getattr(module, name)(conn) + before = self.generation(conn, table) + path = conn.execute("PRAGMA database_list").fetchone()[2] + other = sqlite3.connect(path, timeout=0) + self.connections.append(other) + conn.execute("BEGIN") + function = "_extract_topics" if table == "entity_profiles" else "compute_style_vector" + original = getattr(module, function) + changed = False + def compute(*args: object, **kwargs: object) -> object: + nonlocal changed + self.assertTrue(conn.in_transaction) + if not changed: + other.execute("UPDATE messages SET content='concurrent corrected source' WHERE id=9") + other.commit() + changed = True + return original(*args, **kwargs) + with patch.object(module, function, compute): + with self.assertRaises(sqlite3.OperationalError): + getattr(module, name)(conn) + self.assertTrue(changed) + self.assertTrue(conn.in_transaction) + self.assertEqual(self.generation(conn, table), before) + self.assertNotEqual(conn.execute("SELECT content FROM messages WHERE id=9").fetchone()[0], "concurrent corrected source") + conn.rollback() + self.assertEqual(conn.execute("SELECT content FROM messages WHERE id=9").fetchone()[0], "concurrent corrected source") + self.assertEqual(self.generation(conn, table), before) + + def test_validation_holds_writer_ownership(self) -> None: + for module, name, table in self.builders(): + conn = self.db(file=True) + path = conn.execute("PRAGMA database_list").fetchone()[2] + other = sqlite3.connect(path, timeout=0) + self.connections.append(other) + original = module._l0_validate_source + blocked = [] + def validate(*args: object, **kwargs: object) -> None: + original(*args, **kwargs) + try: + with self.assertRaises(sqlite3.OperationalError): + other.execute("UPDATE messages SET sender='Racing' WHERE id=9") + blocked.append(True) + finally: + other.rollback() + with patch.object(module, "_l0_validate_source", validate): + getattr(module, name)(conn) + self.assertEqual(blocked, [True]) + + def test_cancellation_inside_publication_preserves_caller_generation(self) -> None: + class CancelConnection(sqlite3.Connection): + cancel_table: str | None = None + + def executemany(self, sql: str, parameters: object) -> sqlite3.Cursor: + if self.cancel_table and f"INSERT INTO {self.cancel_table}" in sql: + rows = iter(parameters) + super().execute(sql, next(rows)) + raise KeyboardInterrupt("synthetic publication cancellation") + return super().executemany(sql, parameters) + + for module, name, table in self.builders(): + conn = sqlite3.connect(":memory:", factory=CancelConnection) + self.connections.append(conn) + conn.executescript(SCHEMA) + _seed(conn) + getattr(module, name)(conn) + before = self.generation(conn, table) + conn.execute("INSERT INTO sentinel VALUES ('pending')") + conn.execute("UPDATE messages SET content=content || ' changed'") + conn.cancel_table = table + with self.assertRaises(KeyboardInterrupt): + getattr(module, name)(conn) + self.assertTrue(conn.in_transaction) + conn.commit() + self.assertEqual(self.generation(conn, table), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 1) + + +if __name__ == "__main__": + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + unittest.main() diff --git a/truememory/personality.py b/truememory/personality.py index 6317528e..440be83e 100644 --- a/truememory/personality.py +++ b/truememory/personality.py @@ -32,6 +32,9 @@ from datetime import datetime, timezone from truememory.fts_search import _build_safe_fts_query, _fts_search +from truememory.personality_style_vec import ( + _l0_directive_filter, _l0_source_schema, _l0_transaction, _l0_validate_source, +) # --------------------------------------------------------------------------- @@ -142,20 +145,16 @@ # Internal helpers # --------------------------------------------------------------------------- -def _get_all_messages(conn: sqlite3.Connection) -> list[dict]: - """Fetch all messages ordered by timestamp.""" - rows = conn.execute( - "SELECT id, content, sender, recipient, timestamp, category, modality " - "FROM messages ORDER BY timestamp" - ).fetchall() - return [ - { - "id": r[0], "content": r[1], "sender": r[2], - "recipient": r[3], "timestamp": r[4], - "category": r[5], "modality": r[6], - } - for r in rows - ] +_PROFILE_SOURCE_COLUMNS = ( + "id", "content", "sender", "recipient", "timestamp", "category", "modality", +) + + +def _profile_source_query(schema: tuple[tuple, ...]) -> str: + return ( + "SELECT id, content, sender, recipient, timestamp, category, modality, " + f"{_l0_directive_filter(schema)} AS ordinary FROM messages ORDER BY timestamp" + ) def _get_messages_by_sender(conn: sqlite3.Connection, sender: str, @@ -438,32 +437,7 @@ def _warnings_ctx(): yield -def build_entity_profiles(conn: sqlite3.Connection) -> dict: - """ - Analyze all messages and build personality profiles for key entities. - - For each entity (sender), extracts: - - - **message_count**: total messages sent. - - **topics**: frequent themes (startup, health, food, etc.). - - **communication_style**: average message length, emoji usage, - formality level, and typical greeting. - - **relationships**: who they message most and approximate topic focus - per recipient. - - **traits**: personality descriptors inferred from content analysis. - - Results are stored in the ``entity_profiles`` table and also returned - as a dict keyed by entity name. - - Args: - conn: Open database connection (from :func:`truememory.storage.create_db`). - - Returns: - ``{entity: profile_dict}`` for every sender who has at least one - message in the database. - """ - all_msgs = _get_all_messages(conn) - +def _compute_entity_profiles(all_msgs: list[dict]) -> dict[str, dict]: # Group messages by sender (normalized to lowercase for case-insensitive matching) by_sender: dict[str, list[dict]] = defaultdict(list) for msg in all_msgs: @@ -534,25 +508,65 @@ def build_entity_profiles(conn: sqlite3.Connection) -> dict: } profiles[sender] = profile - # Store in database - now = datetime.now(timezone.utc).isoformat() - conn.execute( - """INSERT OR REPLACE INTO entity_profiles + return profiles + + +def build_entity_profiles(conn: sqlite3.Connection) -> dict: + """ + Analyze all messages and build personality profiles for key entities. + + For each entity (sender), extracts: + + - **message_count**: total messages sent. + - **topics**: frequent themes (startup, health, food, etc.). + - **communication_style**: average message length, emoji usage, + formality level, and typical greeting. + - **relationships**: who they message most and approximate topic focus + per recipient. + - **traits**: personality descriptors inferred from content analysis. + + Results are stored in the ``entity_profiles`` table and also returned + as a dict keyed by entity name. + + Args: + conn: Open database connection (from :func:`truememory.storage.create_db`). + + Returns: + ``{entity: profile_dict}`` for every sender who has at least one + ordinary message in the database. + + CPU computation precedes the builder's write transaction. Publication + checks the exact source and atomically replaces the table, including + obsolete entities. Changed source raises ``sqlite3.OperationalError``. + Existing caller transactions are preserved. + """ + with _l0_transaction(conn): + schema = _l0_source_schema(conn) + query = _profile_source_query(schema) + source_rows = [tuple(row) for row in conn.execute(query)] + # Filtering via the directive index can reorder timestamp ties. Keep the + # original scan order and remove ineligible rows before computation. + all_msgs = [dict(zip(_PROFILE_SOURCE_COLUMNS, row[:-1])) for row in source_rows if row[-1]] + profiles = _compute_entity_profiles(all_msgs) + now = datetime.now(timezone.utc).isoformat() + stored_rows = [ + ( + sender, profile["message_count"], json.dumps(profile["traits"]), + json.dumps(profile["communication_style"]), json.dumps(profile["topics"]), + json.dumps(profile["relationships"]), now, + ) + for sender, profile in profiles.items() + ] + with _l0_transaction(conn, write=True): + _l0_validate_source(conn, schema, query, source_rows) + conn.execute("DELETE FROM entity_profiles") + conn.executemany( + """INSERT INTO entity_profiles (entity, message_count, traits, communication_style, topics, relationships, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)""", - ( - sender, - len(messages), - json.dumps(traits), - json.dumps(comm_style), - json.dumps(topics), - json.dumps(relationships), - now, - ), + stored_rows, ) - - conn.commit() return profiles diff --git a/truememory/personality_style_vec.py b/truememory/personality_style_vec.py index 6caeb779..4cbeb846 100644 --- a/truememory/personality_style_vec.py +++ b/truememory/personality_style_vec.py @@ -26,6 +26,8 @@ import re import sqlite3 import struct +from collections.abc import Iterator +from contextlib import contextmanager from datetime import datetime, timezone @@ -111,6 +113,86 @@ def cosine_similarity(a: list[float], b: list[float]) -> float: return num / (na * nb) +@contextmanager +def _l0_transaction(conn: sqlite3.Connection, *, write: bool = False) -> Iterator[None]: + """Own only this phase, retaining any transaction opened by the caller.""" + owned = not conn.in_transaction + if owned: + conn.execute("BEGIN IMMEDIATE" if write else "BEGIN") + else: + conn.execute("SAVEPOINT truememory_l0") + completed = False + try: + if write and not owned: + # Upgrade before validation: a stale WAL reader must not publish. + conn.execute("UPDATE messages SET sender = sender WHERE 0") + yield + conn.execute("COMMIT" if owned else "RELEASE SAVEPOINT truememory_l0") + completed = True + finally: + if not completed and conn.in_transaction: + if owned: + conn.rollback() + else: + # If rollback fails, never RELEASE potentially partial writes. + conn.execute("ROLLBACK TO SAVEPOINT truememory_l0") + conn.execute("RELEASE SAVEPOINT truememory_l0") + + +def _l0_source_schema(conn: sqlite3.Connection) -> tuple[tuple, ...]: + return tuple(tuple(row) for row in conn.execute("PRAGMA table_info(messages)")) + + +def _l0_directive_filter(schema: tuple[tuple, ...]) -> str: + # Public builders also accept older standalone schemas without directives. + if any(column[1] == "directive" for column in schema): + return "(directive = 0 OR directive IS NULL)" + return "1" + + +def _l0_validate_source( + conn: sqlite3.Connection, + schema: tuple[tuple, ...], + query: str, + source_rows: list[tuple], +) -> None: + """Compare exact relevant source values while holding writer ownership.""" + error = "L0 source changed during computation; retry the rebuild" + if _l0_source_schema(conn) != schema: + raise sqlite3.OperationalError(error) + current = conn.execute(query) + try: + for expected in source_rows: + row = current.fetchone() + if row is None or tuple(row) != expected: + raise sqlite3.OperationalError(error) + if current.fetchone() is not None: + raise sqlite3.OperationalError(error) + finally: + current.close() + + +def _compute_entity_style_vectors( + rows: list[tuple], +) -> tuple[dict[str, list[float]], list[tuple]]: + from collections import defaultdict + + by_sender: dict[str, list[str]] = defaultdict(list) + for sender, content, _timestamp, ordinary in rows: + if ordinary: + by_sender[sender.lower()].append(content) + + result: dict[str, list[float]] = {} + stored_rows: list[tuple] = [] + now = datetime.now(timezone.utc).isoformat() + for sender, contents in by_sender.items(): + vecs = [compute_style_vector(c) for c in contents] + mean_vec = mean_pool_vectors(vecs) + result[sender] = mean_vec + stored_rows.append((sender, json.dumps(mean_vec), len(contents), now)) + return result, stored_rows + + def build_entity_style_vectors(conn: sqlite3.Connection) -> dict[str, list[float]]: """Batch-build style vectors for every entity (sender) in the database. @@ -124,41 +206,39 @@ def build_entity_style_vectors(conn: sqlite3.Connection) -> dict[str, list[float Returns: ``{entity: vector}`` for every sender. - """ - conn.execute( - """CREATE TABLE IF NOT EXISTS entity_style_vectors ( - entity TEXT PRIMARY KEY, - vector TEXT, - message_count INTEGER DEFAULT 0, - updated_at TEXT - )""" - ) - - rows = conn.execute( - "SELECT sender, content FROM messages WHERE sender != '' ORDER BY sender, timestamp" - ).fetchall() - - from collections import defaultdict - by_sender: dict[str, list[str]] = defaultdict(list) - for sender, content in rows: - by_sender[sender.lower()].append(content) - - result: dict[str, list[float]] = {} - now = datetime.now(timezone.utc).isoformat() - for sender, contents in by_sender.items(): - vecs = [compute_style_vector(c) for c in contents] - mean_vec = mean_pool_vectors(vecs) - result[sender] = mean_vec + Only ordinary messages contribute. Computation starts no write transaction; + any transaction already held by the caller remains theirs. Publication + atomically replaces the complete table after checking the source snapshot. + A changed source raises ``sqlite3.OperationalError`` without retrying. + """ + with _l0_transaction(conn): + schema = _l0_source_schema(conn) + query = ( + f"SELECT sender, content, timestamp, {_l0_directive_filter(schema)} AS ordinary " + "FROM messages WHERE sender != '' " + "ORDER BY sender, timestamp" + ) + rows = [tuple(row) for row in conn.execute(query)] + result, stored_rows = _compute_entity_style_vectors(rows) + with _l0_transaction(conn, write=True): + _l0_validate_source(conn, schema, query, rows) conn.execute( - """INSERT OR REPLACE INTO entity_style_vectors + """CREATE TABLE IF NOT EXISTS entity_style_vectors ( + entity TEXT PRIMARY KEY, + vector TEXT, + message_count INTEGER DEFAULT 0, + updated_at TEXT + )""" + ) + conn.execute("DELETE FROM entity_style_vectors") + conn.executemany( + """INSERT INTO entity_style_vectors (entity, vector, message_count, updated_at) VALUES (?, ?, ?, ?)""", - (sender, json.dumps(mean_vec), len(contents), now), + stored_rows, ) - - conn.commit() return result From 9aa264e4d3a119e667b96373e9de341e42fc9ac8 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 09:08:01 -0500 Subject: [PATCH 2/3] test: Preserve style mean normalization in legacy schema assertion --- tests/test-source-invalidation-748.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test-source-invalidation-748.py b/tests/test-source-invalidation-748.py index 2bbe7ce2..83cd2969 100644 --- a/tests/test-source-invalidation-748.py +++ b/tests/test-source-invalidation-748.py @@ -406,7 +406,10 @@ def test_style_builder_accepts_its_original_minimal_source_schema(self) -> None: conn.execute("INSERT INTO messages VALUES ('Alice','hello ordinary content','2026-01-01')") conn.commit() result = self.style.build_entity_style_vectors(conn) - self.assertEqual(result, {"alice": self.style.compute_style_vector("hello ordinary content")}) + expected = self.style.mean_pool_vectors([ + self.style.compute_style_vector("hello ordinary content"), + ]) + self.assertEqual(result, {"alice": expected}) self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], 1) def test_failed_source_read_does_not_leave_owned_transaction(self) -> None: From a1cd4822be8c96c6503ced5fcb286a471f13d199 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:43:07 -0500 Subject: [PATCH 3/3] fix: Preserve raw sums for exact style profile updates --- docs/resource-budgets.md | 45 ++ tests/test-l0-style-accumulator-747.py | 745 +++++++++++++++++++++++++ tests/test_l0_style_vector.py | 9 +- truememory/personality_style_vec.py | 194 +++++-- truememory/storage.py | 32 +- 5 files changed, 959 insertions(+), 66 deletions(-) create mode 100644 tests/test-l0-style-accumulator-747.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index 309f1186..d38cd82d 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -1037,3 +1037,48 @@ repair after source mutations or alter opt-in entity-profile summary sheets. Those integration steps remain necessary to resolve the complete invalidation issue. An A10 runner can wrap a builder in its own source transaction and commit the successful output/checkpoint together once that shared contract is wired. + +### L0 style accumulator format, checkpoint 1 + +Style profiles retain the unnormalized sum of their per-message, unit-length +character n-gram vectors. The `vector_sum` column stores 256 little-endian +float64 components: `256 * 8 = 2048` bytes per entity. The existing JSON `vector` +column remains the normalized mean exposed to readers. `message_count` counts +contributing messages, including messages whose style vector is zero. Each +append reads and writes one entity and performs O(256) arithmetic. Batch +computation retains the existing source snapshot and arithmetic order and uses +one 256-component sum while computing each entity's messages. + +`accumulator_version = 1` identifies this format. Additive schema migration +leaves existing rows at version 0 with a NULL sum; it neither reconstructs lost +magnitude from the old normalized profile nor scans source messages. Repeated +schema initialization performs only the fixed table-column check. These +derived-table schema changes do not advance message source revision counters. +Batch publication creates or upgrades the format only after source validation, +then atomically replaces profiles, sums, counts and versions in the same +transaction. Failed publication preserves the previous generation and any +caller-owned writes. + +Appending to a legacy, unsupported-version or invalid accumulator raises +`StyleVectorRebuildRequired` with a bounded reason (`legacy`, +`unsupported_version` or `invalid`). The previous profile remains readable. +An append uses writer ownership before reading its sum and a savepoint around +schema and row changes. It never commits: a successful standalone call leaves +its transaction pending; an existing caller keeps final commit and rollback +ownership. The engine's existing best-effort style catch can therefore retain +its source/vector add without partially changing the style profile. + +Empty entity/message incremental calls remain no-ops. Batch builds retain +their ordinary-message filter, case folding and existing empty-source behavior; +directives remain excluded by the engine's add caller. Getters, style dimension, +embedding and reranking models, native dtypes, and retrieval result caps are +unchanged. The format does not certify freshness after unrelated source edits. + +This is a preparatory checkpoint, not production activation. Existing legacy +profiles defer style appends until an explicit successful style rebuild. The +production migration bridge and its pending/failed status must be integrated +before release; otherwise legacy profiles could remain stale while source adds +continue. This checkpoint adds no engine routing, automatic rebuild, model +load, corpus scan on open, or user-facing maintenance status. Synthetic +arithmetic and transaction checks do not establish retrieval quality or native +performance acceptance. diff --git a/tests/test-l0-style-accumulator-747.py b/tests/test-l0-style-accumulator-747.py new file mode 100644 index 00000000..290b9af5 --- /dev/null +++ b/tests/test-l0-style-accumulator-747.py @@ -0,0 +1,745 @@ +"""A04 raw-sum regressions using stdlib source execution and in-memory SQLite.""" +from __future__ import annotations + +import ast +import builtins +import itertools +import json +import logging +import math +import random +import sqlite3 +import struct +import sys +import threading +import types +import unittest +import uuid +from collections.abc import Iterator +from contextlib import AbstractContextManager, contextmanager, nullcontext +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] + + +def load_stdlib_source(name: str) -> types.ModuleType: + def safe_import(name: str, globals: dict | None = None, locals: dict | None = None, + fromlist: tuple[str, ...] = (), level: int = 0) -> object: + if name.split(".", 1)[0] not in sys.stdlib_module_names: + raise AssertionError("Non-stdlib import forbidden: " + name) + return builtins.__import__(name, globals, locals, fromlist, level) + + module = types.ModuleType("synthetic_style_" + name) + module.__dict__["__builtins__"] = dict(vars(builtins), __import__=safe_import) + source = ROOT / "truememory" / (name + ".py") + exec(compile(source.read_text(encoding="utf-8"), str(source), "exec"), module.__dict__) + return module + + +STYLE = load_stdlib_source("personality_style_vec") +STORAGE = load_stdlib_source("storage") +LEGACY_SCHEMA = """CREATE TABLE entity_style_vectors ( + entity TEXT PRIMARY KEY, vector TEXT, message_count INTEGER DEFAULT 0, updated_at TEXT +)""" +SOURCES = """CREATE TABLE messages ( + id INTEGER PRIMARY KEY, sender TEXT, content TEXT, timestamp TEXT, directive INTEGER +); CREATE TABLE sentinel (value TEXT);""" + +# Unchanged updater body from dependency base 8d7249c. It is executed with +# synthetic vectors to prove the old behavior fails the new numeric contract. +LEGACY_INCREMENTAL = '''def update_entity_style_vector_incremental( + conn: sqlite3.Connection, entity: str, new_message: str, + *, _pre_computed_vec: list[float] | None = None, +) -> None: + """Incrementally update an entity's style vector with a new message. + + Uses a running weighted average: given the existing mean vector and + its message count, the new mean is:: + + new_mean = (existing * count + new_vec) / (count + 1) + + Then re-L2-normalized. + + Args: + conn: Open database connection. + entity: Entity name (sender). + new_message: The new message text. + _pre_computed_vec: If provided, skip compute_style_vector() call. + """ + if not entity or not new_message: + return + + # Normalize entity name to lowercase for case-insensitive matching (#467) + entity = entity.lower() + + conn.execute( + """CREATE TABLE IF NOT EXISTS entity_style_vectors ( + entity TEXT PRIMARY KEY, + vector TEXT, + message_count INTEGER DEFAULT 0, + updated_at TEXT + )""" + ) + + new_vec = _pre_computed_vec if _pre_computed_vec is not None else compute_style_vector(new_message) + now = datetime.now(timezone.utc).isoformat() + + row = conn.execute( + "SELECT vector, message_count FROM entity_style_vectors WHERE entity = ?", + (entity,), + ).fetchone() + + if row is None or row[0] is None: + conn.execute( + """INSERT OR REPLACE INTO entity_style_vectors + (entity, vector, message_count, updated_at) + VALUES (?, ?, ?, ?)""", + (entity, json.dumps(new_vec), 1, now), + ) + else: + existing_vec = json.loads(row[0]) + count = row[1] or 0 + + new_count = count + 1 + merged = [ + (existing_vec[i] * count + new_vec[i]) / new_count + for i in range(len(existing_vec)) + ] + + norm = math.sqrt(sum(x * x for x in merged)) + if norm > 0: + merged = [x / norm for x in merged] + + conn.execute( + """INSERT OR REPLACE INTO entity_style_vectors + (entity, vector, message_count, updated_at) + VALUES (?, ?, ?, ?)""", + (entity, json.dumps(merged), new_count, now), + ) +''' + + +def axis(index: int) -> list[float]: + return [float(position == index) for position in range(256)] + + +def reference_mean(vectors: list[list[float]]) -> list[float]: + if not vectors: + return [0.0] * 256 + total = [0.0] * len(vectors[0]) + for vector in vectors: + for index in range(len(total)): + total[index] += vector[index] + mean = [value / len(vectors) for value in total] + norm = math.sqrt(sum(value * value for value in mean)) + return [value / norm for value in mean] if norm > 0 else mean + + +def allow_authorized_action(*_args: object) -> int: + return sqlite3.SQLITE_OK + + +class StatementFailure: + """Deny a single execution boundary, independent of SQLite's statement cache.""" + + def __init__(self, conn: sqlite3.Connection, sql: str, denied_action: tuple[int, str, str | None], + *, occurrence: int = 1) -> None: + self.conn = conn + self.sql = sql + self.denied_action = denied_action + self.occurrence = occurrence + self.matches = 0 + self.armed: list[str] = [] + + def __getattr__(self, name: str) -> object: + return getattr(self.conn, name) + + def authorize(self, action: int, first: str | None, second: str | None, *_rest: object) -> int: + return sqlite3.SQLITE_DENY if (action, first, second) == self.denied_action else sqlite3.SQLITE_OK + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + if sql == self.sql: + self.matches += 1 + if self.matches == self.occurrence: + self.armed.append(sql) + # Install outside the pure callback so cached statements must + # reauthorize; reset before the production rollback/RELEASE. + self.conn.set_authorizer(self.authorize) + try: + return self.conn.execute(sql, *args) + finally: + self.conn.set_authorizer(allow_authorized_action) + return self.conn.execute(sql, *args) + + +class StyleAccumulatorTests(unittest.TestCase): + def connection(self, *, legacy: bool = False, full: bool = False, cached_statements: int = 100) -> sqlite3.Connection: + conn = sqlite3.connect(":memory:", timeout=0, cached_statements=cached_statements) + self.addCleanup(conn.close) + if legacy: + conn.execute(LEGACY_SCHEMA) + conn.executescript(STORAGE._SCHEMA_SQL if full else SOURCES) + if full: + conn.execute("CREATE TABLE sentinel (value TEXT)") + STORAGE._initialize_maintenance_tracking(conn) + conn.commit() + return conn + + def columns(self, conn: sqlite3.Connection) -> list[tuple]: + return [tuple(row) for row in conn.execute("PRAGMA table_info(entity_style_vectors)")] + + def row(self, conn: sqlite3.Connection, entity: str = "synthetic") -> tuple | None: + return conn.execute("SELECT * FROM entity_style_vectors WHERE entity=?", (entity,)).fetchone() + + def seed_legacy(self, conn: sqlite3.Connection) -> tuple: + conn.execute("INSERT INTO entity_style_vectors(entity,vector,message_count,updated_at) VALUES (?,?,?,?)", + ("synthetic", json.dumps(axis(0)), 9, "synthetic-old-generation")) + conn.commit() + return self.row(conn) + + def append(self, conn: sqlite3.Connection, vector: list[float], *, entity: str = "synthetic") -> None: + STYLE.update_entity_style_vector_incremental(conn, entity, "synthetic message", _pre_computed_vec=vector) + + def assert_close(self, expected: list[float], actual: list[float]) -> None: + self.assertEqual(len(expected), 256) + self.assertEqual(len(actual), 256) + self.assertTrue(all(math.isclose(a, b, rel_tol=1e-12, abs_tol=1e-12) for a, b in zip(expected, actual))) + + def test_permutations_match_batch_and_preserve_raw_sum(self) -> None: + vectors = [axis(0), axis(1), axis(1)] + for order in itertools.permutations(vectors): + conn = self.connection() + for vector in order: + self.append(conn, vector) + self.assert_close(reference_mean(vectors), STYLE.get_entity_style_vector(conn, "SYNTHETIC")) + raw, count, version = conn.execute("SELECT vector_sum,message_count,accumulator_version FROM entity_style_vectors").fetchone() + self.assertEqual((len(raw), count, version), (2048, 3, 1)) + self.assertEqual(struct.unpack("<256d", raw), (1.0, 2.0) + (0.0,) * 254) + self.assertTrue(conn.in_transaction) + + def test_seeded_orders_match_reference(self) -> None: + rng = random.Random(747) + vectors = [] + for _ in range(128): + vector = [rng.random() for _ in range(256)] + norm = math.sqrt(sum(value * value for value in vector)) + vectors.append([value / norm for value in vector]) + for count in (1, 3, 31, 128): + selected = vectors[:count] + shuffled = list(selected) + rng.shuffle(shuffled) + for order in (selected, list(reversed(selected)), shuffled): + conn = self.connection() + for vector in order: + self.append(conn, vector) + self.assert_close(reference_mean(selected), STYLE.get_entity_style_vector(conn, "synthetic")) + + def test_old_code_negative_controls(self) -> None: + namespace = dict(STYLE.__dict__) + exec(LEGACY_INCREMENTAL, namespace) + old = namespace["update_entity_style_vector_incremental"] + for vectors in ([axis(0), axis(1), axis(1)], [[0.0] * 256, axis(0), axis(1)]): + conn = self.connection() + for vector in vectors: + old(conn, "synthetic", "synthetic message", _pre_computed_vec=vector) + actual = STYLE.get_entity_style_vector(conn, "synthetic") + self.assertGreater(max(abs(a - b) for a, b in zip(actual, reference_mean(vectors))), 0.05) + conn = self.connection(legacy=True) + before = self.seed_legacy(conn) + old(conn, "synthetic", "synthetic message", _pre_computed_vec=axis(1)) + self.assertNotEqual(self.row(conn), before) + + def test_zero_and_noop_semantics(self) -> None: + for vectors in ([[0.0] * 256], [[0.0] * 256, axis(0), axis(1)], + [axis(0), [0.0] * 256, axis(1)], [axis(0), axis(1), [0.0] * 256]): + conn = self.connection() + for vector in vectors: + self.append(conn, vector) + self.assert_close(reference_mean(vectors), STYLE.get_entity_style_vector(conn, "synthetic")) + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], len(vectors)) + conn = self.connection() + STYLE.update_entity_style_vector_incremental(conn, "", "message") + STYLE.update_entity_style_vector_incremental(conn, "synthetic", "", _pre_computed_vec=axis(0)) + self.assertEqual(self.columns(conn), []) + self.assertFalse(conn.in_transaction) + for text in ("a", "ab", " \t\n "): + STYLE.update_entity_style_vector_incremental(conn, "synthetic", text) + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], 3) + self.assertEqual(STYLE.get_entity_style_vector(conn, "synthetic"), [0.0] * 256) + + def test_backup_restart_between_additions(self) -> None: + vectors = [axis(0), axis(1), axis(1)] + conn = self.connection() + for vector in vectors: + self.append(conn, vector) + conn.commit() + before = self.row(conn) + restored = sqlite3.connect(":memory:") + self.addCleanup(restored.close) + conn.backup(restored) + conn.close() + conn = restored + self.assertEqual(self.row(conn), before) + loaded_again = load_stdlib_source("personality_style_vec") + self.assertEqual(loaded_again.get_entity_style_vector(conn, "synthetic"), json.loads(before[1])) + self.assert_close(reference_mean(vectors), STYLE.get_entity_style_vector(conn, "synthetic")) + + def test_batch_arithmetic_and_append_after_rebuild(self) -> None: + conn = self.connection(legacy=True) + texts = ["please deploy the blue service", "I prefer quiet mornings and coffee", "another synthetic sentence"] + for index, text in enumerate(texts): + conn.execute("INSERT INTO messages VALUES (?,?,?,?,?)", (index + 1, "Synthetic", text, str(index), 0)) + conn.commit() + result = STYLE.build_entity_style_vectors(conn) + vectors = [STYLE.compute_style_vector(text) for text in texts] + self.assertEqual(result["synthetic"], reference_mean(vectors)) + self.assertEqual(result["synthetic"], STYLE.mean_pool_vectors(vectors)) + self.assertFalse(conn.in_transaction) + extra = STYLE.compute_style_vector("one more synthetic message") + self.append(conn, extra) + self.assert_close(reference_mean(vectors + [extra]), STYLE.get_entity_style_vector(conn, "synthetic")) + + def test_batch_directives_counts_and_case_folding(self) -> None: + conn = self.connection() + rows = [(1, "Synthetic", "abc", "1", 0), (2, "SYNTHETIC", "", "2", None), + (3, "Synthetic", "a", "3", 0), (4, "Synthetic", "directive text", "4", 1), + (5, "Directive-only", "directive text", "5", 1), (6, "", "no sender", "6", 0)] + conn.executemany("INSERT INTO messages VALUES (?,?,?,?,?)", rows) + conn.commit() + result = STYLE.build_entity_style_vectors(conn) + self.assertEqual(set(result), {"synthetic"}) + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], 3) + self.assertEqual(result["synthetic"], reference_mean([STYLE.compute_style_vector(text) for text in ("", "abc", "a")])) + self.assertEqual(STYLE.get_entity_style_vector(conn, "SYNTHETIC"), result["synthetic"]) + + def test_batch_minimal_source_schema_and_empty_generation(self) -> None: + conn = sqlite3.connect(":memory:") + self.addCleanup(conn.close) + conn.execute("CREATE TABLE messages(sender TEXT, content TEXT, timestamp TEXT)") + conn.execute("INSERT INTO messages VALUES ('Synthetic','abc','1')") + conn.commit() + self.assertEqual(STYLE.build_entity_style_vectors(conn), {"synthetic": reference_mean([STYLE.compute_style_vector("abc")])}) + conn.execute("DELETE FROM messages") + conn.commit() + self.assertEqual(STYLE.build_entity_style_vectors(conn), {}) + self.assertEqual(conn.execute("SELECT count(*) FROM entity_style_vectors").fetchone()[0], 0) + + def test_legacy_append_preserves_row_schema_and_outer_writes(self) -> None: + for caller in (False, True): + conn = self.connection(legacy=True) + before = self.seed_legacy(conn) + columns = self.columns(conn) + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + conn.execute("INSERT INTO messages VALUES (1,'Synthetic','source survives','1',0)") + with self.assertRaises(STYLE.StyleVectorRebuildRequired) as raised: + self.append(conn, axis(1)) + self.assertEqual(raised.exception.reason, "legacy") + self.assertEqual(self.row(conn), before) + self.assertEqual(self.columns(conn), columns) + self.assertEqual(conn.in_transaction, caller) + conn.commit() + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], int(caller)) + self.assertEqual(STYLE.get_entity_style_vector(conn, "synthetic"), axis(0)) + + def test_bad_persisted_accumulators_defer_without_writes(self) -> None: + raw = struct.pack("<256d", *axis(0)) + cases = [(None, 1, 0, "legacy"), (raw, 1, 0, "legacy"), (raw, 1, 2, "unsupported_version"), + (raw, 1, -1, "invalid"), (raw, 1, "bad", "invalid"), (None, 1, 1, "invalid"), + (raw[:-1], 1, 1, "invalid"), (raw.hex(), 1, 1, "invalid"), (raw, 0, 1, "invalid"), + (raw, None, 1, "invalid"), (raw, 1.5, 1, "invalid"), (raw, "bad", 1, "invalid"), + (struct.pack("<256d", float("nan"), *([0.0] * 255)), 1, 1, "invalid"), + (struct.pack("<256d", float("inf"), *([0.0] * 255)), 1, 1, "invalid"), + (struct.pack("<256d", 1e308, *([0.0] * 255)), 1, 1, "invalid")] + for total, count, version, reason in cases: + with self.subTest(count=count, version=version, reason=reason): + conn = self.connection(full=True) + conn.execute("INSERT INTO entity_style_vectors VALUES (?,?,?,?,?,?)", + ("synthetic", json.dumps(axis(0)), count, "unchanged", total, version)) + conn.commit() + before = self.row(conn) + with self.assertRaises(STYLE.StyleVectorRebuildRequired) as raised: + self.append(conn, axis(1)) + self.assertEqual(raised.exception.reason, reason) + self.assertEqual(self.row(conn), before) + self.assertFalse(conn.in_transaction) + + def test_invalid_input_fails_before_schema_or_writer(self) -> None: + for vector in ([], [0.0] * 255, [0.0] * 257, [True] * 256, ["x"] * 256, + [float("nan")] * 256, [float("inf")] * 256, [10**1000] * 256): + conn = self.connection() + statements = [] + conn.set_trace_callback(statements.append) + with self.assertRaises(ValueError): + self.append(conn, vector) + conn.set_trace_callback(None) + self.assertEqual(statements, []) + self.assertFalse(conn.in_transaction) + self.assertEqual(self.columns(conn), []) + + def test_precomputed_vector_skips_hashing(self) -> None: + conn = self.connection() + with patch.object(STYLE, "compute_style_vector", side_effect=AssertionError("must use precomputed vector")): + self.append(conn, axis(0)) + self.assertEqual(STYLE.get_entity_style_vector(conn, "synthetic"), axis(0)) + + def test_append_success_retains_caller_commit_and_rollback(self) -> None: + for caller in (False, True): + conn = self.connection(legacy=True) + columns = self.columns(conn) + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + self.append(conn, axis(0)) + self.assertTrue(conn.in_transaction) + self.assertEqual(STYLE.get_entity_style_vector(conn, "synthetic"), axis(0)) + conn.rollback() + self.assertEqual(self.columns(conn), columns) + self.assertIsNone(self.row(conn)) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 0) + + def test_append_write_failure_rolls_back_its_generation(self) -> None: + for caller in (False, True): + conn = self.connection() + self.append(conn, axis(0)) + conn.commit() + before = self.row(conn) + conn.execute("CREATE TRIGGER fail_style BEFORE INSERT ON entity_style_vectors BEGIN SELECT RAISE(ABORT,'synthetic failure'); END") + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + with self.assertRaisesRegex(sqlite3.IntegrityError, "synthetic failure"): + self.append(conn, axis(1)) + self.assertEqual(conn.in_transaction, caller) + conn.commit() + self.assertEqual(self.row(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_append_savepoint_and_release_failure_preserve_caller(self) -> None: + for caller, operation, cache in itertools.product((False, True), ("BEGIN", "RELEASE"), (0, 5, 100)): + conn = self.connection(cached_statements=cache) + self.append(conn, axis(0)) + conn.commit() + before = self.row(conn) + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + sql = ("SAVEPOINT" if operation == "BEGIN" else "RELEASE SAVEPOINT") + " truememory_style_append" + probe = StatementFailure(conn, sql, (sqlite3.SQLITE_SAVEPOINT, operation, "truememory_style_append")) + with self.subTest(caller=caller, operation=operation, cached_statements=cache): + with self.assertRaisesRegex(sqlite3.DatabaseError, "not authorized"): + self.append(probe, axis(1)) + self.assertEqual(probe.armed, [sql]) + self.assertEqual(conn.in_transaction, caller) + conn.commit() + self.assertEqual(self.row(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_storage_migration_preserves_legacy_and_source_counters(self) -> None: + conn = self.connection(legacy=True, full=True) + before = self.seed_legacy(conn) + token = conn.execute("SELECT * FROM maintenance_source_state").fetchone() + STORAGE._initialize_style_accumulator_schema(conn) + self.assertEqual(self.row(conn), before + (None, 0)) + self.assertEqual(conn.execute("SELECT * FROM maintenance_source_state").fetchone(), token) + self.assertFalse(conn.in_transaction) + fields = {row[1]: row for row in self.columns(conn)} + self.assertEqual(fields["vector_sum"][2:5], ("BLOB", 0, "NULL")) + self.assertEqual(fields["accumulator_version"][2:5], ("INTEGER", 1, "0")) + + def test_create_db_initializes_only_format_for_legacy_profiles(self) -> None: + conn = self.connection(legacy=True, full=True) + before = self.seed_legacy(conn) + conn.execute("INSERT INTO messages(content,sender) VALUES ('synthetic source','Synthetic')") + conn.commit() + token = conn.execute("SELECT * FROM maintenance_source_state").fetchone() + with patch.object(STORAGE.sqlite3, "connect", return_value=conn): + self.assertIs(STORAGE.create_db(":memory:"), conn) + self.assertEqual(self.row(conn), before + (None, 0)) + self.assertEqual(conn.execute("SELECT * FROM maintenance_source_state").fetchone(), token) + with self.assertRaises(STYLE.StyleVectorRebuildRequired) as raised: + self.append(conn, axis(1)) + self.assertEqual(raised.exception.reason, "legacy") + + def test_partial_column_installation_uses_named_fields(self) -> None: + for existing in ("vector_sum BLOB DEFAULT NULL", "accumulator_version INTEGER NOT NULL DEFAULT 0"): + conn = self.connection(legacy=True) + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN " + existing) + conn.commit() + STORAGE._initialize_style_accumulator_schema(conn) + self.append(conn, axis(0)) + self.append(conn, axis(1)) + self.assertEqual(conn.execute("SELECT message_count,accumulator_version,length(vector_sum) FROM entity_style_vectors").fetchone(), + (2, 1, 2048)) + self.assert_close(reference_mean([axis(0), axis(1)]), STYLE.get_entity_style_vector(conn, "synthetic")) + + def test_repeated_schema_check_does_not_read_source_or_write(self) -> None: + conn = self.connection(full=True) + statements = [] + def authorize(action: int, one: str | None, *_rest: object) -> int: + if action in (sqlite3.SQLITE_INSERT, sqlite3.SQLITE_UPDATE, sqlite3.SQLITE_DELETE, sqlite3.SQLITE_TRANSACTION): + return sqlite3.SQLITE_DENY + if action == sqlite3.SQLITE_READ and one in ("messages", "messages_fts", "entity_style_vectors"): + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + conn.set_authorizer(authorize) + conn.set_trace_callback(statements.append) + try: + STORAGE._initialize_style_accumulator_schema(conn) + finally: + conn.set_authorizer(allow_authorized_action) + conn.set_trace_callback(None) + self.assertEqual(statements, ["PRAGMA table_info(entity_style_vectors)"]) + + def test_storage_schema_migration_rolls_back_with_caller(self) -> None: + conn = self.connection(legacy=True) + before = self.seed_legacy(conn) + columns = self.columns(conn) + conn.execute("INSERT INTO sentinel VALUES ('pending')") + STORAGE._initialize_style_accumulator_schema(conn) + self.assertTrue(conn.in_transaction) + self.assertEqual(self.row(conn), before + (None, 0)) + conn.rollback() + self.assertEqual(self.row(conn), before) + self.assertEqual(self.columns(conn), columns) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 0) + + def test_partial_schema_migration_failure_is_atomic(self) -> None: + for caller, cache in itertools.product((False, True), (0, 5, 100)): + conn = self.connection(legacy=True, cached_statements=cache) + before = self.seed_legacy(conn) + columns = self.columns(conn) + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + sql = "ALTER TABLE entity_style_vectors ADD COLUMN accumulator_version INTEGER NOT NULL DEFAULT 0" + probe = StatementFailure(conn, sql, (sqlite3.SQLITE_ALTER_TABLE, "main", "entity_style_vectors")) + statements: list[str] = [] + conn.set_trace_callback(statements.append) + try: + with self.subTest(caller=caller, cached_statements=cache): + with self.assertRaisesRegex(sqlite3.DatabaseError, "not authorized"): + STORAGE._initialize_style_accumulator_schema(probe) + self.assertEqual(probe.armed, [sql]) + finally: + conn.set_trace_callback(None) + self.assertIn("ALTER TABLE entity_style_vectors ADD COLUMN vector_sum BLOB DEFAULT NULL", statements) + self.assertEqual(conn.in_transaction, caller) + self.assertEqual(self.columns(conn), columns) + conn.commit() + self.assertEqual(self.row(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_schema_commit_or_release_failure_restores_legacy_format(self) -> None: + for caller, cache in itertools.product((False, True), (0, 5, 100)): + conn = self.connection(legacy=True, cached_statements=cache) + before = self.seed_legacy(conn) + columns = self.columns(conn) + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + sql = "RELEASE SAVEPOINT truememory_style_schema" if caller else "COMMIT" + denied_action = ((sqlite3.SQLITE_SAVEPOINT, "RELEASE", "truememory_style_schema") if caller else + (sqlite3.SQLITE_TRANSACTION, "COMMIT", None)) + probe = StatementFailure(conn, sql, denied_action) + with self.subTest(caller=caller, cached_statements=cache): + with self.assertRaisesRegex(sqlite3.DatabaseError, "not authorized"): + STORAGE._initialize_style_accumulator_schema(probe) + self.assertEqual(probe.armed, [sql]) + self.assertEqual(conn.in_transaction, caller) + self.assertEqual(self.columns(conn), columns) + conn.commit() + self.assertEqual(self.row(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_batch_schema_creation_occurs_only_after_computation(self) -> None: + conn = self.connection() + conn.execute("INSERT INTO messages VALUES (1,'Synthetic','abc','1',0)") + conn.commit() + compute = STYLE.compute_style_vector + observed = [] + def outside_writer(text: str) -> list[float]: + observed.append((conn.in_transaction, self.columns(conn))) + return compute(text) + with patch.object(STYLE, "compute_style_vector", outside_writer): + STYLE.build_entity_style_vectors(conn) + self.assertEqual(observed, [(False, [])]) + self.assertEqual(conn.execute("SELECT accumulator_version,length(vector_sum) FROM entity_style_vectors").fetchone(), (1, 2048)) + + def test_batch_partial_publication_rolls_back_schema_and_rows(self) -> None: + for caller in (False, True): + conn = self.connection(legacy=True) + before = self.seed_legacy(conn) + columns = self.columns(conn) + conn.executemany("INSERT INTO messages VALUES (?,?,?,?,?)", + [(1, "a", "first content", "1", 0), (2, "b", "second content", "2", 0)]) + conn.commit() + conn.execute("CREATE TRIGGER fail_style BEFORE INSERT ON entity_style_vectors WHEN new.entity='b' BEGIN SELECT RAISE(ABORT,'synthetic failure'); END") + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + with self.assertRaisesRegex(sqlite3.IntegrityError, "synthetic failure"): + STYLE.build_entity_style_vectors(conn) + self.assertEqual(conn.in_transaction, caller) + self.assertEqual(self.columns(conn), columns) + conn.commit() + self.assertEqual(self.row(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM entity_style_vectors").fetchone()[0], 1) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_batch_late_cancellation_retains_prior_accumulator_generation(self) -> None: + conn = self.connection() + conn.executemany("INSERT INTO messages VALUES (?,?,?,?,?)", + [(1, "a", "first", "1", 0), (2, "b", "second", "2", 0)]) + conn.commit() + STYLE.build_entity_style_vectors(conn) + before = conn.execute("SELECT * FROM entity_style_vectors ORDER BY entity").fetchall() + conn.execute("INSERT INTO sentinel VALUES ('pending')") + compute = STYLE.compute_style_vector + calls = 0 + def cancelled(text: str) -> list[float]: + nonlocal calls + calls += 1 + if calls == 2: + raise KeyboardInterrupt("synthetic cancellation") + return compute(text) + with patch.object(STYLE, "compute_style_vector", cancelled): + with self.assertRaises(KeyboardInterrupt): + STYLE.build_entity_style_vectors(conn) + conn.commit() + self.assertEqual(conn.execute("SELECT * FROM entity_style_vectors ORDER BY entity").fetchall(), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 1) + + def test_batch_source_change_rejects_before_schema_upgrade(self) -> None: + conn = self.connection(legacy=True) + before = self.seed_legacy(conn) + columns = self.columns(conn) + conn.execute("INSERT INTO messages VALUES (1,'Synthetic','original','1',0)") + conn.commit() + compute = STYLE.compute_style_vector + def changed(text: str) -> list[float]: + self.assertFalse(conn.in_transaction) + conn.execute("UPDATE messages SET content='corrected' WHERE id=1") + conn.commit() + return compute(text) + with patch.object(STYLE, "compute_style_vector", changed): + with self.assertRaisesRegex(sqlite3.OperationalError, "L0 source changed"): + STYLE.build_entity_style_vectors(conn) + self.assertFalse(conn.in_transaction) + self.assertEqual(self.columns(conn), columns) + self.assertEqual(self.row(conn), before) + + def test_batch_commit_and_release_failure_restore_accumulator(self) -> None: + for caller, cache in itertools.product((False, True), (0, 5, 100)): + conn = self.connection(legacy=True, cached_statements=cache) + before = self.seed_legacy(conn) + columns = self.columns(conn) + conn.execute("INSERT INTO messages VALUES (1,'Synthetic','replacement','1',0)") + conn.commit() + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + sql = "RELEASE SAVEPOINT truememory_l0" if caller else "COMMIT" + denied_action = ((sqlite3.SQLITE_SAVEPOINT, "RELEASE", "truememory_l0") if caller else + (sqlite3.SQLITE_TRANSACTION, "COMMIT", None)) + probe = StatementFailure(conn, sql, denied_action, occurrence=2) + with self.subTest(caller=caller, cached_statements=cache): + with self.assertRaisesRegex(sqlite3.DatabaseError, "not authorized"): + STYLE.build_entity_style_vectors(probe) + self.assertEqual(probe.armed, [sql]) + self.assertEqual(probe.matches, 3 if caller else 2) + self.assertEqual(conn.in_transaction, caller) + self.assertEqual(self.columns(conn), columns) + conn.commit() + self.assertEqual(self.row(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(caller)) + + def test_batch_and_schema_do_not_advance_source_tracking(self) -> None: + conn = self.connection(legacy=True, full=True) + self.seed_legacy(conn) + conn.execute("INSERT INTO messages(content,sender) VALUES ('ordinary content','Synthetic')") + conn.commit() + token = conn.execute("SELECT * FROM maintenance_source_state").fetchone() + layers = conn.execute("SELECT * FROM maintenance_layers ORDER BY layer").fetchall() + STORAGE._initialize_style_accumulator_schema(conn) + STYLE.build_entity_style_vectors(conn) + self.assertEqual(conn.execute("SELECT * FROM maintenance_source_state").fetchone(), token) + self.assertEqual(conn.execute("SELECT * FROM maintenance_layers ORDER BY layer").fetchall(), layers) + + def test_two_connections_cannot_read_then_overwrite_one_another(self) -> None: + uri = "file:synthetic-style-" + uuid.uuid4().hex + "?mode=memory&cache=shared" + first = sqlite3.connect(uri, uri=True, timeout=0, cached_statements=100) + second = sqlite3.connect(uri, uri=True, timeout=0, cached_statements=100) + self.addCleanup(first.close) + self.addCleanup(second.close) + self.append(first, axis(0)) + first.commit() + self.append(first, axis(1)) + statements = [] + second.set_trace_callback(statements.append) + with self.assertRaises(sqlite3.OperationalError): + self.append(second, axis(1)) + second.set_trace_callback(None) + self.assertFalse(second.in_transaction) + self.assertFalse(any("SELECT vector_sum" in sql for sql in statements)) + first.commit() + self.append(second, axis(1)) + second.commit() + self.assertEqual(second.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], 3) + self.assert_close(reference_mean([axis(0), axis(1), axis(1)]), STYLE.get_entity_style_vector(first, "synthetic")) + + def test_getter_still_reads_legacy_json_and_missing_rows(self) -> None: + conn = self.connection(legacy=True) + self.seed_legacy(conn) + self.assertEqual(STYLE.get_entity_style_vector(conn, "SYNTHETIC"), axis(0)) + self.assertIsNone(STYLE.get_entity_style_vector(conn, "unknown")) + self.assertEqual(len(self.columns(conn)), 4) + + def test_actual_engine_add_catches_style_deferral_and_keeps_source(self) -> None: + rebuild = types.ModuleType("synthetic_rebuild") + rebuild.__dict__.update(sqlite3=sqlite3, Iterator=Iterator, AbstractContextManager=AbstractContextManager, + contextmanager=contextmanager, nullcontext=nullcontext) + tree = ast.parse((ROOT / "truememory/rebuild_source.py").read_text(encoding="utf-8")) + selected = [node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "rebuild_transaction"] + exec(compile(ast.Module(body=selected, type_ignores=[]), "rebuild_source.py", "exec"), rebuild.__dict__) + + def safe_import(name: str, globals: dict | None = None, locals: dict | None = None, + fromlist: tuple[str, ...] = (), level: int = 0) -> object: + if name == "truememory.personality_style_vec": + return STYLE + if name == "truememory.rebuild_source": + return rebuild + if name.split(".", 1)[0] not in sys.stdlib_module_names: + raise AssertionError("Unexpected import in engine.add: " + name) + return builtins.__import__(name, globals, locals, fromlist, level) + + tree = ast.parse((ROOT / "truememory/engine.py").read_text(encoding="utf-8")) + engine_class = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "TrueMemoryEngine") + add = next(node for node in engine_class.body if isinstance(node, ast.FunctionDef) and node.name == "add") + namespace = {"__builtins__": dict(vars(builtins), __import__=safe_import), "MAX_CONTENT_LENGTH": 50000, + "insert_message": STORAGE.insert_message, "_update_style_vec": STYLE.update_entity_style_vector_incremental, + "logger": logging.getLogger("synthetic_style_engine")} + exec(compile(ast.Module(body=[add], type_ignores=[]), "engine.py", "exec"), namespace) + for caller, directive in itertools.product((False, True), (False, True)): + conn = self.connection(legacy=True, full=True) + before = self.seed_legacy(conn) + engine = types.SimpleNamespace(conn=conn, _has_vectors=False, _has_personality=False, _has_style_vec=True, + _write_lock=threading.Lock(), _ensure_connection=lambda: None, + _maybe_auto_consolidate=lambda: None) + compute = STYLE.compute_style_vector + locks = [] + def outside_writer(text: str) -> list[float]: + locks.append(engine._write_lock.locked()) + return compute(text) + if caller: + conn.execute("INSERT INTO sentinel VALUES ('pending')") + with patch.object(STYLE, "compute_style_vector", outside_writer): + result = namespace["add"](engine, "source survives legacy deferral", sender="Synthetic", directive=directive) + self.assertEqual(locks, [False]) + self.assertEqual(self.row(conn), before) + self.assertEqual(conn.in_transaction, caller) + self.assertEqual(conn.execute("SELECT content,directive FROM messages WHERE id=?", (result["id"],)).fetchone(), + ("source survives legacy deferral", int(directive))) + if caller: + conn.rollback() + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_l0_style_vector.py b/tests/test_l0_style_vector.py index 72aae705..540f9da6 100644 --- a/tests/test_l0_style_vector.py +++ b/tests/test_l0_style_vector.py @@ -182,7 +182,7 @@ def test_build_for_multiple_entities(self): class TestIncrementalUpdate: def test_incremental_converges(self): - """Add messages one at a time, verify vector converges toward batch-built vector.""" + """Incremental accumulation must equal the batch mean within float64 tolerance.""" conn = _make_test_db() messages = [ "I love hiking in the mountains", @@ -205,10 +205,9 @@ def test_incremental_converges(self): inc_vec = get_entity_style_vector(conn2, "alice") assert inc_vec is not None - # Cosine should be reasonably high (incremental weighted average - # differs slightly from batch mean-pool due to normalization order) - sim = cosine_similarity(batch_vec, inc_vec) - assert sim > 0.8, f"Expected cosine > 0.8, got {sim}" + assert len(inc_vec) == len(batch_vec) == DIM + assert all(math.isclose(expected, actual, abs_tol=1e-12, rel_tol=1e-12) + for expected, actual in zip(batch_vec, inc_vec)) conn.close() conn2.close() diff --git a/truememory/personality_style_vec.py b/truememory/personality_style_vec.py index 4cbeb846..26e0bd89 100644 --- a/truememory/personality_style_vec.py +++ b/truememory/personality_style_vec.py @@ -37,6 +37,16 @@ def _stable_hash(s: str) -> int: DIM = 256 NGRAM_SIZES = (3, 4, 5) +_STYLE_ACCUMULATOR_VERSION = 1 +_STYLE_SUM = struct.Struct("<256d") + + +class StyleVectorRebuildRequired(RuntimeError): + """The stored profile remains readable but cannot accept exact appends.""" + + def __init__(self, reason: str) -> None: + self.reason = reason + super().__init__(f"Style accumulator rebuild required: {reason}") def compute_style_vector(text: str) -> list[float]: @@ -186,20 +196,110 @@ def _compute_entity_style_vectors( stored_rows: list[tuple] = [] now = datetime.now(timezone.utc).isoformat() for sender, contents in by_sender.items(): - vecs = [compute_style_vector(c) for c in contents] - mean_vec = mean_pool_vectors(vecs) + total = [0.0] * DIM + for content in contents: + vec = _validated_style_vector(compute_style_vector(content)) + for index in range(DIM): + total[index] += vec[index] + mean_vec = _style_profile_from_sum(total, len(contents)) result[sender] = mean_vec - stored_rows.append((sender, json.dumps(mean_vec), len(contents), now)) + stored_rows.append((sender, json.dumps(mean_vec), len(contents), now, + _STYLE_SUM.pack(*total), _STYLE_ACCUMULATOR_VERSION)) return result, stored_rows +def _validated_style_vector(vector: list[float]) -> list[float]: + if not isinstance(vector, (list, tuple)) or len(vector) != DIM: + raise ValueError("Style vector must have 256 finite components") + if any(not isinstance(value, (int, float)) or isinstance(value, bool) for value in vector): + raise ValueError("Style vector must have 256 finite components") + try: + values = [float(value) for value in vector] + except (OverflowError, ValueError) as error: + raise ValueError("Style vector must have 256 finite components") from error + if not all(math.isfinite(value) for value in values): + raise ValueError("Style vector must have 256 finite components") + return values + + +def _style_profile_from_sum(total: list[float], count: int) -> list[float]: + # Retain mean_pool_vectors' divide-then-normalize arithmetic exactly. + mean = [value / count for value in total] + norm = math.sqrt(sum(value * value for value in mean)) + if not math.isfinite(norm): + raise ValueError("Style profile magnitude is not finite") + return [value / norm for value in mean] if norm > 0 else mean + + +def _ensure_style_accumulator_schema(conn: sqlite3.Connection) -> None: + """Add format columns inside the caller's publication transaction.""" + conn.execute( + """CREATE TABLE IF NOT EXISTS entity_style_vectors ( + entity TEXT PRIMARY KEY, + vector TEXT, + message_count INTEGER DEFAULT 0, + updated_at TEXT, + vector_sum BLOB DEFAULT NULL, + accumulator_version INTEGER NOT NULL DEFAULT 0 + )""" + ) + columns = {row[1] for row in conn.execute("PRAGMA table_info(entity_style_vectors)")} + if "vector_sum" not in columns: + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN vector_sum BLOB DEFAULT NULL") + if "accumulator_version" not in columns: + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN accumulator_version INTEGER NOT NULL DEFAULT 0") + + +def _decode_style_accumulator(row: tuple) -> tuple[list[float], int]: + raw, count, version = row + if type(version) is not int or version < 0: + raise StyleVectorRebuildRequired("invalid") + if version == 0: + raise StyleVectorRebuildRequired("legacy") + if version != _STYLE_ACCUMULATOR_VERSION: + raise StyleVectorRebuildRequired("unsupported_version") + if not isinstance(raw, bytes) or len(raw) != _STYLE_SUM.size or type(count) is not int or count < 1: + raise StyleVectorRebuildRequired("invalid") + total = list(_STYLE_SUM.unpack(raw)) + if not all(math.isfinite(value) for value in total): + raise StyleVectorRebuildRequired("invalid") + try: + _style_profile_from_sum(total, count) + except ValueError as error: + raise StyleVectorRebuildRequired("invalid") from error + return total, count + + +@contextmanager +def _style_append_transaction(conn: sqlite3.Connection) -> Iterator[None]: + """Serialize read-modify-write without committing an append for its caller.""" + owned = not conn.in_transaction + if owned: + conn.execute("BEGIN IMMEDIATE") + savepoint = False + completed = False + try: + conn.execute("SAVEPOINT truememory_style_append") + savepoint = True + yield + conn.execute("RELEASE SAVEPOINT truememory_style_append") + completed = True + finally: + if not completed and conn.in_transaction: + if owned: + conn.rollback() + elif savepoint: + conn.execute("ROLLBACK TO SAVEPOINT truememory_style_append") + conn.execute("RELEASE SAVEPOINT truememory_style_append") + + def build_entity_style_vectors(conn: sqlite3.Connection) -> dict[str, list[float]]: """Batch-build style vectors for every entity (sender) in the database. For each sender: 1. Compute ``compute_style_vector(msg.content)`` for each message. - 2. Mean-pool all per-message vectors via ``mean_pool_vectors``. - 3. Store the result in the ``entity_style_vectors`` table. + 2. Mean-pool with the same arithmetic as ``mean_pool_vectors``. + 3. Store the profile and its raw sum in ``entity_style_vectors``. Args: conn: Open database connection (from :func:`truememory.storage.create_db`). @@ -224,19 +324,12 @@ def build_entity_style_vectors(conn: sqlite3.Connection) -> dict[str, list[float with _l0_transaction(conn, write=True): _l0_validate_source(conn, schema, query, rows) - conn.execute( - """CREATE TABLE IF NOT EXISTS entity_style_vectors ( - entity TEXT PRIMARY KEY, - vector TEXT, - message_count INTEGER DEFAULT 0, - updated_at TEXT - )""" - ) + _ensure_style_accumulator_schema(conn) conn.execute("DELETE FROM entity_style_vectors") conn.executemany( """INSERT INTO entity_style_vectors - (entity, vector, message_count, updated_at) - VALUES (?, ?, ?, ?)""", + (entity, vector, message_count, updated_at, vector_sum, accumulator_version) + VALUES (?, ?, ?, ?, ?, ?)""", stored_rows, ) return result @@ -248,12 +341,11 @@ def update_entity_style_vector_incremental( ) -> None: """Incrementally update an entity's style vector with a new message. - Uses a running weighted average: given the existing mean vector and - its message count, the new mean is:: - - new_mean = (existing * count + new_vec) / (count + 1) - - Then re-L2-normalized. + Persist the sum of per-message vectors and normalize only the exposed + profile. A legacy, unsupported or invalid accumulator raises + ``StyleVectorRebuildRequired`` without changing its previous profile. + Rebuild explicitly with ``build_entity_style_vectors`` before appending. + This function never commits; successful writes remain the caller's. Args: conn: Open database connection. @@ -267,49 +359,31 @@ def update_entity_style_vector_incremental( # Normalize entity name to lowercase for case-insensitive matching (#467) entity = entity.lower() - conn.execute( - """CREATE TABLE IF NOT EXISTS entity_style_vectors ( - entity TEXT PRIMARY KEY, - vector TEXT, - message_count INTEGER DEFAULT 0, - updated_at TEXT - )""" + new_vec = _validated_style_vector( + _pre_computed_vec if _pre_computed_vec is not None else compute_style_vector(new_message) ) - - new_vec = _pre_computed_vec if _pre_computed_vec is not None else compute_style_vector(new_message) now = datetime.now(timezone.utc).isoformat() - - row = conn.execute( - "SELECT vector, message_count FROM entity_style_vectors WHERE entity = ?", - (entity,), - ).fetchone() - - if row is None or row[0] is None: - conn.execute( - """INSERT OR REPLACE INTO entity_style_vectors - (entity, vector, message_count, updated_at) - VALUES (?, ?, ?, ?)""", - (entity, json.dumps(new_vec), 1, now), - ) - else: - existing_vec = json.loads(row[0]) - count = row[1] or 0 - - new_count = count + 1 - merged = [ - (existing_vec[i] * count + new_vec[i]) / new_count - for i in range(len(existing_vec)) - ] - - norm = math.sqrt(sum(x * x for x in merged)) - if norm > 0: - merged = [x / norm for x in merged] - + with _style_append_transaction(conn): + _ensure_style_accumulator_schema(conn) + # Existing callers may hold only a read snapshot. Upgrade before + # reading the accumulator so another connection cannot lose an append. + conn.execute("UPDATE entity_style_vectors SET entity=entity WHERE 0") + row = conn.execute( + "SELECT vector_sum, message_count, accumulator_version FROM entity_style_vectors WHERE entity = ?", + (entity,), + ).fetchone() + if row is None: + total, count = new_vec, 1 + else: + previous, count = _decode_style_accumulator(row) + total = [previous[index] + new_vec[index] for index in range(DIM)] + count += 1 + profile = _style_profile_from_sum(total, count) conn.execute( """INSERT OR REPLACE INTO entity_style_vectors - (entity, vector, message_count, updated_at) - VALUES (?, ?, ?, ?)""", - (entity, json.dumps(merged), new_count, now), + (entity, vector, message_count, updated_at, vector_sum, accumulator_version) + VALUES (?, ?, ?, ?, ?, ?)""", + (entity, json.dumps(profile), count, now, _STYLE_SUM.pack(*total), _STYLE_ACCUMULATOR_VERSION), ) diff --git a/truememory/storage.py b/truememory/storage.py index fba044f5..c1f2d1ab 100644 --- a/truememory/storage.py +++ b/truememory/storage.py @@ -175,7 +175,9 @@ def _messages_fts_trigger_definitions(conn: sqlite3.Connection | None = None) -> entity TEXT PRIMARY KEY, vector TEXT, message_count INTEGER DEFAULT 0, - updated_at TEXT + updated_at TEXT, + vector_sum BLOB DEFAULT NULL, + accumulator_version INTEGER NOT NULL DEFAULT 0 ); -- Fact timeline (L5 contradiction tracking) @@ -375,6 +377,33 @@ def _messages_fts_trigger_definitions(conn: sqlite3.Connection | None = None) -> _SCHEMA_SQL += _DUNBAR_OWNERSHIP_SQL + ";\n" +def _initialize_style_accumulator_schema(conn: sqlite3.Connection) -> None: + """Add raw-sum storage without upgrading or rebuilding legacy profiles.""" + required = {"vector_sum", "accumulator_version"} + columns = {row[1] for row in conn.execute("PRAGMA table_info(entity_style_vectors)")} + if required <= columns: + return + owned = not conn.in_transaction + conn.execute("BEGIN IMMEDIATE" if owned else "SAVEPOINT truememory_style_schema") + completed = False + try: + conn.execute("UPDATE entity_style_vectors SET entity=entity WHERE 0") + columns = {row[1] for row in conn.execute("PRAGMA table_info(entity_style_vectors)")} + if "vector_sum" not in columns: + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN vector_sum BLOB DEFAULT NULL") + if "accumulator_version" not in columns: + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN accumulator_version INTEGER NOT NULL DEFAULT 0") + conn.execute("COMMIT" if owned else "RELEASE SAVEPOINT truememory_style_schema") + completed = True + finally: + if not completed and conn.in_transaction: + if owned: + conn.rollback() + else: + conn.execute("ROLLBACK TO SAVEPOINT truememory_style_schema") + conn.execute("RELEASE SAVEPOINT truememory_style_schema") + + def _dunbar_ownership_ready(conn: sqlite3.Connection) -> bool: definitions = {name: " ".join(sql.split()) for name, sql in _DUNBAR_OWNERSHIP_TRIGGERS.items()} installed = {row[0]: " ".join(row[1].strip().rstrip(";").split()) for row in conn.execute( @@ -1000,6 +1029,7 @@ def create_db(db_path: str | Path) -> sqlite3.Connection: _migrate_messages_fts_trigger(conn) _initialize_maintenance_tracking(conn) _initialize_dunbar_ownership(conn) + _initialize_style_accumulator_schema(conn) except sqlite3.Error: conn.close() raise