Download src/pcm/planner/personality.py from Dermitio/Planner-Cache: direct link, hf CLI and curl.
- Browser
- Download file 47.5 kB
-
https://huggingface.co/Dermitio/Planner-Cache/resolve/main/src/pcm/planner/personality.py
- Command line
-
hf download hf://Dermitio/Planner-Cache/src/pcm/planner/personality.py
-
curl -L -o personality.py https://huggingface.co/Dermitio/Planner-Cache/resolve/main/src/pcm/planner/personality.py
47.5 kB
| """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" | |
| 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" | |
| 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") | |
| 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 | |
| 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, | |
| } | |
| class PromotionDecision: | |
| evidence_id: str | |
| promoted: bool | |
| entry_id: str | None | |
| action: str | |
| promotion_score: float | |
| threshold: float | |
| reason: str | |
| class PersonalityQuery: | |
| subject: str | |
| interaction_type: str | |
| domain: str | |
| relationship: str | None = None | |
| relation: str | None = None | |
| timestamp: str | None = None | |
| class PersonalityRoute: | |
| entry_ids: tuple[str, ...] | |
| scores: tuple[float, ...] | |
| candidate_count: int | |
| header_bytes_read: int | |
| accepted: int | |
| class PersonalitySelection: | |
| entries: tuple[PersonalityEntry, ...] | |
| route: PersonalityRoute | |
| entry_bytes_read: int | |
| retrieval_latency_seconds: float | |
| 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() | |
| 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() | |
| def package_id(self) -> str: | |
| return self.metadata()["package_id"] | |
| 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) | |
| 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) | |
| 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 | |
| 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=(), | |
| ) | |