fix(context): take the lock in ContextGraph.to_dict() (#929)

* fix(context): take the lock in ContextGraph.to_dict()

to_dict() iterated self.nodes.values() and self.edges without holding
self._lock, so a concurrent writer raised "RuntimeError: dictionary changed
size during iteration". It was the only reader on the class that did not take
the lock -- stats(), density(), find_nodes(), find_edges(), get_neighbors(),
get_nodes_by_label(), state_at() and save_to_file() all hold it.

Commit 1d1ae398 introduced the RLock and added 26 "with self._lock:" blocks;
to_dict already existed and was not among them. save_to_file is safe only
incidentally -- it holds the lock and builds its payload inline rather than
delegating to to_dict, so it never reaches the unguarded loops.

Beyond the RuntimeError, the unguarded body could also return a torn snapshot:
the statistics block reads len(self.nodes)/len(self.edges) after building the
node and edge lists, so a write landing in between yields counts that
contradict the payload they describe.

self._lock is an RLock, so this composes with the callers that already hold it
(build_from_conversation and build_from_documents both return self.to_dict()
from inside a locked block). Neither external caller -- agent_context's
_capture_checkpoint_state nor triplet_store's knowledge-graph conversion --
defines a lock of its own, so there is no ordering inversion.

Add tests/context/test_context_graph_thread_safety.py: a deterministic check
that to_dict() blocks while another thread holds _lock (no race window
needed), a reentrancy check, and three checks under concurrent writes covering
the RuntimeError, statistics/payload agreement, and duplicate node ids. Four
of the five fail against the unfixed method.

Closes #923

* test(context): make to_dict lock tests deterministic and hang-proof

Wait for the worker thread to actually start before asserting to_dict()
blocks on _lock, and run the reentrancy check in a joined worker so a
non-reentrant lock fails the test instead of hanging CI.

* test(context): assert worker threads actually stopped after timed joins

A join(timeout=...) on a daemon thread returns even if the thread is
still running, so a deadlock would leak a live thread into subsequent
tests instead of failing. Assert not is_alive() after each timed join.

---------

Co-authored-by: Pravit Ampapathini <pravitampapathini@Pravits-MacBook-Air-3.local>
Co-authored-by: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com>

semantica/context/context_graph.py

@@ -1769,47 +1769,48 @@ class ContextGraph:
 
     def to_dict(self) -> Dict[str, Any]:
         """Export graph to dictionary format."""
-        nodes_out = []
-        for n in self.nodes.values():
-            entry: Dict[str, Any] = {
-                "id": n.node_id,
-                "type": n.node_type,
-                "content": n.content,
-                "properties": n.properties,
-                "metadata": n.metadata,
-            }
-            if n.valid_from is not None:
-                entry["valid_from"] = n.valid_from
-            if n.valid_until is not None:
-                entry["valid_until"] = n.valid_until
-            nodes_out.append(entry)
+        with self._lock:
+            nodes_out = []
+            for n in self.nodes.values():
+                entry: Dict[str, Any] = {
+                    "id": n.node_id,
+                    "type": n.node_type,
+                    "content": n.content,
+                    "properties": n.properties,
+                    "metadata": n.metadata,
+                }
+                if n.valid_from is not None:
+                    entry["valid_from"] = n.valid_from
+                if n.valid_until is not None:
+                    entry["valid_until"] = n.valid_until
+                nodes_out.append(entry)
 
-        edges_out = []
-        for e in self.edges:
-            entry = {
-                "id": e.edge_id,
-                "familyId": e.family_id or e.edge_id,
-                "source": e.source_id,
-                "target": e.target_id,
-                "type": e.edge_type,
-                "weight": e.weight,
-            }
-            if e.metadata:
-                entry["metadata"] = e.metadata
-            if e.valid_from is not None:
-                entry["valid_from"] = e.valid_from
-            if e.valid_until is not None:
-                entry["valid_until"] = e.valid_until
-            edges_out.append(entry)
+            edges_out = []
+            for e in self.edges:
+                entry = {
+                    "id": e.edge_id,
+                    "familyId": e.family_id or e.edge_id,
+                    "source": e.source_id,
+                    "target": e.target_id,
+                    "type": e.edge_type,
+                    "weight": e.weight,
+                }
+                if e.metadata:
+                    entry["metadata"] = e.metadata
+                if e.valid_from is not None:
+                    entry["valid_from"] = e.valid_from
+                if e.valid_until is not None:
+                    entry["valid_until"] = e.valid_until
+                edges_out.append(entry)
 
-        return {
-            "nodes": nodes_out,
-            "edges": edges_out,
-            "statistics": {
-                "node_count": len(self.nodes),
-                "edge_count": len(self.edges),
-            },
-        }
+            return {
+                "nodes": nodes_out,
+                "edges": edges_out,
+                "statistics": {
+                    "node_count": len(self.nodes),
+                    "edge_count": len(self.edges),
+                },
+            }
 
     def from_dict(self, graph_dict: Dict[str, Any]) -> None:
         """Load graph from dictionary format."""

tests/context/test_context_graph_thread_safety.py

@@ -0,0 +1,168 @@
+#!/usr/bin/env python3
+"""Regression tests for ``ContextGraph.to_dict()`` thread safety.
+
+``ContextGraph`` guards its state with ``self._lock`` (an ``RLock``), and every
+reader on the class takes it -- ``stats``, ``density``, ``find_nodes``,
+``find_edges``, ``get_neighbors``, ``get_nodes_by_label``, ``state_at`` and
+``save_to_file`` all do. ``to_dict`` was the one exception: it iterated
+``self.nodes.values()`` and ``self.edges`` unguarded, so a concurrent writer
+raised ``RuntimeError: dictionary changed size during iteration``.
+
+``save_to_file`` was safe only incidentally -- it holds the lock and builds its
+payload inline rather than delegating to ``to_dict``.
+"""
+
+import threading
+import time
+
+from semantica.context.context_graph import ContextGraph
+
+
+def _seeded_graph(node_count: int = 200) -> ContextGraph:
+    graph = ContextGraph(advanced_analytics=False)
+    for i in range(node_count):
+        graph.add_node(f"seed{i}", "seed")
+    return graph
+
+
+class TestToDictHoldsTheLock:
+    """``to_dict`` must take ``_lock``, like every sibling reader."""
+
+    def test_to_dict_waits_for_the_lock(self):
+        """Deterministic proof the lock is held -- no race window needed.
+
+        With the lock held elsewhere, ``to_dict`` must block. Without the fix it
+        returns immediately, since it never asks for the lock at all.
+        """
+        graph = _seeded_graph(10)
+        started = threading.Event()
+        finished = threading.Event()
+
+        def snapshot():
+            started.set()
+            graph.to_dict()
+            finished.set()
+
+        with graph._lock:
+            worker = threading.Thread(target=snapshot, daemon=True)
+            worker.start()
+            assert started.wait(timeout=5.0), "the worker thread never started running"
+            # The worker is now running and cannot finish while this thread
+            # owns the lock.
+            assert not finished.wait(timeout=0.5), (
+                "to_dict() completed while another thread held _lock, so it is "
+                "reading graph state unguarded"
+            )
+
+        assert finished.wait(timeout=5.0), "to_dict() did not complete after _lock was released"
+        worker.join(timeout=5.0)
+        assert not worker.is_alive(), "the worker thread is still running after to_dict() finished"
+
+    def test_to_dict_is_reentrant_for_a_caller_holding_the_lock(self):
+        """``_lock`` is an ``RLock``, so lock-holding callers must not deadlock.
+
+        The nested acquisition runs in a daemon worker joined with a timeout so
+        that a non-reentrant lock fails the test instead of hanging it.
+        """
+        graph = _seeded_graph(10)
+        result = {}
+
+        def nested_snapshot():
+            with graph._lock:
+                result["snapshot"] = graph.to_dict()
+
+        worker = threading.Thread(target=nested_snapshot, daemon=True)
+        worker.start()
+        worker.join(timeout=5.0)
+
+        assert not worker.is_alive(), (
+            "to_dict() deadlocked when called by a thread already holding "
+            "_lock -- the lock is no longer reentrant"
+        )
+        assert len(result["snapshot"]["nodes"]) == 10
+
+
+class TestToDictUnderConcurrentWrites:
+    """The reported race: snapshot one thread, mutate from another."""
+
+    def _run_race(self, graph: ContextGraph, reader, duration: float = 1.0):
+        """Hammer ``reader`` while a writer adds nodes. Returns (errors, reads)."""
+        stop = threading.Event()
+        errors = []
+        reads = []
+
+        def writer():
+            i = 0
+            while not stop.is_set():
+                try:
+                    graph.add_node(f"w{i}", "written")
+                except Exception as exc:  # pragma: no cover - writer must stay healthy
+                    errors.append(exc)
+                    return
+                i += 1
+
+        def reader_loop():
+            while not stop.is_set():
+                try:
+                    reads.append(reader())
+                except Exception as exc:
+                    errors.append(exc)
+                    stop.set()
+                    return
+
+        threads = [
+            threading.Thread(target=writer, daemon=True),
+            threading.Thread(target=reader_loop, daemon=True),
+        ]
+        for thread in threads:
+            thread.start()
+        time.sleep(duration)
+        stop.set()
+        for thread in threads:
+            thread.join(timeout=5.0)
+            assert not thread.is_alive(), (
+                "a worker thread was still running 5s after the stop signal -- "
+                "a hang here would otherwise leak into subsequent tests"
+            )
+
+        return errors, reads
+
+    def test_to_dict_does_not_raise_during_concurrent_writes(self):
+        graph = _seeded_graph()
+        errors, reads = self._run_race(graph, graph.to_dict)
+
+        assert not errors, f"to_dict() raised under concurrent writes: {errors[0]!r}"
+        assert reads, "the reader thread never completed a to_dict() call"
+
+    def test_to_dict_snapshot_is_internally_consistent(self):
+        """The reported statistics must describe the payload actually emitted.
+
+        ``to_dict`` builds ``nodes``/``edges`` and then reads ``len(self.nodes)``
+        and ``len(self.edges)`` for its ``statistics`` block. Unguarded, a write
+        landing between those steps yields counts that contradict the lists.
+        """
+        graph = _seeded_graph()
+        errors, reads = self._run_race(graph, graph.to_dict)
+
+        assert not errors, f"to_dict() raised under concurrent writes: {errors[0]!r}"
+        assert reads, "the reader thread never completed a to_dict() call"
+        for snapshot in reads:
+            stats = snapshot["statistics"]
+            assert stats["node_count"] == len(snapshot["nodes"]), (
+                f"statistics.node_count={stats['node_count']} contradicts the "
+                f"{len(snapshot['nodes'])} nodes in the same snapshot"
+            )
+            assert stats["edge_count"] == len(snapshot["edges"]), (
+                f"statistics.edge_count={stats['edge_count']} contradicts the "
+                f"{len(snapshot['edges'])} edges in the same snapshot"
+            )
+
+    def test_snapshot_node_ids_are_unique(self):
+        """A torn read can emit the same node twice; a locked one cannot."""
+        graph = _seeded_graph()
+        errors, reads = self._run_race(graph, graph.to_dict)
+
+        assert not errors, f"to_dict() raised under concurrent writes: {errors[0]!r}"
+        for snapshot in reads:
+            ids = [node["id"] for node in snapshot["nodes"]]
+            assert len(ids) == len(set(ids)), "to_dict() emitted duplicate node ids"