"""Knowtation memory event schema — Phase 4.1, locked at v1.0.0. This module is the **canonical Python source of truth** for the 15 memory event kinds that Knowtation emits and that Muse persists in commit metadata. It mirrors the JavaScript enum in :file:`/Users/aaronrenecarvajal/knowtation/lib/memory-event.mjs` field-for-field and value-for-value. Any drift between the two is a correctness bug and is detected by the data-integrity tests in :file:`tests/test_knowtation_events.py` (Tier 5). Schema lock policy ------------------ :data:`EVENTS_SCHEMA_VERSION` is **immutable once shipped**. Bumping it requires a written migration path that converts existing commit metadata records. Renaming, removing, or reordering :class:`MemoryEventKind` members is treated as a breaking schema change and **must** be paired with a version bump and a migration. Adding a new kind value is also a breaking change because :data:`EVENTS_SCHEMA_HASH` is content-addressed over the sorted value list — readers pinned to ``EVENTS_SCHEMA_HASH`` would refuse the new value. Sensitive-data handling ----------------------- :func:`MemoryEventRecord.__post_init__` rejects any ``data`` payload that contains a key matching the same regular expression used by the JavaScript ``hasSensitiveKeys`` function: ``api[_-]?key``, ``secret``, ``password``, ``token``, ``credential``, ``authorization``, ``bearer``, ``private[_-]?key``, case-insensitive. The scan is recursive into nested dicts and lists up to a maximum depth of 8, matching the JS implementation byte-for-byte. Beyond that depth the scan returns ``False`` rather than recursing further, which is a deliberate trade-off between thoroughness and amplification-attack resistance. Public API ---------- * :data:`EVENTS_SCHEMA_VERSION` — immutable schema version string. * :data:`EVENTS_SCHEMA_HASH` — SHA-256 of the canonical sorted value list. * :class:`MemoryEventKind` — :class:`enum.StrEnum` of the 15 canonical kinds. * :class:`MemoryEventRecord` — frozen dataclass holding one validated event. * :func:`is_valid_event_kind` — non-raising predicate for runtime validation. * :func:`from_dict` — strict deserialiser; raises :class:`ValueError`. * :func:`to_dict` — minimal serialiser; omits ``None``-valued fields. Round-trip identity ------------------- For every well-formed :class:`MemoryEventRecord` ``r``, ``from_dict(to_dict(r)) == r`` holds. The serialiser emits Python field names (``event_id``, ``kind``, ...) rather than the JavaScript-side names (``id``, ``type``, ...) so the canonical Python dict shape is unambiguous. Cross-runtime adapters live outside this module so the schema lock is single-purpose. """ from __future__ import annotations import copy import hashlib import json import re from dataclasses import dataclass, field from datetime import datetime from enum import StrEnum from typing import Any, Final, Literal # ────────────────────────────────────────────────────────────────────────────── # Schema version — IMMUTABLE # ────────────────────────────────────────────────────────────────────────────── #: The locked schema version for the memory-event contract. Bumping this #: constant requires a paired migration script for prior commit metadata. EVENTS_SCHEMA_VERSION: Final[str] = "1.0.0" # ────────────────────────────────────────────────────────────────────────────── # Canonical event-kind enumeration # ────────────────────────────────────────────────────────────────────────────── class MemoryEventKind(StrEnum): """The 15 canonical memory event kinds for the Knowtation domain. Mirrored from ``MEMORY_EVENT_TYPES`` in :file:`knowtation/lib/memory-event.mjs`. The string *value* of each member is the wire-format name that appears in Muse commit metadata; the Python member name is uppercase to satisfy PEP 8 and (for ``IMPORT``) to avoid colliding with the Python keyword. Use the value (e.g. ``MemoryEventKind.SEARCH``) directly in string contexts — :class:`enum.StrEnum` makes ``str(MemoryEventKind.SEARCH) == "search"`` and ``MemoryEventKind.SEARCH == "search"`` both true. """ SEARCH = "search" EXPORT = "export" WRITE = "write" IMPORT = "import" INDEX = "index" PROPOSE = "propose" AGENT_INTERACTION = "agent_interaction" CAPTURE = "capture" ERROR = "error" SESSION_SUMMARY = "session_summary" USER = "user" CONSOLIDATION = "consolidation" CONSOLIDATION_PASS = "consolidation_pass" MAINTENANCE = "maintenance" INSIGHT = "insight" #: Frozen set of every valid wire-format kind value. Used by #: :func:`is_valid_event_kind` for non-raising lookups in hot paths. _VALID_KIND_VALUES: Final[frozenset[str]] = frozenset(k.value for k in MemoryEventKind) #: Allowed values for :attr:`MemoryEventRecord.status`. Mirrors #: ``MEMORY_EVENT_STATUSES`` in ``memory-event.mjs``. _VALID_STATUSES: Final[frozenset[str]] = frozenset({"success", "failed"}) # ────────────────────────────────────────────────────────────────────────────── # Validation primitives # ────────────────────────────────────────────────────────────────────────────── #: Event-id format: literal ``mem_`` prefix + exactly 12 lowercase hex chars. #: Matches the output of ``crypto.randomBytes(6).toString('hex')`` in JS. _EVENT_ID_RE: Final[re.Pattern[str]] = re.compile(r"^mem_[0-9a-f]{12}$") #: Sensitive-key pattern. Mirrors the regex literal in ``memory-event.mjs``:: #: #: /(api[_-]?key|secret|password|token|credential|authorization|bearer|private[_-]?key)/i #: #: Note: this uses :func:`re.search` (substring match), not :func:`re.fullmatch`. #: A key like ``"my_api_key_for_x"`` is therefore rejected, exactly as the JS #: side rejects it. _SENSITIVE_VALUE_RE: Final[re.Pattern[str]] = re.compile( r"(api[_-]?key|secret|password|token|credential|authorization|bearer|private[_-]?key)", re.IGNORECASE, ) #: Maximum recursion depth for the sensitive-key scan. Matches the #: ``depth > 8`` short-circuit in JS ``hasSensitiveKeys``. _MAX_SENSITIVE_SCAN_DEPTH: Final[int] = 8 def _has_sensitive_keys(obj: Any, depth: int = 0) -> bool: """Return ``True`` if *obj* contains a sensitive key within depth bounds. Mirrors :func:`hasSensitiveKeys` in :file:`knowtation/lib/memory-event.mjs`. The traversal walks dicts and lists; scalars are skipped (only *keys* are checked, not values, matching the JS implementation). When ``depth`` is strictly greater than :data:`_MAX_SENSITIVE_SCAN_DEPTH` the function returns ``False`` immediately — this is a deliberate amplification-attack safeguard, not a correctness defect, and is observable from the test suite (Tier 7 security tests). Args: obj: Arbitrary value to scan; only ``dict`` and ``list`` are recursed. depth: Current recursion depth. Callers should leave at the default 0. Returns: ``True`` if any key (case-insensitive) at depth ≤ 8 matches the sensitive pattern; ``False`` otherwise. """ if depth > _MAX_SENSITIVE_SCAN_DEPTH: return False if obj is None: return False if isinstance(obj, dict): for k, v in obj.items(): if isinstance(k, str) and _SENSITIVE_VALUE_RE.search(k): return True if isinstance(v, (dict, list)) and _has_sensitive_keys(v, depth + 1): return True return False if isinstance(obj, list): for item in obj: if isinstance(item, (dict, list)) and _has_sensitive_keys(item, depth + 1): return True return False return False def is_valid_event_kind(kind: str) -> bool: """Return ``True`` iff *kind* is one of the 15 canonical event kinds. Non-raising predicate suitable for hot-path validation (e.g. while streaming events from disk). Strict equality on the string value; no aliasing or case folding. Args: kind: Candidate string to test. Returns: ``True`` iff *kind* is a member of :class:`MemoryEventKind`'s value set. Examples: >>> is_valid_event_kind("search") True >>> is_valid_event_kind("recall") # historical name in the issue text False >>> is_valid_event_kind(123) # type: ignore[arg-type] False """ return isinstance(kind, str) and kind in _VALID_KIND_VALUES # ────────────────────────────────────────────────────────────────────────────── # Event record — frozen dataclass # ────────────────────────────────────────────────────────────────────────────── @dataclass(frozen=True, slots=True) class MemoryEventRecord: """One validated memory event as it appears in Muse commit metadata. Constructing an instance runs every validation rule (see :meth:`__post_init__`); a successfully constructed record is therefore guaranteed wire-safe. The dataclass is :py:obj:`frozen` so records cannot be mutated after creation — any field change must reconstruct the record with :func:`dataclasses.replace`, which re-runs validation. Fields: event_id: ``"mem_" + 12 lowercase hex chars``. Matches :data:`_EVENT_ID_RE`. kind: One of the 15 :class:`MemoryEventKind` members. ts: ISO-8601 timestamp string parseable by :meth:`datetime.datetime.fromisoformat`. vault_id: Non-empty vault identifier. agent_id: Optional ``--agent-id`` value passed to ``muse commit``; may be an opaque string today and an identity-domain handle in Phase 7.7. model_id: Optional ``--model-id`` value passed to ``muse commit``. status: ``"success"`` (default) or ``"failed"``. ttl: Optional time-to-live hint (free-form string; consolidator interprets it). air_id: Optional ICP/AIR attestation receipt identifier added by Phase 7.x. data: Arbitrary JSON-serialisable payload. **Must not** contain any sensitive keys (recursive scan, depth ≤ 8). Raises: ValueError: When any field violates the schema. See :meth:`__post_init__` for the full list of rules. """ event_id: str kind: MemoryEventKind ts: str vault_id: str agent_id: str | None = None model_id: str | None = None status: Literal["success", "failed"] = "success" ttl: str | None = None air_id: str | None = None data: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """Validate every field; raise :class:`ValueError` on any violation. The Python type system enforces some of these constraints at static check time, but the dataclass constructor accepts arbitrary inputs at runtime (e.g. ``MemoryEventRecord(**json_dict)``), so all checks are repeated here. Order of checks is shallow→deep so the cheapest rejections happen first. This matters for the Tier 7 security tests where we feed in adversarial 10 000-character ``kind`` strings. """ if not isinstance(self.event_id, str) or not _EVENT_ID_RE.match(self.event_id): raise ValueError( f"event_id must match ^mem_[0-9a-f]{{12}}$, got {self.event_id!r}" ) if not isinstance(self.kind, MemoryEventKind): raise ValueError( f"kind must be a MemoryEventKind, got {type(self.kind).__name__}" ) if self.kind.value not in _VALID_KIND_VALUES: raise ValueError(f"unknown MemoryEventKind value: {self.kind.value!r}") if not isinstance(self.ts, str): raise ValueError(f"ts must be a string, got {type(self.ts).__name__}") try: datetime.fromisoformat(self.ts) except ValueError as exc: raise ValueError( f"ts must be parseable by datetime.fromisoformat: {exc}" ) from None if not isinstance(self.vault_id, str) or not self.vault_id: raise ValueError("vault_id must be a non-empty string") if self.status not in _VALID_STATUSES: raise ValueError( f"status must be one of {sorted(_VALID_STATUSES)!r}, got {self.status!r}" ) for opt_name in ("agent_id", "model_id", "ttl", "air_id"): opt_val = getattr(self, opt_name) if opt_val is not None and not isinstance(opt_val, str): raise ValueError( f"{opt_name} must be str or None, got {type(opt_val).__name__}" ) if not isinstance(self.data, dict): raise ValueError( f"data must be a dict, got {type(self.data).__name__}" ) if _has_sensitive_keys(self.data): raise ValueError( "data contains sensitive key patterns " "(api_key, secret, password, token, credential, authorization, " "bearer, private_key); remove secrets before constructing the event" ) # ────────────────────────────────────────────────────────────────────────────── # Serialisation # ────────────────────────────────────────────────────────────────────────────── def from_dict(d: dict[str, Any]) -> MemoryEventRecord: """Deserialise a plain dict into a :class:`MemoryEventRecord`. Required keys: ``event_id``, ``kind``, ``ts``, ``vault_id``. Optional keys: ``agent_id``, ``model_id``, ``status``, ``ttl``, ``air_id``, ``data``. Any other key is silently ignored for forward-compatibility with future schema additions — callers that care about strict validation should compare ``set(d.keys())`` against the expected schema themselves. Args: d: Dict matching the canonical Python field shape. Returns: Validated :class:`MemoryEventRecord`. Raises: ValueError: For any of the following: * *d* is not a :class:`dict`. * Any required key is missing. * ``kind`` is not a string or is not one of the 15 canonical values. * ``data`` is present and is not a :class:`dict`. * Any constructor-level validation fails (see :meth:`MemoryEventRecord.__post_init__`). """ if not isinstance(d, dict): raise ValueError(f"from_dict expects a dict, got {type(d).__name__}") required_keys = ("event_id", "kind", "ts", "vault_id") missing = [k for k in required_keys if k not in d] if missing: raise ValueError(f"missing required field(s): {', '.join(missing)}") raw_kind = d["kind"] if not isinstance(raw_kind, str): raise ValueError(f"kind must be a string, got {type(raw_kind).__name__}") if not is_valid_event_kind(raw_kind): snippet = raw_kind if len(raw_kind) <= 64 else (raw_kind[:64] + "…") raise ValueError( f"kind {snippet!r} is not one of the {len(_VALID_KIND_VALUES)} " "canonical MemoryEventKind values" ) kind = MemoryEventKind(raw_kind) raw_data = d.get("data", {}) if raw_data is None: raw_data = {} if not isinstance(raw_data, dict): raise ValueError(f"data must be a dict, got {type(raw_data).__name__}") return MemoryEventRecord( event_id=d["event_id"], kind=kind, ts=d["ts"], vault_id=d["vault_id"], agent_id=d.get("agent_id"), model_id=d.get("model_id"), status=d.get("status", "success"), ttl=d.get("ttl"), air_id=d.get("air_id"), data=raw_data, ) def to_dict(record: MemoryEventRecord) -> dict[str, Any]: """Serialise a :class:`MemoryEventRecord` into a minimal JSON-friendly dict. Optional fields whose value is ``None`` are *omitted* from the output to keep the wire format small and to make round-trip equality unambiguous. The ``kind`` field is emitted as its string value, not the enum object, so the result is JSON-serialisable as-is (``json.dumps(to_dict(r))``). The ``data`` payload is deep-copied so callers cannot mutate the record's frozen state through the returned dict. This is the only O(n)-in-data-size operation in serialisation; for the perf budget see :file:`tests/test_knowtation_events.py` Tier 6. Args: record: Record to serialise. Returns: A new dict containing the canonical wire-format keys. """ out: dict[str, Any] = { "event_id": record.event_id, "kind": record.kind.value, "ts": record.ts, "vault_id": record.vault_id, "status": record.status, "data": copy.deepcopy(record.data), } if record.agent_id is not None: out["agent_id"] = record.agent_id if record.model_id is not None: out["model_id"] = record.model_id if record.ttl is not None: out["ttl"] = record.ttl if record.air_id is not None: out["air_id"] = record.air_id return out # ────────────────────────────────────────────────────────────────────────────── # Schema content-address — computed at module load # ────────────────────────────────────────────────────────────────────────────── def _compute_schema_hash() -> str: """Return SHA-256 hex of the canonical JSON of sorted enum values. The output is deterministic given the enum membership and order-stable JSON formatting (sorted list, no whitespace). Callers may compare this against a pinned value to detect schema drift before reading commit metadata they did not write themselves. """ sorted_values = sorted(k.value for k in MemoryEventKind) canonical = json.dumps(sorted_values, sort_keys=True, separators=(",", ":")) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() #: Deterministic SHA-256 hex digest over the sorted list of canonical kind #: values. Stable across Python interpreter restarts and re-imports; #: changes only when :class:`MemoryEventKind`'s membership changes. EVENTS_SCHEMA_HASH: Final[str] = _compute_schema_hash() __all__ = [ "EVENTS_SCHEMA_HASH", "EVENTS_SCHEMA_VERSION", "MemoryEventKind", "MemoryEventRecord", "from_dict", "is_valid_event_kind", "to_dict", ]