fix(core): exclude only the edge's own sources from the weight sum

Follow-up hardening of the edge-weight fix for the reprocess/partial-write path.
The weight filter must exclude sources ALREADY reflected in the stored scalar
weight. That scalar comes from the graph edge (already_weights), so filter on
the edge's own source_ids (already_source_ids) — NOT existing_full_source_ids,
which can come from relation_chunks_storage, a SEPARATE store.

relation_chunks_storage and the graph edge can diverge on a partial write
(chunks upserted, upsert_edge not yet):
- chunks present, edge absent -> filtering on chunks recovered the edge with
  weight 0 (raised in the Codex review of #3373);
- chunks AHEAD of the edge -> filtering on chunks skipped a genuinely new
  source, under-counting the weight.

Filtering on already_source_ids keeps the exclusion consistent with the stored
scalar in both cases: no stored edge -> nothing excluded (a recovered edge keeps
its full weight); a re-fed already-stored source -> dropped (no reprocess
double-count).

Regression tests: partial-write recovery keeps weight 1 (not 0); a new source
re-fed while relation_chunks is ahead of the edge still adds weight (2, not 1);
same-source reprocess stays fixed; distinct sources still accumulate.

Author: ysys143

lightrag/operate.py

@@ -2510,12 +2510,23 @@ async def _merge_edges_then_upsert(
         # per reprocess (the #3367 sibling of description accumulation).
         # Genuinely new sources still add their weight, preserving legitimate
         # multi-document growth.
-        already_source_id_set = set(existing_full_source_ids)
+        #
+        # Filter on the EDGE's own source_ids (already_source_ids) — the same
+        # source as already_weights — NOT existing_full_source_ids, which can
+        # come from relation_chunks_storage, a SEPARATE store. relation_chunks
+        # can run ahead of the graph edge on a partial write (chunks upserted,
+        # upsert_edge not yet), so filtering on it would wrongly skip a genuinely
+        # new source (under-count), or, with no stored weight at all, recover the
+        # edge with weight 0. Filtering on already_source_ids keeps the exclusion
+        # consistent with the stored scalar: when there is no stored edge
+        # (already_source_ids empty) nothing is excluded, so a recovered edge
+        # keeps its full weight; a re-fed already-stored source is dropped.
+        already_edge_source_set = set(already_source_ids)
         weight = sum(
             [
                 dp["weight"]
                 for dp in edges_data
-                if dp.get("source_id") not in already_source_id_set
+                if dp.get("source_id") not in already_edge_source_set
             ]
             + already_weights
         )

tests/extraction/test_summary_description_dedup.py

@@ -21,6 +21,7 @@
     _handle_entity_relation_summary,
     _merge_edges_then_upsert,
     _merge_nodes_then_upsert,
+    make_relation_chunk_key,
 )
 from lightrag.constants import GRAPH_FIELD_SEP
 
@@ -257,3 +258,113 @@ def _edge(src: str) -> dict:
         await _merge_edges_then_upsert("A", "B", [_edge(src)], g, None, None, cfg)
 
     assert await g.get_edge("A", "B") is not None
+
+
+class _MemKV:
+    """No-op-ish KV that just serves a preset get_by_id."""
+
+    def __init__(self, data: dict):
+        self.data = dict(data)
+
+    async def get_by_id(self, key):
+        return self.data.get(key)
+
+    async def upsert(self, data):
+        self.data.update(data)
+
+
+@pytest.mark.offline
+@pytest.mark.asyncio
+async def test_edge_weight_recovers_to_one_when_edge_missing_but_chunks_stored():
+    """Partial-write recovery: relation_chunks_storage already has the source but
+    the graph edge does not exist yet (a crash between the two writes). The weight
+    filter must NOT treat those sources as already-weighted — already_weights is
+    empty, so filtering them would recover the edge with weight 0. With no stored
+    scalar, sum all contributions (weight = 1, not 0)."""
+    g = _MemGraph()
+    cfg = _cfg()
+    for name in ("A", "B"):
+        await g.upsert_node(
+            name,
+            {
+                "entity_id": name,
+                "description": name,
+                "source_id": "c1",
+                "entity_type": "X",
+                "file_path": "f",
+            },
+        )
+    rcs = _MemKV({make_relation_chunk_key("A", "B"): {"chunk_ids": ["c1"]}})
+    await _merge_edges_then_upsert(
+        "A",
+        "B",
+        [
+            {
+                "weight": 1.0,
+                "source_id": "c1",
+                "description": "rel",
+                "keywords": "k",
+                "file_path": "f",
+            }
+        ],
+        g,
+        None,
+        None,
+        cfg,
+        relation_chunks_storage=rcs,
+    )
+    edge = await g.get_edge("A", "B")
+    assert edge is not None
+    assert edge["weight"] == 1.0
+
+
+@pytest.mark.offline
+@pytest.mark.asyncio
+async def test_edge_weight_adds_new_source_when_chunks_store_is_ahead():
+    """relation_chunks_storage can run AHEAD of the graph edge (chunks upserted,
+    edge not yet updated). A genuinely new source re-fed then must still add
+    weight — the filter uses the EDGE's own source_ids (consistent with the
+    stored scalar), not the chunk store, so the new source is not wrongly skipped
+    (which would under-count the weight)."""
+    g = _MemGraph()
+    cfg = _cfg()
+    for name in ("A", "B"):
+        await g.upsert_node(
+            name,
+            {
+                "entity_id": name,
+                "description": name,
+                "source_id": "c1",
+                "entity_type": "X",
+                "file_path": "f",
+            },
+        )
+    # Existing edge reflects only c1 (weight 1)...
+    g.edges[("A", "B")] = {
+        "weight": 1.0,
+        "source_id": "c1",
+        "description": "rel",
+        "keywords": "k",
+        "file_path": "f",
+    }
+    # ...but relation_chunks already lists c1 AND c2 (ahead of the edge).
+    rcs = _MemKV({make_relation_chunk_key("A", "B"): {"chunk_ids": ["c1", "c2"]}})
+    await _merge_edges_then_upsert(
+        "A",
+        "B",
+        [
+            {
+                "weight": 1.0,
+                "source_id": "c2",
+                "description": "rel",
+                "keywords": "k",
+                "file_path": "f",
+            }
+        ],
+        g,
+        None,
+        None,
+        cfg,
+        relation_chunks_storage=rcs,
+    )
+    assert (await g.get_edge("A", "B"))["weight"] == 2.0