From 9fc72121c86a18a35ac7c61d7f9bdf77852b8eaf Mon Sep 17 00:00:00 2001 From: MrNirmata <110922639+MrNirmata@users.noreply.github.com> Date: Sat, 4 Jul 2026 23:41:06 +0200 Subject: [PATCH] fix(community): bound label propagation iterations to guarantee termination label_propagation() ran a `while True` loop that only exited on full convergence. Synchronous (Jacobi-style) label propagation is not guaranteed to converge: on symmetric or tie-heavy graphs the assignment can settle into a period-two oscillation and flip between two states forever, so the loop never terminates. build_communities() then hangs, pinning a CPU with no progress and no error. Cap the propagation at MAX_LABEL_PROPAGATION_ITERATIONS (1000) and return the latest assignment when it has not converged, logging a warning. Graphs that do converge are unaffected: the `no_change` early-exit still fires exactly as before, so results are identical for every converging input. Add unit tests for a small non-convergent graph that previously hung, the non-convergence warning, and a separable graph that still converges to the expected communities. --- .../utils/maintenance/community_operations.py | 18 +++- .../maintenance/test_community_operations.py | 87 +++++++++++++++++++ 2 files changed, 103 insertions(+), 2 deletions(-) create mode 100644 tests/utils/maintenance/test_community_operations.py diff --git a/graphiti_core/utils/maintenance/community_operations.py b/graphiti_core/utils/maintenance/community_operations.py index 8c96bd79f4..7b4d955f8c 100644 --- a/graphiti_core/utils/maintenance/community_operations.py +++ b/graphiti_core/utils/maintenance/community_operations.py @@ -18,6 +18,7 @@ from graphiti_core.utils.text_utils import MAX_SUMMARY_CHARS, truncate_at_sentence MAX_COMMUNITY_BUILD_CONCURRENCY = 10 +MAX_LABEL_PROPAGATION_ITERATIONS = 1000 logger = logging.getLogger(__name__) @@ -95,11 +96,16 @@ def label_propagation(projection: dict[str, list[Neighbor]]) -> list[list[str]]: # 1. Start with each node being assigned its own community # 2. Each node will take on the community of the plurality of its neighbors # 3. Ties are broken by going to the largest community - # 4. Continue until no communities change during propagation + # 4. Continue until no communities change during propagation, or until a + # bounded number of iterations is reached. Synchronous label propagation + # is not guaranteed to converge on symmetric or tie-heavy graphs, where + # the assignment can oscillate indefinitely, so the iteration count is + # capped to guarantee termination. community_map = {uuid: i for i, uuid in enumerate(projection.keys())} - while True: + converged = False + for _ in range(MAX_LABEL_PROPAGATION_ITERATIONS): no_change = True new_community_map: dict[str, int] = {} @@ -126,10 +132,18 @@ def label_propagation(projection: dict[str, list[Neighbor]]) -> list[list[str]]: no_change = False if no_change: + converged = True break community_map = new_community_map + if not converged: + logger.warning( + 'label_propagation did not converge within %d iterations; ' + 'returning the latest community assignment', + MAX_LABEL_PROPAGATION_ITERATIONS, + ) + community_cluster_map = defaultdict(list) for uuid, community in community_map.items(): community_cluster_map[community].append(uuid) diff --git a/tests/utils/maintenance/test_community_operations.py b/tests/utils/maintenance/test_community_operations.py new file mode 100644 index 0000000000..8b232e79d1 --- /dev/null +++ b/tests/utils/maintenance/test_community_operations.py @@ -0,0 +1,87 @@ +import logging +import threading + +from graphiti_core.utils.maintenance.community_operations import Neighbor, label_propagation + + +def _call_with_timeout(func, timeout: float = 10.0): + """Run ``func`` in a daemon thread and return (result, finished). + + ``label_propagation`` is a pure, CPU-bound function, so a thread that is + stuck in an unbounded loop cannot be interrupted; the daemon thread is + abandoned when the test process exits. This lets a non-terminating call + fail the assertion quickly instead of hanging the whole test session. + """ + box: dict = {} + + def target(): + box['result'] = func() + + thread = threading.Thread(target=target, daemon=True) + thread.start() + thread.join(timeout) + if thread.is_alive(): + return None, False + return box['result'], True + + +def _non_convergent_projection() -> dict[str, list[Neighbor]]: + """A small symmetric weighted graph on which synchronous label propagation + oscillates with period two and never converges. + + Starting from the one-community-per-node seed, every pass flips the two + halves of the graph, so ``no_change`` is never reached. Without an + iteration bound the propagation loop runs forever on this input. + """ + edges = { + 'n0': [('n1', 2), ('n2', 2), ('n3', 1)], + 'n1': [('n0', 2), ('n2', 3), ('n3', 2)], + 'n2': [('n0', 2), ('n1', 3)], + 'n3': [('n0', 1), ('n1', 2)], + } + return { + uuid: [Neighbor(node_uuid=nb, edge_count=weight) for nb, weight in neighbors] + for uuid, neighbors in edges.items() + } + + +def test_label_propagation_terminates_on_non_convergent_graph(): + projection = _non_convergent_projection() + + clusters, finished = _call_with_timeout(lambda: label_propagation(projection)) + + assert finished, 'label_propagation did not terminate on a non-convergent graph' + + # The result must still be a valid partition: every node appears in exactly + # one cluster and no cluster is empty. + assigned = [uuid for cluster in clusters for uuid in cluster] + assert sorted(assigned) == sorted(projection.keys()) + assert all(len(cluster) > 0 for cluster in clusters) + + +def test_label_propagation_warns_when_iteration_cap_is_reached(caplog): + projection = _non_convergent_projection() + + with caplog.at_level( + logging.WARNING, logger='graphiti_core.utils.maintenance.community_operations' + ): + _, finished = _call_with_timeout(lambda: label_propagation(projection)) + + assert finished + assert any(record.levelno == logging.WARNING for record in caplog.records) + + +def test_label_propagation_still_converges_on_separable_graph(): + # Two disconnected pairs form two obvious communities; propagation converges + # normally, so the iteration bound must not change the result. + projection: dict[str, list[Neighbor]] = { + 'a': [Neighbor(node_uuid='b', edge_count=1)], + 'b': [Neighbor(node_uuid='a', edge_count=1)], + 'c': [Neighbor(node_uuid='d', edge_count=1)], + 'd': [Neighbor(node_uuid='c', edge_count=1)], + } + + clusters = label_propagation(projection) + + partition = {frozenset(cluster) for cluster in clusters} + assert partition == {frozenset({'a', 'b'}), frozenset({'c', 'd'})}