From eadcfe6bd6abc9e181f9e2ed620e0b924317a716 Mon Sep 17 00:00:00 2001 From: LZL0 <12474488+LZL0@users.noreply.github.com> Date: Mon, 22 Jun 2026 05:03:50 +0200 Subject: [PATCH 1/3] perf: make provenance/alias walks and multi-sort index-driven Replace repeated full-scan loops and per-command re-clones with one-shot indexes, turning several O(n^2) graph operations into roughly O(n) while keeping behavior (results, ordering, and emitted event sequences) identical. - query: tuple-key _multi_sort (numeric fields) replaces cmp_to_key (~4-5x faster, byte-identical ordering incl. stable tie-breaks); add shared build_children_index helper. - integrity: cascade_retract now walks the children index and retracts in a single pass (clone items once, lazy reverse edge index) instead of one apply_command per node; get_dependents(transitive) and get_alias_group use one-shot indexes. - transplant: export_slice include_children/include_aliases walk one-shot indexes. Add tests/test_perf_equivalence.py pinning the optimized paths against inline pre-optimization references (notably byte-identical cascade event sequences), and benchmarks/perf.py contrasting naive vs optimized scaling. Measured (n=1600): cascade_retract 51x, get_alias_group 355x, get_dependents(transitive) 125x; multi-sort 5x at n=8000. --- benchmarks/perf.py | 234 +++++++++++++++++++++++++ src/memex/integrity.py | 99 +++++++---- src/memex/query.py | 42 +++-- src/memex/transplant.py | 20 ++- tests/test_perf_equivalence.py | 307 +++++++++++++++++++++++++++++++++ 5 files changed, 656 insertions(+), 46 deletions(-) create mode 100644 benchmarks/perf.py create mode 100644 tests/test_perf_equivalence.py diff --git a/benchmarks/perf.py b/benchmarks/perf.py new file mode 100644 index 0000000..807894a --- /dev/null +++ b/benchmarks/perf.py @@ -0,0 +1,234 @@ +"""Micro-benchmarks for the graph hot paths. + +Standalone (not collected by pytest — ``testpaths`` is ``tests``). Run with:: + + python benchmarks/perf.py + +Each benchmark contrasts the shipped, index-driven implementation with an inline +"naive" reference mirroring the pre-optimization code, so the speedup — and the +shift from quadratic to roughly linear scaling — is visible directly. +""" + +from __future__ import annotations + +import sys +import time +from collections.abc import Callable +from functools import cmp_to_key +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +from memex import ( # noqa: E402 + GraphState, + MemoryItem, + apply_command, + cascade_retract, + create_graph_state, + create_memory_item, + get_alias_group, + get_dependents, + get_items, +) +from memex.models import Edge # noqa: E402 +from memex.query import get_children, get_edges, get_sort_value # noqa: E402 + + +def _time(fn: Callable[[], Any], repeat: int = 1) -> float: + """Return best-of-``repeat`` wall time in milliseconds.""" + best = float("inf") + for _ in range(repeat): + t = time.perf_counter() + fn() + best = min(best, time.perf_counter() - t) + return best * 1000 + + +def _bench(title: str, sizes: list[int], variants: dict[str, Callable[[int], float]]) -> None: + print(f"\n{title}") + cols = "".join(f"{name:>18}" for name in variants) + print(f"{'n':>8}{cols}{'speedup':>12}") + for n in sizes: + times = {name: fn(n) for name, fn in variants.items()} + row = "".join(f"{times[name]:>15.2f}ms" for name in variants) + vals = list(times.values()) + speedup = (vals[0] / vals[1]) if len(vals) == 2 and vals[1] else float("nan") + tail = f"{speedup:>11.1f}x" if speedup == speedup else " " * 12 # noqa: PLR0124 + print(f"{n:>8}{row}{tail}") + + +# --------------------------------------------------------------------------- +# Builders +# --------------------------------------------------------------------------- + + +def _chain_state(n: int) -> tuple[GraphState, list[str]]: + """Linear provenance chain of n items (item k is the parent of item k+1).""" + state = create_graph_state() + ids: list[str] = [] + prev: str | None = None + for k in range(n): + item = create_memory_item( + scope="s", kind="observation", content={"k": k}, + author="a", source_kind="observed", authority=0.5, + parents=[prev] if prev else None, + ) + state = apply_command(state, {"type": "memory.create", "item": item}).state + ids.append(item.id) + prev = item.id + return state, ids + + +def _scored_state(n: int) -> GraphState: + import random + + rng = random.Random(1) + state = create_graph_state() + for k in range(n): + item = create_memory_item( + scope="s", kind="observation", content={"k": k}, + author="a", source_kind="observed", + authority=round(rng.random(), 2), importance=round(rng.random(), 2), + ) + state = apply_command(state, {"type": "memory.create", "item": item}).state + return state + + +def _alias_chain_state(n: int) -> tuple[GraphState, str]: + items = {} + edges = {} + ids = [f"a{k:06d}" for k in range(n)] + for k, id_ in enumerate(ids): + items[id_] = MemoryItem( + id=id_, scope="s", kind="observation", content={}, + author="a", source_kind="observed", authority=0.5, + ) + if k > 0: + eid = f"e{k:06d}" + edges[eid] = Edge( + edge_id=eid, from_=ids[k - 1], to=id_, kind="ALIAS", + author="a", source_kind="derived_deterministic", authority=1.0, active=True, + ) + return GraphState(items=items, edges=edges), ids[0] + + +# --------------------------------------------------------------------------- +# Naive references (pre-optimization behavior) +# --------------------------------------------------------------------------- + + +def _naive_cascade(state: GraphState, item_id: str) -> None: + visited = {item_id} + order: list[str] = [] + stack = [(c.id, "enter") for c in get_children(state, item_id)] + while stack: + fid, phase = stack.pop() + if phase == "exit": + order.append(fid) + continue + if fid in visited: + continue + visited.add(fid) + stack.append((fid, "exit")) + for c in get_children(state, fid): + if c.id not in visited: + stack.append((c.id, "enter")) + current = state + for dep in [*order, item_id]: + if dep not in current.items: + continue + current = apply_command(current, {"type": "memory.retract", "item_id": dep, "author": "x"}).state + + +def _naive_multi_sort(items: list[MemoryItem], sorts: list[dict[str, str]]) -> None: + def _cmp(a: MemoryItem, b: MemoryItem) -> int: + for s in sorts: + va = get_sort_value(a, s["field"]) + vb = get_sort_value(b, s["field"]) + if va < vb: + return -1 if s["order"] == "asc" else 1 + if va > vb: + return 1 if s["order"] == "asc" else -1 + return 0 + + sorted(items, key=cmp_to_key(_cmp)) + + +def _naive_alias_group(state: GraphState, item_id: str) -> None: + visited: set[str] = set() + queue = [item_id] + while queue: + node_id = queue.pop() + if node_id in visited: + continue + visited.add(node_id) + for edge in get_edges(state, {"from": node_id, "kind": "ALIAS", "active_only": True}): + queue.append(edge.to) + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + + +def main() -> None: + # cascade_retract over a deep chain: naive is O(depth*(N+E)); batched is ~O(N+E). + cascade_states = {n: _chain_state(n) for n in (200, 400, 800, 1600)} + _bench( + "cascade_retract - full chain retraction", + [200, 400, 800, 1600], + { + "naive": lambda n: _time(lambda: _naive_cascade(cascade_states[n][0], cascade_states[n][1][0])), + "optimized": lambda n: _time(lambda: cascade_retract(cascade_states[n][0], cascade_states[n][1][0], "x")), + }, + ) + + # multi-sort: cmp_to_key vs tuple key. + sort_states = {n: _scored_state(n) for n in (1000, 2000, 4000, 8000)} + spec = [{"field": "authority", "order": "desc"}, {"field": "importance", "order": "asc"}] + sort_items = {n: list(sort_states[n].items.values()) for n in sort_states} + _bench( + "get_items multi-sort (authority desc, importance asc)", + [1000, 2000, 4000, 8000], + { + "naive": lambda n: _time(lambda: _naive_multi_sort(sort_items[n], spec), repeat=5), + "optimized": lambda n: _time(lambda: get_items(sort_states[n], None, {"sort": spec}), repeat=5), + }, + ) + + # get_alias_group over an alias chain: naive rescans all edges per node. + alias_states = {n: _alias_chain_state(n) for n in (200, 400, 800, 1600)} + _bench( + "get_alias_group - alias chain traversal", + [200, 400, 800, 1600], + { + "naive": lambda n: _time(lambda: _naive_alias_group(*alias_states[n])), + "optimized": lambda n: _time(lambda: get_alias_group(*alias_states[n])), + }, + ) + + # get_dependents(transitive) over the chain: naive rescans all items per node. + _bench( + "get_dependents(transitive) - chain", + [200, 400, 800, 1600], + { + "naive": lambda n: _time(lambda: _naive_dependents(cascade_states[n][0], cascade_states[n][1][0])), + "optimized": lambda n: _time(lambda: get_dependents(cascade_states[n][0], cascade_states[n][1][0], True)), + }, + ) + + +def _naive_dependents(state: GraphState, item_id: str) -> None: + visited: set[str] = set() + queue = list(get_children(state, item_id)) + while queue: + item = queue.pop() + if item.id in visited: + continue + visited.add(item.id) + queue.extend(get_children(state, item.id)) + + +if __name__ == "__main__": + main() diff --git a/src/memex/integrity.py b/src/memex/integrity.py index e9e955f..b55d0ad 100644 --- a/src/memex/integrity.py +++ b/src/memex/integrity.py @@ -10,7 +10,7 @@ from .factories import create_edge from .graph import GraphState from .models import Edge, MemoryFilter, MemoryItem, MemoryLifecycleEvent, ScoredItem, ScoreWeights -from .query import get_children, get_edges, get_scored_items +from .query import build_children_index, get_children, get_edges, get_scored_items from .reducer import CommandResult, apply_command __all__ = [ @@ -149,6 +149,9 @@ def get_dependents(state: GraphState, item_id: str, transitive: bool = False) -> if not transitive: return direct + # Walk the whole dependent subtree off one children index instead of + # re-scanning every item per node (O(N + dependents), not O(dependents x N)). + children_index = build_children_index(state) visited: set[str] = set() result: list[MemoryItem] = [] queue = list(direct) @@ -158,7 +161,7 @@ def get_dependents(state: GraphState, item_id: str, transitive: bool = False) -> continue visited.add(item.id) result.append(item) - queue.extend(get_children(state, item.id)) + queue.extend(children_index.get(item.id, [])) return result @@ -176,14 +179,26 @@ def cascade_retract( ) -> CascadeResult: """Retract an item and all transitive dependents in post-order (leaves first). - Iterative post-order DFS: cycle-safe, DAG-safe (shared children), and does - not consume the call stack on deep dependency chains. The root is pre-marked - visited so a cycle pointing back to it is ignored — it's retracted last. + Iterative post-order DFS over a one-shot children index: cycle-safe, DAG-safe + (shared children), and does not consume the call stack on deep dependency + chains. The root is pre-marked visited so a cycle pointing back to it is + ignored — it's retracted last. + + The retraction itself is done in a single pass — the items dict is cloned + once and a reverse edge index is built lazily on the first retract — instead + of issuing one ``apply_command`` per dependent (each of which would re-clone + the whole state and re-scan every edge, making a deep cascade quadratic). + ``author``/``reason`` carry no information into the emitted events or state, + so the events produced here are byte-identical to the per-command path. """ + children_index = build_children_index(state) + visited: set[str] = {item_id} order: list[str] = [] - stack: list[tuple[str, str]] = [(child.id, "enter") for child in get_children(state, item_id)] + stack: list[tuple[str, str]] = [ + (child.id, "enter") for child in children_index.get(item_id, []) + ] while stack: frame_id, phase = stack.pop() if phase == "exit": @@ -193,35 +208,53 @@ def cascade_retract( continue visited.add(frame_id) stack.append((frame_id, "exit")) # processed after all children (post-order) - for child in get_children(state, frame_id): + for child in children_index.get(frame_id, []): if child.id not in visited: stack.append((child.id, "enter")) - current = state + # Dependents first (post-order), then the root. + items = dict(state.items) + edges: dict[str, Edge] | None = None + edges_by_endpoint: dict[str, list[str]] | None = None all_events: list[MemoryLifecycleEvent] = [] retracted: list[str] = [] - for dep_id in order: - if dep_id not in current.items: + for rid in (*order, item_id): + existing = items.pop(rid, None) + if existing is None: continue - r = apply_command( - current, - {"type": "memory.retract", "item_id": dep_id, "author": author, - "reason": reason if reason is not None else f"parent {item_id} retracted"}, - ) - current = r.state - all_events.extend(r.events) - retracted.append(dep_id) - - if item_id in current.items: - r = apply_command( - current, {"type": "memory.retract", "item_id": item_id, "author": author, "reason": reason} + all_events.append( + MemoryLifecycleEvent(type="memory.retracted", item=existing, cause_type="memory.retract") ) - current = r.state - all_events.extend(r.events) - retracted.append(item_id) - - return CascadeResult(current, all_events, retracted) + retracted.append(rid) + if state.edges: + if edges is None: + edges = dict(state.edges) + if edges_by_endpoint is None: + edges_by_endpoint = {} + for edge_id, edge in state.edges.items(): + edges_by_endpoint.setdefault(edge.from_, []).append(edge_id) + if edge.from_ != edge.to: + edges_by_endpoint.setdefault(edge.to, []).append(edge_id) + incident_ids = edges_by_endpoint.get(rid) + if incident_ids: + for edge_id in incident_ids: + incident_edge = edges.get(edge_id) + if incident_edge is None: + continue # already removed via its other endpoint + del edges[edge_id] + all_events.append( + MemoryLifecycleEvent(type="edge.retracted", edge=incident_edge, cause_type="memory.retract") + ) + + if not retracted: + return CascadeResult(state, all_events, retracted) + + return CascadeResult( + GraphState(items, edges if edges is not None else state.edges), + all_events, + retracted, + ) # --------------------------------------------------------------------------- @@ -273,6 +306,14 @@ def get_aliases(state: GraphState, item_id: str) -> list[MemoryItem]: def get_alias_group(state: GraphState, item_id: str) -> list[MemoryItem]: + # Index active outbound ALIAS edges once (from_ -> [to]) so the BFS does not + # re-scan every edge per node; per-source order matches get_edges (edge + # insertion order), keeping the walk deterministic. + alias_out: dict[str, list[str]] = {} + for edge in state.edges.values(): + if edge.kind == "ALIAS" and edge.active: + alias_out.setdefault(edge.from_, []).append(edge.to) + visited: set[str] = set() result: list[MemoryItem] = [] queue = [item_id] @@ -284,8 +325,8 @@ def get_alias_group(state: GraphState, item_id: str) -> list[MemoryItem]: item = state.items.get(node_id) if item is not None: result.append(item) - for edge in get_edges(state, {"from": node_id, "kind": "ALIAS", "active_only": True}): - queue.append(edge.to) + for to in alias_out.get(node_id, []): + queue.append(to) return result diff --git a/src/memex/query.py b/src/memex/query.py index b3d9c78..717032f 100644 --- a/src/memex/query.py +++ b/src/memex/query.py @@ -9,7 +9,6 @@ from __future__ import annotations import math -from functools import cmp_to_key from typing import Any from pydantic import BaseModel @@ -43,6 +42,7 @@ "get_related_items", "get_parents", "get_children", + "build_children_index", "compute_decay_multiplier", "compute_score", "get_sort_value", @@ -284,17 +284,18 @@ def get_sort_value(item: MemoryItem, field: str) -> float: def _multi_sort(items: list[MemoryItem], sorts: list[SortOption]) -> list[MemoryItem]: - def _cmp(a: MemoryItem, b: MemoryItem) -> int: - for s in sorts: - va = get_sort_value(a, s.field) - vb = get_sort_value(b, s.field) - if va < vb: - return -1 if s.order == "asc" else 1 - if va > vb: - return 1 if s.order == "asc" else -1 - return 0 + # All sort fields are numeric, so a tuple key reproduces the JS comparator + # exactly — negating descending fields gives the same ordering, and Python's + # stable sort preserves insertion order on full ties (the comparator's ``0``). + # This is ~4x faster than ``cmp_to_key`` (no per-comparison Python call, and + # ``get_sort_value`` is evaluated once per item instead of O(n log n) times). + def _key(item: MemoryItem) -> tuple[float, ...]: + return tuple( + get_sort_value(item, s.field) if s.order == "asc" else -get_sort_value(item, s.field) + for s in sorts + ) - return sorted(items, key=cmp_to_key(_cmp)) + return sorted(items, key=_key) # --------------------------------------------------------------------------- @@ -433,3 +434,22 @@ def get_children(state: GraphState, item_id: str) -> list[MemoryItem]: for item in state.items.values() if item.parents and item_id in item.parents ] + + +def build_children_index(state: GraphState) -> dict[str, list[MemoryItem]]: + """Map ``parent_id -> [child items]`` in one pass over the items. + + Equivalent to calling :func:`get_children` for every id, but O(N) total + instead of O(N) per id — callers that walk provenance in a loop (cascade + retraction, transitive dependents, child-inclusive export) build this once + and look up O(1). Per-bucket order follows item insertion order, matching + ``get_children``; duplicate parent ids on a single item are de-duplicated so + a child still appears at most once per parent. + """ + index: dict[str, list[MemoryItem]] = {} + for item in state.items.values(): + if not item.parents: + continue + for pid in dict.fromkeys(item.parents): + index.setdefault(pid, []).append(item) + return index diff --git a/src/memex/transplant.py b/src/memex/transplant.py index eed0e67..de82f44 100644 --- a/src/memex/transplant.py +++ b/src/memex/transplant.py @@ -18,7 +18,7 @@ from .graph import GraphState from .intent import Intent, IntentState, apply_intent_command from .models import Edge, MemoryItem -from .query import extract_timestamp, get_children, get_edges +from .query import build_children_index, extract_timestamp from .reducer import apply_command from .task import Task, TaskState, apply_task_command @@ -133,18 +133,26 @@ def export_slice( memory_id_set.add(pid) queue.append(pid) - # walk children down-graph + # walk children down-graph (one children index, not a full scan per node) if include_children: + children_index = build_children_index(mem_state) queue = list(memory_id_set) while queue: id_ = queue.pop() - for child in get_children(mem_state, id_): + for child in children_index.get(id_, []): if child.id not in memory_id_set: memory_id_set.add(child.id) queue.append(child.id) - # walk aliases (both directions) + # walk aliases (both directions) off a one-shot ALIAS adjacency index if include_aliases: + alias_out: dict[str, list[Edge]] = {} + alias_in: dict[str, list[Edge]] = {} + for edge in mem_state.edges.values(): + if edge.kind == "ALIAS" and edge.active: + alias_out.setdefault(edge.from_, []).append(edge) + alias_in.setdefault(edge.to, []).append(edge) + queue = list(memory_id_set) visited: set[str] = set() while queue: @@ -152,12 +160,12 @@ def export_slice( if id_ in visited: continue visited.add(id_) - for edge in get_edges(mem_state, {"from": id_, "kind": "ALIAS", "active_only": True}): + for edge in alias_out.get(id_, []): edge_id_set.add(edge.edge_id) if edge.to not in memory_id_set: memory_id_set.add(edge.to) queue.append(edge.to) - for edge in get_edges(mem_state, {"to": id_, "kind": "ALIAS", "active_only": True}): + for edge in alias_in.get(id_, []): edge_id_set.add(edge.edge_id) if edge.from_ not in memory_id_set: memory_id_set.add(edge.from_) diff --git a/tests/test_perf_equivalence.py b/tests/test_perf_equivalence.py new file mode 100644 index 0000000..880e5a7 --- /dev/null +++ b/tests/test_perf_equivalence.py @@ -0,0 +1,307 @@ +"""Equivalence guards for the performance optimizations. + +Each optimization replaced an O(n^2) loop (repeated full scans / per-command +re-clones) with an index-driven single pass. Behavior — result *and* ordering, +and for ``cascade_retract`` the full emitted event sequence — must stay +byte-identical to the straightforward implementation. Each test below pins the +optimized function against an inline reference that mirrors the original code. +""" + +from __future__ import annotations + +from functools import cmp_to_key +from typing import Any + +from memex import ( + Edge, + GraphState, + MemoryItem, + apply_command, + cascade_retract, + get_alias_group, + get_dependents, + get_items, +) +from memex.query import build_children_index, get_children, get_edges, get_sort_value + + +def make_item(id: str, **overrides: Any) -> MemoryItem: + base: dict[str, Any] = { + "id": id, "scope": "test", "kind": "observation", "content": {}, + "author": "user:laz", "source_kind": "observed", "authority": 0.8, + } + base.update(overrides) + return MemoryItem(**base) + + +def make_edge(edge_id: str, from_: str, to: str, kind: str = "SUPPORTS") -> Edge: + return Edge( + edge_id=edge_id, from_=from_, to=to, kind=kind, + author="system:rule", source_kind="derived_deterministic", + authority=0.8, active=True, + ) + + +def state_with(items: list[MemoryItem], edges: list[Edge] | None = None) -> GraphState: + return GraphState( + items={i.id: i for i in items}, + edges={e.edge_id: e for e in (edges or [])}, + ) + + +def _event_key(e: Any) -> tuple[str, str]: + return (e.type, e.item.id if e.item is not None else e.edge.edge_id) + + +# --------------------------------------------------------------------------- +# Reference (pre-optimization) implementations +# --------------------------------------------------------------------------- + + +def _naive_cascade(state: GraphState, item_id: str, author: str) -> tuple[GraphState, list[Any], list[str]]: + """The original per-command cascade: one apply_command per retraction.""" + visited: set[str] = {item_id} + order: list[str] = [] + stack = [(c.id, "enter") for c in get_children(state, item_id)] + while stack: + fid, phase = stack.pop() + if phase == "exit": + order.append(fid) + continue + if fid in visited: + continue + visited.add(fid) + stack.append((fid, "exit")) + for c in get_children(state, fid): + if c.id not in visited: + stack.append((c.id, "enter")) + + current = state + events: list[Any] = [] + retracted: list[str] = [] + for dep in order: + if dep not in current.items: + continue + r = apply_command(current, {"type": "memory.retract", "item_id": dep, "author": author}) + current = r.state + events.extend(r.events) + retracted.append(dep) + if item_id in current.items: + r = apply_command(current, {"type": "memory.retract", "item_id": item_id, "author": author}) + current = r.state + events.extend(r.events) + retracted.append(item_id) + return current, events, retracted + + +def _naive_alias_group(state: GraphState, item_id: str) -> list[MemoryItem]: + visited: set[str] = set() + result: list[MemoryItem] = [] + queue = [item_id] + while queue: + node_id = queue.pop() + if node_id in visited: + continue + visited.add(node_id) + item = state.items.get(node_id) + if item is not None: + result.append(item) + for edge in get_edges(state, {"from": node_id, "kind": "ALIAS", "active_only": True}): + queue.append(edge.to) + return result + + +def _naive_dependents(state: GraphState, item_id: str) -> list[MemoryItem]: + visited: set[str] = set() + result: list[MemoryItem] = [] + queue = list(get_children(state, item_id)) + while queue: + item = queue.pop() + if item.id in visited: + continue + visited.add(item.id) + result.append(item) + queue.extend(get_children(state, item.id)) + return result + + +# --------------------------------------------------------------------------- +# cascade_retract — full state + event-sequence equivalence +# --------------------------------------------------------------------------- + + +def _diamond_state_with_edges() -> GraphState: + # a -> {b, c} -> d (diamond), plus survivor s. Edges exercise: an edge + # between two doomed items, an edge from a doomed item to the survivor, + # a self-edge on the root, and an untouched survivor self-edge. + items = [ + make_item("a"), + make_item("b", parents=["a"]), + make_item("c", parents=["a"]), + make_item("d", parents=["b", "c"]), + make_item("s"), + ] + edges = [ + make_edge("e1", "a", "b"), + make_edge("e2", "b", "c", kind="CONTRADICTS"), + make_edge("e3", "d", "s", kind="ABOUT"), + make_edge("e4", "a", "a", kind="ALIAS"), # self-edge on doomed root + make_edge("e5", "s", "s", kind="ABOUT"), # survivor self-edge, never touched + ] + return state_with(items, edges) + + +def test_cascade_retract_matches_naive_oracle_with_edges() -> None: + state = _diamond_state_with_edges() + + new_state, new_events, new_retracted = cascade_retract(state, "a", "system:cleanup") + ref_state, ref_events, ref_retracted = _naive_cascade(state, "a", "system:cleanup") + + # Identical retraction order, surviving items, surviving edges... + assert new_retracted == ref_retracted + assert list(new_state.items.keys()) == list(ref_state.items.keys()) + assert list(new_state.edges.keys()) == list(ref_state.edges.keys()) + # ...and an identical emitted event sequence (memory + edge events, in order). + assert [_event_key(e) for e in new_events] == [_event_key(e) for e in ref_events] + + # Concretely: everything but the survivor is gone, and the only surviving + # edge is the survivor's self-edge. + assert set(new_state.items.keys()) == {"s"} + assert set(new_state.edges.keys()) == {"e5"} + + +def test_cascade_retract_shared_edge_emitted_once() -> None: + # e2 connects b and c, both retracted. The naive path removes it when the + # first endpoint is retracted; the batched path must do the same (emit once). + state = _diamond_state_with_edges() + _, events, _ = cascade_retract(state, "a", "system:cleanup") + edge_events = [e for e in events if e.type == "edge.retracted"] + edge_ids = [e.edge.edge_id for e in edge_events] + assert sorted(edge_ids) == ["e1", "e2", "e3", "e4"] + assert len(edge_ids) == len(set(edge_ids)) # no double-emission + + +def test_cascade_retract_no_edges_keeps_edges_identity() -> None: + # With no edges the optimized path must not needlessly clone the edges dict. + state = state_with([make_item("a"), make_item("b", parents=["a"])]) + new_state, events, retracted = cascade_retract(state, "a", "system:cleanup") + assert retracted == ["b", "a"] + assert new_state.edges is state.edges # untouched, same object + assert all(e.type == "memory.retracted" for e in events) + + +def test_cascade_retract_orphan_root_not_retracted() -> None: + # item_id absent but referenced as a (stale) parent: dependents are still + # cascaded, the missing root is silently skipped — same as the naive path. + state = state_with([make_item("child", parents=["ghost"])]) + new_state, _events, retracted = cascade_retract(state, "ghost", "system:cleanup") + ref_state, _, ref_retracted = _naive_cascade(state, "ghost", "system:cleanup") + assert retracted == ref_retracted == ["child"] + assert list(new_state.items.keys()) == list(ref_state.items.keys()) + + +# --------------------------------------------------------------------------- +# build_children_index +# --------------------------------------------------------------------------- + + +def test_build_children_index_matches_get_children() -> None: + state = state_with([ + make_item("p1"), + make_item("p2"), + make_item("c1", parents=["p1"]), + make_item("c2", parents=["p1", "p2"]), + make_item("c3", parents=["p2"]), + ]) + index = build_children_index(state) + for pid in ("p1", "p2", "c1", "missing"): + assert index.get(pid, []) == get_children(state, pid) + + +def test_build_children_index_dedupes_duplicate_parents() -> None: + # A child listing the same parent twice still appears once (matches the + # boolean membership test in get_children). + state = state_with([make_item("p"), make_item("c", parents=["p", "p"])]) + index = build_children_index(state) + assert [i.id for i in index["p"]] == ["c"] + assert index["p"] == get_children(state, "p") + + +# --------------------------------------------------------------------------- +# get_dependents (transitive) + get_alias_group ordering +# --------------------------------------------------------------------------- + + +def test_get_dependents_transitive_matches_naive() -> None: + state = state_with([ + make_item("r"), + make_item("a", parents=["r"]), + make_item("b", parents=["r"]), + make_item("c", parents=["a", "b"]), + make_item("d", parents=["c"]), + ]) + assert get_dependents(state, "r", True) == _naive_dependents(state, "r") + + +def test_get_alias_group_order_matches_naive() -> None: + # Branching alias network; BFS visitation order must match the reference. + items = [make_item(x) for x in ("a", "b", "c", "d", "e")] + edges = [ + make_edge("e1", "a", "b", kind="ALIAS"), + make_edge("e2", "a", "c", kind="ALIAS"), + make_edge("e3", "b", "d", kind="ALIAS"), + make_edge("e4", "c", "e", kind="ALIAS"), + make_edge("e5", "x", "y", kind="ALIAS"), # unrelated component + ] + state = state_with(items, edges) + assert [i.id for i in get_alias_group(state, "a")] == [i.id for i in _naive_alias_group(state, "a")] + + +# --------------------------------------------------------------------------- +# _multi_sort (tuple key) vs the original cmp_to_key comparator +# --------------------------------------------------------------------------- + + +def _cmp_reference(items: list[MemoryItem], sorts: list[dict[str, str]]) -> list[MemoryItem]: + def _cmp(a: MemoryItem, b: MemoryItem) -> int: + for s in sorts: + va = get_sort_value(a, s["field"]) + vb = get_sort_value(b, s["field"]) + if va < vb: + return -1 if s["order"] == "asc" else 1 + if va > vb: + return 1 if s["order"] == "asc" else -1 + return 0 + + return sorted(items, key=cmp_to_key(_cmp)) + + +def test_multi_sort_matches_cmp_reference_with_ties() -> None: + # Coarse score values force many ties so the stable insertion-order + # tie-break is exercised; created_at drives `recency`. + items: list[MemoryItem] = [] + for k in range(60): + items.append(make_item( + f"m{k:02d}", + authority=round((k * 7 % 5) / 4, 2), + conviction=round((k * 3 % 4) / 3, 2), + importance=round((k % 3) / 2, 2), + created_at=1_700_000_000_000 + (k % 6), + )) + state = state_with(items) + + sort_specs = [ + [{"field": "authority", "order": "desc"}], + [{"field": "importance", "order": "asc"}, {"field": "authority", "order": "desc"}], + [{"field": "authority", "order": "desc"}, {"field": "recency", "order": "desc"}], + [ + {"field": "authority", "order": "desc"}, + {"field": "conviction", "order": "asc"}, + {"field": "importance", "order": "desc"}, + {"field": "recency", "order": "asc"}, + ], + ] + for spec in sort_specs: + got = get_items(state, None, {"sort": spec}) + expected = _cmp_reference(items, spec) + assert [i.id for i in got] == [i.id for i in expected], spec From 16f7f5af70fbd6f812a48d3cdd94f608750ba3f3 Mon Sep 17 00:00:00 2001 From: LZL0 <12474488+LZL0@users.noreply.github.com> Date: Mon, 22 Jun 2026 05:08:33 +0200 Subject: [PATCH 2/3] Create API.md --- API.md | 923 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 923 insertions(+) create mode 100644 API.md diff --git a/API.md b/API.md new file mode 100644 index 0000000..7a2ea1f --- /dev/null +++ b/API.md @@ -0,0 +1,923 @@ +# memex-python — API reference + +Complete reference for the public surface re-exported from the top-level `memex` +package. Everything below is importable directly: + +```python +from memex import create_graph_state, apply_command, get_scored_items, MemexStore +``` + +This document covers the **functional core** (pure `state -> (new_state, events)` +functions), the typed **Pydantic models**, and the optional **`MemexStore`** +facade. For the conceptual overview see [`README.md`](README.md); for design +rationale see [`PLAN.md`](PLAN.md). + +## Contents + +- [Conventions](#conventions) +- [Core concepts](#core-concepts) +- [Graph state](#graph-state) +- [Entities & factories](#entities--factories) +- [Commands & the reducer](#commands--the-reducer) +- [Querying & filtering](#querying--filtering) +- [Scoring, decay & sorting](#scoring-decay--sorting) +- [Retrieval](#retrieval) +- [Integrity: contradictions, aliases, stale, cascade](#integrity) +- [Bulk operations](#bulk-operations) +- [Intent graph](#intent-graph) +- [Task graph](#task-graph) +- [Statistics](#statistics) +- [Replay](#replay) +- [Serialization](#serialization) +- [Event envelopes](#event-envelopes) +- [Transplant: export / import](#transplant) +- [Validation](#validation) +- [`MemexStore` facade](#memexstore-facade) +- [UUID helpers](#uuid-helpers) +- [Errors](#errors) +- [Type aliases](#type-aliases) + +--- + +## Conventions + +**Pure & immutable.** Every mutation has the shape +`f(state, ...) -> (new_state, events)`. The input `state` is never modified; a +fresh `GraphState` is returned with the relevant dict cloned. Entities +(`MemoryItem`, `Edge`, `Intent`, `Task`) are **frozen** Pydantic models — an +"edit" produces a new instance via `model_copy(update=...)`. + +**Dicts or models.** Public functions that take a filter, options, or weights +accept either a plain `dict` or the typed model; dicts are validated through the +model. The examples mix both styles freely. + +**Validation is always on.** Constructing any model with an out-of-range score +(`authority`/`conviction`/`importance`/`weight`/`priority` must be `0..1`) raises +`pydantic.ValidationError`. Use `Model.model_construct(...)` to deliberately +bypass validation (e.g. test fixtures). + +**Keyword-only constructors.** All `create_*` factories take keyword arguments +only. + +**Reserved-word fields.** `from`, `not`, and `or` are Python keywords, so the +attributes are `from_`, `not_`, and `or_`; each serializes to/accepts its bare +JSON alias (`"from"`, `"not"`, `"or"`). + +**Wire compatibility.** Command tags (`"memory.create"`), enum values +(`"derived_deterministic"`, `"DERIVED_FROM"`), and JSON keys are byte-identical +to the TypeScript `@ai2070/memex`, so a Python and a TS service can share one +event log. + +--- + +## Core concepts + +Three independent graphs, each driven by the same `commands -> reducer -> +lifecycle events` pattern: + +| Graph | State | Core type | Reducer | Namespace | +|--------|--------------|--------------|--------------------------|------------| +| Memory | `GraphState` | `MemoryItem` | `apply_command` | `"memory"` | +| Intent | `IntentState`| `Intent` | `apply_intent_command` | `"intent"` | +| Task | `TaskState` | `Task` | `apply_task_command` | `"task"` | + +Each `MemoryItem` carries three orthogonal `0..1` scores — **authority** (trust), +**conviction** (author confidence), **importance** (current salience) — plus +provenance (`parents`) and typed `edges`. + +```python +from memex import create_graph_state, create_memory_item, apply_command, get_scored_items + +state = create_graph_state() +item = create_memory_item( + scope="user:laz/general", kind="observation", + content={"key": "login_count", "value": 42}, + author="agent:monitor", source_kind="observed", + authority=0.9, importance=0.7, +) +state, events = apply_command(state, {"type": "memory.create", "item": item}) +top = get_scored_items(state, {"authority": 1.0, "importance": 0.5}) +``` + +--- + +## Graph state + +### `class GraphState` + +Frozen dataclass holding the memory graph. Not a Pydantic model (re-validating +every item per command would be unusably slow). + +| Field | Type | Description | +|---------|-------------------------|------------------------| +| `items` | `dict[str, MemoryItem]` | id → item (insertion-ordered) | +| `edges` | `dict[str, Edge]` | id → edge (insertion-ordered) | + +### `create_graph_state() -> GraphState` + +A new empty memory graph. + +### `clone_graph_state(state: GraphState) -> GraphState` + +Shallow clone — new dicts, shared (immutable) item/edge instances. + +--- + +## Entities & factories + +### `class MemoryItem` + +Frozen. The core memory node. + +| Field | Type | Notes | +|---------------|-----------------------|-----------------------------------------| +| `id` | `str` | usually a UUIDv7 | +| `scope` | `str` | namespacing key, e.g. `"user:laz/general"` | +| `kind` | `str` | open union; see [`KnownMemoryKind`](#type-aliases) | +| `content` | `dict[str, Any]` | arbitrary payload | +| `author` | `str` | | +| `source_kind` | `str` | open union; see [`KnownSourceKind`](#type-aliases) | +| `parents` | `list[str] \| None` | provenance ids | +| `authority` | `float` (0..1) | required | +| `conviction` | `float \| None` (0..1)| | +| `importance` | `float \| None` (0..1)| | +| `created_at` | `int \| None` | ms epoch; falls back to the UUIDv7 ts | +| `intent_id` | `str \| None` | cross-graph link | +| `task_id` | `str \| None` | cross-graph link | +| `meta` | `dict[str, Any] \| None` | | + +### `class Edge` + +Frozen, `populate_by_name=True`. + +| Field | Type | Notes | +|---------------|-----------------------|-----------------------------------------| +| `edge_id` | `str` | | +| `from_` | `str` | JSON alias `"from"` | +| `to` | `str` | | +| `kind` | `str` | see [`KnownEdgeKind`](#type-aliases) | +| `weight` | `float \| None` (0..1)| | +| `author` | `str` | | +| `source_kind` | `str` | | +| `authority` | `float` (0..1) | required | +| `active` | `bool` | | +| `meta` | `dict[str, Any] \| None` | | + +### `create_memory_item(...) -> MemoryItem` + +```python +create_memory_item(*, scope, kind, content, author, source_kind, authority, + id=None, parents=None, conviction=None, importance=None, + created_at=None, intent_id=None, task_id=None, meta=None) +``` + +Mints `id` (UUIDv7) when omitted. `created_at` is the explicit value, else the +id's UUIDv7 timestamp, else the wall clock. Validates score bounds. + +### `create_edge(...) -> Edge` + +```python +create_edge(*, from_, to, kind, author, source_kind, authority, + edge_id=None, active=None, weight=None, meta=None) +``` + +`active` defaults to `True`, `edge_id` to a fresh UUIDv7. + +--- + +## Commands & the reducer + +Commands are a Pydantic discriminated union keyed on `type`. `apply_command` +accepts a command **model** or a plain **dict** (validated via +`MemoryCommandAdapter`). + +### Memory commands + +| Model | `type` | Fields (besides `type`) | +|-----------------|--------------------|-----------------------------------------------| +| `MemoryCreate` | `"memory.create"` | `item: MemoryItem` | +| `MemoryUpdate` | `"memory.update"` | `item_id`, `partial: dict`, `author`, `reason?`, `basis?` | +| `MemoryRetract` | `"memory.retract"` | `item_id`, `author`, `reason?` | +| `EdgeCreate` | `"edge.create"` | `edge: Edge` | +| `EdgeUpdate` | `"edge.update"` | `edge_id`, `partial: dict`, `author`, `reason?` | +| `EdgeRetract` | `"edge.retract"` | `edge_id`, `author`, `reason?` | + +- `MemoryCommand` — the `Annotated[Union[...], discriminator="type"]` alias. +- `MemoryCommandAdapter` — `TypeAdapter[MemoryCommand]` for validating raw dicts. + +### `apply_command(state, cmd) -> CommandResult` + +`cmd: MemoryCommand | dict`. Returns a `CommandResult` NamedTuple +`(state: GraphState, events: list[MemoryLifecycleEvent])`. The result is +iterable, so `state, events = apply_command(...)` works. + +Behavior per command: + +- **create** — raises `DuplicateMemoryError` / `DuplicateEdgeError` on id clash. +- **update** — shallow-merges `partial` (for `content`/`meta`, merges keys; keys + cannot be deleted via update); `id`/`created_at` are immutable for items, + `edge_id`/`from`/`to` for edges. Raises `MemoryNotFoundError` / + `EdgeNotFoundError` if absent. Updates do **not** re-validate scores. +- **retract (memory)** — removes the item *and* every incident edge, emitting one + `edge.retracted` per removed edge. + +Emits [`MemoryLifecycleEvent`](#class-memorylifecycleevent)s. + +### `merge_item(existing: MemoryItem, partial: dict) -> MemoryItem` +### `merge_edge(existing: Edge, partial: dict) -> Edge` + +The merge primitives used by the reducer (no validation, mirroring the TS +"updates don't validate" guarantee). Exposed for advanced/bulk callers. + +### `class MemoryLifecycleEvent` + +| Field | Type | Notes | +|--------------|-------------------------------|-------------------------------| +| `namespace` | `Literal["memory"]` | always `"memory"` | +| `type` | [`LifecycleEventType`](#type-aliases) | e.g. `"memory.created"` | +| `item` | `MemoryItem \| None` | set for memory events | +| `edge` | `Edge \| None` | set for edge events | +| `cause_type` | `str \| None` | the command tag that caused it| + +--- + +## Querying & filtering + +### `get_items(state, filter=None, options=None) -> list[MemoryItem]` + +Filter, then sort, then page. `filter: MemoryFilter | dict | None`, +`options: QueryOptions | dict | None`. With no filter, returns all items in +insertion order. + +### `get_item_by_id(state, id) -> MemoryItem | None` + +### `matches_filter(item, f: MemoryFilter) -> bool` + +The predicate behind `get_items` — useful for ad-hoc filtering. + +### `class MemoryFilter` + +All present conditions are AND-combined (except `or_`). `populate_by_name=True`. + +| Field | Type | Matches when… | +|----------------|----------------------------|--------------------------------------------------| +| `ids` | `list[str]` | `item.id` is in the list | +| `scope` | `str` | exact scope | +| `scope_prefix` | `str` | `item.scope.startswith(...)` | +| `author` | `str` | exact | +| `kind` | `str` | exact | +| `source_kind` | `str` | exact | +| `range` | `ScoreRanges` | each score within its `Range` | +| `intent_id` | `str` | exact | +| `intent_ids` | `list[str]` | `item.intent_id` in the list | +| `task_id` | `str` | exact | +| `task_ids` | `list[str]` | `item.task_id` in the list | +| `has_parent` | `str` | id is in `item.parents` | +| `is_root` | `bool` | `True`: no parents; `False`: has parents | +| `parents` | `ParentsFilter` | see below | +| `decay` | `DecayFilter` | decay multiplier ≥ `min` | +| `created` | `CreatedFilter` | timestamp window | +| `not_` | `MemoryFilter` (alias `not`) | sub-filter does **not** match | +| `meta` | `dict[str, Any]` | each dotted path equals the value | +| `meta_has` | `list[str]` | each dotted path exists | +| `or_` | `list[MemoryFilter]` (alias `or`) | at least one sub-filter matches | + +Supporting models: + +- **`Range`** — `min: float | None`, `max: float | None` (inclusive bounds). +- **`ScoreRanges`** — `authority`, `conviction`, `importance`, each a `Range`. +- **`ParentsFilter`** — `includes: str`, `includes_any: list[str]`, + `includes_all: list[str]`, `count: Range` (range over the number of parents). +- **`DecayFilter`** — `config: DecayConfig`, `min: float` (0..1). +- **`CreatedFilter`** — `before: int | None` (exclusive upper, `ts < before`), + `after: int | None` (inclusive lower, `ts >= after`); both ms epoch. + +`meta`/`meta_has` paths are dotted (`"a.b.c"`) and walk nested dicts. + +### `class QueryOptions` + +| Field | Type | Notes | +|----------|-----------------------------------|--------------------------------| +| `sort` | `SortOption \| list[SortOption] \| None` | multi-key, applied in order | +| `limit` | `int \| None` (≥0) | applied after sort | +| `offset` | `int \| None` (≥0) | applied after sort | + +### `class SortOption` + +`field: "authority" | "conviction" | "importance" | "recency"`, +`order: "asc" | "desc"`. `recency` sorts by item timestamp. Ties preserve +insertion order (stable). An unknown field raises `ValueError`. + +### Edges + +- **`get_edges(state, filter=None) -> list[Edge]`** — `filter: EdgeFilter | dict | None`. + When no filter is given, only **active** edges are returned. +- **`get_edge_by_id(state, edge_id) -> Edge | None`** + +#### `class EdgeFilter` + +| Field | Type | Notes | +|--------------|-----------------|--------------------------------------| +| `from_` | `str` (alias `from`) | exact source | +| `to` | `str` | exact target | +| `kind` | `str` | exact | +| `min_weight` | `float` | `edge.weight >= min_weight` | +| `active_only`| `bool \| None` | default `True`; `False` includes retracted edges | + +### Navigation + +- **`get_parents(state, item_id) -> list[MemoryItem]`** — resolves `item.parents` + to the items that exist. +- **`get_children(state, item_id) -> list[MemoryItem]`** — items listing `item_id` + in their `parents`. +- **`get_related_items(state, item_id, direction="both") -> list[MemoryItem]`** — + items connected by active edges. `direction: "from" | "to" | "both"`. Returns + a deduplicated, insertion-ordered list excluding the item itself. + +--- + +## Scoring, decay & sorting + +### `class ScoreWeights` + +Multipliers (intentionally unbounded — not `0..1`). + +| Field | Type | Notes | +|--------------|-----------------------|--------------------------------| +| `authority` | `float \| None` | weight on `item.authority` | +| `conviction` | `float \| None` | weight on `item.conviction` | +| `importance` | `float \| None` | weight on `item.importance` | +| `decay` | `DecayConfig \| None` | if set, multiplies the score | + +### `class DecayConfig` + +| Field | Type | Notes | +|------------|----------------|-----------------------------------------| +| `rate` | `float` (0..1) | per-interval decay rate | +| `interval` | `str` | `"hour"` \| `"day"` \| `"week"` | +| `type` | `str` | `"exponential"` \| `"linear"` \| `"step"` | + +### `compute_decay_multiplier(item, decay: DecayConfig) -> float` + +`intervals = age_ms / interval_ms`, where `age_ms = now - item_timestamp`. + +- **exponential** → `(1 - rate) ** intervals` +- **linear** → `max(0, 1 - rate * intervals)` +- **step** → `(1 - rate) ** floor(intervals)` + +Future-dated items (age ≤ 0, clock skew) return `1.0`. Unknown `interval`/`type` +raises `ValueError`. + +### `compute_score(item, weights: ScoreWeights) -> float` + +`authority*w.authority + conviction*w.conviction + importance*w.importance`, +then `* compute_decay_multiplier(...)` if `weights.decay` is set. Missing weights +and missing scores are treated as `0`. + +### `get_scored_items(state, weights, options=None) -> list[ScoredItem]` + +`weights: ScoreWeights | dict`, `options: ScoredQueryOptions | dict | None`. +Pipeline: `pre`-filter → score → sort by score descending → `min_score` → +`post`-filter → `offset`/`limit`. + +#### `class ScoredQueryOptions` + +| Field | Type | Notes | +|-------------|----------------------|------------------------------------| +| `pre` | `MemoryFilter \| None` | filter applied before scoring | +| `post` | `MemoryFilter \| None` | filter applied after scoring | +| `min_score` | `float \| None` | drop items scoring below this | +| `limit` | `int \| None` | | +| `offset` | `int \| None` | | + +#### `class ScoredItem` + +`item: MemoryItem`, `score: float`, `contradicted_by: list[MemoryItem] | None`. +Not frozen — `surface_contradictions` annotates `contradicted_by` in place. + +### `extract_timestamp(uuid_id: str) -> int` + +Extract the ms timestamp from a UUIDv7 id. Raises `InvalidTimestampError` on +anything that isn't a valid v7 UUID. + +--- + +## Retrieval + +### Provenance walks + +- **`get_support_tree(state, item_id) -> SupportNode | None`** — full provenance + tree, deduplicating on cycles. Returns `None` if the item is absent. +- **`get_support_set(state, item_id) -> list[MemoryItem]`** — flattened set of + items that justify a claim. +- **`class SupportNode`** — dataclass `item: MemoryItem`, `parents: list[SupportNode]`. + +### Contradiction policies + +Both remove **superseded** items (targets of active `SUPERSEDES` edges). + +- **`filter_contradictions(state, scored) -> list[ScoredItem]`** — drops the + lower-scoring side of each unresolved `CONTRADICTS` pair (deterministic + tie-breaks by score then `edge_id`). +- **`surface_contradictions(state, scored) -> list[ScoredItem]`** — keeps both + sides, annotating each via `contradicted_by`. Self-edges are ignored. + +### Diversity + +- **`class DiversityOptions`** — `author_penalty`, `parent_penalty`, + `source_penalty` (each `float | None`). +- **`apply_diversity(scored, options) -> list[ScoredItem]`** — subtracts a + per-duplicate penalty (cumulative per repeated author / parent / source), + clamps at `0`, and re-sorts by score descending. + +### `smart_retrieve(...) -> list[ScoredItem]` + +```python +smart_retrieve(state, *, budget, cost_fn, weights, + filter=None, contradictions=None, diversity=None) +``` + +Score → contradiction policy → diversity → greedy budget pack. `cost_fn: +Callable[[MemoryItem], float]` must return a finite, non-negative number +(otherwise `ValueError`). `contradictions: "filter" | "surface" | None`. +`diversity: DiversityOptions | dict | None`. Greedily appends items whose cost +fits the remaining `budget`. + +```python +packed = smart_retrieve( + state, budget=2000, cost_fn=lambda i: len(str(i.content)), + weights={"authority": 0.6, "importance": 0.4}, + contradictions="surface", diversity={"author_penalty": 0.2}, +) +``` + +### `get_items_by_budget(...) -> list[ScoredItem]` + +```python +get_items_by_budget(state, *, budget, cost_fn, weights, filter=None) +``` + +The budget-pack step without contradiction/diversity passes. + +--- + +## Integrity + +Contradiction & alias management, stale detection, and cascade retraction. + +### Contradictions + +- **`get_contradictions(state) -> list[Contradiction]`** — active `CONTRADICTS` + pairs whose endpoints both exist. +- **`mark_contradiction(state, item_id_a, item_id_b, author, meta=None) -> CommandResult`** + — creates a `CONTRADICTS` edge. +- **`resolve_contradiction(state, winner_id, loser_id, author, reason=None) -> CommandResult`** + — retracts the `CONTRADICTS` edge(s) between the pair, adds a `SUPERSEDES` + edge (winner → loser), and drops the loser's `authority` to 10%. A stale call + with no matching edge is a no-op. +- **`class Contradiction`** — `a: MemoryItem`, `b: MemoryItem`, `edge: Edge | None`. + +### Stale items & dependents + +- **`get_stale_items(state) -> list[StaleItem]`** — items whose `parents` + reference ids no longer present. +- **`class StaleItem`** — `item: MemoryItem`, `missing_parents: list[str]`. +- **`get_dependents(state, item_id, transitive=False) -> list[MemoryItem]`** — + direct children, or the whole dependent subtree when `transitive=True` + (cycle-safe). + +### Cascade retraction + +- **`cascade_retract(state, item_id, author, reason=None) -> CascadeResult`** — + retracts an item and all transitive dependents in post-order (leaves first), + cleaning incident edges. Cycle- and DAG-safe; iterative (no recursion limit). +- **`class CascadeResult`** — `state: GraphState`, + `events: list[MemoryLifecycleEvent]`, `retracted: list[str]` (ids in + retraction order). + +### Aliases (identity) + +- **`mark_alias(state, item_id_a, item_id_b, author, meta=None) -> CommandResult`** + — creates bidirectional `ALIAS` edges. Self-alias is a no-op. +- **`get_aliases(state, item_id) -> list[MemoryItem]`** — direct alias targets. +- **`get_alias_group(state, item_id) -> list[MemoryItem]`** — the full connected + alias component (transitive closure), including the item itself. + +--- + +## Bulk operations + +Single-pass transforms that clone the state once instead of per command. + +### `apply_many(state, filter, transform, author, reason=None, options=None) -> CommandResult` + +Applies `transform: Callable[[MemoryItem], dict | None]` to every item matching +`filter` (with optional `QueryOptions`). The transform returns: + +- `None` → **retract** the item (and clean incident edges), +- an **empty dict** → skip, +- a **partial dict** → update. + +```python +ItemTransform = Callable[[MemoryItem], dict[str, Any] | None] +``` + +### `bulk_adjust_scores(state, criteria, delta, author, reason=None) -> CommandResult` + +Adds a `ScoreAdjustment` to matching items, clamping each result to `0..1`. + +- **`class ScoreAdjustment`** — `authority`, `conviction`, `importance` (each + `float | None`); only the provided deltas are applied. + +### `decay_importance(state, older_than_ms, factor, author, reason=None) -> CommandResult` + +Multiplies `importance` by `factor` for items created before +`now - older_than_ms`. Items with zero/absent importance are skipped. + +--- + +## Intent graph + +Active goals with a status machine: `active ⇄ paused → completed / cancelled`. + +### `class Intent` + +Frozen. `id`, `parent_id?`, `label`, `description?`, `priority` (0..1), `owner`, +`status: IntentStatus`, `context?`, `root_memory_ids?`, `meta?`. + +- **`IntentStatus`** — `"active" | "paused" | "completed" | "cancelled"`. +- **`class IntentState`** — frozen dataclass `intents: dict[str, Intent]`. + +### State & factories + +- **`create_intent_state() -> IntentState`** +- **`create_intent(*, label, priority, owner, id=None, parent_id=None, + description=None, status=None, context=None, root_memory_ids=None, meta=None) + -> Intent`** — `status` defaults to `"active"`. + +### `apply_intent_command(state, cmd) -> IntentResult` + +`cmd: IntentCommand | dict`. `IntentResult` is `(state: IntentState, +events: list[IntentLifecycleEvent])`. + +| `type` | Fields | Effect | +|---------------------|------------------------------------------|---------------------------------| +| `"intent.create"` | `intent: Intent` | add (dup → `DuplicateIntentError`) | +| `"intent.update"` | `intent_id`, `partial`, `author`, `reason?` | merge (`id`/`status` ignored) | +| `"intent.complete"` | `intent_id`, `author`, `reason?` | → `completed` (from active/paused) | +| `"intent.cancel"` | `intent_id`, `author`, `reason?` | → `cancelled` (from active/paused) | +| `"intent.pause"` | `intent_id`, `author`, `reason?` | → `paused` (from active) | +| `"intent.resume"` | `intent_id`, `author`, `reason?` | → `active` (from paused) | + +Invalid transitions raise `InvalidIntentTransitionError`; missing id raises +`IntentNotFoundError`. `IntentCommand` is the discriminated-union alias. + +### Queries + +- **`get_intents(state, filter=None) -> list[Intent]`** — `filter: IntentFilter | dict | None`. +- **`get_intent_by_id(state, id) -> Intent | None`** +- **`get_child_intents(state, parent_id) -> list[Intent]`** + +#### `class IntentFilter` + +`owner`, `status`, `statuses: list[IntentStatus]`, `min_priority`, +`has_memory_id` (in `root_memory_ids`), `parent_id`, `is_root` (bool). + +### Events + +- **`class IntentLifecycleEvent`** — `namespace="intent"`, `type` + (`"intent.created"` …), `intent: Intent`, `cause_type: str`. + +--- + +## Task graph + +Units of work tied to intents: `pending → running → completed`, with +`running → failed → running` retry and `cancel` from any non-terminal state. + +### `class Task` + +Frozen. `id`, `intent_id`, `parent_id?`, `action`, `label?`, `status: +TaskStatus`, `priority` (0..1), `context?`, `result?`, `error?`, +`input_memory_ids?`, `output_memory_ids?`, `agent_id?`, `attempt?`, `meta?`. + +- **`TaskStatus`** — `"pending" | "running" | "completed" | "failed" | "cancelled"`. +- **`class TaskState`** — frozen dataclass `tasks: dict[str, Task]`. + +### State & factories + +- **`create_task_state() -> TaskState`** +- **`create_task(*, intent_id, action, priority, id=None, parent_id=None, + label=None, status=None, context=None, result=None, error=None, + input_memory_ids=None, output_memory_ids=None, agent_id=None, attempt=None, + meta=None) -> Task`** — `status` defaults to `"pending"`, `attempt` to `0`. + +### `apply_task_command(state, cmd) -> TaskResult` + +`cmd: TaskCommand | dict`. `TaskResult` is `(state: TaskState, +events: list[TaskLifecycleEvent])`. + +| `type` | Fields | Effect | +|-------------------|------------------------------------------|-------------------------------------| +| `"task.create"` | `task: Task` | add (dup → `DuplicateTaskError`) | +| `"task.update"` | `task_id`, `partial`, `author` | merge (`id`/`status` ignored) | +| `"task.start"` | `task_id`, `agent_id?` | → `running` (from pending/failed), `attempt++` | +| `"task.complete"` | `task_id`, `result?`, `output_memory_ids?` | → `completed` (from running) | +| `"task.fail"` | `task_id`, `error`, `retryable?` | → `failed` (from running) | +| `"task.cancel"` | `task_id`, `reason?` | → `cancelled` (from non-terminal) | + +Invalid transitions raise `InvalidTaskTransitionError`; missing id raises +`TaskNotFoundError`. `TaskCommand` is the discriminated-union alias. + +### Queries + +- **`get_tasks(state, filter=None) -> list[Task]`** — `filter: TaskFilter | dict | None`. +- **`get_task_by_id(state, id) -> Task | None`** +- **`get_tasks_by_intent(state, intent_id) -> list[Task]`** +- **`get_child_tasks(state, parent_id) -> list[Task]`** + +#### `class TaskFilter` + +`intent_id`, `action`, `status`, `statuses: list[TaskStatus]`, `agent_id`, +`min_priority`, `has_input_memory_id`, `has_output_memory_id`, `parent_id`, +`is_root`. + +### Events + +- **`class TaskLifecycleEvent`** — `namespace="task"`, `type` (`"task.created"` + …), `task: Task`, `cause_type: str`. + +--- + +## Statistics + +### `get_stats(state) -> GraphStats` + +Aggregate counts over a `GraphState`. + +- **`class GraphStats`** — `items: ItemStats`, `edges: EdgeStats`. +- **`class ItemStats`** — `total`, `by_kind`, `by_source_kind`, `by_author`, + `by_scope` (each `dict[str, int]`), `with_parents: int`, `root: int`. +- **`class EdgeStats`** — `total: int`, `active: int`, `by_kind: dict[str, int]`. + +--- + +## Replay + +Rebuild a `GraphState` from an event/command log. Integrity-tolerant: per-item +failures are collected, not raised. + +### `replay_commands(commands: list) -> ReplayResult` + +Fold a list of commands (models or dicts) in order. + +### `replay_from_envelopes(envelopes: list) -> ReplayResult` + +Sort envelopes by their `ts` (strict ISO-8601, ms precision, explicit offset or +`Z`) and fold their payloads. Each envelope may be a dict or an `EventEnvelope`. + +- **`class ReplayResult`** — `state: GraphState`, + `events: list[MemoryLifecycleEvent]`, `skipped: list[ReplayFailure]`. +- **`class ReplayFailure`** — dataclass `index: int`, `error: Exception`, + `command=None`, `envelope=None`. + +--- + +## Serialization + +On-disk shape matches the TS library — +`{"items": [[id, item], ...], "edges": [[id, edge], ...]}` — with unset +optionals omitted and edge `from` under its alias. + +- **`to_json(state) -> SerializedGraphState`** — `dict[str, list[list[Any]]]`. +- **`from_json(data) -> GraphState`** — tolerates missing `items`/`edges` keys. +- **`stringify(state, pretty=False) -> str`** — compact, or 2-space indented. +- **`parse(json_str) -> GraphState`** +- **`SerializedGraphState`** — the serialized-dict type alias. + +```python +from memex import stringify, parse +snapshot = stringify(state, pretty=True) +state = parse(snapshot) +``` + +--- + +## Event envelopes + +### `class EventEnvelope` (generic over `payload`) + +`id`, `namespace`, `type`, `ts` (ISO string), `trace_id: str | None`, `payload: T`. + +### `create_event_envelope(type, payload, *, trace_id=None, namespace="memory") -> EventEnvelope[Any]` + +Mints `id` (UUIDv7) and `ts` (`now_iso`). + +### Wrappers + +Build envelopes from reducer output for an append-only log: + +- **`wrap_lifecycle_event(event, cause_id, trace_id=None) -> EventEnvelope[dict]`** + — wraps a `MemoryLifecycleEvent`; payload carries the set fields plus `cause_id`. +- **`wrap_state_event(item, cause_id, trace_id=None) -> EventEnvelope[dict]`** + — a `"state.memory"` snapshot of an item. +- **`wrap_edge_state_event(edge, cause_id, trace_id=None) -> EventEnvelope[dict]`** + — a `"state.edge"` snapshot of an edge. + +--- + +## Transplant + +Export a slice of all three graphs and import it elsewhere, optionally re-id'ing +on collision. + +### `export_slice(mem_state, intent_state, task_state, *, ...) -> MemexExport` + +Keyword options: `memory_ids`, `intent_ids`, `task_ids` (seed sets); +`include_parents`, `include_children`, `include_aliases`, +`include_related_tasks`, `include_related_intents` (all `bool`, default +`False`). Walks the requested relationships and collects edges between included +memories. + +- **`class MemexExport`** — `memories: list[MemoryItem]`, `edges: list[Edge]`, + `intents: list[Intent]`, `tasks: list[Task]`. +- **`class ExportOptions`** — the same options as a model (for callers that + prefer to pass a struct). + +### `import_slice(mem_state, intent_state, task_state, slice, *, ...) -> ImportResult` + +`slice: MemexExport | dict`. Options: `skip_existing_ids=True`, +`shallow_compare_existing=False`, `re_id_on_difference=False`. When re-id'ing, +colliding-but-different entities get fresh UUIDv7-shaped ids (1ms after the +original), and all cross-references (`parents`, `parent_id`, `intent_id`, +memory-id lists) are remapped via a per-graph pre-pass. + +- **`class ImportResult`** — `mem_state`, `intent_state`, `task_state`, + `report: ImportReport`. +- **`class ImportReport`** — `created`, `updated`, `skipped`, `conflicts`, each + an `ImportBucket`. +- **`class ImportBucket`** — `memories`, `intents`, `tasks`, `edges` (each + `list[str]` of ids). +- **`class ImportOptions`** — the options as a model. + +--- + +## Validation + +`memex.schemas` is the validation entry point (the parity shim for +`@ai2070/memex/schemas`). In Pydantic the models *are* the schema. + +```python +from memex.schemas import validate_command +from memex import apply_command + +cmd = validate_command(raw) # raises pydantic.ValidationError on bad shape +state = apply_command(state, cmd).state +``` + +| Function | Returns | +|-----------------------------------|----------------| +| `validate_command(raw)` | `MemoryCommand`| +| `validate_intent_command(raw)` | `IntentCommand`| +| `validate_task_command(raw)` | `TaskCommand` | +| `validate_memory_item(raw)` | `MemoryItem` | +| `validate_edge(raw)` | `Edge` | + +Schema aliases (the model is the schema): `MemoryItemSchema`, `EdgeSchema`, +`IntentSchema`, `TaskSchema`. Adapters: `MemoryCommandAdapter`, +`IntentCommandAdapter`, `TaskCommandAdapter`. + +--- + +## `MemexStore` facade + +A mutable, object-oriented container over the three graphs. It holds the states, +rebinds them on each mutation, and returns the emitted events — convenient for +agents and daemons that don't want to thread `state =` through every call. The +functional API remains the backbone; the facade just wraps it. + +```python +from memex import MemexStore + +store = MemexStore() +a = store.create(scope="user:laz", kind="observation", content={"v": 1}, + author="agent:x", source_kind="observed", authority=0.9) +store.mark_contradiction(a.id, b_id, "system:detector") +snapshot = store.dumps(pretty=True) +restored = MemexStore.loads(snapshot) +``` + +### Constructor + +`MemexStore(mem=None, intents=None, tasks=None)` — start empty or from existing +states. Attributes `store.mem`, `store.intents`, `store.tasks` expose the live +states. + +### Memory + +| Method | Returns | Notes | +|--------|---------|-------| +| `apply(cmd)` | `list[MemoryLifecycleEvent]` | apply any memory command | +| `create(**kwargs)` | `MemoryItem` | forwards to `create_memory_item`, then creates | +| `add(item)` | `MemoryItem` | create from an existing item | +| `update(item_id, partial, author, reason=None)` | events | | +| `retract(item_id, author, reason=None)` | events | | +| `add_edge(**kwargs)` | `Edge` | forwards to `create_edge`, then creates | +| `items(filter=None, options=None)` | `list[MemoryItem]` | | +| `item(id)` | `MemoryItem \| None` | | +| `scored(weights, options=None)` | `list[ScoredItem]` | | +| `edges(filter=None)` | `list[Edge]` | | +| `parents(item_id)` / `children(item_id)` | `list[MemoryItem]` | | +| `related(item_id, direction="both")` | `list[MemoryItem]` | | +| `smart_retrieve(**kwargs)` | `list[ScoredItem]` | | +| `support_tree(item_id)` | `SupportNode \| None` | | +| `support_set(item_id)` | `list[MemoryItem]` | | +| `stats()` | `GraphStats` | | + +### Integrity + +| Method | Returns | +|--------|---------| +| `mark_contradiction(a, b, author, meta=None)` | events | +| `resolve_contradiction(winner, loser, author, reason=None)` | events | +| `mark_alias(a, b, author, meta=None)` | events | +| `cascade_retract(item_id, author, reason=None)` | `list[str]` (retracted ids) | +| `contradictions()` | `list[Contradiction]` | +| `stale_items()` | `list[StaleItem]` | +| `aliases(item_id)` / `alias_group(item_id)` | `list[MemoryItem]` | + +### Bulk + +`apply_many(filter, transform, author, reason=None, options=None)`, +`bulk_adjust_scores(criteria, delta, author, reason=None)`, +`decay_importance(older_than_ms, factor, author, reason=None)` — each returns +events. + +### Intent / Task + +`apply_intent(cmd)`, `create_intent(**kwargs)`, `get_intents(filter=None)`; +`apply_task(cmd)`, `create_task(**kwargs)`, `get_tasks(filter=None)`. + +### Transplant & serialization + +`export_slice(**kwargs)` → `MemexExport`; `import_slice(slice, **kwargs)` → +`ImportReport` (mutates the store's states); `to_json()` → +`SerializedGraphState`; `dumps(pretty=False)` → `str`; `MemexStore.loads(json_str)` +→ `MemexStore` (classmethod, restores the memory graph). + +--- + +## UUID helpers + +- **`uuid7(ms=None) -> str`** — generate a UUIDv7 string (RFC 9562) for `ms` + (defaults to now). Encodes a 48-bit big-endian ms timestamp in the first six + bytes. +- **`safe_extract_timestamp(value: str) -> int | None`** — decode the ms + timestamp from a UUIDv7; returns `None` for non-v7 input or a non-positive + timestamp (unlike [`extract_timestamp`](#extract_timestampuuid_id-str---int), + which raises). + +> Install the optional `fast-uuid` extra (`uuid-utils`) for faster generation. + +--- + +## Errors + +All domain errors derive from `MemexError`. + +| Exception | Raised by | Attributes | +|-----------|-----------|------------| +| `MemexError` | base class | — | +| `MemoryNotFoundError` | update/retract of an absent item | `item_id` | +| `EdgeNotFoundError` | update/retract of an absent edge | `edge_id` | +| `DuplicateMemoryError` | create with an existing id | `item_id` | +| `DuplicateEdgeError` | create with an existing edge id | `edge_id` | +| `InvalidTimestampError` | bad UUIDv7 / envelope timestamp | — | +| `IntentNotFoundError` | intent reducer | `intent_id` | +| `DuplicateIntentError` | intent reducer | `intent_id` | +| `InvalidIntentTransitionError` | intent reducer | `intent_id`, `from_status`, `to_status` | +| `TaskNotFoundError` | task reducer | `task_id` | +| `DuplicateTaskError` | task reducer | `task_id` | +| `InvalidTaskTransitionError` | task reducer | `task_id`, `from_status`, `to_status` | + +Out-of-range scores and malformed command shapes surface as +`pydantic.ValidationError`. `cost_fn` contract violations and unknown +sort/decay enums surface as `ValueError`. + +--- + +## Type aliases + +Open string unions documented as `Literal` aliases — fields accept any `str`, +but these are the canonical values. + +| Alias | Values | +|-------|--------| +| `KnownMemoryKind` | `observation`, `assertion`, `assumption`, `hypothesis`, `derivation`, `simulation`, `policy`, `trait` | +| `KnownSourceKind` | `user_explicit`, `observed`, `derived_deterministic`, `agent_inferred`, `simulated`, `imported` | +| `KnownEdgeKind` | `DERIVED_FROM`, `CONTRADICTS`, `SUPPORTS`, `ABOUT`, `SUPERSEDES`, `ALIAS` | +| `KnownNamespace` | `memory`, `task`, `agent`, `tool`, `net`, `app`, `chat`, `system`, `debug` | +| `LifecycleEventType` | `memory.created`, `memory.updated`, `memory.retracted`, `edge.created`, `edge.updated`, `edge.retracted` | +| `SortField` | `authority`, `conviction`, `importance`, `recency` | +| `DecayInterval` | `hour`, `day`, `week` | +| `DecayType` | `exponential`, `linear`, `step` | From 863e5878a7bb0e9da641f616148cc816466eb8c6 Mon Sep 17 00:00:00 2001 From: LZL0 <12474488+LZL0@users.noreply.github.com> Date: Mon, 22 Jun 2026 05:08:38 +0200 Subject: [PATCH 3/3] Update README.md --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index e2775c9..70f34a9 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,8 @@ MemEX stores beliefs, evidence, conflicts, and updates — not just retrieved te - **Pydantic-native** — models validate on construction; the discriminated-union commands *are* the schema. - **Wire-compatible** — command tags, enum values, and JSON keys are byte-identical to the TS library, so a Python service and a TS service can share one event log. +> 📖 Full API reference: [`API.md`](API.md). + ## Install ```bash