diff --git a/services/api/app/routes/runtime.py b/services/api/app/routes/runtime.py index ca338cf..2ee4c6c 100644 --- a/services/api/app/routes/runtime.py +++ b/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} diff --git a/services/api/app/services/backup_service.py b/services/api/app/services/backup_service.py index 5a8c9bb..d5e2d6e 100644 --- a/services/api/app/services/backup_service.py +++ b/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"]} diff --git a/services/api/app/services/qdrant_store.py b/services/api/app/services/qdrant_store.py index 8ce15b4..b9c5a97 100644 --- a/services/api/app/services/qdrant_store.py +++ b/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]) diff --git a/services/api/app/services/runtime_pipeline.py b/services/api/app/services/runtime_pipeline.py index 5283822..e0f9250 100644 --- a/services/api/app/services/runtime_pipeline.py +++ b/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