mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(context): add retraction and purge to ContextGraph (#957)

Browse Source 
* feat(context): add retraction and purge to ContextGraph

ContextGraph had 56 public methods and none that removed anything: the only
option was clear(), which discards the whole graph. Removing one entity meant
exporting to a dict, filtering by hand and rebuilding, losing provenance.

Add two operations with deliberately different contracts.

retract_node/retract_edge close the 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. This reuses the
valid_from/valid_until machinery already present rather than adding a new
subsystem.

purge_node/purge_edge remove the entity outright, from history as well as from
the active view, leaving a tombstone that records that a purge happened and why
but never the purged content. Scope is this graph only; copies in AgentMemory
or a bound vector store are not reached, so it is one step of an erasure
workflow rather than the whole of it.

Both record themselves through the existing mutation_callback path.
MutationRecord already documented REMOVE_NODE/REMOVE_EDGE in its operation
vocabulary, so retraction emits UPDATE_NODE and purge emits REMOVE_NODE with no
changes required to change_management.

Incident-edge lookup scans self.edges rather than _adjacency, which is keyed by
source only and would otherwise leave inbound edges pointing at a removed node.
Purge updates edges, edge_type_index and _adjacency together so the indexes
cannot drift, and clear() now resets the retraction and tombstone records.

* fix(context): address review findings on retraction and purge

* fix(context): close every duplicate when retracting/purging by edge_id

edge_id is content-derived and not yet guaranteed unique (#922, fix
pending in #926): two identical add_edge() calls produce two edge
objects sharing one id. retract_edge()/purge_edge() resolved "the
edge" via the first matching object only, so a duplicate was silently
left untouched (still live, still active) while the call returned
True and recorded a tombstone/retraction claiming it 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 once nothing remained to purge.

retract_node()'s cascade had the same root cause from the other
direction: it checked the live _retractions dict mid-loop, so the
first duplicate's just-written record made the second look already
handled and it was skipped outright, left permanently active.

retract_edge()/purge_edge() now act on every edge matching the id
under a single record; the cascade's dedup check is snapshotted
before the loop starts so within-call duplicates are still closed
rather than skipped.

Adds TestDuplicateEdgeId (5 tests) reproducing all three paths.

* docs(changelog): document retraction/purge feature

Adds an Unreleased/Added entry for #955/#957 covering retract_node,
retract_edge, purge_node, purge_edge and the get/list accessors, plus
the duplicate-edge_id fix caught and applied during review.

---------

Co-authored-by: Pravit Ampapathini <pravitampapathini@Pravits-MacBook-Air-3.local>
Co-authored-by: KaifAhmad1 <kaifahmad087@gmail.com>
This commit is contained in:
pravit-amp
2026-08-15 17:12:10 +05:30
committed by GitHub
co-authored by Pravit Ampapathini KaifAhmad1
parent 6df97cf0a0
commit f1e7e64ad1
3 changed files with 1146 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+10
View File
@@ -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
semantica/context/context_graph.py
+541
View File
@@ -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(
tests/context/test_context_graph_retraction.py
+595
View File
@@ -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()