From 1a5d46cd5816a40f8871b3e4a2874d1cec0576c5 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:39:22 +0200 Subject: [PATCH 1/7] fix: Evidence committed 'pending' but ProcessingJob commit can fail In _enqueue the EvidenceObject is committed with processing_state='pending', then a separate db.commit() creates the ProcessingJob. If the process dies or the second commit fails between the two commits, the evidence row exists forever in 'pending' with no ProcessingJob; the only retry path (/runtime/jobs/{id}/retry) requires a job, so the evidence is never processed. Found by the GLM relay (memorygate__services__api_p0__fail__0), confirmed by GLM votes (round 2). Regression test to add: Test that when auto_process is true, the evidence row and its ProcessingJob are committed atomically (e.g. patch the session so the single commit succeeds and verify both rows exist, or patch commit to fail and verify neither row persists). Co-Authored-By: Claude Opus 5.5 --- services/api/app/routes/runtime.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/services/api/app/routes/runtime.py b/services/api/app/routes/runtime.py index ca338cf..6e7e41f 100644 --- a/services/api/app/routes/runtime.py +++ b/services/api/app/routes/runtime.py @@ -52,13 +52,13 @@ def _enqueue(db, source: EvidenceSource, payload: IngestEventRequest, agent_id: ) db.add(row) 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} From bd8bf5cc656f7db68d5dd46a7f1c8f939ece7ca2 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:39:22 +0200 Subject: [PATCH 2/7] fix: Transcript create is non-idempotent across two commits create_transcript commits the SessionTranscript, then commits the MemoryAudit in a second transaction. If the second commit fails (or the client times out and retries), the caller gets a 500 but the transcript row is already stored; a retry inserts a duplicate transcript since there is no idempotency key. Found by the GLM relay (memorygate__services__api_p0__fail__1), confirmed by GLM votes (round 2). Regression test to add: Test that calling POST /transcripts twice with the same agent_id and session_id returns the same transcript id without inserting a second row, and that a failure during the single commit rolls back both the SessionTranscript and its MemoryAudit entry. Co-Authored-By: Claude Opus 5.5 --- services/api/app/routes/transcript.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/services/api/app/routes/transcript.py b/services/api/app/routes/transcript.py index 441e1fa..53646e3 100644 --- a/services/api/app/routes/transcript.py +++ b/services/api/app/routes/transcript.py @@ -43,9 +43,15 @@ def create_transcript(payload: TranscriptCreateRequest, header_agent_id: str = D session_end=parse_timestamp(payload.session_end, "session_end"), word_count=payload.word_count if payload.word_count is not None else len(payload.transcript.split()), ) + existing = db.execute( + select(SessionTranscript).where( + SessionTranscript.agent_id == agent_id, + SessionTranscript.session_id == payload.session_id, + ) + ).scalar_one_or_none() + if existing: + return {"status": "ok", "transcript": _summary_dict(existing)} db.add(row) - db.commit() - db.refresh(row) db.add(MemoryAudit( action="transcript_write", memory_id=None, @@ -56,6 +62,7 @@ def create_transcript(payload: TranscriptCreateRequest, header_agent_id: str = D }), )) db.commit() + db.refresh(row) return {"status": "ok", "transcript": _summary_dict(row)} finally: db.close() From 1de9d5240204f84e8afef082acd4a1296166c408 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:39:23 +0200 Subject: [PATCH 3/7] fix: Partial backup file is listed as a valid backup create_backup writes the JSON with a single write_bytes with no fsync; if the disk write or process fails partway, a truncated memorygate-*.json (whose filename already contains the full sha256 of the intended content) remains and list_backups/download_backup serve it as a restorable backup. Found by the GLM relay (memorygate__services__api_p0__fail__2), confirmed by GLM votes (round 2). Regression test to add: Simulate a crash mid-write (e.g., patch write_bytes/write to raise after partial bytes) and assert that list_backups() does not include any memorygate-*.json entry for the interrupted backup. Co-Authored-By: Claude Opus 5.5 --- services/api/app/services/backup_service.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/services/api/app/services/backup_service.py b/services/api/app/services/backup_service.py index 5a8c9bb..90f7a0d 100644 --- a/services/api/app/services/backup_service.py +++ b/services/api/app/services/backup_service.py @@ -55,7 +55,13 @@ 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() + import os + os.fsync(handle.fileno()) + temp_path.replace(destination / filename) return {"filename": filename, "size": len(encoded), "sha256": digest, "created_at": payload["created_at"]} From 00e9576e311cf053f6d210b8a6921e4b76cc3ec1 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:39:23 +0200 Subject: [PATCH 4/7] fix: Observation/entity vector delete fails on never-created collection An observation or entity is forgotten before any successful upsert (its Qdrant collection was never created because every earlier write failed); delete_observation_embedding/delete_entity_embedding call client.delete on a nonexistent collection, which raises, unlike delete_memory_embedding which guards with a get_collections check. The outbox row is retried with backoff forever and the forget receipt stays 'degraded'/'pending' permanently. Found by the GLM relay (memorygate__services__api_p1__fail__0), confirmed by GLM votes (round 2). Regression test to add: Test that delete_observation_embedding and delete_entity_embedding return without error when their Qdrant collections do not exist yet, mirroring the existing guard on delete_memory_embedding. Co-Authored-By: Claude Opus 5.5 --- services/api/app/services/qdrant_store.py | 6 ++++++ 1 file changed, 6 insertions(+) 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]) From 9bcdff13cb736077cd2e1505a1a46f2ed11ffcc5 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:39:24 +0200 Subject: [PATCH 5/7] fix: Stage-wise commits mean a retried job duplicates analysis rows process_evidence commits after each stage. If a failure (e.g., vector/outbox or DB error) occurs at the memory or final stage, the episode and AnalysisObject rows from earlier stages are already committed, and the failed job is requeued (requeue/lease expiry). On retry, the analysis stage runs again and inserts a second AnalysisObject (and possibly a second local_ai_proposal) for the same evidence, with no dedup check like the episode and memory stages have. Found by the GLM relay (memorygate__services__api_p1__fail__2), confirmed by GLM votes (round 2). Regression test to add: A regression test should run process_evidence twice against the same evidence (simulating a retry after a failure at the memory/final stage) and assert only one deterministic_signal_extraction AnalysisObject and one local_ai_proposal AnalysisObject row exist for that evidence. Co-Authored-By: Claude Opus 5.5 --- services/api/app/services/runtime_pipeline.py | 21 ++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/services/api/app/services/runtime_pipeline.py b/services/api/app/services/runtime_pipeline.py index 5283822..9b9d4e2 100644 --- a/services/api/app/services/runtime_pipeline.py +++ b/services/api/app/services/runtime_pipeline.py @@ -80,9 +80,15 @@ 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], + 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), @@ -94,8 +100,12 @@ def process_evidence( 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 From 1b9a030fc73fc66be0f5271ceb9e3bf6a2a01ca8 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:40:20 +0200 Subject: [PATCH 6/7] review: drop the transcript replay match, tidy two fixes Reverted bd8bf5c: session_id defaults to "" and is not unique, so matching on (agent_id, session_id) would fold every session-less transcript into the first one, and scalar_one_or_none raises once two rows share a session. Kept with tidying: the analysis constructor sits inside its new if-block; os is imported at the top of backup_service. Co-Authored-By: Claude Opus 5.5 --- services/api/app/routes/transcript.py | 11 ++--------- services/api/app/services/backup_service.py | 2 +- services/api/app/services/runtime_pipeline.py | 12 ++++++------ 3 files changed, 9 insertions(+), 16 deletions(-) diff --git a/services/api/app/routes/transcript.py b/services/api/app/routes/transcript.py index 53646e3..441e1fa 100644 --- a/services/api/app/routes/transcript.py +++ b/services/api/app/routes/transcript.py @@ -43,15 +43,9 @@ def create_transcript(payload: TranscriptCreateRequest, header_agent_id: str = D session_end=parse_timestamp(payload.session_end, "session_end"), word_count=payload.word_count if payload.word_count is not None else len(payload.transcript.split()), ) - existing = db.execute( - select(SessionTranscript).where( - SessionTranscript.agent_id == agent_id, - SessionTranscript.session_id == payload.session_id, - ) - ).scalar_one_or_none() - if existing: - return {"status": "ok", "transcript": _summary_dict(existing)} db.add(row) + db.commit() + db.refresh(row) db.add(MemoryAudit( action="transcript_write", memory_id=None, @@ -62,7 +56,6 @@ def create_transcript(payload: TranscriptCreateRequest, header_agent_id: str = D }), )) db.commit() - db.refresh(row) return {"status": "ok", "transcript": _summary_dict(row)} finally: db.close() diff --git a/services/api/app/services/backup_service.py b/services/api/app/services/backup_service.py index 90f7a0d..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 @@ -59,7 +60,6 @@ def create_backup(db) -> dict: with open(temp_path, "wb") as handle: handle.write(encoded) handle.flush() - import os 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/runtime_pipeline.py b/services/api/app/services/runtime_pipeline.py index 9b9d4e2..e0f9250 100644 --- a/services/api/app/services/runtime_pipeline.py +++ b/services/api/app/services/runtime_pipeline.py @@ -89,12 +89,12 @@ def process_evidence( 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() + 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" From a6431d42300cdba209b0ebbbaf849a2faece8794 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:14:46 +0200 Subject: [PATCH 7/7] review: flush the evidence row before its job takes the id The single-commit _enqueue built the ProcessingJob with evidence_id=row.id before the row was flushed, so the id was empty and /runtime/ingest returned 500 (10 tests). Co-Authored-By: Claude Opus 5.5 --- services/api/app/routes/runtime.py | 1 + 1 file changed, 1 insertion(+) diff --git a/services/api/app/routes/runtime.py b/services/api/app/routes/runtime.py index 6e7e41f..2ee4c6c 100644 --- a/services/api/app/routes/runtime.py +++ b/services/api/app/routes/runtime.py @@ -51,6 +51,7 @@ 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 job = None if payload.auto_process: