diff --git a/CHANGELOG.md b/CHANGELOG.md index ae6ccf95..758fa407 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`ContextGraph` gains retraction and purge — the graph previously had no way to remove a node or edge without discarding everything via `clear()`** (#957, closes #955) by @pravit-amp, reviewed by @KaifAhmad1 + - `retract_node()`/`retract_edge()` close an entity's validity window rather than deleting it, reusing the existing `valid_from`/`valid_until`/`state_at()` machinery: the entity drops out of `find_active_nodes()` and future `state_at()` queries going forward, but `state_at()` calls before the retraction time still return it, so decisions recorded against it stay explainable. A `("kind", id)`-keyed retraction record captures who/why/when, retrievable via `get_retraction()`/`list_retractions()` + - `purge_node()`/`purge_edge()` are the destructive counterpart: the entity is removed outright, from history as well as the active view, for erasure obligations retraction alone cannot satisfy (e.g. GDPR Article 17). Only a tombstone remains — that a purge happened, when, and why — deliberately never the purged content, via `get_tombstone()`/`list_tombstones()`. Purge is graph-scope only: `AgentMemory` and any bound vector store are not reached, so it is one step of an erasure workflow rather than the whole of it + - Both operations default to `cascade=True` (also touching every incident edge, and for `purge_node`, the marker node of any cross-graph link the node exits through) since leaving edges active around an inactive/removed node produces an inconsistent active view or dangling endpoints; both accept `cascade=False` for callers that want to handle edges themselves + - Both are idempotent: retracting/purging an already-retracted/purged entity returns `False` rather than raising, and a repeat retraction preserves the original record's reason rather than overwriting it + - Retraction/purge closing a validity window never widens an existing one — a node or edge added with `valid_until` already in the past keeps that earlier bound rather than being pushed later by a subsequent retraction time + - Reuses the existing audit-trail path with no changes to `change_management`: `MutationRecord` already documented `REMOVE_NODE`/`REMOVE_EDGE` in its operation vocabulary; retraction now emits `UPDATE_NODE`/`UPDATE_EDGE`, purge emits `REMOVE_NODE`/`REMOVE_EDGE`, matching the documented contract. Mutation payloads are snapshotted inside the lock and the callback fires after it is released, so a callback that itself mutates the graph (e.g. `clear()`) can't observe or lose in-flight records + - **Fixed during review** (@KaifAhmad1): `retract_edge()`/`purge_edge()` resolved "the edge" for a given `edge_id` via the first matching object only. `edge_id` is content-derived and, prior to #926, was not guaranteed unique — a graph holding two identical `add_edge()` calls had two edge objects sharing one id. A direct `retract_edge()`/`purge_edge()` call would silently leave the second duplicate untouched (still live, still active) while returning `True` and recording a tombstone/retraction that claimed the edge was fully handled; repeat `purge_edge()` calls also silently overwrote the tombstone's `reason`/`purged_at` on each partial attempt instead of no-op'ing. The same gap let `retract_node()`'s cascade skip a duplicate outright, since it checked the live `_retractions` dict mid-loop and treated the first duplicate's just-written record as proof the second was already handled. `#926` (merged) stops *new* duplicates from being created, but any graph already holding one — loaded from a save made before that fix, or built during the window before it landed — could still trigger this. Now `retract_edge()`/`purge_edge()` act on every edge matching the id under one record, and the cascade's dedup check is snapshotted before the loop starts so within-call duplicates are still closed rather than skipped. 5 new regression tests in `TestDuplicateEdgeId` + - New `tests/context/test_context_graph_retraction.py`: 49 tests, covering retraction/purge semantics, cascade, idempotency, validity-window narrowing, id-keyspace collisions between node and edge ids, cross-graph link teardown, `clear()`/`load_from_file()` resetting retraction/tombstone state, audit-trail integration against a real `TemporalVersionManager`, mutation-emission ordering under a concurrent `clear()`, and concurrent purges + - Full `tests/context/` suite: 533 passed - **`DistanceExporter.compute_pairs()` gains an opt-in `metric_errors` column to distinguish legitimate `None` results from computation failures** (#960, follow-up to #879) by @Karunasagar12 - Previously, a `None` in `hop_count`/`weighted_distance`/`semantic_similarity`/betweenness could mean either "no path exists" or "the underlying computation raised" — logged as a warning per #879, but not otherwise surfaced, so the two cases were indistinguishable in exported CSV/JSONL/DataFrame data. `include=["metric_errors"]` now adds a `metric_errors` field per row: `""` when all requested metrics succeeded, or a comma-separated list of metric names that raised (e.g. `"hop_count,weighted_distance"`) - Opt-in only — default `compute_pairs()`/`to_csv()`/`to_dataframe()`/`to_jsonl()` schema is unchanged unless `"metric_errors"` is explicitly requested diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index 4cc66e70..28b431ef 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -188,6 +188,26 @@ def _normalize_temporal_input(value: Optional[Union[str, int, float, datetime]]) raise ValueError("Temporal values must be datetime, epoch seconds, ISO strings, or None") +def _closing_valid_until(current: Optional[str], at_iso: str) -> str: + """Return the earlier of an existing end bound and a retraction time. + + Retraction closes a validity window and must never widen one: an entity + added with ``valid_until`` already in the past would otherwise be reported + active by ``is_active``/``state_at`` for the span between its original end + and the retraction. An unparseable ``current`` imposes no end bound at all + (see :func:`_parse_iso_dt`), so ``at_iso`` still closes it. + """ + if current is None: + return at_iso + existing = _parse_iso_dt(current) + if existing is None: + return at_iso + requested = _parse_iso_dt(at_iso) + if requested is None or existing <= requested: + return current + return at_iso + + def _pick_first(*values: Any) -> Any: for value in values: if value is None: @@ -476,6 +496,15 @@ class ContextGraph: self._unresolved_links: Dict[str, Dict[str, str]] = {} + # Retraction closes an entity's validity window but keeps it in the + # graph; a tombstone records that an entity was purged outright, + # without retaining the purged content. Keyed by + # ``(entity_kind, entity_id)`` -- node ids are caller-supplied strings + # and edge ids are UUID strings, so a single id keyspace would let a + # node record mask an edge of the same id, and vice versa. + self._retractions: Dict[Tuple[str, str], Dict[str, Any]] = {} + self._tombstones: Dict[Tuple[str, str], Dict[str, Any]] = {} + self.progress_tracker = get_progress_tracker() @@ -1127,6 +1156,10 @@ class ContextGraph: self.edge_type_index.clear() self._linked_graphs.clear() self._unresolved_links.clear() + # Deletion metadata belongs to the graph being replaced; keeping it + # would make entities in the loaded graph read as already retracted. + self._retractions.clear() + self._tombstones.clear() if "graph_id" in data: self.graph_id = data["graph_id"] @@ -1520,6 +1553,371 @@ class ContextGraph: max_edges = n * (n - 1) return len(self.edges) / max_edges + def retract_node( + self, + node_id: str, + reason: Optional[str] = None, + at: Optional[Union[str, datetime]] = None, + cascade: bool = True, + ) -> bool: + """Retract a node: no longer active, but still visible in history. + + Closes the node's validity window rather than deleting it, so + :meth:`state_at` before ``at`` still returns the node and any decision + recorded against it remains explainable. Use :meth:`purge_node` when + the data itself has to be gone. + + Args: + node_id: Node to retract. + reason: Why it was retracted, stored on the retraction record. + at: When the retraction takes effect (ISO string or datetime). + Defaults to now, UTC. + cascade: Also retract every edge touching the node. Leaving edges + active around an inactive node means :meth:`find_active_nodes` + drops the node while its relationships still read as current, + so the default keeps the active view self-consistent. + + Retraction is expressed through the temporal window, so it is visible + to the activity-aware views -- :meth:`find_active_nodes`, + :meth:`state_at`, ``ContextNode.is_active`` -- and not to membership + checks like :meth:`has_node` or :meth:`stats`, which continue to count + the retained record. That matches how ``valid_until`` already behaved + before retraction existed. + + A node whose ``valid_until`` is already earlier than ``at`` keeps that + earlier bound: retraction only ever closes a validity window, never + widens one. + + Returns: + True if the node was retracted; False if it does not exist or was + already retracted. + + Note: + Emits ``UPDATE_NODE`` to the audit-trail callback, since retraction + changes the validity window rather than removing the record. + """ + at_iso = _normalize_temporal_input(at) or datetime.now(timezone.utc).isoformat() + with self._lock: + node = self.nodes.get(node_id) + if node is None: + self.logger.warning("Cannot retract unknown node: %r", node_id) + return False + if ("node", node_id) in self._retractions: + return False + + node.valid_until = _closing_valid_until(node.valid_until, at_iso) + record = { + "entity_id": node_id, + "entity_kind": "node", + "retracted_at": at_iso, + "reason": reason, + } + self._retractions[("node", node_id)] = record + node_payload = {**node.to_dict(), "retraction": dict(record)} + + cascaded: List[Tuple[str, Dict[str, Any]]] = [] + if cascade: + # Snapshotted once, before the loop: edge_id is content-derived + # and not guaranteed unique (#922), so two distinct edge objects + # can share one id. Checking the live _retractions dict inside + # the loop would let the first duplicate's record block the + # second from ever being closed, leaving it active indefinitely + # while its retraction record claimed otherwise. + already_retracted_edge_ids = { + key[1] for key in self._retractions if key[0] == "edge" + } + for edge in self._incident_edges(node_id): + if edge.edge_id in already_retracted_edge_ids: + continue + edge.valid_until = _closing_valid_until(edge.valid_until, at_iso) + edge_record = { + "entity_id": edge.edge_id, + "entity_kind": "edge", + "retracted_at": at_iso, + "reason": reason, + "cascaded_from": node_id, + } + self._retractions[("edge", edge.edge_id)] = edge_record + # Payloads are snapshotted here, not read back after the + # lock is released: a concurrent clear() would otherwise + # wipe the record out from under the emission below. + cascaded.append( + ( + edge.edge_id, + {**edge.to_dict(), "retraction": dict(edge_record)}, + ) + ) + + self._emit_mutation("UPDATE_NODE", node_id, node_payload) + for edge_id, edge_payload in cascaded: + self._emit_mutation("UPDATE_EDGE", edge_id, edge_payload) + self.logger.info( + "Retracted node %r at %s (cascaded %d edge(s))", + node_id, + at_iso, + len(cascaded), + ) + return True + + def retract_edge( + self, + edge_id: str, + reason: Optional[str] = None, + at: Optional[Union[str, datetime]] = None, + ) -> bool: + """Retract a single edge, leaving its endpoints untouched. + + An edge whose ``valid_until`` is already earlier than ``at`` keeps that + earlier bound; retraction never widens a validity window. + + Args: + edge_id: Edge to retract. + reason: Why it was retracted. + at: When the retraction takes effect. Defaults to now, UTC. + + Returns: + True if the edge was retracted; False if it does not exist or was + already retracted. + + Note: + ``edge_id`` is content-derived and not guaranteed unique (#922): + two distinct edge objects can share one id. Every edge matching + ``edge_id`` is closed under a single retraction record, so a + duplicate can never be left silently active while the record + claims it was retracted. + """ + at_iso = _normalize_temporal_input(at) or datetime.now(timezone.utc).isoformat() + with self._lock: + edges = [e for e in self.edges if e.edge_id == edge_id] + if not edges: + self.logger.warning("Cannot retract unknown edge: %r", edge_id) + return False + if ("edge", edge_id) in self._retractions: + return False + + record = { + "entity_id": edge_id, + "entity_kind": "edge", + "retracted_at": at_iso, + "reason": reason, + } + self._retractions[("edge", edge_id)] = record + for edge in edges: + edge.valid_until = _closing_valid_until(edge.valid_until, at_iso) + payload = {**edges[0].to_dict(), "retraction": dict(record)} + + self._emit_mutation("UPDATE_EDGE", edge_id, payload) + self.logger.info( + "Retracted edge %r at %s (%d underlying record(s))", + edge_id, + at_iso, + len(edges), + ) + return True + + def purge_node( + self, + node_id: str, + reason: Optional[str] = None, + at: Optional[Union[str, datetime]] = None, + cascade: bool = True, + ) -> bool: + """Permanently remove a node; history no longer contains it. + + Unlike :meth:`retract_node` this is destructive: the node disappears + from :meth:`state_at` as well as from the active view. Only a tombstone + remains, recording that a purge happened and why -- deliberately + without the purged content, since retaining it would defeat the point. + + Scope is this graph only. Copies held elsewhere (``AgentMemory``, a + bound vector store, an exported file) are not reached, so this is one + step of an erasure workflow, not the whole of it. + + Args: + node_id: Node to purge. + reason: Why it was purged, e.g. an erasure-request reference. + at: When the purge takes effect, recorded as the tombstone's + ``purged_at`` (ISO string or datetime). Defaults to now, UTC. + cascade: Also purge every edge touching the node, and the marker + node of any cross-graph link it exits through. Defaults to True + because leaving edges pointing at a removed node produces + dangling endpoints. + + Cross-graph links registered by :meth:`link_graph` out of this node are + deregistered either way -- a link whose source no longer exists would + still resolve through :meth:`navigate_to` and still be serialized by + :meth:`save_to_file`. + + Returns: + True if the node was purged; False if it does not exist. + + Note: + Emits ``REMOVE_NODE``/``REMOVE_EDGE`` to the audit-trail callback. + """ + purged_at = ( + _normalize_temporal_input(at) or datetime.now(timezone.utc).isoformat() + ) + with self._lock: + if node_id not in self.nodes: + self.logger.warning("Cannot purge unknown node: %r", node_id) + return False + + # The link marker node is scaffolding reachable only from the node + # being purged, so it goes with the cascade rather than surviving as + # an orphan. Resolve the markers before deregistering the links they + # are derived from. + targets = [node_id] + if cascade: + targets.extend(self._cross_graph_marker_nodes(node_id)) + for link_id in self._cross_graph_links_for(node_id): + self._linked_graphs.pop(link_id, None) + self._unresolved_links.pop(link_id, None) + + # Tombstones are snapshotted into locals before the lock is + # released; reading them back afterwards would race a clear(). + purged_edges: List[Tuple[str, Dict[str, Any]]] = [] + purged_nodes: List[Tuple[str, Dict[str, Any]]] = [] + for target in targets: + cascaded_from = None if target == node_id else node_id + if cascade: + for edge in self._incident_edges(target): + self._drop_edge_from_indexes(edge) + edge_record = { + "entity_id": edge.edge_id, + "entity_kind": "edge", + "purged_at": purged_at, + "reason": reason, + "cascaded_from": node_id, + } + self._tombstones[("edge", edge.edge_id)] = edge_record + self._retractions.pop(("edge", edge.edge_id), None) + purged_edges.append((edge.edge_id, dict(edge_record))) + + self._drop_node_from_indexes(target) + node_record = { + "entity_id": target, + "entity_kind": "node", + "purged_at": purged_at, + "reason": reason, + } + if cascaded_from is not None: + node_record["cascaded_from"] = cascaded_from + self._tombstones[("node", target)] = node_record + self._retractions.pop(("node", target), None) + purged_nodes.append((target, dict(node_record))) + + for edge_id, payload in purged_edges: + self._emit_mutation("REMOVE_EDGE", edge_id, payload) + for purged_id, payload in purged_nodes: + self._emit_mutation("REMOVE_NODE", purged_id, payload) + self.logger.info( + "Purged node %r (cascaded %d edge(s), %d node(s))", + node_id, + len(purged_edges), + len(purged_nodes) - 1, + ) + return True + + def purge_edge( + self, + edge_id: str, + reason: Optional[str] = None, + at: Optional[Union[str, datetime]] = None, + ) -> bool: + """Permanently remove a single edge, leaving its endpoints in place. + + If the edge is the bridge of a cross-graph link, the link is also + deregistered -- :meth:`navigate_to` should not keep resolving a link + whose bridge is gone. The marker node itself is an endpoint and is left + in place; purge it directly, or purge the link's source node, to remove + it too. + + Args: + edge_id: Edge to purge. + reason: Why it was purged. + at: When the purge takes effect, recorded as the tombstone's + ``purged_at``. Defaults to now, UTC. + + Returns: + True if the edge was purged; False if it does not exist. + + Note: + ``edge_id`` is content-derived and not guaranteed unique (#922): + two distinct edge objects can share one id. Every edge matching + ``edge_id`` is dropped under a single tombstone, so a duplicate + can never be left live in the graph while the tombstone claims + the edge is gone. + """ + purged_at = ( + _normalize_temporal_input(at) or datetime.now(timezone.utc).isoformat() + ) + with self._lock: + edges = [e for e in self.edges if e.edge_id == edge_id] + if not edges: + self.logger.warning("Cannot purge unknown edge: %r", edge_id) + return False + for edge in edges: + self._drop_edge_from_indexes(edge) + link_id = (edge.metadata or {}).get("link_id") + if (edge.metadata or {}).get("cross_graph") and link_id: + self._linked_graphs.pop(link_id, None) + self._unresolved_links.pop(link_id, None) + self._retractions.pop(("edge", edge_id), None) + record = { + "entity_id": edge_id, + "entity_kind": "edge", + "purged_at": purged_at, + "reason": reason, + } + self._tombstones[("edge", edge_id)] = record + payload = dict(record) + + self._emit_mutation("REMOVE_EDGE", edge_id, payload) + self.logger.info( + "Purged edge %r (%d underlying record(s))", edge_id, len(edges) + ) + return True + + def get_retraction( + self, entity_id: str, entity_kind: Optional[str] = None + ) -> Optional[Dict[str, Any]]: + """Return the retraction record for a node or edge, or None. + + Args: + entity_id: Node id or edge id. + entity_kind: ``"node"`` or ``"edge"``. Records are keyed by kind as + well as id, so pass this when a node id and an edge id could + collide; without it a node record is preferred over an edge one. + """ + with self._lock: + return self._find_removal_record(self._retractions, entity_id, entity_kind) + + def get_tombstone( + self, entity_id: str, entity_kind: Optional[str] = None + ) -> Optional[Dict[str, Any]]: + """Return the purge tombstone for a node or edge, or None. + + The tombstone records that a purge happened, when, and why. It never + contains the purged content. + + Args: + entity_id: Node id or edge id. + entity_kind: ``"node"`` or ``"edge"``; disambiguates a node id that + collides with an edge id, as for :meth:`get_retraction`. + """ + with self._lock: + return self._find_removal_record(self._tombstones, entity_id, entity_kind) + + def list_retractions(self) -> List[Dict[str, Any]]: + """Return every retraction record.""" + with self._lock: + return [dict(record) for record in self._retractions.values()] + + def list_tombstones(self) -> List[Dict[str, Any]]: + """Return every purge tombstone.""" + with self._lock: + return [dict(record) for record in self._tombstones.values()] + def clear(self) -> None: """Fully reset the graph state and indexes.""" with self._lock: @@ -1531,6 +1929,8 @@ class ContextGraph: self.edge_type_index.clear() self._linked_graphs.clear() self._unresolved_links.clear() + self._retractions.clear() + self._tombstones.clear() self.logger.debug("Graph state fully cleared.") # --- Internal Helpers --- @@ -1629,6 +2029,147 @@ class ContextGraph: ) return True + def _emit_mutation( + self, operation: str, entity_id: str, payload: Dict[str, Any] + ) -> None: + """Fire the audit-trail callback, mirroring the add paths. + + Kept in one place so retraction and purge record themselves the same + way ``_add_internal_node``/``_add_internal_edge`` already do, including + the ``_suspend_mutation_callback`` guard used during restores. + """ + if not getattr(self, "mutation_callback", None): + return + if getattr(self, "_suspend_mutation_callback", False): + return + try: + self.mutation_callback(operation, entity_id, payload) + except Exception as e: + self.logger.warning( + f"Audit trail callback failed for {operation} {entity_id}: {e}" + ) + + def _incident_edges(self, node_id: str) -> List[ContextEdge]: + """Every edge touching ``node_id``, in either direction. + + ``_adjacency`` is keyed by source only, so incoming edges have to come + from a scan of ``self.edges``; relying on ``_adjacency`` alone would + silently leave inbound edges pointing at a removed node. + """ + return [ + edge + for edge in self.edges + if edge.source_id == node_id or edge.target_id == node_id + ] + + @staticmethod + def _find_removal_record( + store: Dict[Tuple[str, str], Dict[str, Any]], + entity_id: str, + entity_kind: Optional[str], + ) -> Optional[Dict[str, Any]]: + """Look a retraction/tombstone up by id, optionally narrowed by kind. + + The caller must hold ``self._lock``. Records are keyed by + ``(entity_kind, entity_id)``; with no kind given, both keyspaces are + tried so callers that know an id is unambiguous can pass it alone. + """ + if entity_kind is not None: + if entity_kind not in ("node", "edge"): + raise ValueError( + f"entity_kind must be 'node', 'edge' or None, got {entity_kind!r}" + ) + kinds: Tuple[str, ...] = (entity_kind,) + else: + kinds = ("node", "edge") + for kind in kinds: + record = store.get((kind, entity_id)) + if record is not None: + return dict(record) + return None + + def _cross_graph_links_for(self, node_id: str) -> List[str]: + """Link ids that ``node_id`` participates in, as exit point or marker. + + The caller must hold ``self._lock``. :meth:`link_graph` registers a link + in three places -- ``_linked_graphs``, a marker node and the bridge edge + -- so removing only the node would leave :meth:`navigate_to` resolving a + link whose source is gone. + """ + link_ids = [ + link_id + for link_id, (_, source_node_id, _) in self._linked_graphs.items() + if source_node_id == node_id + ] + link_ids.extend( + link_id + for link_id, meta in self._unresolved_links.items() + if meta.get("source_node_id") == node_id + ) + node = self.nodes.get(node_id) + metadata = getattr(node, "metadata", None) or {} + if metadata.get("cross_graph") and metadata.get("link_id"): + link_ids.append(metadata["link_id"]) + return list(dict.fromkeys(link_ids)) + + def _cross_graph_marker_nodes(self, node_id: str) -> List[str]: + """Marker nodes of the cross-graph links ``node_id`` exits through. + + The caller must hold ``self._lock``. + """ + return [ + marker_id + for marker_id in ( + f"__cross_graph_{link_id}" + for link_id in self._cross_graph_links_for(node_id) + ) + if marker_id != node_id and marker_id in self.nodes + ] + + def _drop_node_from_indexes(self, node_id: str) -> None: + """Remove one node from ``nodes``, ``node_type_index`` and ``_adjacency``. + + The caller must hold ``self._lock``. Incident edges are not touched -- + see :meth:`_drop_edge_from_indexes`. + """ + node = self.nodes.pop(node_id, None) + if node is None: + return + bucket = self.node_type_index.get(node.node_type) + if bucket is not None: + bucket.discard(node_id) + if not bucket: + del self.node_type_index[node.node_type] + self._adjacency.pop(node_id, None) + + def _drop_edge_from_indexes(self, edge: ContextEdge) -> None: + """Remove one edge from every structure that references it. + + The caller must hold ``self._lock``. ``edges``, ``edge_type_index`` and + ``_adjacency`` must be updated together or the indexes drift out of + step with the edge list. + """ + try: + self.edges.remove(edge) + except ValueError: + pass + bucket = self.edge_type_index.get(edge.edge_type) + if bucket is not None: + try: + bucket.remove(edge) + except ValueError: + pass + if not bucket: + del self.edge_type_index[edge.edge_type] + adjacent = self._adjacency.get(edge.source_id) + if adjacent is not None: + try: + adjacent.remove(edge) + except ValueError: + pass + if not adjacent: + del self._adjacency[edge.source_id] + # --- Builder Methods (Legacy/Utility) --- def build_from_conversations( diff --git a/tests/context/test_context_graph_retraction.py b/tests/context/test_context_graph_retraction.py new file mode 100644 index 00000000..fa0fc158 --- /dev/null +++ b/tests/context/test_context_graph_retraction.py @@ -0,0 +1,595 @@ +"""Tests for ContextGraph retraction and purge (issue #955). + +``ContextGraph`` had 56 public methods and none that removed anything: the only +option was ``clear()``, which discards the whole graph. Two operations are +added, with deliberately different contracts. + +Retraction closes an entity's validity window. The entity stops being active +going forward, but ``state_at()`` before the retraction still returns it, so +decisions recorded against it remain explainable. Purge is destructive: the +entity is gone from history too, leaving only a tombstone recording that a +purge happened and why -- never the purged content. + +The audit-trail assertions run against a real ``TemporalVersionManager`` rather +than a mock callback, since the behaviour under test is precisely that these +operations reach the existing mutation-recording path. +""" + +import json +import os +import tempfile +import threading +import unittest +from datetime import datetime + +from semantica.change_management import TemporalVersionManager +from semantica.context import ContextEdge, ContextGraph + +BEFORE = "2025-06-01T00:00:00Z" +BETWEEN = "2025-09-01T00:00:00Z" +CUTOFF = "2026-01-01T00:00:00Z" +AFTER = "2026-06-01T00:00:00Z" + + +def _graph(): + """alice --works_at--> acme, plus an unrelated bob.""" + graph = ContextGraph(advanced_analytics=False) + graph.add_node("alice", "person") + graph.add_node("acme", "org") + graph.add_node("bob", "person") + graph.add_edge("alice", "acme", "works_at") + return graph + + +def _ids_at(graph, when): + return {node.get("id") for node in graph.state_at(when).get("nodes", [])} + + +def _index_totals(graph): + return { + "nodes": len(graph.nodes), + "node_index": sum(len(v) for v in graph.node_type_index.values()), + "edges": len(graph.edges), + "edge_index": sum(len(v) for v in graph.edge_type_index.values()), + "adjacency": sum(len(v) for v in graph._adjacency.values()), + } + + +class TestRetractNode(unittest.TestCase): + def test_retracted_node_leaves_the_active_view(self): + graph = _graph() + self.assertTrue(graph.retract_node("alice", at=CUTOFF)) + active = {node["id"] for node in graph.find_active_nodes()} + self.assertNotIn("alice", active) + self.assertIn("bob", active) + + def test_history_before_the_retraction_is_preserved(self): + graph = _graph() + graph.retract_node("alice", at=CUTOFF) + self.assertIn("alice", _ids_at(graph, BEFORE)) + self.assertNotIn("alice", _ids_at(graph, AFTER)) + + def test_retraction_record_captures_reason_and_time(self): + graph = _graph() + graph.retract_node("alice", reason="employment ended", at=CUTOFF) + record = graph.get_retraction("alice") + self.assertEqual(record["entity_id"], "alice") + self.assertEqual(record["entity_kind"], "node") + self.assertEqual(record["reason"], "employment ended") + self.assertIn("2026-01-01", record["retracted_at"]) + + def test_retracting_twice_is_a_no_op(self): + graph = _graph() + self.assertTrue(graph.retract_node("alice", reason="first", at=CUTOFF)) + self.assertFalse(graph.retract_node("alice", reason="second")) + self.assertEqual(graph.get_retraction("alice")["reason"], "first") + + def test_retracting_an_unknown_node_returns_false(self): + graph = _graph() + self.assertFalse(graph.retract_node("nobody")) + self.assertIsNone(graph.get_retraction("nobody")) + + def test_cascade_retracts_incident_edges_in_both_directions(self): + graph = _graph() + graph.add_edge("bob", "alice", "knows") # inbound, not in _adjacency['alice'] + graph.retract_node("alice", at=CUTOFF) + for edge in graph.edges: + self.assertIsNotNone( + graph.get_retraction(edge.edge_id), + f"edge {edge.edge_type} touching alice was not retracted", + ) + + def test_cascade_can_be_disabled(self): + graph = _graph() + graph.retract_node("alice", at=CUTOFF, cascade=False) + edge = graph.edges[0] + self.assertIsNone(graph.get_retraction(edge.edge_id)) + + def test_retraction_does_not_remove_the_record(self): + """Retraction is a temporal change, not a deletion.""" + graph = _graph() + graph.retract_node("alice", at=CUTOFF) + self.assertTrue(graph.has_node("alice")) + self.assertIsNotNone(graph.find_node("alice")) + + +class TestRetractEdge(unittest.TestCase): + def test_edge_is_retracted_without_touching_endpoints(self): + graph = _graph() + edge_id = graph.edges[0].edge_id + self.assertTrue(graph.retract_edge(edge_id, reason="wrong extraction")) + self.assertIsNotNone(graph.get_retraction(edge_id)) + active = {node["id"] for node in graph.find_active_nodes()} + self.assertIn("alice", active) + self.assertIn("acme", active) + + def test_retracting_an_unknown_edge_returns_false(self): + self.assertFalse(_graph().retract_edge("no-such-edge")) + + def test_retracting_an_edge_twice_is_a_no_op(self): + graph = _graph() + edge_id = graph.edges[0].edge_id + self.assertTrue(graph.retract_edge(edge_id)) + self.assertFalse(graph.retract_edge(edge_id)) + + +class TestPurge(unittest.TestCase): + def test_purged_node_is_absent_from_history(self): + graph = _graph() + self.assertTrue(graph.purge_node("alice", reason="erasure request #1")) + self.assertNotIn("alice", _ids_at(graph, BEFORE)) + self.assertFalse(graph.has_node("alice")) + + def test_tombstone_records_the_purge_without_the_content(self): + graph = ContextGraph(advanced_analytics=False) + graph.add_node("alice", "person", email="alice@example.com") + graph.purge_node("alice", reason="erasure request #1") + + tombstone = graph.get_tombstone("alice") + self.assertEqual(tombstone["entity_id"], "alice") + self.assertEqual(tombstone["reason"], "erasure request #1") + self.assertIn("purged_at", tombstone) + self.assertNotIn( + "alice@example.com", + str(tombstone), + "tombstone retained purged content, defeating the purpose of a purge", + ) + + def test_purge_cascades_to_incident_edges(self): + graph = _graph() + graph.add_edge("bob", "alice", "knows") + graph.purge_node("alice") + remaining = {(e.source_id, e.target_id) for e in graph.edges} + self.assertEqual(remaining, set()) + + def test_purge_keeps_every_index_consistent(self): + """The invariant clear() already upholds must hold here too.""" + graph = ContextGraph(advanced_analytics=False) + for i in range(5): + graph.add_node(f"n{i}", f"t{i % 2}") + graph.add_edge("n0", "n1", "a") + graph.add_edge("n1", "n2", "b") + graph.add_edge("n2", "n0", "a") + graph.add_edge("n3", "n0", "b") + + graph.purge_node("n0") + + totals = _index_totals(graph) + self.assertEqual(totals["node_index"], totals["nodes"]) + self.assertEqual(totals["edge_index"], totals["edges"]) + self.assertEqual(totals["adjacency"], totals["edges"]) + self.assertEqual(totals["edges"], 1) # only n1->n2 survives + + def test_purge_edge_leaves_endpoints_in_place(self): + graph = _graph() + edge_id = graph.edges[0].edge_id + self.assertTrue(graph.purge_edge(edge_id)) + self.assertEqual(len(graph.edges), 0) + self.assertTrue(graph.has_node("alice")) + self.assertTrue(graph.has_node("acme")) + totals = _index_totals(graph) + self.assertEqual(totals["edge_index"], 0) + self.assertEqual(totals["adjacency"], 0) + + def test_purging_unknown_entities_returns_false(self): + graph = _graph() + self.assertFalse(graph.purge_node("nobody")) + self.assertFalse(graph.purge_edge("no-such-edge")) + + def test_purge_supersedes_an_earlier_retraction(self): + graph = _graph() + graph.retract_node("alice", reason="left", at=CUTOFF) + graph.purge_node("alice", reason="erasure request #2") + self.assertIsNone(graph.get_retraction("alice")) + self.assertIsNotNone(graph.get_tombstone("alice")) + + +class TestRetractionNeverWidensTheWindow(unittest.TestCase): + """Retraction closes a validity window; it must never extend one. + + An entity added with ``valid_until`` already in the past was inactive from + that point on. Overwriting the bound with a later retraction time would + make ``state_at`` report it active over a span it previously was not. + """ + + def test_a_node_keeps_an_earlier_valid_until(self): + graph = ContextGraph(advanced_analytics=False) + graph.add_node("alice", "person", valid_until=BEFORE) + self.assertTrue(graph.retract_node("alice", at=AFTER)) + self.assertEqual(graph.nodes["alice"].valid_until, BEFORE) + self.assertNotIn("alice", _ids_at(graph, BETWEEN)) + + def test_an_edge_keeps_an_earlier_valid_until(self): + graph = ContextGraph(advanced_analytics=False) + graph.add_node("alice", "person") + graph.add_node("acme", "org") + graph.add_edge("alice", "acme", "works_at", valid_until=BEFORE) + edge = graph.edges[0] + self.assertTrue(graph.retract_edge(edge.edge_id, at=AFTER)) + self.assertEqual(edge.valid_until, BEFORE) + self.assertFalse(edge.is_active(datetime(2025, 9, 1))) + + def test_cascade_keeps_an_earlier_edge_bound(self): + graph = ContextGraph(advanced_analytics=False) + graph.add_node("alice", "person") + graph.add_node("acme", "org") + graph.add_edge("alice", "acme", "works_at", valid_until=BEFORE) + graph.retract_node("alice", at=AFTER) + self.assertEqual(graph.edges[0].valid_until, BEFORE) + + def test_an_open_window_is_still_closed_at_the_retraction_time(self): + graph = _graph() + graph.retract_node("alice", at=CUTOFF) + self.assertEqual(graph.nodes["alice"].valid_until, "2026-01-01T00:00:00") + + +class TestPurgeTimestamp(unittest.TestCase): + """Purge accepts an explicit effective time, as retraction does.""" + + def test_node_tombstone_records_the_supplied_time(self): + graph = _graph() + graph.purge_node("alice", reason="erasure request #4", at=CUTOFF) + self.assertEqual( + graph.get_tombstone("alice")["purged_at"], "2026-01-01T00:00:00" + ) + + def test_edge_tombstone_records_the_supplied_time(self): + graph = _graph() + edge_id = graph.edges[0].edge_id + graph.purge_edge(edge_id, at=CUTOFF) + self.assertEqual( + graph.get_tombstone(edge_id)["purged_at"], "2026-01-01T00:00:00" + ) + + def test_cascaded_edge_tombstones_share_the_supplied_time(self): + graph = _graph() + edge_id = graph.edges[0].edge_id + graph.purge_node("alice", at=CUTOFF) + self.assertEqual( + graph.get_tombstone(edge_id)["purged_at"], "2026-01-01T00:00:00" + ) + + def test_purge_time_defaults_to_now(self): + graph = _graph() + graph.purge_node("alice") + self.assertIn("purged_at", graph.get_tombstone("alice")) + + +class TestIdKeyspaces(unittest.TestCase): + """Node ids are caller-supplied and edge ids are UUIDs, so they can collide.""" + + def _colliding(self): + graph = _graph() + edge_id = graph.edges[0].edge_id + graph.add_node(edge_id, "person") + return graph, edge_id + + def test_an_edge_retraction_does_not_block_a_colliding_node(self): + graph, edge_id = self._colliding() + self.assertTrue(graph.retract_edge(edge_id, reason="edge")) + self.assertTrue(graph.retract_node(edge_id, reason="node")) + self.assertEqual(graph.get_retraction(edge_id, "edge")["reason"], "edge") + self.assertEqual(graph.get_retraction(edge_id, "node")["reason"], "node") + + def test_purging_a_node_leaves_a_colliding_edge_alone(self): + graph, edge_id = self._colliding() + self.assertTrue(graph.purge_node(edge_id)) + self.assertEqual(len(graph.edges), 1) + self.assertIsNone(graph.get_tombstone(edge_id, "edge")) + self.assertIsNotNone(graph.get_tombstone(edge_id, "node")) + + def test_an_unknown_entity_kind_is_rejected(self): + with self.assertRaises(ValueError): + _graph().get_retraction("alice", "vertex") + + +class TestDuplicateEdgeId(unittest.TestCase): + """``edge_id`` is content-derived; before #926, two identical ``add_edge`` + calls produced two edge objects sharing one id. #926 stops *new* + duplicates through ``add_edge``/``add_edges``, but a graph can still carry + one from a save made before that fix, or from any other path that builds + a ``ContextEdge`` directly -- so retraction/purge must still handle it. + Every duplicate must be reached, or a retraction/tombstone record can + claim an edge is gone/inactive while a live copy remains in the graph. + """ + + def _duplicated(self): + """A graph with two distinct ``ContextEdge`` objects sharing one + edge_id, reproducing pre-#926 (or any hand-built) duplicate state + without going through the now-deduping ``add_edge``. + """ + graph = _graph() + original = graph.edges[0] + duplicate = ContextEdge( + source_id=original.source_id, + target_id=original.target_id, + edge_type=original.edge_type, + weight=original.weight, + ) + self.assertEqual(duplicate.edge_id, original.edge_id) + graph.edges.append(duplicate) + graph.edge_type_index[duplicate.edge_type].append(duplicate) + graph._adjacency[duplicate.source_id].append(duplicate) + edge_id = original.edge_id + self.assertEqual({e.edge_id for e in graph.edges}, {edge_id}) + self.assertEqual(len(graph.edges), 2) + return graph, edge_id + + def test_retract_edge_closes_every_duplicate(self): + graph, edge_id = self._duplicated() + self.assertTrue(graph.retract_edge(edge_id, reason="dup", at=CUTOFF)) + for edge in graph.edges: + self.assertEqual(edge.valid_until, "2026-01-01T00:00:00") + self.assertFalse(edge.is_active(datetime(2026, 6, 1))) + + def test_retract_node_cascade_closes_every_duplicate(self): + graph, edge_id = self._duplicated() + self.assertTrue(graph.retract_node("alice", at=CUTOFF)) + for edge in graph.edges: + self.assertEqual(edge.valid_until, "2026-01-01T00:00:00") + + def test_purge_edge_removes_every_duplicate(self): + graph, edge_id = self._duplicated() + self.assertTrue(graph.purge_edge(edge_id, reason="dup")) + self.assertFalse(any(e.edge_id == edge_id for e in graph.edges)) + + def test_purge_node_cascade_removes_every_duplicate(self): + graph, edge_id = self._duplicated() + self.assertTrue(graph.purge_node("alice")) + self.assertFalse(any(e.edge_id == edge_id for e in graph.edges)) + + def test_repeat_purge_edge_does_not_overwrite_the_tombstone(self): + """Once every duplicate is gone, a second call must no-op, not + silently 'complete' the purge again and clobber the original record.""" + graph, edge_id = self._duplicated() + self.assertTrue(graph.purge_edge(edge_id, reason="first")) + self.assertFalse(graph.purge_edge(edge_id, reason="second")) + self.assertEqual(graph.get_tombstone(edge_id)["reason"], "first") + + +class TestPurgeCrossGraphLinks(unittest.TestCase): + """link_graph() registers a link, a marker node and a bridge edge.""" + + def _linked(self): + graph = _graph() + other = ContextGraph(advanced_analytics=False) + other.add_node("target", "topic") + return graph, other, graph.link_graph(other, "alice", "target") + + def test_purging_the_source_removes_link_marker_and_registration(self): + graph, _, link_id = self._linked() + graph.purge_node("alice", reason="erasure request #5") + self.assertFalse(graph.has_node(f"__cross_graph_{link_id}")) + with self.assertRaises(KeyError): + graph.navigate_to(link_id) + totals = _index_totals(graph) + self.assertEqual(totals["node_index"], totals["nodes"]) + self.assertEqual(totals["edge_index"], totals["edges"]) + self.assertEqual(totals["adjacency"], totals["edges"]) + + def test_the_marker_purge_is_recorded_as_cascaded(self): + graph, _, link_id = self._linked() + graph.purge_node("alice") + tombstone = graph.get_tombstone(f"__cross_graph_{link_id}") + self.assertEqual(tombstone["cascaded_from"], "alice") + + def test_a_purged_link_is_not_serialized(self): + graph, _, _ = self._linked() + graph.purge_node("alice") + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "graph.json") + graph.save_to_file(path) + with open(path, encoding="utf-8") as handle: + data = json.load(handle) + self.assertEqual(data["links"], []) + + def test_cascade_disabled_still_deregisters_the_link(self): + """The source node is gone either way, so the link cannot resolve.""" + graph, _, link_id = self._linked() + graph.purge_node("alice", cascade=False) + with self.assertRaises(KeyError): + graph.navigate_to(link_id) + self.assertTrue(graph.has_node(f"__cross_graph_{link_id}")) + + def test_purging_the_bridge_edge_deregisters_the_link(self): + graph, _, link_id = self._linked() + bridge = next( + edge for edge in graph.edges if edge.metadata.get("link_id") == link_id + ) + self.assertTrue(graph.purge_edge(bridge.edge_id)) + with self.assertRaises(KeyError): + graph.navigate_to(link_id) + + def test_purging_the_marker_node_deregisters_the_link(self): + graph, _, link_id = self._linked() + self.assertTrue(graph.purge_node(f"__cross_graph_{link_id}")) + with self.assertRaises(KeyError): + graph.navigate_to(link_id) + self.assertTrue(graph.has_node("alice")) + + def test_an_unrelated_link_survives(self): + graph, other, link_id = self._linked() + graph.purge_node("bob") + self.assertEqual(graph.navigate_to(link_id), (other, "target")) + + +class TestClearResetsRecords(unittest.TestCase): + def test_clear_drops_retractions_and_tombstones(self): + graph = _graph() + graph.retract_node("alice", at=CUTOFF) + graph.purge_node("bob") + graph.clear() + self.assertEqual(graph.list_retractions(), []) + self.assertEqual(graph.list_tombstones(), []) + + def test_load_from_file_drops_records_from_the_previous_graph(self): + source = _graph() + graph = ContextGraph(advanced_analytics=False) + graph.add_node("alice", "person") + graph.add_node("carol", "person") + graph.retract_node("alice", at=CUTOFF) + graph.purge_node("carol") + + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "graph.json") + source.save_to_file(path) + graph.load_from_file(path) + + self.assertEqual(graph.list_retractions(), []) + self.assertEqual(graph.list_tombstones(), []) + # The reloaded alice is a fresh record, not one already retracted. + self.assertTrue(graph.retract_node("alice", at=CUTOFF)) + + +class TestAuditTrailIntegration(unittest.TestCase): + """Against the real TemporalVersionManager, not a mock callback.""" + + def _attached(self): + manager = TemporalVersionManager() + graph = _graph() + manager.attach_to_graph(graph) + return manager, graph + + def _ops(self, manager, entity_id): + history = manager.storage.get_entity_history(entity_id) or [] + return [entry.get("operation") for entry in history] + + def test_retraction_is_recorded_as_an_update(self): + manager, graph = self._attached() + graph.retract_node("alice", reason="left", at=CUTOFF) + self.assertIn("UPDATE_NODE", self._ops(manager, "alice")) + + def test_purge_is_recorded_as_a_removal(self): + manager, graph = self._attached() + graph.purge_node("acme", reason="erasure request #3") + self.assertIn("REMOVE_NODE", self._ops(manager, "acme")) + + def test_operations_use_the_documented_mutation_vocabulary(self): + """MutationRecord documents ADD/UPDATE/REMOVE for nodes and edges.""" + manager, graph = self._attached() + graph.retract_node("alice", at=CUTOFF) + graph.purge_node("bob") + allowed = { + "ADD_NODE", + "UPDATE_NODE", + "REMOVE_NODE", + "ADD_EDGE", + "UPDATE_EDGE", + "REMOVE_EDGE", + } + seen = set() + for entity_id in ("alice", "acme", "bob"): + seen.update(self._ops(manager, entity_id)) + self.assertTrue(seen) + self.assertTrue( + seen <= allowed, f"undocumented mutation operation(s): {seen - allowed}" + ) + + +class TestMutationEmissionIsSelfContained(unittest.TestCase): + """Audit payloads must be snapshotted before the lock is released. + + The callback fires outside the lock, so anything read from + ``_retractions``/``_tombstones`` at emission time can already have been + wiped by a concurrent ``clear()``. A callback that clears the graph on its + first call stands in for that interleaving deterministically. + """ + + def _clearing_callback(self, graph, seen): + def callback(operation, entity_id, payload): + seen.append((operation, entity_id, payload)) + if len(seen) == 1: + graph.clear() + + return callback + + def test_purge_emits_every_mutation_after_a_concurrent_clear(self): + graph = _graph() + graph.add_edge("bob", "alice", "knows") + seen = [] + graph.mutation_callback = self._clearing_callback(graph, seen) + + self.assertTrue(graph.purge_node("alice", reason="erasure request #6")) + + self.assertEqual( + [operation for operation, _, _ in seen], + ["REMOVE_EDGE", "REMOVE_EDGE", "REMOVE_NODE"], + ) + for _, entity_id, payload in seen: + self.assertEqual(payload["entity_id"], entity_id) + self.assertEqual(payload["reason"], "erasure request #6") + + def test_retraction_emits_every_mutation_after_a_concurrent_clear(self): + graph = _graph() + graph.add_edge("bob", "alice", "knows") + seen = [] + graph.mutation_callback = self._clearing_callback(graph, seen) + + self.assertTrue(graph.retract_node("alice", reason="left", at=CUTOFF)) + + self.assertEqual( + [operation for operation, _, _ in seen], + ["UPDATE_NODE", "UPDATE_EDGE", "UPDATE_EDGE"], + ) + for _, _, payload in seen: + self.assertEqual(payload["retraction"]["reason"], "left") + + +class TestConcurrency(unittest.TestCase): + def test_concurrent_purges_keep_indexes_consistent(self): + """Post-condition, not timing: threads must finish and indexes agree.""" + graph = ContextGraph(advanced_analytics=False) + for i in range(60): + graph.add_node(f"n{i}", "t") + for i in range(59): + graph.add_edge(f"n{i}", f"n{i + 1}", "rel") + + errors = [] + + def purge(start): + try: + for i in range(start, 60, 4): + graph.purge_node(f"n{i}") + except Exception as exc: # surfaced below, never swallowed + errors.append(f"{type(exc).__name__}: {exc}") + + threads = [ + threading.Thread(target=purge, args=(offset,)) for offset in range(4) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + self.assertEqual([t.name for t in threads if t.is_alive()], []) + self.assertEqual(errors, []) + self.assertEqual(len(graph.nodes), 0) + totals = _index_totals(graph) + self.assertEqual(totals["node_index"], 0) + self.assertEqual(totals["edge_index"], 0) + self.assertEqual(totals["adjacency"], 0) + self.assertEqual(totals["edges"], 0) + + +if __name__ == "__main__": + unittest.main()