diff --git a/backend/app/knowledge_graph.py b/backend/app/knowledge_graph.py index d0bf890c5..82e57b636 100644 --- a/backend/app/knowledge_graph.py +++ b/backend/app/knowledge_graph.py @@ -623,7 +623,7 @@ async def post_knowledge_graph( from post_summary_semantic_relationship where post_id = $1 order by relation_ordinal - limit $2 + limit ($2 + 1) """, post_id, relation_limit, @@ -633,7 +633,7 @@ def semantic_key(node_type: str, name: str) -> str: digest = hashlib.sha256(f"{node_type}\0{name}".encode()).hexdigest()[:16] return f"semantic:{post_id}:{digest}" - for row in relation_rows: + for row in relation_rows[:relation_limit]: source = semantic_key(row["subject_type"], row["subject_name"]) target = semantic_key(row["object_type"], row["object_name"]) for key, node_type, name in ( @@ -670,7 +670,7 @@ def semantic_key(node_type: str, name: str) -> str: "post_id": post_id, "nodes": list(nodes.values()), "edges": edges, - "truncated": len(catalog_edges) > relation_limit or len(relation_rows) >= relation_limit, + "truncated": len(catalog_edges) > relation_limit or len(relation_rows) > relation_limit, } diff --git a/backend/app/main.py b/backend/app/main.py index c928b691b..d1f9e0a60 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -2645,12 +2645,13 @@ async def read_post_summary( and load_settings().orchestrator_api_key ), ) - job = await ensure_post_content_job( - conn, - post_id, - raw_body, - content_complete=content_complete, - ) + async with conn.transaction(): + job = await ensure_post_content_job( + conn, + post_id, + raw_body, + content_complete=content_complete, + ) if job.should_publish: queue_event = (job.post_id, job.source_body_sha256) summary_waiting_for_images = not await post_content_summary_is_ready(conn, post_id) diff --git a/docker/postgres-init/migrate.sh b/docker/postgres-init/migrate.sh index 11425ea6c..e76d3c051 100644 --- a/docker/postgres-init/migrate.sh +++ b/docker/postgres-init/migrate.sh @@ -18,7 +18,7 @@ for migration in /opt/lineageweave/migrations/*.sql; do migration_name=${migration##*/} case "$migration_name" in 0012_*|0013_*|0014_*|0015_*|0016_*|0017_*|0018_*|0019_*|0020_*|0021_*|0022_*|0023_*|0024_*|0025_*|0026_*|0027_*|0028_*|0029_*|0030_*|0031_*|0032_*|0033_*|0034_*|0035_*|0036_*|0037_*|0038_*|0039_*|0040_*|0041_*|0042_*|0043_*|0044_*|0045_*|0046_*|0047_*|0048_*|0049_*|0050_*) ;; - 0060_*|0100_*|0101_*|0102_*|0103_*|0104_*|0106_*|0107_*|0108_*|0109_*|0110_*|0111_*|0112_*|0113_*|0114_*) ;; + 0060_*|0100_*|0101_*|0102_*|0103_*|0104_*|0105_*|0106_*|0107_*|0108_*|0109_*|0110_*|0111_*|0112_*|0113_*|0114_*) ;; *) continue ;; esac printf 'Applying %s\n' "$migration_name" diff --git a/tests/test_knowledge_graph.py b/tests/test_knowledge_graph.py index b2d023871..ef6afb5a3 100644 --- a/tests/test_knowledge_graph.py +++ b/tests/test_knowledge_graph.py @@ -16,6 +16,8 @@ from __future__ import annotations +import asyncio + import pytest from lineageweave.knowledge_graph import ( @@ -30,6 +32,7 @@ random_walk_with_restart, select_related_nodes, ) +from backend.app.knowledge_graph import post_knowledge_graph @pytest.fixture @@ -72,6 +75,58 @@ def test_adaptive_depth_hub_reaches_more_nodes_than_a_sparse_node(synthetic_grap assert len(hub_related) > len(loner_related) +@pytest.mark.parametrize(("overflow", "expected_truncated"), [(False, False), (True, True)]) +def test_post_knowledge_graph_relation_limit_boundary( + overflow: bool, expected_truncated: bool +) -> None: + """A look-ahead row distinguishes an exact page from an overflow page.""" + post_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1" + organization_id = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbb1" + + class Connection: + def __init__(self) -> None: + self.semantic_query = "" + self.semantic_rows = [ + { + "relation_ordinal": 1, + "subject_name": "Synthetic source", + "subject_type": "organization", + "predicate_code": "rel_voc", + "object_name": "Synthetic customer", + "object_type": "organization", + "evidence_text": "Synthetic evidence", + "relation_confidence": 0.9, + } + ] + if overflow: + self.semantic_rows.append({**self.semantic_rows[0], "relation_ordinal": 2}) + + async def fetch(self, query: str, *args: object) -> list[dict[str, object]]: + normalized = " ".join(query.lower().split()) + if "select distinct person_id" in normalized: + return [] + if "select distinct team_id" in normalized: + return [] + if "select distinct corporate_entity_id" in normalized: + return [{"corporate_entity_id": organization_id}] + if "from knowledge_graph_edge edge" in normalized: + return [] + if "from post_summary_semantic_relationship" in normalized: + self.semantic_query = normalized + return self.semantic_rows + return [] + + async def fetchval(self, query: str, *args: object) -> str | None: + return None + + conn = Connection() + result = asyncio.run(post_knowledge_graph(conn, post_id, relation_limit=1)) + + assert "limit ($2 + 1)" in conn.semantic_query + assert result["truncated"] is expected_truncated + assert len(result["edges"]) == 1 + + def test_start_node_absent_from_graph_returns_only_itself() -> None: scores = random_walk_with_restart({}, start_node="nowhere") assert scores == {"nowhere": 1.0}