"""Disk-resident, model-independent durable personality memory. The package is deliberately mechanical: evidence is accumulated on CPU/disk, promotion is deterministic, and only selected canonical entries are activated through the existing P translation boundary. """ from __future__ import annotations from dataclasses import asdict, dataclass, replace from datetime import datetime, timezone from enum import Enum import hashlib import json import math import os from pathlib import Path import sqlite3 from typing import Iterable, Sequence import torch from torch import Tensor import torch.nn.functional as F from pcm.planner.canonical import CanonicalPConfig, CanonicalPStore from pcm.planner.cache import Freshness, Persistence, SlotSource, SlotType from pcm.planner.representation import CANONICAL, FactorizedStateRepresentation from pcm.planner.split_translator import ByteEntityEncoder from pcm.planner.canonical import CANONICAL_P_PROTOCOL PPKG_FORMAT = "pcm-personality-package-v1" PPKG_PROTOCOL = "pcm-canonical-personality-v1" class PersonalityType(str, Enum): TRAIT = "trait" PREFERENCE = "preference" RELATIONSHIP_PATTERN = "relationship_pattern" BEHAVIORAL_PATTERN = "behavioral_pattern" INTERACTION_STYLE = "interaction_style" TERMINOLOGY = "terminology" HABIT = "habit" RESPONSE_TENDENCY = "response_tendency" CONTEXTUAL_TENDENCY = "contextual_tendency" class EvidenceAuthority(str, Enum): EXPLICIT_USER = "explicit_user_correction_or_statement" EXTERNALLY_VERIFIED = "externally_verified_observation" REPEATED_OBSERVED = "repeated_observed_interaction_behavior" SINGLE_OBSERVED = "single_observed_behavior" MODEL_INFERENCE = "model_inference" MODEL_UNSUPPORTED = "model_generated_unsupported_claim" @property def weight(self) -> float: return { EvidenceAuthority.EXPLICIT_USER: 1.0, EvidenceAuthority.EXTERNALLY_VERIFIED: 0.9, EvidenceAuthority.REPEATED_OBSERVED: 0.7, EvidenceAuthority.SINGLE_OBSERVED: 0.45, EvidenceAuthority.MODEL_INFERENCE: 0.15, EvidenceAuthority.MODEL_UNSUPPORTED: 0.0, }[self] class PersonalityStatus(str, Enum): ACTIVE = "active" CONTRADICTED = "contradicted" SUPERSEDED = "superseded" INVALIDATED = "invalidated" @dataclass(frozen=True) class EvidenceRecord: id: str entry_type: str subject: str relation: str value: str context: str scope: str confidence: float source_authority: str timestamp: str archive_reference: str polarity: int = 1 relationship: str | None = None connected_evidence_ids: tuple[str, ...] = () note: str | None = None def __post_init__(self) -> None: if not self.id or not self.subject or not self.relation or not self.value: raise ValueError("evidence id, subject, relation, and value are required") if not 0 <= self.confidence <= 1: raise ValueError("evidence confidence must be in [0, 1]") if self.polarity not in (-1, 1): raise ValueError("evidence polarity must be -1 or 1") EvidenceAuthority(self.source_authority) if not self.archive_reference: raise ValueError("evidence must reference an archive/source event") @dataclass(frozen=True) class PersonalityEntry: id: str entry_type: str subject: str relation: str value: str scope: str relationship: str | None strength: float confidence: float importance: float evidence_count: int context_diversity: int last_reinforced: str created_at: str updated_at: str source_authority: str supporting_evidence_ids: tuple[str, ...] contradicting_evidence_ids: tuple[str, ...] status: str = PersonalityStatus.ACTIVE.value extension: dict[str, object] | None = None @dataclass(frozen=True) class PromotionPolicy: """Public coefficients for the conservative v1 promotion score.""" promotion_threshold: float = 1.25 context_diversity_coefficient: float = 0.8 connectivity_coefficient: float = 0.25 contradiction_weight: float = 0.75 confidence_prior: float = 0.5 explicit_override_confidence: float = 0.85 def score(self, evidence: Sequence[EvidenceRecord]) -> dict[str, float]: supporting = [record for record in evidence if record.polarity > 0] weighted_support = sum( record.confidence * EvidenceAuthority(record.source_authority).weight for record in supporting ) diversity = len({record.context for record in supporting}) linked = len({ linked_id for record in supporting for linked_id in record.connected_evidence_ids }) diversity_factor = 1 + self.context_diversity_coefficient * math.log1p( max(0, diversity - 1) ) connectivity_factor = 1 + self.connectivity_coefficient * math.log1p(linked) promotion_score = weighted_support * diversity_factor * connectivity_factor return { "weighted_support": weighted_support, "context_diversity": float(diversity), "connected_evidence": float(linked), "diversity_factor": diversity_factor, "connectivity_factor": connectivity_factor, "promotion_score": promotion_score, } @dataclass(frozen=True) class PromotionDecision: evidence_id: str promoted: bool entry_id: str | None action: str promotion_score: float threshold: float reason: str @dataclass(frozen=True) class PersonalityQuery: subject: str interaction_type: str domain: str relationship: str | None = None relation: str | None = None timestamp: str | None = None @dataclass(frozen=True) class PersonalityRoute: entry_ids: tuple[str, ...] scores: tuple[float, ...] candidate_count: int header_bytes_read: int accepted: int @dataclass(frozen=True) class PersonalitySelection: entries: tuple[PersonalityEntry, ...] route: PersonalityRoute entry_bytes_read: int retrieval_latency_seconds: float @property def logical_bytes_read(self) -> int: return self.route.header_bytes_read + self.entry_bytes_read def _canonical_json(value: object) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="microseconds") def _parse_time(value: str) -> datetime: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) def _stable_id(prefix: str, *fields: object) -> str: digest = hashlib.sha256(_canonical_json(fields).encode()).hexdigest()[:24] return f"{prefix}_{digest}" def _entry_to_row(entry: PersonalityEntry) -> tuple[object, ...]: return ( entry.id, entry.entry_type, entry.subject, entry.relation, entry.value, entry.scope, entry.relationship, entry.strength, entry.confidence, entry.importance, entry.evidence_count, entry.context_diversity, entry.last_reinforced, entry.created_at, entry.updated_at, entry.source_authority, _canonical_json(entry.supporting_evidence_ids), _canonical_json(entry.contradicting_evidence_ids), entry.status, _canonical_json(entry.extension or {}), ) def _row_to_entry(row: Sequence[object]) -> PersonalityEntry: return PersonalityEntry( id=str(row[0]), entry_type=str(row[1]), subject=str(row[2]), relation=str(row[3]), value=str(row[4]), scope=str(row[5]), relationship=None if row[6] is None else str(row[6]), strength=float(row[7]), confidence=float(row[8]), importance=float(row[9]), evidence_count=int(row[10]), context_diversity=int(row[11]), last_reinforced=str(row[12]), created_at=str(row[13]), updated_at=str(row[14]), source_authority=str(row[15]), supporting_evidence_ids=tuple(json.loads(str(row[16]))), contradicting_evidence_ids=tuple(json.loads(str(row[17]))), status=str(row[18]), extension=json.loads(str(row[19])), ) ENTRY_COLUMNS = ( "id,type,subject,relation,value,scope,relationship,strength,confidence,importance," "evidence_count,context_diversity,last_reinforced,created_at,updated_at," "source_authority,supporting_ids,contradicting_ids,status,extension_json" ) class PersonalityPackage: """A SQLite-backed `.ppkg`; opening it allocates no CUDA tensors.""" def __init__(self, path: str | Path, *, validate: bool = True) -> None: self.path = Path(path) if not self.path.is_file(): raise FileNotFoundError(self.path) # Web sessions create the package on the launcher thread and execute # serialized turns on the HTTP worker thread. SQLite's transactional # behavior is unchanged. Cross-thread access is enabled only so the # gateway's single session lock can own that serialization boundary. self._connection = sqlite3.connect(self.path, check_same_thread=False) self._connection.row_factory = sqlite3.Row self._connection.execute("PRAGMA query_only = ON") self._closed = False self._validate_header() self._dirty = self.metadata().get("integrity_state", "clean") != "clean" if validate: self.validate_checksum() @classmethod def create( cls, path: str | Path, *, package_id: str, created_at: str | None = None, metadata: dict[str, object] | None = None, overwrite: bool = False, ) -> "PersonalityPackage": path = Path(path) if path.exists() and not overwrite: raise FileExistsError(path) if path.exists(): path.unlink() path.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(path) connection.executescript( """ PRAGMA page_size = 4096; PRAGMA journal_mode = DELETE; PRAGMA synchronous = FULL; CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); CREATE TABLE entries ( id TEXT PRIMARY KEY, type TEXT NOT NULL, subject TEXT NOT NULL, relation TEXT NOT NULL, value TEXT NOT NULL, scope TEXT NOT NULL, relationship TEXT, strength REAL NOT NULL, confidence REAL NOT NULL, importance REAL NOT NULL, evidence_count INTEGER NOT NULL, context_diversity INTEGER NOT NULL, last_reinforced TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, source_authority TEXT NOT NULL, supporting_ids TEXT NOT NULL, contradicting_ids TEXT NOT NULL, status TEXT NOT NULL, extension_json TEXT NOT NULL ); CREATE INDEX entry_lookup ON entries(status, subject, relation, scope, relationship); CREATE INDEX entry_subject_route ON entries( status,subject,importance DESC,confidence DESC,id ); CREATE INDEX entry_scope_route ON entries( status,scope,importance DESC,confidence DESC,id ); CREATE INDEX entry_relationship_route ON entries( status,relationship,importance DESC,confidence DESC,id ); CREATE TABLE evidence ( id TEXT PRIMARY KEY, entry_type TEXT NOT NULL, subject TEXT NOT NULL, relation TEXT NOT NULL, value TEXT NOT NULL, context TEXT NOT NULL, scope TEXT NOT NULL, confidence REAL NOT NULL, source_authority TEXT NOT NULL, timestamp TEXT NOT NULL, archive_reference TEXT NOT NULL, polarity INTEGER NOT NULL, relationship TEXT, connected_ids TEXT NOT NULL, note TEXT ); CREATE INDEX evidence_candidate ON evidence(entry_type, subject, relation, scope, relationship, value); CREATE TABLE changes ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, transaction_id TEXT NOT NULL, timestamp TEXT NOT NULL, action TEXT NOT NULL, entry_id TEXT NOT NULL, before_json TEXT, after_json TEXT ); """ ) stamp = created_at or _utc_now() values = { "format": PPKG_FORMAT, "protocol": PPKG_PROTOCOL, "canonical_p_protocol": CANONICAL_P_PROTOCOL, "package_id": package_id, "created_at": stamp, "updated_at": stamp, "schema_version": "1", "extensions": _canonical_json(metadata or {}), "integrity_state": "clean", "content_sha256": "pending", } connection.executemany( "INSERT INTO metadata(key,value) VALUES (?,?)", sorted(values.items()) ) connection.commit() connection.close() package = cls(path, validate=False) package._dirty = True package.checkpoint(updated_at=stamp) return package def close(self, *, checkpoint: bool = True) -> None: if self._closed: return if checkpoint and self._dirty: self.checkpoint() self._connection.close() self._closed = True def __enter__(self) -> "PersonalityPackage": return self def __exit__(self, *_exc) -> None: self.close() @property def package_id(self) -> str: return self.metadata()["package_id"] @property def size_bytes(self) -> int: return self.path.stat().st_size def metadata(self) -> dict[str, str]: return { str(row[0]): str(row[1]) for row in self._connection.execute("SELECT key,value FROM metadata") } def _validate_header(self) -> None: try: metadata = self.metadata() except sqlite3.DatabaseError as error: raise ValueError("corrupt personality package") from error if metadata.get("format") != PPKG_FORMAT: raise ValueError("unsupported personality package format") if metadata.get("protocol") != PPKG_PROTOCOL: raise ValueError("incompatible personality protocol") if metadata.get("canonical_p_protocol") != CANONICAL_P_PROTOCOL: raise ValueError("incompatible canonical P protocol") if metadata.get("schema_version") != "1": raise ValueError("unsupported personality package schema") def _semantic_checksum(self) -> str: digest = hashlib.sha256() metadata = [ tuple(row) for row in self._connection.execute( "SELECT key,value FROM metadata WHERE key != 'content_sha256' ORDER BY key" ) ] digest.update(_canonical_json(metadata).encode()) for table, order in ( ("entries", "id"), ("evidence", "id"), ("changes", "sequence") ): for row in self._connection.execute(f"SELECT * FROM {table} ORDER BY {order}"): digest.update(_canonical_json(tuple(row)).encode()) return digest.hexdigest() def validate_checksum(self) -> None: if self._dirty or self.metadata().get("integrity_state", "clean") != "clean": raise ValueError("personality package has uncheckpointed changes") expected = self.metadata().get("content_sha256") try: actual = self._semantic_checksum() except sqlite3.DatabaseError as error: raise ValueError("corrupt personality package") from error if not expected or expected != actual: raise ValueError("personality package checksum does not match") def verify(self) -> None: """Explicit full-package integrity boundary.""" self.validate_checksum() def _writable(self) -> None: self._connection.execute("PRAGMA query_only = OFF") def _readonly(self) -> None: self._connection.execute("PRAGMA query_only = ON") def _mark_dirty(self, updated_at: str) -> None: self._connection.execute( "UPDATE metadata SET value=? WHERE key='updated_at'", (updated_at,) ) self._connection.execute( "INSERT OR REPLACE INTO metadata(key,value) VALUES ('integrity_state','dirty')" ) self._dirty = True def checkpoint(self, *, updated_at: str | None = None) -> str: """Atomically seal all committed mutations with one semantic checksum.""" if self._closed: raise RuntimeError("personality package is closed") if not self._dirty: return self.metadata()["content_sha256"] self._writable() try: self._connection.execute("BEGIN IMMEDIATE") self._connection.execute( "UPDATE metadata SET value=? WHERE key='updated_at'", (updated_at or self.metadata()["updated_at"],), ) self._connection.execute( "INSERT OR REPLACE INTO metadata(key,value) VALUES ('integrity_state','clean')" ) self._connection.execute( "UPDATE metadata SET value='pending' WHERE key='content_sha256'" ) checksum = self._semantic_checksum() self._connection.execute( "UPDATE metadata SET value=? WHERE key='content_sha256'", (checksum,) ) self._connection.commit() except Exception: self._connection.rollback() self._readonly() raise self._dirty = False self._readonly() return checksum def export(self, path: str | Path) -> Path: """Checkpoint, snapshot with SQLite backup, and verify the snapshot.""" self.checkpoint() destination = Path(path) if destination.exists(): raise FileExistsError(destination) target = sqlite3.connect(destination) try: self._connection.backup(target) finally: target.close() with PersonalityPackage(destination, validate=True): pass return destination def _evidence_rows(self, record: EvidenceRecord) -> None: self._connection.execute( """INSERT INTO evidence VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( record.id, record.entry_type, record.subject, record.relation, record.value, record.context, record.scope, record.confidence, record.source_authority, record.timestamp, record.archive_reference, record.polarity, record.relationship, _canonical_json(record.connected_evidence_ids), record.note, ), ) def evidence(self, evidence_id: str) -> EvidenceRecord: row = self._connection.execute( "SELECT * FROM evidence WHERE id=?", (evidence_id,) ).fetchone() if row is None: raise KeyError(evidence_id) return EvidenceRecord( id=row[0], entry_type=row[1], subject=row[2], relation=row[3], value=row[4], context=row[5], scope=row[6], confidence=row[7], source_authority=row[8], timestamp=row[9], archive_reference=row[10], polarity=row[11], relationship=row[12], connected_evidence_ids=tuple(json.loads(row[13])), note=row[14], ) def _candidate_records(self, record: EvidenceRecord) -> list[EvidenceRecord]: rows = self._connection.execute( """SELECT id FROM evidence WHERE entry_type=? AND subject=? AND relation=? AND scope=? AND relationship IS ? ORDER BY id""", (record.entry_type, record.subject, record.relation, record.scope, record.relationship), ) return [self.evidence(str(row[0])) for row in rows] def entry_count(self, *, status: str | None = None) -> int: if status is None: row = self._connection.execute("SELECT COUNT(*) FROM entries").fetchone() else: row = self._connection.execute( "SELECT COUNT(*) FROM entries WHERE status=?", (status,) ).fetchone() assert row is not None return int(row[0]) def entries( self, *, status: str | None = None, limit: int | None = None, offset: int = 0, ) -> list[PersonalityEntry]: if limit is not None and limit <= 0: raise ValueError("entry limit must be positive") if offset < 0: raise ValueError("entry offset cannot be negative") where = "" if status is None else " WHERE status=?" parameters: list[object] = [] if status is None else [status] pagination = "" if limit is not None: pagination = " LIMIT ? OFFSET ?" parameters.extend((limit, offset)) elif offset: pagination = " LIMIT -1 OFFSET ?" parameters.append(offset) rows = self._connection.execute( f"SELECT {ENTRY_COLUMNS} FROM entries{where} ORDER BY id{pagination}", parameters, ) return [_row_to_entry(tuple(row)) for row in rows] def entry(self, entry_id: str) -> PersonalityEntry: row = self._connection.execute( f"SELECT {ENTRY_COLUMNS} FROM entries WHERE id=?", (entry_id,) ).fetchone() if row is None: raise KeyError(entry_id) return _row_to_entry(tuple(row)) def routing_headers( self, *, subject: str, scopes: Sequence[str], relationship: str | None, candidate_limit: int, ) -> list[sqlite3.Row]: """Indexed/coarse prefilter returning bounded canonical header rows.""" if candidate_limit <= 0: raise ValueError("candidate limit must be positive") columns = ( "id,type,subject,relation,value,scope,relationship,strength," "confidence,importance,updated_at" ) per_bucket = max(32, candidate_limit // 4) available_indexes = { str(row[0]) for row in self._connection.execute( "SELECT name FROM sqlite_master WHERE type='index'" ) } subject_hint = ( "INDEXED BY entry_subject_route" if "entry_subject_route" in available_indexes else "" ) scope_hint = ( "INDEXED BY entry_scope_route" if "entry_scope_route" in available_indexes else "" ) relationship_hint = ( "INDEXED BY entry_relationship_route" if "entry_relationship_route" in available_indexes else "" ) statements: list[tuple[str, tuple[object, ...]]] = [( f"""SELECT {columns} FROM entries {subject_hint} WHERE status=? AND subject=? ORDER BY importance DESC,confidence DESC,id LIMIT ?""", (PersonalityStatus.ACTIVE.value, subject, per_bucket), )] for scope in dict.fromkeys((*scopes, "global")): statements.append(( f"""SELECT {columns} FROM entries {scope_hint} WHERE status=? AND scope=? ORDER BY importance DESC,confidence DESC,id LIMIT ?""", (PersonalityStatus.ACTIVE.value, scope, per_bucket), )) if relationship is not None: statements.append(( f"""SELECT {columns} FROM entries {relationship_hint} WHERE status=? AND relationship=? ORDER BY importance DESC,confidence DESC,id LIMIT ?""", (PersonalityStatus.ACTIVE.value, relationship, per_bucket), )) rows: dict[str, sqlite3.Row] = {} for statement, parameters in statements: for row in self._connection.execute(statement, parameters): rows.setdefault(str(row[0]), row) return list(rows.values()) def _entry_json(self, entry: PersonalityEntry | None) -> str | None: return None if entry is None else _canonical_json(asdict(entry)) def _upsert_entry( self, entry: PersonalityEntry, *, transaction_id: str, action: str, previous: PersonalityEntry | None, ) -> None: self._connection.execute( f"INSERT OR REPLACE INTO entries({ENTRY_COLUMNS}) VALUES ({','.join('?' for _ in range(20))})", _entry_to_row(entry), ) self._connection.execute( """INSERT INTO changes(transaction_id,timestamp,action,entry_id,before_json,after_json) VALUES (?,?,?,?,?,?)""", ( transaction_id, entry.updated_at, action, entry.id, self._entry_json(previous), self._entry_json(entry), ), ) def ingest( self, record: EvidenceRecord, *, policy: PromotionPolicy = PromotionPolicy(), importance: float = 0.5, ) -> PromotionDecision: """Persist evidence, then deterministically recompute its candidate family.""" self._writable() try: self._evidence_rows(record) except sqlite3.IntegrityError as error: self._connection.rollback() self._readonly() raise ValueError(f"duplicate evidence id: {record.id}") from error family = self._candidate_records(record) support = [item for item in family if item.value == record.value and item.polarity > 0] direct_negative = [ item for item in family if item.value == record.value and item.polarity < 0 ] opposing = [ item for item in family if item.value != record.value and item.polarity > 0 ] + direct_negative score = policy.score(support) opposing_weight = sum( item.confidence * EvidenceAuthority(item.source_authority).weight for item in opposing ) net_score = max(0.0, score["promotion_score"] - policy.contradiction_weight * opposing_weight) authorities = {EvidenceAuthority(item.source_authority) for item in support} explicit_override = any( EvidenceAuthority(item.source_authority) == EvidenceAuthority.EXPLICIT_USER and item.confidence >= policy.explicit_override_confidence for item in support ) inference_only = bool(authorities) and authorities <= { EvidenceAuthority.MODEL_INFERENCE, EvidenceAuthority.MODEL_UNSUPPORTED } promotable = ( bool(support) and not inference_only and ( explicit_override or (len(support) >= 3 and net_score >= policy.promotion_threshold) ) ) entry_id = _stable_id( "personality", record.entry_type, record.subject, record.relation, record.value, record.scope, record.relationship, ) transaction_id = _stable_id("change", record.id, record.timestamp) existing_row = self._connection.execute( f"SELECT {ENTRY_COLUMNS} FROM entries WHERE id=?", (entry_id,) ).fetchone() existing = None if existing_row is None else _row_to_entry(tuple(existing_row)) action = "evidence_only" reason = "promotion threshold not reached" promoted_entry_id: str | None = None if promotable: weighted_support = score["weighted_support"] confidence = weighted_support / ( weighted_support + opposing_weight + policy.confidence_prior ) if explicit_override: confidence = max(confidence, 0.9) strength = min(1.0, net_score / (2 * policy.promotion_threshold)) strongest = max( support, key=lambda item: ( EvidenceAuthority(item.source_authority).weight, item.confidence ), ) created_at = existing.created_at if existing else record.timestamp entry = PersonalityEntry( id=entry_id, entry_type=record.entry_type, subject=record.subject, relation=record.relation, value=record.value, scope=record.scope, relationship=record.relationship, strength=strength, confidence=confidence, importance=importance, evidence_count=len(support), context_diversity=int(score["context_diversity"]), last_reinforced=max(item.timestamp for item in support), created_at=created_at, updated_at=record.timestamp, source_authority=strongest.source_authority, supporting_evidence_ids=tuple(sorted(item.id for item in support)), contradicting_evidence_ids=tuple(sorted(item.id for item in opposing)), status=PersonalityStatus.ACTIVE.value, extension=(existing.extension if existing else {}), ) # A correction never erases the old conclusion; it supersedes it. active_opponents = self._connection.execute( f"""SELECT {ENTRY_COLUMNS} FROM entries WHERE type=? AND subject=? AND relation=? AND scope=? AND relationship IS ? AND status=? AND id != ?""", ( record.entry_type, record.subject, record.relation, record.scope, record.relationship, PersonalityStatus.ACTIVE.value, entry_id, ), ).fetchall() for opponent_row in active_opponents: opponent = _row_to_entry(tuple(opponent_row)) new_status = ( PersonalityStatus.SUPERSEDED.value if explicit_override else PersonalityStatus.CONTRADICTED.value ) changed = replace( opponent, confidence=max(0.0, opponent.confidence - opposing_weight / (1 + opposing_weight)), contradicting_evidence_ids=tuple(sorted(set( opponent.contradicting_evidence_ids + tuple(item.id for item in support) ))), status=new_status, updated_at=record.timestamp, ) self._upsert_entry( changed, transaction_id=transaction_id, action="supersede" if explicit_override else "contradict", previous=opponent, ) self._upsert_entry( entry, transaction_id=transaction_id, action="create" if existing is None else "reinforce", previous=existing, ) action = "create" if existing is None else "reinforce" reason = "explicit authority override" if explicit_override else "promotion threshold reached" promoted_entry_id = entry.id elif existing is not None and opposing: lowered = replace( existing, confidence=max(0.0, existing.confidence - opposing_weight / (1 + opposing_weight)), contradicting_evidence_ids=tuple(sorted(set( existing.contradicting_evidence_ids + tuple(item.id for item in opposing) ))), status=( PersonalityStatus.CONTRADICTED.value if existing.confidence < 0.5 else existing.status ), updated_at=record.timestamp, ) self._upsert_entry( lowered, transaction_id=transaction_id, action="lower_confidence", previous=existing, ) action = "lower_confidence" reason = "contradictory evidence recorded" promoted_entry_id = existing.id self._mark_dirty(record.timestamp) self._connection.commit() self._readonly() return PromotionDecision( evidence_id=record.id, promoted=promotable, entry_id=promoted_entry_id, action=action, promotion_score=net_score, threshold=policy.promotion_threshold, reason=reason, ) def changes(self) -> list[dict[str, object]]: return [dict(row) for row in self._connection.execute( "SELECT * FROM changes ORDER BY sequence" )] def undo_last(self, *, timestamp: str | None = None) -> str: last = self._connection.execute( "SELECT transaction_id FROM changes ORDER BY sequence DESC LIMIT 1" ).fetchone() if last is None: raise ValueError("personality package has no reversible changes") transaction_id = str(last[0]) rows = self._connection.execute( "SELECT * FROM changes WHERE transaction_id=? ORDER BY sequence DESC", (transaction_id,), ).fetchall() self._writable() for row in rows: before = row[5] if before is None: self._connection.execute("DELETE FROM entries WHERE id=?", (row[4],)) else: data = json.loads(before) data["supporting_evidence_ids"] = tuple(data["supporting_evidence_ids"]) data["contradicting_evidence_ids"] = tuple(data["contradicting_evidence_ids"]) restored = PersonalityEntry(**data) self._connection.execute( f"INSERT OR REPLACE INTO entries({ENTRY_COLUMNS}) VALUES ({','.join('?' for _ in range(20))})", _entry_to_row(restored), ) self._connection.execute("DELETE FROM changes WHERE transaction_id=?", (transaction_id,)) self._mark_dirty(timestamp or _utc_now()) self._connection.commit() self._readonly() return transaction_id def bulk_insert_entries( self, entries: Iterable[PersonalityEntry], *, updated_at: str ) -> int: """Benchmark/import path; it never bypasses canonical schema validation.""" rows = [] for entry in entries: PersonalityType(entry.entry_type) PersonalityStatus(entry.status) EvidenceAuthority(entry.source_authority) rows.append(_entry_to_row(entry)) self._writable() self._connection.executemany( f"INSERT INTO entries({ENTRY_COLUMNS}) VALUES ({','.join('?' for _ in range(20))})", rows, ) self._mark_dirty(updated_at) self._connection.commit() self._readonly() return len(rows) class PersonalityRouter: """Canonical CPU/disk router; no model hidden width or CUDA state.""" def __init__( self, *, acceptance_threshold: float = 5.0, candidate_limit: int = 512, entity_width: int = 128, ) -> None: self.acceptance_threshold = acceptance_threshold self.candidate_limit = candidate_limit self.encoder = ByteEntityEncoder(entity_width) @staticmethod def _header_bytes(rows: Sequence[sqlite3.Row]) -> int: return sum(len(_canonical_json(tuple(row)).encode()) for row in rows) def route( self, package: PersonalityPackage, query: PersonalityQuery, *, top_k: int = 4 ) -> PersonalityRoute: if top_k not in (1, 4, 8): raise ValueError("P-package router supports top_k 1, 4, or 8") rows = package.routing_headers( subject=query.subject, scopes=(query.interaction_type, query.domain), relationship=query.relationship, candidate_limit=self.candidate_limit, ) if not rows: return PersonalityRoute((), (), 0, 0, 0) query_semantic = self.encoder.encode_one( " ".join(filter(None, ( query.subject, query.relation or "", query.interaction_type, query.domain, query.relationship or "", ))) ) now = _parse_time(query.timestamp) if query.timestamp else datetime.now(timezone.utc) scored: list[tuple[float, str]] = [] for row in rows: semantic = float(F.cosine_similarity( query_semantic, self.encoder.encode_one(" ".join((row[2], row[3], row[4]))), dim=0, )) identity = 1.0 if row[2].casefold() == query.subject.casefold() else 0.15 if row[5] == query.interaction_type: context = 1.0 elif row[5] == query.domain: context = 0.85 elif row[5] == "global": context = 0.4 else: context = 0.05 if row[6] is None: relationship = 0.35 elif query.relationship and row[6].casefold() == query.relationship.casefold(): relationship = 1.0 else: relationship = 0.0 relation = 0.5 if query.relation: relation = 1.0 if row[3].casefold() == query.relation.casefold() else 0.0 age_days = max(0.0, (now - _parse_time(row[10])).total_seconds() / 86400) freshness = math.exp(-age_days / 3650) score = ( 1.5 * semantic + 2.0 * identity + 2.0 * context + 1.5 * relationship + relation + float(row[7]) + float(row[8]) + float(row[9]) + 0.25 * freshness ) scored.append((score, str(row[0]))) scored.sort(key=lambda item: (-item[0], item[1])) accepted = [item for item in scored if item[0] >= self.acceptance_threshold][:top_k] return PersonalityRoute( entry_ids=tuple(item[1] for item in accepted), scores=tuple(item[0] for item in accepted), candidate_count=len(rows), header_bytes_read=self._header_bytes(rows), accepted=len(accepted), ) def retrieve( self, package: PersonalityPackage, query: PersonalityQuery, *, top_k: int = 4 ) -> PersonalitySelection: import time started = time.perf_counter() route = self.route(package, query, top_k=top_k) entries = tuple(package.entry(entry_id) for entry_id in route.entry_ids) entry_bytes = sum(len(_canonical_json(asdict(entry)).encode()) for entry in entries) return PersonalitySelection( entries=entries, route=route, entry_bytes_read=entry_bytes, retrieval_latency_seconds=time.perf_counter() - started, ) class FactorizedPersonalityCanonicalizer: """Model-independent personality entry -> existing canonical-P protocol.""" def __init__( self, representation: FactorizedStateRepresentation, *, value_labels: Sequence[str], ) -> None: self.representation = representation.cpu().eval() self.value_labels = tuple(value_labels) @staticmethod def _index(value: str, size: int) -> int: return int.from_bytes(hashlib.sha256(value.casefold().encode()).digest()[:8], "big") % size def ids(self, entry: PersonalityEntry) -> tuple[int, int, int, int]: value_lookup = {label.casefold(): index for index, label in enumerate(self.value_labels)} relation_aliases = { "owner": 0, "preferred_persona": 0, "response_style": 0, "location": 1, "status": 2, } return ( self._index(entry.subject, self.representation.entity.num_embeddings), relation_aliases.get( entry.relation.casefold(), self._index(entry.relation, self.representation.relation.num_embeddings), ), value_lookup.get( entry.value.casefold(), self._index(entry.value, self.representation.value.num_embeddings), ), CANONICAL, ) def encode(self, entry: PersonalityEntry) -> tuple[Tensor, tuple[int, int, int, int]]: ids = self.ids(entry) fields = [torch.tensor([value]) for value in ids] with torch.inference_mode(): vector = self.representation.encode(*fields)[0].detach().cpu() return vector, ids @dataclass(frozen=True) class PersonalityActivation: selection: PersonalitySelection store: CanonicalPStore | None canonical_bytes: int inactive_vram_bytes: int = 0 class PersonalityTranslateSession: """Validate once, then reuse one read connection for bounded top-k activation.""" def __init__( self, package_path: str | Path, router: PersonalityRouter, canonicalizer: FactorizedPersonalityCanonicalizer, *, validate_on_open: bool = True, ) -> None: self.package_path = Path(package_path) self.router = router self.canonicalizer = canonicalizer self.validated_checksum: str | None = None self._validated_stat: tuple[int, int] | None = None if validate_on_open: with PersonalityPackage(self.package_path, validate=True) as package: self.validated_checksum = package.metadata()["content_sha256"] stat = self.package_path.stat() self._validated_stat = (stat.st_size, stat.st_mtime_ns) self._package = PersonalityPackage(self.package_path, validate=False) def close(self) -> None: self._package.close(checkpoint=False) def __enter__(self) -> "PersonalityTranslateSession": return self def __exit__(self, *_exc) -> None: self.close() def activate( self, query: PersonalityQuery, *, top_k: int = 4, device: str | torch.device = "cpu", dtype: torch.dtype = torch.float16, ) -> PersonalityActivation: if self._validated_stat is not None: stat = self.package_path.stat() if (stat.st_size, stat.st_mtime_ns) != self._validated_stat: raise ValueError("personality package changed after integrity validation") # Full semantic validation is a cold-open operation. Per-generation # activation reuses the read connection and reads bounded router headers # plus selected full rows. selection = self.router.retrieve(self._package, query, top_k=top_k) if not selection.entries: return PersonalityActivation(selection, None, 0) capacity = len(selection.entries) store = CanonicalPStore(CanonicalPConfig( slots=capacity, width=512, dtype=dtype, device=device, merge_similarity=1.0, )) for entry in selection.entries: vector, ids = self.canonicalizer.encode(entry) store.create( vector, entity_id=ids[0], relation_id=ids[1], value_id=ids[2], metadata_id=ids[3], slot_type=SlotType.FACT, confidence=entry.confidence, importance=entry.importance, freshness=Freshness.FRESH, persistence=Persistence.DURABLE, source=SlotSource.CONVERSATION, label=entry.subject, ) tensors = ( store.cache.values, store.cache.valid, store.cache.slot_type, store.cache.confidence, store.cache.importance, store.cache.freshness, store.cache.persistence, store.cache.last_updated, store.cache.source, store.entity_id, store.relation_id, store.value_id, store.canonical_metadata_id, ) canonical_bytes = sum(tensor.numel() * tensor.element_size() for tensor in tensors) return PersonalityActivation(selection, store, canonical_bytes) def merge_active_personality_with_p_cache( p_cache: CanonicalPStore | None, personality: CanonicalPStore | None ) -> CanonicalPStore: """Construct a temporary combined view without mutating either sibling.""" p_count = 0 if p_cache is None else p_cache.cache.occupied personality_count = 0 if personality is None else personality.cache.occupied capacity = max(1, p_count + personality_count) if personality is not None: device = personality.cache.device dtype = personality.cache.values.dtype elif p_cache is not None: device = p_cache.cache.device dtype = p_cache.cache.values.dtype else: device = torch.device("cpu") dtype = torch.float16 merged = CanonicalPStore(CanonicalPConfig( slots=capacity, width=512, dtype=dtype, device=device, merge_similarity=1.0, )) for source in (p_cache, personality): if source is None: continue for index in source.valid.nonzero(as_tuple=False).flatten().tolist(): merged.create( source.canonical_values[index], entity_id=int(source.entity_id[index]), relation_id=int(source.relation_id[index]), value_id=int(source.value_id[index]), metadata_id=int(source.canonical_metadata_id[index]), slot_type=SlotType(int(source.cache.slot_type[index])), confidence=float(source.cache.confidence[index]), importance=float(source.cache.importance[index]), freshness=Freshness(int(source.cache.freshness[index])), persistence=Persistence(int(source.cache.persistence[index])), source=SlotSource(int(source.cache.source[index])), label=source.cache.labels[index], ) return merged def evidence_from_p_cache( store: CanonicalPStore, slot: int, *, evidence_id: str, entry_type: str, relation: str, value: str, context: str, scope: str, timestamp: str, archive_reference: str, behavioral: bool, relationship: str | None = None, ) -> EvidenceRecord | None: """Explicitly gated P->evidence flow; ordinary mutable facts return None.""" if not behavioral or not bool(store.valid[slot]): return None return EvidenceRecord( id=evidence_id, entry_type=entry_type, subject=store.cache.labels[slot] or f"entity:{int(store.entity_id[slot])}", relation=relation, value=value, context=context, scope=scope, confidence=float(store.cache.confidence[slot]), source_authority=EvidenceAuthority.SINGLE_OBSERVED.value, timestamp=timestamp, archive_reference=archive_reference, relationship=relationship, ) def synthetic_entry( index: int, *, timestamp: str = "2026-01-01T00:00:00+00:00", ) -> PersonalityEntry: """Deterministic growth-benchmark fixture.""" evidence_id = f"archive-evidence-{index}" return PersonalityEntry( id=f"personality-{index:08d}", entry_type=PersonalityType.CONTEXTUAL_TENDENCY.value, subject=f"subject-{index % 4096}", relation=f"trait-{index % 97}", value=f"value-{index}", scope=f"domain-{index % 31}", relationship=None, strength=0.75, confidence=0.8, importance=0.5, evidence_count=3, context_diversity=2, last_reinforced=timestamp, created_at=timestamp, updated_at=timestamp, source_authority=EvidenceAuthority.REPEATED_OBSERVED.value, supporting_evidence_ids=(evidence_id,), contradicting_evidence_ids=(), )