From bd3303f7fa5d0699081d50467a5b5db1a4ed03a6 Mon Sep 17 00:00:00 2001
From: rock <locartrock@gmail.com>
Date: Thu, 10 Sep 2026 10:17:42 +0900
Subject: [PATCH] fix: entity upsert dedup on (project_id, name) +
 observability

Entity save now uses ON CONFLICT (project_id, name) DO UPDATE:
- Merges description (keep non-empty)
- Keeps highest confidence
- Increments source_count
- Updates t_updated timestamp

Prevents duplicate entities across ingests (was 26 rows, now 11).
Unique index added to production DB.

Compaction (T3.1 exact dedup + T3.2 semantic) already wired at
POST /memory/compact endpoint. Cache alignment + chunk optimizer
wired through full_pipeline.rs + query_orchestrator.rs.

781 tests pass.
---
 crates/mem-cli/src/ingest_worker.rs | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/crates/mem-cli/src/ingest_worker.rs b/crates/mem-cli/src/ingest_worker.rs
index bc4a5bb..8d8e7e8 100644
--- a/crates/mem-cli/src/ingest_worker.rs
+++ b/crates/mem-cli/src/ingest_worker.rs
@@ -10,6 +10,7 @@ use uuid::Uuid;
 use std::sync::Arc;
 use pgvector::Vector;
 
+
 /// Ingest worker — processes queued records through entity/fact extraction pipeline
 pub struct IngestWorker {
     pool: PgPool,
@@ -135,9 +136,15 @@ impl IngestWorker {
             .await?;
 
         tracing::info!(
-            "Ingest completed: {} (entities={}, edges={}, reviews={})",
-            ingest_id, total_entities, total_edges, total_reviews
+            target: "observability",
+            event = "ingest_complete",
+            ingest_id = ingest_id,
+            entities = total_entities,
+            edges = total_edges,
+            reviews = total_reviews,
+            "Ingest completed"
         );
+
         Ok(())
     }
 
@@ -187,7 +194,12 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) ->
     sqlx::query(
         "INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence)
          VALUES ($1, $2, $3, $4, $5, $6::TIMESTAMPTZ, $7::TIMESTAMPTZ, $8)
-         ON CONFLICT (id) DO NOTHING"
+         ON CONFLICT (project_id, name) DO UPDATE SET
+           entity_type = EXCLUDED.entity_type,
+           description = COALESCE(NULLIF(EXCLUDED.description, ''), memory_entity.description),
+           t_updated = NOW(),
+           confidence = GREATEST(memory_entity.confidence, EXCLUDED.confidence),
+           source_count = memory_entity.source_count + 1"
     )
     .bind(&entity.id)
     .bind(&entity.project_id)