Publish sync conflict vectors safely

Author: Coding-Dev-Tools

engraphis/core/sync.py

@@ -1350,9 +1350,11 @@ def _apply_one(self, d: dict, rec, report: dict, accepted: dict, known: dict,
         if existing is not None and provenance_is_approved(existing.provenance):
             content_changed = rec.content != existing.content
             self._rehome_external_record(rec, src_device=src_device)
-            self._preserve_hlc_conflict(
+            conflict_action = self._preserve_hlc_conflict(
                 existing, rec, report=report, known=known, dry_run=dry_run,
             )
+            if conflict_action is not None:
+                pending_index_actions.append(conflict_action)
             if not dry_run and content_changed:
                 self.store.audit(
                     "sync:%s" % _clamp_str(src_device or "peer", 128),
@@ -1412,9 +1414,11 @@ def _apply_one(self, d: dict, rec, report: dict, accepted: dict, known: dict,
             rec.valid_to_recorded_at = now_ts()
             rec.embedding = None
         if existing is not None:
-            self._preserve_hlc_conflict(
+            conflict_action = self._preserve_hlc_conflict(
                 existing, rec, report=report, known=known, dry_run=dry_run,
             )
+            if conflict_action is not None:
+                pending_index_actions.append(conflict_action)
         if existing is None:
             if not dry_run:
                 index_action = self._write(rec, commit=False)
@@ -1743,11 +1747,11 @@ def _preserve_hlc_conflict(
         report: dict,
         known: dict,
         dry_run: bool,
-    ) -> None:
+    ) -> Optional[_VectorIndexAction]:
         """Keep the losing concurrent edit as one deterministic untrusted successor."""
         conflict = self._hlc_conflict(existing, incoming)
         if conflict is None:
-            return
+            return None
         physical, logical, existing_hash, incoming_hash = conflict
         winner = (
             existing
@@ -1844,9 +1848,10 @@ def _preserve_hlc_conflict(
                 or not _same_sync_payload(already_preserved, preserved)
             ):
                 raise SyncError("sync conflict identity collision")
-            return
+            return None
+        index_action = None
         if not dry_run:
-            self._write(preserved, commit=False)
+            index_action = self._write(preserved, commit=False)
             self.store.audit(
                 "sync",
                 "sync_conflict_preserved",
@@ -1860,6 +1865,7 @@ def _preserve_hlc_conflict(
             )
         known[conflict_id] = preserved
         report["conflicts_preserved"] += 1
+        return index_action
 
     def _write(
         self, rec: MemoryRecord, *, commit: bool = True,

engraphis/service.py

@@ -3162,6 +3162,8 @@ def _apply_postgres_schema_snapshot(
                 claim_kind="catalog_snapshot_chunk",
                 resolve_conflicts=True,
             ))
+        if not stored_rows:
+            return {"workspace": workspace, "stored": 0, "entities": 0, "relations": 0}
         stored = stored_rows[0]
         wid, rid = self._require_scope(workspace, repo)
         actual_ids: dict[str, str] = {}

tests/test_postgres_schema.py

@@ -6,7 +6,8 @@
 
 from engraphis.backends import postgres_schema
 from engraphis.core.interfaces import SchemaSnapshot, SearchFilter
-from engraphis.service import MemoryService
+import engraphis.service as service_module
+from engraphis.service import MAX_CONTENT_CHARS, MemoryService
 
 
 class _Cursor:
@@ -360,6 +361,32 @@ def inspect(self, supplied, *, schemas=None):
     assert "secret" not in serialized
 
 
+def test_empty_postgres_chunk_result_returns_without_indexing(monkeypatch):
+    snapshot = SchemaSnapshot(
+        title="PostgreSQL schema: empty",
+        text="x" * (MAX_CONTENT_CHARS + 1),
+        metadata={"database": "empty", "source_digest": "digest"},
+    )
+
+    class _Introspector:
+        def inspect(self, supplied, *, schemas=None):
+            return snapshot
+
+    class _EmptyExtractor:
+        def extract(self, _text):
+            return []
+
+    monkeypatch.setattr(
+        postgres_schema, "get_postgres_introspector", lambda: _Introspector()
+    )
+    monkeypatch.setattr(service_module, "ChunkingExtractor", _EmptyExtractor)
+    service = MemoryService.create(":memory:")
+
+    assert service.import_postgres_schema(
+        "postgresql://local/empty", workspace="acme"
+    ) == {"workspace": "acme", "stored": 0, "entities": 0, "relations": 0}
+
+
 def test_large_postgres_snapshot_keeps_every_chunk_distinct(monkeypatch):
     snapshot = SchemaSnapshot(
         title="PostgreSQL schema: large",

tests/test_sync.py

@@ -2672,6 +2672,52 @@ def fail_late(actor, action, target, detail="", *, commit=True):
     assert publications == []
 
 
+def test_hlc_conflict_variant_publishes_external_vector():
+    engine = MemoryEngine.create(":memory:", vector_backend="numpy")
+    publications = []
+
+    class RecordingExternalIndex:
+        def upsert(self, ids, _vecs, meta=None, *, commit=True):
+            publications.extend(ids)
+
+        def delete(self, ids, *, commit=True):
+            del ids, commit
+
+    lower_node = f"dev_{'0' * 26}"
+    higher_node = f"dev_{'1' * 26}"
+
+    def bundle(content, node):
+        return {
+            "format": SYNC_FORMAT,
+            "version": 2,
+            "device_id": node,
+            "workspace_name": "w",
+            "repos": {},
+            "memories": [{
+                "id": "same-hlc-id",
+                "content": content,
+                "ingested_at": 42.0,
+                "valid_from": 42.0,
+                "modified_hlc": format_modified_hlc(42, 1, node),
+            }],
+            "mem_links": [],
+        }
+
+    syncer = SyncEngine(
+        engine.store,
+        embedder=engine.embedder,
+        vector_index=RecordingExternalIndex(),
+    )
+    syncer.apply_bundle(bundle("lower-node edit", lower_node), into_workspace="w")
+    publications.clear()
+    syncer.apply_bundle(bundle("higher-node edit", higher_node), into_workspace="w")
+
+    conflict_id = engine.store.conn.execute(
+        "SELECT id FROM memories WHERE id <> 'same-hlc-id'"
+    ).fetchone()["id"]
+    assert conflict_id in publications
+
+
 def test_sync_configured_embedder_failure_aborts_before_memory_write(caplog):
     engine = MemoryEngine.create(":memory:", vector_backend="numpy")