fix(07.1): use savepoints for KG entity/relationship inserts and rollback on failure

- Wrap each entity/relationship insert in db.begin_nested() savepoint to prevent
  PendingRollbackError cascade when a single insert fails
- Add db.rollback() in pipeline KG Builder except block so db.commit() doesn't
  cascade-fail and kill the entire pipeline
- Guard compute_entity_degrees behind entities_written > 0 check

Author: Siddharth-Khattar

backend/app/agents/kg_builder.py

@@ -248,26 +248,29 @@ async def write_kg_from_llm_output(
             # Use first domain for backward-compat 'domain' column
             primary_domain = entity.domains[0] if entity.domains else "unknown"
 
-            kg_entity = KgEntity(
-                case_id=case_uuid,
-                name=entity.name,
-                name_normalized=normalize_entity_name(entity.name),
-                entity_type=entity.entity_type,
-                domain=primary_domain,
-                confidence=entity.confidence,
-                properties=properties_dict,
-                context=entity.description_detailed,
-                aliases=entity.aliases if entity.aliases else None,
-                description_brief=entity.description_brief,
-                description_detailed=entity.description_detailed,
-                domains=entity.domains if entity.domains else None,
-                source_finding_ids=(
-                    entity.source_finding_ids if entity.source_finding_ids else None
-                ),
-                source_execution_id=execution_id,
-            )
-            db.add(kg_entity)
-            await db.flush()
+            # Use a savepoint so a single malformed entity doesn't poison
+            # the entire session transaction (PendingRollbackError cascade).
+            async with db.begin_nested():
+                kg_entity = KgEntity(
+                    case_id=case_uuid,
+                    name=entity.name,
+                    name_normalized=normalize_entity_name(entity.name),
+                    entity_type=entity.entity_type,
+                    domain=primary_domain,
+                    confidence=entity.confidence,
+                    properties=properties_dict,
+                    context=entity.description_detailed,
+                    aliases=entity.aliases if entity.aliases else None,
+                    description_brief=entity.description_brief,
+                    description_detailed=entity.description_detailed,
+                    domains=entity.domains if entity.domains else None,
+                    source_finding_ids=(
+                        entity.source_finding_ids if entity.source_finding_ids else None
+                    ),
+                    source_execution_id=execution_id,
+                )
+                db.add(kg_entity)
+                await db.flush()
 
             llm_id_to_db_id[entity.id] = kg_entity.id
             entities_written += 1
@@ -302,23 +305,31 @@ async def write_kg_from_llm_output(
 
             corroboration = len(rel.source_finding_ids) if rel.source_finding_ids else 1
 
-            kg_rel = KgRelationship(
-                case_id=case_uuid,
-                source_entity_id=source_db_id,
-                target_entity_id=target_db_id,
-                relationship_type=rel.relationship_type,
-                label=rel.label,
-                strength=rel.strength,
-                evidence_excerpt=rel.evidence_excerpt if rel.evidence_excerpt else None,
-                source_finding_ids=(
-                    rel.source_finding_ids if rel.source_finding_ids else None
-                ),
-                temporal_context=rel.temporal_context if rel.temporal_context else None,
-                confidence=rel.confidence,
-                corroboration_count=corroboration,
-                source_execution_id=execution_id,
-            )
-            db.add(kg_rel)
+            # Savepoint per relationship for the same reason as entities
+            async with db.begin_nested():
+                kg_rel = KgRelationship(
+                    case_id=case_uuid,
+                    source_entity_id=source_db_id,
+                    target_entity_id=target_db_id,
+                    relationship_type=rel.relationship_type,
+                    label=rel.label,
+                    strength=rel.strength,
+                    evidence_excerpt=(
+                        rel.evidence_excerpt if rel.evidence_excerpt else None
+                    ),
+                    source_finding_ids=(
+                        rel.source_finding_ids if rel.source_finding_ids else None
+                    ),
+                    temporal_context=(
+                        rel.temporal_context if rel.temporal_context else None
+                    ),
+                    confidence=rel.confidence,
+                    corroboration_count=corroboration,
+                    source_execution_id=execution_id,
+                )
+                db.add(kg_rel)
+                await db.flush()
+
             relationships_written += 1
         except Exception:
             logger.warning(
@@ -331,11 +342,10 @@ async def write_kg_from_llm_output(
                 exc_info=True,
             )
 
-    await db.flush()
-
     # Compute entity degrees after all relationships are written
-    await compute_entity_degrees(case_uuid, db)
-    await db.flush()
+    if entities_written > 0:
+        await compute_entity_degrees(case_uuid, db)
+        await db.flush()
 
     logger.info(
         "Wrote curated KG data for case=%s: %d entities, %d relationships "

backend/app/services/pipeline.py

@@ -1022,6 +1022,9 @@ async def run_analysis_workflow(
                 logger.exception("KG Builder failed for case=%s: %s", case_id, exc)
                 kg_entities_created = 0
                 kg_relationships_created = 0
+                # Clear poisoned session state so db.commit() below doesn't
+                # cascade-fail with PendingRollbackError and kill the pipeline.
+                await db.rollback()
                 await emit_agent_error(
                     case_id=case_id,
                     agent_type="kg_builder",