From 8a24b678e6ac4fc2cbf98804a28595d4e8fae921 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:09:26 +0200 Subject: [PATCH 1/3] fix: TOCTOU in exact-match dedup creates duplicate memories Two concurrent process_evidence calls with identical content for the same agent both run the ilike lookup before either commits, both pass the `if not memory` check, and both insert a Memory row and enqueue separate Qdrant upserts; if the owner then forgets one copy, the duplicate stays stored and vector-indexed, so 'forgotten' content is still recallable. Found by the GLM relay (memorygate__services__api_p1__sec__0), confirmed by 3 votes and Corti. Regression test to add: Run two concurrent process_evidence calls with identical content for the same agent and assert exactly one Memory row and one enqueued Qdrant upsert are produced, so a later forget of that memory leaves nothing recallable. Co-Authored-By: Claude Opus 5.5 --- services/api/app/services/runtime_pipeline.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/services/api/app/services/runtime_pipeline.py b/services/api/app/services/runtime_pipeline.py index a532b29..33f0fe1 100644 --- a/services/api/app/services/runtime_pipeline.py +++ b/services/api/app/services/runtime_pipeline.py @@ -14,7 +14,7 @@ from app.services.processing_lease import LeaseHeartbeat, LeaseLostError, fence_job_lease from app.services.signal_filter import score_value from app.services.vector_outbox import enqueue_upsert -from sqlalchemy import select +from sqlalchemy import select, text def _link(db, source_type, source_id, target_type, target_id, relationship, confidence=1.0): @@ -111,6 +111,7 @@ def process_evidence( indexing = {"status": "ok"} if value >= 0.3: # Case-insensitive exact match: reuse the stored memory when the listener text is already there. + db.execute(text("SELECT pg_advisory_xact_lock(hashtext(:agent_id) ^ hashtext(:memory_text))"), {"agent_id": evidence.agent_id, "memory_text": content.strip()}) memory = db.execute(select(Memory).where(Memory.agent_id == evidence.agent_id, Memory.text.ilike(escape_like(content.strip()), escape="/"))).scalars().first() if not memory: memory = Memory( From 6e178e06d2bc0e78b8c21efafebc81a828dce46c Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:09:26 +0200 Subject: [PATCH 2/3] fix: episode.occurred_end overwritten with earlier or None timestamp The reuse query only requires occurred_start >= evidence.occurred_at - 30min, so evidence dated up to 30 minutes before the episode's start sets occurred_end < occurred_start; and if evidence.occurred_at is None (the fallback at line 59), occurred_end is wiped to None, destroying a previously stored end time. Found by the GLM relay (memorygate__services__api_p1__bugs__2), confirmed by 3 votes and Corti. Regression test to add: Add a regression test that processes evidence with occurred_at earlier than an open episode's occurred_end (and separately with occurred_at None) and asserts the episode's occurred_end is unchanged rather than overwritten with an earlier or None timestamp. Co-Authored-By: Claude Opus 5.5 --- services/api/app/services/runtime_pipeline.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/services/api/app/services/runtime_pipeline.py b/services/api/app/services/runtime_pipeline.py index 33f0fe1..b6cbe91 100644 --- a/services/api/app/services/runtime_pipeline.py +++ b/services/api/app/services/runtime_pipeline.py @@ -72,7 +72,8 @@ def process_evidence( db.add(episode) db.flush() else: - episode.occurred_end = evidence.occurred_at + if evidence.occurred_at is not None and (episode.occurred_end is None or evidence.occurred_at > episode.occurred_end): + episode.occurred_end = evidence.occurred_at _link(db, "evidence", evidence.id, "episode", episode.id, "grouped_into", evidence.integrity_confidence) job.stage = "analysis" From 99ca59bec8c54c07c846ece63e6b705cdaa2bf27 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:13:10 +0200 Subject: [PATCH 3/3] review: make the dedup advisory lock valid Postgres and skip it on SQLite In Postgres `^` is exponentiation, so `hashtext(a) ^ hashtext(b)` is a float and pg_advisory_xact_lock has no overload for it. Use the two-int form instead, key it on the lowercased text (the lookup is case-insensitive), and only run it on PostgreSQL: the SQLite test database has no advisory locks and the commit broke tests/test_like_literal.py. Co-Authored-By: Claude Opus 5.5 --- services/api/app/services/runtime_pipeline.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/services/api/app/services/runtime_pipeline.py b/services/api/app/services/runtime_pipeline.py index b6cbe91..5283822 100644 --- a/services/api/app/services/runtime_pipeline.py +++ b/services/api/app/services/runtime_pipeline.py @@ -112,7 +112,12 @@ def process_evidence( indexing = {"status": "ok"} if value >= 0.3: # Case-insensitive exact match: reuse the stored memory when the listener text is already there. - db.execute(text("SELECT pg_advisory_xact_lock(hashtext(:agent_id) ^ hashtext(:memory_text))"), {"agent_id": evidence.agent_id, "memory_text": content.strip()}) + if db.get_bind().dialect.name == "postgresql": + # Serialise concurrent promotions of the same text until this transaction commits. + db.execute( + text("SELECT pg_advisory_xact_lock(hashtext(:agent_id), hashtext(:memory_text))"), + {"agent_id": evidence.agent_id, "memory_text": content.strip().lower()}, + ) memory = db.execute(select(Memory).where(Memory.agent_id == evidence.agent_id, Memory.text.ilike(escape_like(content.strip()), escape="/"))).scalars().first() if not memory: memory = Memory(