fix: relay round 3, batch 2 (one-commit ingest, atomic backups, retry dedupe)

services/api/app/routes/runtime.py

@@ -51,14 +51,15 @@ def _enqueue(db, source: EvidenceSource, payload: IngestEventRequest, agent_id:
         processing_state="pending", occurred_at=parse_timestamp(payload.occurred_at, "occurred_at") or datetime.now(UTC),
     )
     db.add(row)
+    db.flush()  # the job below needs the evidence id; both commit together
     source.last_ingested_at = row.occurred_at
-    db.commit()
-    db.refresh(row)
     job = None
     if payload.auto_process:
         job = ProcessingJob(agent_id=agent_id, evidence_id=row.id, status="pending", stage="queued")
         db.add(job)
-        db.commit()
+    db.commit()
+    db.refresh(row)
+    if job:
         db.refresh(job)
     return {"status": "queued" if job else "accepted", "evidence_id": row.id, "job_id": job.id if job else None}
 

services/api/app/services/backup_service.py

@@ -1,6 +1,7 @@
 """Safe, portable logical backups for MemoryGate's user memory data."""
 import hashlib
 import json
+import os
 from datetime import date, datetime
 from pathlib import Path
 
@@ -55,7 +56,12 @@ def create_backup(db) -> dict:
     encoded = json.dumps(payload, ensure_ascii=True, indent=2, sort_keys=True).encode("utf-8")
     digest = hashlib.sha256(encoded).hexdigest()
     filename = f"memorygate-{datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')}-{digest[:12]}.json"
-    (destination / filename).write_bytes(encoded)
+    temp_path = destination / (filename + ".tmp")
+    with open(temp_path, "wb") as handle:
+        handle.write(encoded)
+        handle.flush()
+        os.fsync(handle.fileno())
+    temp_path.replace(destination / filename)
     return {"filename": filename, "size": len(encoded), "sha256": digest, "created_at": payload["created_at"]}
 
 

services/api/app/services/qdrant_store.py

@@ -214,6 +214,9 @@ def upsert_observation_embedding(observation_id: str, text: str, payload: dict |
 
 def delete_observation_embedding(observation_id: str) -> None:
     client = get_qdrant_client()
+    # Forgetting before the first successful vector write is already satisfied.
+    if OBSERVATION_COLLECTION not in {collection.name for collection in client.get_collections().collections}:
+        return
     client.delete(collection_name=OBSERVATION_COLLECTION, points_selector=[observation_id])
 
 
@@ -259,6 +262,9 @@ def upsert_entity_embedding(entity_id: str, text: str, payload: dict | None = No
 
 def delete_entity_embedding(entity_id: str) -> None:
     client = get_qdrant_client()
+    # Forgetting before the first successful vector write is already satisfied.
+    if ENTITY_COLLECTION not in {collection.name for collection in client.get_collections().collections}:
+        return
     client.delete(collection_name=ENTITY_COLLECTION, points_selector=[entity_id])
 
 

services/api/app/services/runtime_pipeline.py

@@ -80,22 +80,32 @@ def process_evidence(
         _commit_job(db, job, lease)
         value = score_value(content)
         classification = classify_memory(content, "automatic_listener")
-        analysis = AnalysisObject(
-            agent_id=evidence.agent_id, analysis_type="deterministic_signal_extraction",
-            evidence_ids_json=json.dumps([evidence.id]), input_summary=content[:500],
-            output_summary=f"Value {value:.2f}; classified as {classification['memory_type']} ({classification['confidence']}).",
-            steps_json=json.dumps(["normalize", "value_score", "memory_classification"]),
-            confidence=min(evidence.integrity_confidence, 0.85 if value >= 0.3 else 0.6),
-        )
-        db.add(analysis)
-        db.flush()
+        analysis = db.execute(select(AnalysisObject).where(
+            AnalysisObject.agent_id == evidence.agent_id,
+            AnalysisObject.analysis_type == "deterministic_signal_extraction",
+            AnalysisObject.evidence_ids_json == json.dumps([evidence.id]),
+        )).scalars().first()
+        if not analysis:
+            analysis = AnalysisObject(
+                agent_id=evidence.agent_id, analysis_type="deterministic_signal_extraction",
+                evidence_ids_json=json.dumps([evidence.id]), input_summary=content[:500],
+                output_summary=f"Value {value:.2f}; classified as {classification['memory_type']} ({classification['confidence']}).",
+                steps_json=json.dumps(["normalize", "value_score", "memory_classification"]),
+                confidence=min(evidence.integrity_confidence, 0.85 if value >= 0.3 else 0.6),
+            )
+            db.add(analysis)
+            db.flush()
         _link(db, "episode", episode.id, "analysis", analysis.id, "analyzed_into", analysis.confidence)
 
         job.stage = "local_ai"
         _commit_job(db, job, lease)
         ai_result = analyze_evidence(content, evidence.source_type)
-        ai_analysis = None
-        if ai_result:
+        ai_analysis = db.execute(select(AnalysisObject).where(
+            AnalysisObject.agent_id == evidence.agent_id,
+            AnalysisObject.analysis_type == "local_ai_proposal",
+            AnalysisObject.evidence_ids_json == json.dumps([evidence.id]),
+        )).scalars().first()
+        if not ai_analysis and ai_result:
             ai_analysis = AnalysisObject(
                 agent_id=evidence.agent_id, analysis_type="local_ai_proposal",
                 evidence_ids_json=json.dumps([evidence.id]), input_summary=content[:500],
@@ -106,6 +116,7 @@ def process_evidence(
             )
             db.add(ai_analysis)
             db.flush()
+        if ai_analysis:
             _link(db, "episode", episode.id, "analysis", ai_analysis.id, "proposed_by", ai_analysis.confidence)
 
         memory = None