From 6f918a177a2c30c4bbd732cad3f71f762f1da632 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Mon, 31 Aug 2026 17:25:19 +0300 Subject: [PATCH 01/40] fix(profile-b): expand tenacity retry deadline to support bounded provider timeouts --- mesa_memory/adapter/live.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mesa_memory/adapter/live.py b/mesa_memory/adapter/live.py index f71300f..8fd0b4e 100644 --- a/mesa_memory/adapter/live.py +++ b/mesa_memory/adapter/live.py @@ -26,7 +26,7 @@ _OPENAI_CONNECTION_ERRORS = (openai.APIConnectionError,) if openai is not None else () _OPENAI_NOT_FOUND_ERRORS = (openai.NotFoundError,) if openai is not None else () _RETRYABLE_OPENAI_ERRORS = _OPENAI_RATE_LIMIT_ERRORS + _OPENAI_CONNECTION_ERRORS -_OPENAI_RETRY_STOP = stop_after_attempt(3) | stop_after_delay(45) +_OPENAI_RETRY_STOP = stop_after_attempt(3) | stop_after_delay(300) class OpenAICompatibleAdapter(BaseUniversalLLMAdapter): From 48aea9ffb1b35c1bbfc8465a0095b3fdccec5dba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 00:52:08 +0300 Subject: [PATCH 02/40] fix(live-adapter): expand retry stop delay to 1800s for deep-reasoning extraction --- mesa_memory/adapter/live.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mesa_memory/adapter/live.py b/mesa_memory/adapter/live.py index 8fd0b4e..642ae5f 100644 --- a/mesa_memory/adapter/live.py +++ b/mesa_memory/adapter/live.py @@ -26,7 +26,7 @@ _OPENAI_CONNECTION_ERRORS = (openai.APIConnectionError,) if openai is not None else () _OPENAI_NOT_FOUND_ERRORS = (openai.NotFoundError,) if openai is not None else () _RETRYABLE_OPENAI_ERRORS = _OPENAI_RATE_LIMIT_ERRORS + _OPENAI_CONNECTION_ERRORS -_OPENAI_RETRY_STOP = stop_after_attempt(3) | stop_after_delay(300) +_OPENAI_RETRY_STOP = stop_after_attempt(3) | stop_after_delay(1800) class OpenAICompatibleAdapter(BaseUniversalLLMAdapter): From 135dbf2982a4ae2fd8e1e692c59ef13efb824e3f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 00:55:48 +0300 Subject: [PATCH 03/40] fix(dao): set claim_raw_log default lease to 1800s for deep-reasoning extraction --- mesa_storage/dao.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 4d32a1a..425acaa 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -8344,7 +8344,7 @@ async def update_raw_log_status( ) async def claim_raw_log( - self, agent_id: str, log_id: int, *, worker_id: str, lease_seconds: int = 300 + self, agent_id: str, log_id: int, *, worker_id: str, lease_seconds: int = 1800 ) -> dict[str, Any] | None: """Atomically claim a deferred or expired cold-path job.""" _assert_valid_agent_id(agent_id) From 22a7412532868992cb1644cc9a0cd83f612a2cf1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 01:20:35 +0300 Subject: [PATCH 04/40] fix(server): refresh config from environment at startup in _runtime_lifespan --- mesa_memory/api/server.py | 1 + 1 file changed, 1 insertion(+) diff --git a/mesa_memory/api/server.py b/mesa_memory/api/server.py index c237e33..6eb87ff 100644 --- a/mesa_memory/api/server.py +++ b/mesa_memory/api/server.py @@ -298,6 +298,7 @@ async def _run_combined_durable_consumer( @asynccontextmanager async def _runtime_lifespan(app: FastAPI, runtime: RuntimeProfileConfig): + refresh_config_from_environment() state.is_ready = False state.obs_layer = ObservabilityLayer() From f3b99fc63882712332cb9395dba1e8a7d7c12353 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 01:28:05 +0300 Subject: [PATCH 05/40] feat(config): raise default queue admission limits to 50000 records to support large legal corpus release delivery --- mesa_memory/config.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/mesa_memory/config.py b/mesa_memory/config.py index 63d497f..f327833 100644 --- a/mesa_memory/config.py +++ b/mesa_memory/config.py @@ -222,14 +222,14 @@ def load_explicit_dotenv(runtime: RuntimeProfileConfig) -> None: class QueueAdmissionPolicy(BaseModel): """Fail-closed, server-side admission limits for durable cold-path work.""" - queue_max_pending_records: int = 10_000 - queue_max_pending_bytes: int = 536_870_912 - queue_max_pending_records_per_tenant: int = 2_000 - queue_max_pending_bytes_per_tenant: int = 134_217_728 + queue_max_pending_records: int = 50_000 + queue_max_pending_bytes: int = 2_147_483_648 + queue_max_pending_records_per_tenant: int = 50_000 + queue_max_pending_bytes_per_tenant: int = 2_147_483_648 queue_max_in_flight_records: int = 32 queue_max_in_flight_records_per_tenant: int = 8 - queue_max_retry_pending_records: int = 2_000 - queue_max_retry_pending_records_per_tenant: int = 500 + queue_max_retry_pending_records: int = 10_000 + queue_max_retry_pending_records_per_tenant: int = 10_000 queue_max_single_record_bytes: int = 8_388_608 queue_retry_after_seconds: int = 5 @@ -661,16 +661,16 @@ def vector_worker_limit(self) -> int: # WAVE-004B: bounded durable queue admission. Each operator value is # independently environment-configurable and exposed as one typed policy. queue_max_pending_records: int = Field( - 10_000, validation_alias="MESA_QUEUE_MAX_PENDING_RECORDS" + 50_000, validation_alias="MESA_QUEUE_MAX_PENDING_RECORDS" ) queue_max_pending_bytes: int = Field( - 536_870_912, validation_alias="MESA_QUEUE_MAX_PENDING_BYTES" + 2_147_483_648, validation_alias="MESA_QUEUE_MAX_PENDING_BYTES" ) queue_max_pending_records_per_tenant: int = Field( - 2_000, validation_alias="MESA_QUEUE_MAX_PENDING_RECORDS_PER_TENANT" + 50_000, validation_alias="MESA_QUEUE_MAX_PENDING_RECORDS_PER_TENANT" ) queue_max_pending_bytes_per_tenant: int = Field( - 134_217_728, validation_alias="MESA_QUEUE_MAX_PENDING_BYTES_PER_TENANT" + 2_147_483_648, validation_alias="MESA_QUEUE_MAX_PENDING_BYTES_PER_TENANT" ) queue_max_in_flight_records: int = Field( 32, validation_alias="MESA_QUEUE_MAX_IN_FLIGHT_RECORDS" @@ -679,10 +679,10 @@ def vector_worker_limit(self) -> int: 8, validation_alias="MESA_QUEUE_MAX_IN_FLIGHT_RECORDS_PER_TENANT" ) queue_max_retry_pending_records: int = Field( - 2_000, validation_alias="MESA_QUEUE_MAX_RETRY_PENDING_RECORDS" + 10_000, validation_alias="MESA_QUEUE_MAX_RETRY_PENDING_RECORDS" ) queue_max_retry_pending_records_per_tenant: int = Field( - 500, validation_alias="MESA_QUEUE_MAX_RETRY_PENDING_RECORDS_PER_TENANT" + 10_000, validation_alias="MESA_QUEUE_MAX_RETRY_PENDING_RECORDS_PER_TENANT" ) queue_max_single_record_bytes: int = Field( 8_388_608, validation_alias="MESA_QUEUE_MAX_SINGLE_RECORD_BYTES" From 98a62031fa63be7cd937d14fc02f0fa6df35b28f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 01:34:39 +0300 Subject: [PATCH 06/40] fix(dao): add exception logging when admit_v4_memory encounters SQLite/OS error --- mesa_storage/dao.py | 1 + 1 file changed, 1 insertion(+) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 425acaa..58ad014 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -2100,6 +2100,7 @@ async def admit_v4_memory( ) await db.commit() except (aiosqlite.Error, OSError) as exc: + logger.exception("admit_v4_memory failed with SQLite/OS error: %s", exc) raise QueueUnavailableError("durable admission is unavailable") from exc return {"outcome": "ADMITTED", "response": response} From d03104131b1ae28d86314607868fba0dc351dd25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 01:37:30 +0300 Subject: [PATCH 07/40] debug(dao): print traceback on admit_v4_memory failure --- mesa_storage/dao.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 58ad014..bd56e5a 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -2100,6 +2100,8 @@ async def admit_v4_memory( ) await db.commit() except (aiosqlite.Error, OSError) as exc: + import traceback + traceback.print_exc() logger.exception("admit_v4_memory failed with SQLite/OS error: %s", exc) raise QueueUnavailableError("durable admission is unavailable") from exc return {"outcome": "ADMITTED", "response": response} From 90eb971cc61336588c28413dc4eaa6a3876f942a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 01:55:15 +0300 Subject: [PATCH 08/40] perf(runtime): optimize Mode 0 validation path and scale combined consumer batch limits --- mesa_memory/api/server.py | 8 ++++---- mesa_workers/ingestion_worker.py | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/mesa_memory/api/server.py b/mesa_memory/api/server.py index 6eb87ff..7a54708 100644 --- a/mesa_memory/api/server.py +++ b/mesa_memory/api/server.py @@ -220,7 +220,7 @@ async def _consume_combined_durable_work_once( ) -> dict[str, int]: """Consume bounded durable work in the single storage-owner runtime.""" worker_id = "combined-runtime" - claimed = await dao.claim_dispatch_queue(worker_id=worker_id, limit=1) + claimed = await dao.claim_dispatch_queue(worker_id=worker_id, limit=20) for dispatch in claimed: log_id = int(dispatch["payload_reference"]) agent_id = str(dispatch["agent_id"]) @@ -259,7 +259,7 @@ async def _consume_combined_durable_work_once( outcome=status[:120], side_effect_verified=status.split(":", 1)[0] in {"processed", "rejected"}, ) - finalizations = await dao.list_pending_session_finalizations(limit=1) + finalizations = await dao.list_pending_session_finalizations(limit=5) for finalization in finalizations: await process_session_finalization( str(finalization["agent_id"]), @@ -270,8 +270,8 @@ async def _consume_combined_durable_work_once( projections = {"completed": 0} cleanup = {"completed": 0} if type(dao) is MemoryDAO: - projections = await process_projection_outbox_once(dao, worker_id=worker_id) - cleanup = await process_artifact_cleanup_once(dao, worker_id=worker_id) + projections = await process_projection_outbox_once(dao, worker_id=worker_id, limit=50) + cleanup = await process_artifact_cleanup_once(dao, worker_id=worker_id, limit=50) return { "dispatches": len(claimed), "finalizations": len(finalizations), diff --git a/mesa_workers/ingestion_worker.py b/mesa_workers/ingestion_worker.py index 553c383..22e9f2f 100644 --- a/mesa_workers/ingestion_worker.py +++ b/mesa_workers/ingestion_worker.py @@ -413,7 +413,7 @@ async def _transition( dao, "record_mutation", candidate_record, raw_log_id=log_id ) - if consolidation_loop is not None: + if consolidation_loop is not None and effective_validation_mode > 0: async with _tier3_semaphore: outcome = await consolidation_loop.run_batch([candidate_record]) else: From f87dfd618cd7750c3e60485d2d329e66ce972282 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 01:56:00 +0300 Subject: [PATCH 09/40] perf(server): parallelize combined dispatch consumption and increase projection batch to 100 --- mesa_memory/api/server.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/mesa_memory/api/server.py b/mesa_memory/api/server.py index 7a54708..d7adb45 100644 --- a/mesa_memory/api/server.py +++ b/mesa_memory/api/server.py @@ -220,8 +220,9 @@ async def _consume_combined_durable_work_once( ) -> dict[str, int]: """Consume bounded durable work in the single storage-owner runtime.""" worker_id = "combined-runtime" - claimed = await dao.claim_dispatch_queue(worker_id=worker_id, limit=20) - for dispatch in claimed: + claimed = await dao.claim_dispatch_queue(worker_id=worker_id, limit=50) + + async def _handle_one_dispatch(dispatch: dict[str, Any]) -> None: log_id = int(dispatch["payload_reference"]) agent_id = str(dispatch["agent_id"]) processing = asyncio.create_task( @@ -259,7 +260,11 @@ async def _consume_combined_durable_work_once( outcome=status[:120], side_effect_verified=status.split(":", 1)[0] in {"processed", "rejected"}, ) - finalizations = await dao.list_pending_session_finalizations(limit=5) + + if claimed: + await asyncio.gather(*(_handle_one_dispatch(d) for d in claimed), return_exceptions=True) + + finalizations = await dao.list_pending_session_finalizations(limit=10) for finalization in finalizations: await process_session_finalization( str(finalization["agent_id"]), @@ -270,8 +275,8 @@ async def _consume_combined_durable_work_once( projections = {"completed": 0} cleanup = {"completed": 0} if type(dao) is MemoryDAO: - projections = await process_projection_outbox_once(dao, worker_id=worker_id, limit=50) - cleanup = await process_artifact_cleanup_once(dao, worker_id=worker_id, limit=50) + projections = await process_projection_outbox_once(dao, worker_id=worker_id, limit=100) + cleanup = await process_artifact_cleanup_once(dao, worker_id=worker_id, limit=100) return { "dispatches": len(claimed), "finalizations": len(finalizations), From 852aee11f716cef6568d69d56f76f5144aa6c214 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 02:00:17 +0300 Subject: [PATCH 10/40] fix(server): wrap combined durable consumer in try-except loop with 50ms polling --- mesa_memory/api/server.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/mesa_memory/api/server.py b/mesa_memory/api/server.py index d7adb45..2bfc52e 100644 --- a/mesa_memory/api/server.py +++ b/mesa_memory/api/server.py @@ -293,12 +293,17 @@ async def _run_combined_durable_consumer( ) -> None: """Poll the durable journal without introducing a second storage writer.""" while True: - await _consume_combined_durable_work_once( - dao, - consolidation_loop=consolidation_loop, - model_processing_enabled=model_processing_enabled, - ) - await asyncio.sleep(0.25) + try: + await _consume_combined_durable_work_once( + dao, + consolidation_loop=consolidation_loop, + model_processing_enabled=model_processing_enabled, + ) + except asyncio.CancelledError: + break + except Exception as exc: + logger.exception("Combined durable consumer iteration failed: %s", exc) + await asyncio.sleep(0.05) @asynccontextmanager From 7b52a320e0b096dc1be9d39b38784889bdb45464 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 02:02:52 +0300 Subject: [PATCH 11/40] perf(worker): bypass legacy ECOD gate when require_tier3_validation is True --- mesa_workers/ingestion_worker.py | 37 +++++++++++--------------------- 1 file changed, 13 insertions(+), 24 deletions(-) diff --git a/mesa_workers/ingestion_worker.py b/mesa_workers/ingestion_worker.py index 22e9f2f..6e8e1b8 100644 --- a/mesa_workers/ingestion_worker.py +++ b/mesa_workers/ingestion_worker.py @@ -316,30 +316,19 @@ async def _transition( # ============================================================== # 3. TIER-1: ECOD ANOMALY DETECTION (Novelty Gate) # ============================================================== - ecod_passed = await _run_ecod_gate(dao, payload_agent_id, content) - - # A low novelty score is useful as a cheap duplicate heuristic for - # the legacy projection path. It is not a reliable rejection for - # the full-cognitive path: corrections and contradictions are - # deliberately similar to the memory they update. Let the selected - # validation policy make the final STORE/DISCARD decision there. - if not ecod_passed and not require_tier3_validation: - await _transition( - "rejected", - error_reason="ecod_novelty_below_threshold", - target_agent_id=payload_agent_id, - ) - logger.info( - "COLD_PATH_REJECTED | log_id=%d reason=ecod_novelty_gate", - log_id, - ) - return - if not ecod_passed: - logger.info( - "COLD_PATH_ECOD_DEFERRED_TO_TIER3 | log_id=%d agent_id=%s", - log_id, - payload_agent_id, - ) + if not require_tier3_validation: + ecod_passed = await _run_ecod_gate(dao, payload_agent_id, content) + if not ecod_passed: + await _transition( + "rejected", + error_reason="ecod_novelty_below_threshold", + target_agent_id=payload_agent_id, + ) + logger.info( + "COLD_PATH_REJECTED | log_id=%d reason=ecod_novelty_gate", + log_id, + ) + return _write_cold_path_trace(f"BEFORE REBEL {log_id}") # ============================================================== From 2f3b913e664906b47259f27b5e8ac3790d9f0706 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 02:05:38 +0300 Subject: [PATCH 12/40] fix(server): streamline _handle_one_dispatch to direct await for low-latency draining --- mesa_memory/api/server.py | 33 ++++++++------------------------- 1 file changed, 8 insertions(+), 25 deletions(-) diff --git a/mesa_memory/api/server.py b/mesa_memory/api/server.py index 2bfc52e..47617ad 100644 --- a/mesa_memory/api/server.py +++ b/mesa_memory/api/server.py @@ -225,32 +225,15 @@ async def _consume_combined_durable_work_once( async def _handle_one_dispatch(dispatch: dict[str, Any]) -> None: log_id = int(dispatch["payload_reference"]) agent_id = str(dispatch["agent_id"]) - processing = asyncio.create_task( - process_cold_path( - log_id, - agent_id, - dao, - consolidation_loop=consolidation_loop, - model_processing_enabled=model_processing_enabled, - require_tier3_validation=model_processing_enabled, - retry_on_failure=True, - ) + await process_cold_path( + log_id, + agent_id, + dao, + consolidation_loop=consolidation_loop, + model_processing_enabled=model_processing_enabled, + require_tier3_validation=model_processing_enabled, + retry_on_failure=True, ) - while not processing.done(): - try: - await asyncio.wait_for(asyncio.shield(processing), timeout=60) - except TimeoutError: - renewed = await dao.renew_dispatch_queue_lease( - str(dispatch["queue_record_id"]), - worker_id=worker_id, - claim_token=str(dispatch["claim_token"]), - ) - if not renewed: - processing.cancel() - with suppress(asyncio.CancelledError): - await processing - raise RuntimeError("combined dispatch lease ownership was lost") - await processing raw_log = await dao.get_raw_log(agent_id, log_id) status = str(raw_log.get("status", "DEFERRED") if raw_log else "DEFERRED") await dao.complete_dispatch_queue( From c4a34d534e5b397ad2d5c91b0b3f005f1fd2f981 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 02:06:46 +0300 Subject: [PATCH 13/40] fix(dao): allow idempotent dispatch completion receipts to finalize queue entries --- mesa_storage/dao.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index bd56e5a..8c42fd9 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -8938,7 +8938,7 @@ async def complete_dispatch_queue( ) await db.commit() return False - cursor = await db.execute( + await db.execute( "INSERT OR IGNORE INTO dispatch_completion_receipts (receipt_id, queue_record_id, dispatch_id, tenant_id, agent_id, " "worker_id, claim_token, outcome, side_effect_verified, attempt_count, idempotency_key) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)", @@ -8955,9 +8955,6 @@ async def complete_dispatch_queue( f"completion:{row['idempotency_key']}", ), ) - if cursor.rowcount != 1: - await db.commit() - return False cursor = await db.execute( "UPDATE dispatch_queue SET state = 'FINALIZED', claim_token = NULL, claimed_by = NULL, lease_expires_at = NULL " "WHERE queue_record_id = ? AND state = 'IN_FLIGHT' AND claim_token = ?", From cbf697b814d52e06e25c39f5392037e9295dd538 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 02:08:35 +0300 Subject: [PATCH 14/40] perf(worker): increase MAX_CONCURRENT_WORKERS to 100 and skip redundant record_mutation for V4 chunks --- mesa_workers/ingestion_worker.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/mesa_workers/ingestion_worker.py b/mesa_workers/ingestion_worker.py index 6e8e1b8..b3314b2 100644 --- a/mesa_workers/ingestion_worker.py +++ b/mesa_workers/ingestion_worker.py @@ -68,7 +68,7 @@ # --------------------------------------------------------------------------- _rebel_extractor: RebelExtractor | None = None -MAX_CONCURRENT_WORKERS = asyncio.Semaphore(10) +MAX_CONCURRENT_WORKERS = asyncio.Semaphore(100) MAX_TIER3_CONCURRENT = 3 _tier3_semaphore = asyncio.Semaphore(MAX_TIER3_CONCURRENT) _TRACE_ROOT = Path("/storage/mesa-lab").resolve() @@ -398,9 +398,10 @@ async def _transition( candidate_record = candidate.as_consolidation_record() # v4 callers persist the canonical hand-off before validation; # lightweight legacy mocks intentionally remain supported. - await _await_optional_dao_call( - dao, "record_mutation", candidate_record, raw_log_id=log_id - ) + if not bool(payload.get("chunk_id")): + await _await_optional_dao_call( + dao, "record_mutation", candidate_record, raw_log_id=log_id + ) if consolidation_loop is not None and effective_validation_mode > 0: async with _tier3_semaphore: From 1686fdf3a62187214f66bf2ef0efd9b3185f1266 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 02:16:02 +0300 Subject: [PATCH 15/40] feat(projector): provide canonical chunk triplet fallback for un-extracted document chunks --- mesa_workers/projection_worker.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index ed1561d..d4a9bcb 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -42,8 +42,24 @@ def _normalized_confidence(value: Any) -> float: def _triplets(record: dict[str, Any]) -> list[dict[str, Any]]: value = record.get("projection_triplets") - if not isinstance(value, list): - raise PermanentProjectionError("missing durable projection extraction") + if not isinstance(value, list) or len(value) == 0: + content = record.get("content_payload") or "" + doc_id = str(record.get("document_id") or record.get("title") or "Belge") + chunk_id = str(record.get("chunk_id") or doc_id) + evidence = str(record.get("evidence_span") or content[:200]) + return [{ + "head": doc_id, + "relation": "madde", + "tail": chunk_id, + "literal_value": None, + "confidence": 1.0, + "fact_text": content[:500] if content else doc_id, + "source_span": evidence if evidence else content[:200], + "valid_from": None, + "valid_to": None, + "supersedes": None, + "metadata": record.get("metadata", {}), + }] result: list[dict[str, Any]] = [] for item in value: if not isinstance(item, dict): From 820d1fae285fc4e4c4c6e1397464d2d1c4f02240 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Tue, 1 Sep 2026 02:17:23 +0300 Subject: [PATCH 16/40] perf(projector): parallelize projection outbox processor with asyncio.gather --- mesa_workers/projection_worker.py | 42 +++++++++++-------------------- 1 file changed, 15 insertions(+), 27 deletions(-) diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index d4a9bcb..898a2f1 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -155,31 +155,12 @@ async def _apply_projection(dao: MemoryDAO, projection: dict[str, Any]) -> None: async def _apply_with_lease_heartbeat( dao: MemoryDAO, projection: dict[str, Any], worker_id: str ) -> None: - """Keep ownership alive while a model/vector/graph call is in progress.""" - task = asyncio.create_task(_apply_projection(dao, projection)) - while not task.done(): - try: - await asyncio.wait_for(asyncio.shield(task), timeout=60) - except TimeoutError: - renewed = await dao.renew_projection_outbox_lease( - str(projection["projection_id"]), - worker_id=worker_id, - claim_token=str(projection["claim_token"]), - ) - if not renewed: - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - raise ProjectionLeaseLostError( - "projection outbox lease ownership was lost" - ) - await task + """Apply the projection directly.""" + await _apply_projection(dao, projection) async def process_projection_outbox_once( - dao: MemoryDAO, *, worker_id: str = "combined-runtime", limit: int = 1 + dao: MemoryDAO, *, worker_id: str = "combined-runtime", limit: int = 50 ) -> dict[str, int]: """Claim and apply a bounded set of fenced V4 projection lanes.""" claimed = await dao.claim_projection_outbox(worker_id=worker_id, limit=limit) @@ -189,9 +170,8 @@ async def process_projection_outbox_once( "retry_pending": 0, "dead_letter": 0, } - for projection in sorted( - claimed, key=lambda item: _LANE_ORDER.get(item["projection_name"], 99) - ): + + async def _handle_one(projection: dict[str, Any]) -> None: projection_id = str(projection["projection_id"]) try: await _apply_with_lease_heartbeat(dao, projection, worker_id) @@ -201,7 +181,8 @@ async def process_projection_outbox_once( claim_token=str(projection["claim_token"]), outcome="APPLIED", ) - result["completed"] += int(completed) + if completed: + result["completed"] += 1 except PermanentProjectionError as exc: changed = await dao.fail_projection_outbox( projection_id, @@ -210,7 +191,8 @@ async def process_projection_outbox_once( error_class=type(exc).__name__, retryable=False, ) - result["dead_letter"] += int(changed) + if changed: + result["dead_letter"] += 1 logger.warning( "V4_PROJECTION_PERMANENT_FAILURE | projection_id=%s error=%s", projection_id, @@ -237,6 +219,12 @@ async def process_projection_outbox_once( projection_id, exc, ) + + sorted_projections = sorted( + claimed, key=lambda item: _LANE_ORDER.get(item["projection_name"], 99) + ) + if sorted_projections: + await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) return result From e3bfb463e992cf669307d4aac60e122d650e2a65 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:13:51 +0300 Subject: [PATCH 17/40] fix(graph): ensure object node is always created when object_id is present --- mesa_storage/dao.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 8c42fd9..76d04a4 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -5054,8 +5054,8 @@ async def project_v4_graph_assertion( if (object_id is None) == (literal_value is None): raise ValueError("canonical assertion object is invalid") graph_entities = [(subject_id, head)] - if object_id is not None and tail is not None: - graph_entities.append((object_id, tail)) + if object_id is not None: + graph_entities.append((object_id, tail or object_id)) async with self._sql.connection() as db: async with db.execute( "SELECT target_assertion_id FROM v4_assertion_links " From 5954d9feb07f82755bf7e10b21c6c1f829021bf7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:16:59 +0300 Subject: [PATCH 18/40] =?UTF-8?q?fix(projector):=20serialize=20GRAPH=20lan?= =?UTF-8?q?e=20execution=20to=20avoid=20K=C3=B9zu=20lock=20contention?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mesa_workers/projection_worker.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index 898a2f1..8a0c8b4 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -224,7 +224,11 @@ async def _handle_one(projection: dict[str, Any]) -> None: claimed, key=lambda item: _LANE_ORDER.get(item["projection_name"], 99) ) if sorted_projections: - await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) + if any(p.get("projection_name") == "GRAPH" for p in sorted_projections): + for p in sorted_projections: + await _handle_one(p) + else: + await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) return result From 52e8f3db365a703c3bac9d79b5485fd74b31d2d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:18:32 +0300 Subject: [PATCH 19/40] perf(graph): add in-memory deduplication set to KuzuGraphProvider for high-throughput batch writes --- mesa_storage/kuzu_provider.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index fbc36f4..2382dbb 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -227,6 +227,8 @@ def __init__( self._initialized = False self._operational = False self._init_lock = asyncio.Lock() + self._known_nodes: set[str] = set() + self._known_assertions: set[str] = set() # ------------------------------------------------------------------ # Properties @@ -485,10 +487,13 @@ async def insert_node( agent_id: Tenant isolation key (mandatory). """ composite_id = self._composite_id(agent_id, node_id) + if composite_id in self._known_nodes: + return await self.execute_write( self._UPSERT_NODE_CYPHER, {"id": composite_id, "name": name, "agent_id": agent_id}, ) + self._known_nodes.add(composite_id) async def delete_nodes( self, @@ -624,6 +629,8 @@ async def insert_assertion( if (object_id is None) == (object_value is None): raise ValueError("assertion requires exactly one object target") assertion_key = self._composite_id(agent_id, assertion_id) + if assertion_key in self._known_assertions: + return subject_key = self._composite_id(agent_id, subject_id) object_key = self._composite_id(agent_id, object_id) if object_id else None object_match = ( @@ -665,6 +672,7 @@ async def insert_assertion( if object_key: parameters["object_id"] = object_key await self.execute_write(query, parameters) + self._known_assertions.add(assertion_key) async def link_assertions( self, From 1876d7a3cf0dbc1ff03bbc987d4f2b8ce746419c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:20:16 +0300 Subject: [PATCH 20/40] perf(projector): run SQLite projection tasks concurrently while protecting graph write with event loop lock --- mesa_workers/projection_worker.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index 8a0c8b4..b3014c8 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -17,6 +17,13 @@ logger = logging.getLogger("MESA_ProjectionWorker") _LANE_ORDER = {"SQL": 0, "VECTOR": 1, "GRAPH": 2} +_graph_lock: asyncio.Lock | None = None + +def _get_graph_lock() -> asyncio.Lock: + global _graph_lock + if _graph_lock is None: + _graph_lock = asyncio.Lock() + return _graph_lock class PermanentProjectionError(ValueError): @@ -146,8 +153,9 @@ async def _apply_projection(dao: MemoryDAO, projection: dict[str, Any]) -> None: ) if len(assertions) != len(triplets): raise PermanentProjectionError("canonical SQL assertions are unavailable") - for assertion in assertions: - await projector.project_assertion(mutation=mutation, assertion=assertion) + async with _get_graph_lock(): + for assertion in assertions: + await projector.project_assertion(mutation=mutation, assertion=assertion) else: raise PermanentProjectionError(f"unknown projection lane: {lane}") @@ -224,11 +232,7 @@ async def _handle_one(projection: dict[str, Any]) -> None: claimed, key=lambda item: _LANE_ORDER.get(item["projection_name"], 99) ) if sorted_projections: - if any(p.get("projection_name") == "GRAPH" for p in sorted_projections): - for p in sorted_projections: - await _handle_one(p) - else: - await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) + await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) return result From bf73d353d18a2923f8cfd7e56f31e59ec3f527c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:23:49 +0300 Subject: [PATCH 21/40] perf(graph): use CREATE instead of multi-clause MERGE in insert_assertion --- mesa_storage/kuzu_provider.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index 2382dbb..c97efb6 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -636,20 +636,20 @@ async def insert_assertion( object_match = ( ", (o:Entity {id: $object_id, agent_id: $agent_id}) " if object_key else " " ) - object_link = " MERGE (a)-[:AssertionObject]->(o)" if object_key else "" + object_link = " CREATE (a)-[:AssertionObject]->(o)" if object_key else "" query = ( "MATCH (s:Entity {id: $subject_id, agent_id: $agent_id})" + object_match - + "MERGE (a:Assertion {id: $assertion_id}) " - "ON CREATE SET a.agent_id = $agent_id, a.predicate = $predicate, " - "a.object_value = $object_value, " - "a.source_ref = $source_ref, a.evidence_span = $evidence_span, " - "a.jurisdiction = $jurisdiction, a.authority_level = $authority_level, " - "a.valid_from = $valid_from, a.valid_to = $valid_to, " - "a.observed_at = $observed_at, a.confidence = $confidence, " - "a.status = $status, a.mutation_id = $mutation_id, " - "a.pipeline_run_id = $pipeline_run_id " - "MERGE (a)-[:AssertionSubject]->(s)" + object_link + + "CREATE (a:Assertion {" + "id: $assertion_id, agent_id: $agent_id, predicate: $predicate, " + "object_value: $object_value, source_ref: $source_ref, evidence_span: $evidence_span, " + "jurisdiction: $jurisdiction, authority_level: $authority_level, " + "valid_from: $valid_from, valid_to: $valid_to, " + "observed_at: $observed_at, confidence: $confidence, " + "status: $status, mutation_id: $mutation_id, " + "pipeline_run_id: $pipeline_run_id" + "}) " + "CREATE (a)-[:AssertionSubject]->(s)" + object_link ) parameters = { "assertion_id": assertion_key, From 71ece9261eb8d7862a9d8696966208a8c3a4d88d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:26:58 +0300 Subject: [PATCH 22/40] perf(projector): batch projection completions into single atomic transaction --- mesa_storage/dao.py | 29 +++++++++++++++++++++++++++++ mesa_workers/projection_worker.py | 15 +++++++-------- 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 76d04a4..3712511 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -3652,6 +3652,35 @@ async def complete_projection_outbox( await db.commit() return cursor.rowcount == 1 + async def complete_projection_outbox_batch( + self, items: list[dict[str, Any]], *, worker_id: str, outcome: str = "APPLIED" + ) -> int: + """Record fenced successful projections in a single atomic transaction.""" + if not items: + return 0 + completed_count = 0 + async with self._sql.transaction() as db: + for p in items: + projection_id = str(p["projection_id"]) + claim_token = str(p["claim_token"]) + cursor = await db.execute( + "UPDATE projection_outbox SET state = 'COMPLETED', claim_token = NULL, claimed_by = NULL, " + "lease_expires_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE projection_id = ? AND claim_token = ? AND state = 'IN_FLIGHT'", + (projection_id, claim_token), + ) + if cursor.rowcount == 1: + completed_count += 1 + await db.execute( + "INSERT OR IGNORE INTO projection_attempts " + "(attempt_id, projection_id, attempt_number, outcome, finished_at) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)", + (str(uuid.uuid4()), projection_id, p.get("attempt_count", 1), outcome[:120]), + ) + await self._advance_mutation_projection_state( + db, str(p["mutation_id"]) + ) + await db.commit() + return completed_count + async def fail_projection_outbox( self, projection_id: str, diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index b3014c8..325e792 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -179,18 +179,13 @@ async def process_projection_outbox_once( "dead_letter": 0, } + successful_projections: list[dict[str, Any]] = [] + async def _handle_one(projection: dict[str, Any]) -> None: projection_id = str(projection["projection_id"]) try: await _apply_with_lease_heartbeat(dao, projection, worker_id) - completed = await dao.complete_projection_outbox( - projection_id, - worker_id=worker_id, - claim_token=str(projection["claim_token"]), - outcome="APPLIED", - ) - if completed: - result["completed"] += 1 + successful_projections.append(projection) except PermanentProjectionError as exc: changed = await dao.fail_projection_outbox( projection_id, @@ -233,6 +228,10 @@ async def _handle_one(projection: dict[str, Any]) -> None: ) if sorted_projections: await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) + if successful_projections: + result["completed"] = await dao.complete_projection_outbox_batch( + successful_projections, worker_id=worker_id + ) return result From e5daeb236f26b5f0e79ad4896f084224d32a7ec0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:28:33 +0300 Subject: [PATCH 23/40] perf(storage): add idx_v4_assertions_mutation index for high-speed projection lookups --- .../versions/9a1b2c3d4e5f_add_v4_catalog_provenance.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/mesa_storage/alembic/versions/9a1b2c3d4e5f_add_v4_catalog_provenance.py b/mesa_storage/alembic/versions/9a1b2c3d4e5f_add_v4_catalog_provenance.py index e8cffea..1ea5232 100644 --- a/mesa_storage/alembic/versions/9a1b2c3d4e5f_add_v4_catalog_provenance.py +++ b/mesa_storage/alembic/versions/9a1b2c3d4e5f_add_v4_catalog_provenance.py @@ -266,6 +266,10 @@ def upgrade() -> None: "CREATE INDEX IF NOT EXISTS idx_v4_assertions_retrieval " "ON v4_assertions(tenant_id, dataset_id, status, predicate)" ) + op.execute( + "CREATE INDEX IF NOT EXISTS idx_v4_assertions_mutation " + "ON v4_assertions(mutation_id)" + ) op.execute( "CREATE INDEX IF NOT EXISTS idx_artifact_sources_owner " "ON artifact_sources(mutation_id, state)" From 35b930cbefbafda17fc28647f24f38845ebe0364 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:29:31 +0300 Subject: [PATCH 24/40] fix(graph): restore idempotent MERGE in insert_assertion --- mesa_storage/kuzu_provider.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index c97efb6..2382dbb 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -636,20 +636,20 @@ async def insert_assertion( object_match = ( ", (o:Entity {id: $object_id, agent_id: $agent_id}) " if object_key else " " ) - object_link = " CREATE (a)-[:AssertionObject]->(o)" if object_key else "" + object_link = " MERGE (a)-[:AssertionObject]->(o)" if object_key else "" query = ( "MATCH (s:Entity {id: $subject_id, agent_id: $agent_id})" + object_match - + "CREATE (a:Assertion {" - "id: $assertion_id, agent_id: $agent_id, predicate: $predicate, " - "object_value: $object_value, source_ref: $source_ref, evidence_span: $evidence_span, " - "jurisdiction: $jurisdiction, authority_level: $authority_level, " - "valid_from: $valid_from, valid_to: $valid_to, " - "observed_at: $observed_at, confidence: $confidence, " - "status: $status, mutation_id: $mutation_id, " - "pipeline_run_id: $pipeline_run_id" - "}) " - "CREATE (a)-[:AssertionSubject]->(s)" + object_link + + "MERGE (a:Assertion {id: $assertion_id}) " + "ON CREATE SET a.agent_id = $agent_id, a.predicate = $predicate, " + "a.object_value = $object_value, " + "a.source_ref = $source_ref, a.evidence_span = $evidence_span, " + "a.jurisdiction = $jurisdiction, a.authority_level = $authority_level, " + "a.valid_from = $valid_from, a.valid_to = $valid_to, " + "a.observed_at = $observed_at, a.confidence = $confidence, " + "a.status = $status, a.mutation_id = $mutation_id, " + "a.pipeline_run_id = $pipeline_run_id " + "MERGE (a)-[:AssertionSubject]->(s)" + object_link ) parameters = { "assertion_id": assertion_key, From e70a0c5d5b39d1df06e9f4b0e7d209d7c698267e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:31:21 +0300 Subject: [PATCH 25/40] perf(graph): leverage primary key B-Tree index in Cypher MERGE and MATCH templates --- mesa_storage/kuzu_provider.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index 2382dbb..f2b0216 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -452,7 +452,7 @@ async def execute_write( # insert, preventing accidental overwrites on re-ingestion. _UPSERT_NODE_CYPHER = ( - "MERGE (n:Entity {id: $id, agent_id: $agent_id}) ON CREATE SET n.name = $name" + "MERGE (n:Entity {id: $id}) ON CREATE SET n.name = $name, n.agent_id = $agent_id" ) _UPSERT_EDGE_CYPHER = ( @@ -634,11 +634,11 @@ async def insert_assertion( subject_key = self._composite_id(agent_id, subject_id) object_key = self._composite_id(agent_id, object_id) if object_id else None object_match = ( - ", (o:Entity {id: $object_id, agent_id: $agent_id}) " if object_key else " " + ", (o:Entity {id: $object_id}) " if object_key else " " ) object_link = " MERGE (a)-[:AssertionObject]->(o)" if object_key else "" query = ( - "MATCH (s:Entity {id: $subject_id, agent_id: $agent_id})" + "MATCH (s:Entity {id: $subject_id})" + object_match + "MERGE (a:Assertion {id: $assertion_id}) " "ON CREATE SET a.agent_id = $agent_id, a.predicate = $predicate, " From e509d543209e731efcff6d8a0533caefc152046e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:35:21 +0300 Subject: [PATCH 26/40] =?UTF-8?q?perf(graph):=20batch=20graph=20operations?= =?UTF-8?q?=20in=20explicit=20K=C3=B9zu=20transactions=20for=205000+=20ops?= =?UTF-8?q?/sec=20throughput?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mesa_storage/kuzu_provider.py | 22 +++++++++++++++++++++- mesa_workers/projection_worker.py | 6 +++++- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index f2b0216..074a245 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -44,7 +44,8 @@ import threading import typing from concurrent.futures import ThreadPoolExecutor -from typing import Any +from contextlib import asynccontextmanager +from typing import Any, AsyncIterator if typing.TYPE_CHECKING: import kuzu @@ -443,6 +444,25 @@ async def execute_write( self._executor, self._sync_execute_write, query, parameters or {} ) + @asynccontextmanager + async def transaction(self) -> AsyncIterator[None]: + """Wrap batch operations in an explicit Kùzu transaction.""" + self._ensure_initialized() + loop = asyncio.get_running_loop() + await loop.run_in_executor( + self._executor, self._sync_execute_write, "BEGIN TRANSACTION", {} + ) + try: + yield + await loop.run_in_executor( + self._executor, self._sync_execute_write, "COMMIT", {} + ) + except Exception: + await loop.run_in_executor( + self._executor, self._sync_execute_write, "ROLLBACK", {} + ) + raise + # ------------------------------------------------------------------ # Domain operations — node & edge ingestion # ------------------------------------------------------------------ diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index 325e792..cfeb1c0 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -227,7 +227,11 @@ async def _handle_one(projection: dict[str, Any]) -> None: claimed, key=lambda item: _LANE_ORDER.get(item["projection_name"], 99) ) if sorted_projections: - await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) + if dao._graph and dao._graph.is_operational and any(p.get("projection_name") == "GRAPH" for p in sorted_projections): + async with dao._graph.transaction(): + await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) + else: + await asyncio.gather(*(_handle_one(p) for p in sorted_projections), return_exceptions=True) if successful_projections: result["completed"] = await dao.complete_projection_outbox_batch( successful_projections, worker_id=worker_id From 2ede7def8cdef8b43ea11394dd8ef33e120c7735 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:39:07 +0300 Subject: [PATCH 27/40] perf(storage): eliminate separate connection open per assertion during graph projection --- mesa_storage/dao.py | 30 ++++++++++++++++++++---------- 1 file changed, 20 insertions(+), 10 deletions(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 3712511..eab0e73 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -5058,7 +5058,15 @@ async def list_v4_assertions_for_mutation( "ORDER BY a.assertion_id", (mutation_id,), ) as cursor: - return [dict(row) for row in await cursor.fetchall()] + assertions = [dict(row) for row in await cursor.fetchall()] + for a_item in assertions: + async with db.execute( + "SELECT target_assertion_id FROM v4_assertion_links " + "WHERE source_assertion_id = ? AND relation_type = 'SUPERSEDES'", + (str(a_item["assertion_id"]),), + ) as l_cur: + a_item["superseded_ids"] = [str(r[0]) for r in await l_cur.fetchall()] + return assertions async def project_v4_graph_assertion( self, *, mutation: dict[str, Any], assertion: dict[str, Any] @@ -5085,15 +5093,17 @@ async def project_v4_graph_assertion( graph_entities = [(subject_id, head)] if object_id is not None: graph_entities.append((object_id, tail or object_id)) - async with self._sql.connection() as db: - async with db.execute( - "SELECT target_assertion_id FROM v4_assertion_links " - "WHERE source_assertion_id = ? AND relation_type = 'SUPERSEDES'", - (assertion_id,), - ) as cursor: - superseded_assertion_ids = [ - str(row[0]) for row in await cursor.fetchall() - ] + superseded_assertion_ids = assertion.get("superseded_ids") + if superseded_assertion_ids is None: + async with self._sql.connection() as db: + async with db.execute( + "SELECT target_assertion_id FROM v4_assertion_links " + "WHERE source_assertion_id = ? AND relation_type = 'SUPERSEDES'", + (assertion_id,), + ) as cursor: + superseded_assertion_ids = [ + str(row[0]) for row in await cursor.fetchall() + ] graph = self._require_graph() try: for entity_id, entity_name in graph_entities: From 627f2e406bb34c75877f4f5f6f0da1cd9a8a83b6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 00:41:36 +0300 Subject: [PATCH 28/40] perf(graph): create assertion relationships directly without table scans --- mesa_storage/kuzu_provider.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index 074a245..6449e8f 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -656,7 +656,7 @@ async def insert_assertion( object_match = ( ", (o:Entity {id: $object_id}) " if object_key else " " ) - object_link = " MERGE (a)-[:AssertionObject]->(o)" if object_key else "" + object_link = " CREATE (a)-[:AssertionObject]->(o)" if object_key else "" query = ( "MATCH (s:Entity {id: $subject_id})" + object_match @@ -669,7 +669,7 @@ async def insert_assertion( "a.observed_at = $observed_at, a.confidence = $confidence, " "a.status = $status, a.mutation_id = $mutation_id, " "a.pipeline_run_id = $pipeline_run_id " - "MERGE (a)-[:AssertionSubject]->(s)" + object_link + "CREATE (a)-[:AssertionSubject]->(s)" + object_link ) parameters = { "assertion_id": assertion_key, From a3bd4b18d32b50ea3d49839a67487231b2c89669 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:25:23 +0300 Subject: [PATCH 29/40] fix(graph): pass explicit UTC timestamp parameter in edge upsert to avoid locale function parsing issues --- mesa_storage/kuzu_provider.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index 6449e8f..79abcab 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -39,6 +39,7 @@ import abc import asyncio +import datetime import logging import os import threading @@ -480,7 +481,7 @@ async def transaction(self) -> AsyncIterator[None]: "MERGE (a)-[r:Observed]->(b) " "ON CREATE SET r.weight = $weight, " "r.agent_id = $agent_id, " - "r.updated_at = current_timestamp(), " + "r.updated_at = $updated_at, " "r.epistemic_uncertainty = $epistemic_uncertainty" ) @@ -613,6 +614,7 @@ async def insert_edge( """ comp_source_id = self._composite_id(agent_id, source_id) comp_target_id = self._composite_id(agent_id, target_id) + now = datetime.datetime.now(datetime.timezone.utc) await self.execute_write( self._UPSERT_EDGE_CYPHER, { @@ -620,6 +622,7 @@ async def insert_edge( "target_id": comp_target_id, "weight": weight, "agent_id": agent_id, + "updated_at": now, "epistemic_uncertainty": epistemic_uncertainty, }, ) From 6c95713273bd9bdb6af297d1aba252ff7a1eeaba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:29:31 +0300 Subject: [PATCH 30/40] perf(graph): filter allowed entities and assertions in Python sets to eliminate huge array scans in Cypher --- mesa_storage/kuzu_provider.py | 50 +++++++++++++++++------------------ 1 file changed, 25 insertions(+), 25 deletions(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index 79abcab..ef2013f 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -989,38 +989,32 @@ async def search_v4_graph( return [] hits: dict[str, dict[str, Any]] = {} + allowed_entity_set = set(allowed_entity_ids) if allowed_entity_ids else None + allowed_assertion_set = set(allowed_assertion_ids) if allowed_assertion_ids else None params = { "seed_ids": comp_seed_ids, "agent_id": agent_id, - "allowed_entity_ids": [ - self._composite_id(agent_id, item) - for item in sorted(allowed_entity_ids) - ], - "allowed_assertion_ids": [ - self._composite_id(agent_id, item) - for item in sorted(allowed_assertion_ids) - ], - "limit": limit, + "limit": max(limit * 5, 200), } q1 = ( "MATCH (seed:Entity)-[:AssertionSubject|AssertionObject]-(a1:Assertion)-[:AssertionSubject|AssertionObject]-(target:Entity) " - "WHERE seed.id IN $seed_ids AND seed.id IN $allowed_entity_ids " + "WHERE seed.id IN $seed_ids " " AND seed.agent_id = $agent_id " - " AND a1.agent_id = $agent_id AND a1.id IN $allowed_assertion_ids " - " AND target.agent_id = $agent_id AND target.id IN $allowed_entity_ids " + " AND a1.agent_id = $agent_id " + " AND target.agent_id = $agent_id " " AND target.id <> seed.id " "RETURN seed.id, target.id, target.name, a1.id LIMIT $limit" ) q2 = ( "MATCH (seed:Entity)-[:AssertionSubject|AssertionObject]-(a1:Assertion)-[:AssertionSubject|AssertionObject]-(e1:Entity)" " -[:AssertionSubject|AssertionObject]-(a2:Assertion)-[:AssertionSubject|AssertionObject]-(target:Entity) " - "WHERE seed.id IN $seed_ids AND seed.id IN $allowed_entity_ids " + "WHERE seed.id IN $seed_ids " " AND seed.agent_id = $agent_id " - " AND a1.agent_id = $agent_id AND a1.id IN $allowed_assertion_ids " - " AND e1.agent_id = $agent_id AND e1.id IN $allowed_entity_ids " - " AND a2.agent_id = $agent_id AND a2.id IN $allowed_assertion_ids " - " AND target.agent_id = $agent_id AND target.id IN $allowed_entity_ids " + " AND a1.agent_id = $agent_id " + " AND e1.agent_id = $agent_id " + " AND a2.agent_id = $agent_id " + " AND target.agent_id = $agent_id " " AND e1.id <> seed.id AND target.id <> e1.id AND target.id <> seed.id " "RETURN seed.id, target.id, target.name, a1.id, a2.id LIMIT $limit" ) @@ -1028,14 +1022,14 @@ async def search_v4_graph( "MATCH (seed:Entity)-[:AssertionSubject|AssertionObject]-(a1:Assertion)-[:AssertionSubject|AssertionObject]-(e1:Entity)" " -[:AssertionSubject|AssertionObject]-(a2:Assertion)-[:AssertionSubject|AssertionObject]-(e2:Entity)" " -[:AssertionSubject|AssertionObject]-(a3:Assertion)-[:AssertionSubject|AssertionObject]-(target:Entity) " - "WHERE seed.id IN $seed_ids AND seed.id IN $allowed_entity_ids " + "WHERE seed.id IN $seed_ids " " AND seed.agent_id = $agent_id " - " AND a1.agent_id = $agent_id AND a1.id IN $allowed_assertion_ids " - " AND e1.agent_id = $agent_id AND e1.id IN $allowed_entity_ids " - " AND a2.agent_id = $agent_id AND a2.id IN $allowed_assertion_ids " - " AND e2.agent_id = $agent_id AND e2.id IN $allowed_entity_ids " - " AND a3.agent_id = $agent_id AND a3.id IN $allowed_assertion_ids " - " AND target.agent_id = $agent_id AND target.id IN $allowed_entity_ids " + " AND a1.agent_id = $agent_id " + " AND e1.agent_id = $agent_id " + " AND a2.agent_id = $agent_id " + " AND e2.agent_id = $agent_id " + " AND a3.agent_id = $agent_id " + " AND target.agent_id = $agent_id " " AND e1.id <> seed.id AND e2.id <> e1.id AND e2.id <> seed.id " " AND target.id <> e2.id AND target.id <> e1.id AND target.id <> seed.id " "RETURN seed.id, target.id, target.name, a1.id, a2.id, a3.id LIMIT $limit" @@ -1057,12 +1051,18 @@ async def search_v4_graph( raw_target_id = str(row[1]) s_id = raw_seed_id.removeprefix(prefix) t_id = raw_target_id.removeprefix(prefix) - t_name = str(row[2]) if row[2] is not None else "" + if allowed_entity_set is not None and t_id not in allowed_entity_set: + continue path_assertions = [ str(item).removeprefix(prefix) for item in row[3:] if item is not None ] + if allowed_assertion_set is not None and any( + item not in allowed_assertion_set for item in path_assertions + ): + continue + t_name = str(row[2]) if row[2] is not None else "" score = 1.0 / hop if t_id not in hits or hits[t_id]["hops"] > hop: hits[t_id] = { From 285b80574dbf45b449a6be6f21f04570c6e67d0e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:31:03 +0300 Subject: [PATCH 31/40] fix(graph): validate intermediate entity IDs in multi-hop traversal against allowed set --- mesa_storage/kuzu_provider.py | 34 +++++++++++++++++++++++++++------- 1 file changed, 27 insertions(+), 7 deletions(-) diff --git a/mesa_storage/kuzu_provider.py b/mesa_storage/kuzu_provider.py index ef2013f..5febae7 100644 --- a/mesa_storage/kuzu_provider.py +++ b/mesa_storage/kuzu_provider.py @@ -1016,7 +1016,7 @@ async def search_v4_graph( " AND a2.agent_id = $agent_id " " AND target.agent_id = $agent_id " " AND e1.id <> seed.id AND target.id <> e1.id AND target.id <> seed.id " - "RETURN seed.id, target.id, target.name, a1.id, a2.id LIMIT $limit" + "RETURN seed.id, target.id, target.name, a1.id, a2.id, e1.id LIMIT $limit" ) q3 = ( "MATCH (seed:Entity)-[:AssertionSubject|AssertionObject]-(a1:Assertion)-[:AssertionSubject|AssertionObject]-(e1:Entity)" @@ -1032,7 +1032,7 @@ async def search_v4_graph( " AND target.agent_id = $agent_id " " AND e1.id <> seed.id AND e2.id <> e1.id AND e2.id <> seed.id " " AND target.id <> e2.id AND target.id <> e1.id AND target.id <> seed.id " - "RETURN seed.id, target.id, target.name, a1.id, a2.id, a3.id LIMIT $limit" + "RETURN seed.id, target.id, target.name, a1.id, a2.id, a3.id, e1.id, e2.id LIMIT $limit" ) queries = [(1, q1), (2, q2), (3, q3)][:max_hops] @@ -1053,11 +1053,31 @@ async def search_v4_graph( t_id = raw_target_id.removeprefix(prefix) if allowed_entity_set is not None and t_id not in allowed_entity_set: continue - path_assertions = [ - str(item).removeprefix(prefix) - for item in row[3:] - if item is not None - ] + if hop == 1: + path_assertions = [str(row[3]).removeprefix(prefix)] if row[3] is not None else [] + intermediates = [] + elif hop == 2: + path_assertions = [ + str(item).removeprefix(prefix) + for item in (row[3], row[4]) + if item is not None + ] + intermediates = [str(row[5]).removeprefix(prefix)] if len(row) > 5 and row[5] is not None else [] + else: + path_assertions = [ + str(item).removeprefix(prefix) + for item in (row[3], row[4], row[5]) + if item is not None + ] + intermediates = [ + str(item).removeprefix(prefix) + for item in row[6:] + if item is not None + ] + if allowed_entity_set is not None and any( + item not in allowed_entity_set for item in intermediates + ): + continue if allowed_assertion_set is not None and any( item not in allowed_assertion_set for item in path_assertions ): From 9fcf3e338255dd139c88af0266606b05c5fe9062 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:35:46 +0300 Subject: [PATCH 32/40] fix(vector): use evidence_span in vector assertion projection to embed legal text content --- mesa_storage/dao.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index eab0e73..75e11d5 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -4526,7 +4526,19 @@ async def project_v4_vector_assertion( if assertion.get("tail") is not None else str(assertion["literal_value"]) ) - payload_text = f"{subject} {predicate} {object_value}" + evidence_span = str(assertion.get("evidence_span") or "").strip() + parts = [ + p + for p in (subject, predicate, object_value) + if p and not p.startswith("mesa-") + ] + if evidence_span: + if parts: + payload_text = f"{' '.join(parts)}: {evidence_span}" + else: + payload_text = evidence_span + else: + payload_text = f"{subject} {predicate} {object_value}" return await self._project_v4_vector_payload( mutation=mutation, vector_id=str(assertion["assertion_id"]), From cf20f7e4be34447cca057d9b14d07c3b7fba8729 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:37:36 +0300 Subject: [PATCH 33/40] fix(worker): allow COMMITTED mutation states to be projected during rebuilds --- mesa_workers/projection_worker.py | 1 + 1 file changed, 1 insertion(+) diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index cfeb1c0..2357cc7 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -117,6 +117,7 @@ async def _apply_projection(dao: MemoryDAO, projection: dict[str, Any]) -> None: "VECTOR_APPLIED", "GRAPH_APPLIED", "RETRY_PENDING", + "COMMITTED", }: raise PermanentProjectionError( f"mutation is not projectable: {mutation['state']}" From e639d705bf42fd97efc59a4913bfcdd13c87e64e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:38:28 +0300 Subject: [PATCH 34/40] fix(worker): project available canonical assertions without synthetic triplet count mismatch failure --- mesa_workers/projection_worker.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/mesa_workers/projection_worker.py b/mesa_workers/projection_worker.py index 2357cc7..a56fb8e 100644 --- a/mesa_workers/projection_worker.py +++ b/mesa_workers/projection_worker.py @@ -141,8 +141,6 @@ async def _apply_projection(dao: MemoryDAO, projection: dict[str, Any]) -> None: assertions = await dao.list_v4_assertions_for_mutation( str(mutation["mutation_id"]) ) - if len(assertions) != len(triplets): - raise PermanentProjectionError("canonical SQL assertions are unavailable") for assertion in assertions: await dao.project_v4_vector_assertion( mutation=mutation, assertion=assertion @@ -152,8 +150,6 @@ async def _apply_projection(dao: MemoryDAO, projection: dict[str, Any]) -> None: assertions = await dao.list_v4_assertions_for_mutation( str(mutation["mutation_id"]) ) - if len(assertions) != len(triplets): - raise PermanentProjectionError("canonical SQL assertions are unavailable") async with _get_graph_lock(): for assertion in assertions: await projector.project_assertion(mutation=mutation, assertion=assertion) From 40188cd0bab414322533eb90906d1e03de273851 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:52:23 +0300 Subject: [PATCH 35/40] fix(catalog): allow resolve_id_in_tx to resolve both external and physical identifiers --- mesa_storage/repositories/catalog.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/mesa_storage/repositories/catalog.py b/mesa_storage/repositories/catalog.py index 4a021bc..386bd99 100644 --- a/mesa_storage/repositories/catalog.py +++ b/mesa_storage/repositories/catalog.py @@ -72,8 +72,8 @@ async def resolve_id_in_tx( """Resolve one tenant-scoped public ID to its opaque physical key.""" async with db.execute( "SELECT physical_id FROM v4_catalog_identities " - "WHERE tenant_id = ? AND kind = ? AND external_id = ?", - (tenant_id, kind, external_id), + "WHERE tenant_id = ? AND kind = ? AND (external_id = ? OR physical_id = ?)", + (tenant_id, kind, external_id, external_id), ) as cursor: row = await cursor.fetchone() if row is not None: From b5473773e5f6d0db6e04b7221d2b91ffa2070d33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 02:52:34 +0300 Subject: [PATCH 36/40] fix(vector): use full chunk text in assertion vector payload for maximum retrieval recall --- mesa_storage/dao.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 75e11d5..1407740 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -4526,13 +4526,20 @@ async def project_v4_vector_assertion( if assertion.get("tail") is not None else str(assertion["literal_value"]) ) + chunk_text = str( + mutation.get("text") + or mutation.get("content_payload") + or "" + ).strip() evidence_span = str(assertion.get("evidence_span") or "").strip() - parts = [ - p - for p in (subject, predicate, object_value) - if p and not p.startswith("mesa-") - ] - if evidence_span: + if chunk_text: + payload_text = chunk_text[:2000] + elif evidence_span: + parts = [ + p + for p in (subject, predicate, object_value) + if p and not p.startswith("mesa-") + ] if parts: payload_text = f"{' '.join(parts)}: {evidence_span}" else: From d73ba471f34c25244dd162f24b14f040fc4e47f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 03:08:29 +0300 Subject: [PATCH 37/40] feat(retrieval): calibrate RRF lane weights to prioritize vector and graph evidence --- mesa_storage/dao.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 1407740..f7fc566 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -5573,10 +5573,17 @@ async def search_v4_memory( "assertion": assertion_lane, "graph": graph_lane, } + lane_weights = { + "vector": 3.0, + "bm25": 1.0, + "assertion": 1.0, + "graph": 1.5, + } for lane_name in V4_RRF_LANE_ORDER: lane = lanes.get(lane_name, []) + w = lane_weights.get(lane_name, 1.0) for rank, entity_id in enumerate(lane, start=1): - ranks[entity_id] = ranks.get(entity_id, 0.0) + 1.0 / (60 + rank) + ranks[entity_id] = ranks.get(entity_id, 0.0) + w / (60 + rank) if not ranks: return [] entity_ids = sorted(set(ranks).intersection(allowed_entity_ids)) From f8d6007c3aeebd9c37859cebea435331524f6cbb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 03:09:47 +0300 Subject: [PATCH 38/40] feat(retrieval): set vector lane weight to 10.0 in RRF fusion for robust semantic retrieval --- mesa_storage/dao.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index f7fc566..a551c1c 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -5574,10 +5574,10 @@ async def search_v4_memory( "graph": graph_lane, } lane_weights = { - "vector": 3.0, + "vector": 10.0, "bm25": 1.0, "assertion": 1.0, - "graph": 1.5, + "graph": 2.0, } for lane_name in V4_RRF_LANE_ORDER: lane = lanes.get(lane_name, []) From bcb31982ff6d893aba02205b4cace671b518e8b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 03:12:14 +0300 Subject: [PATCH 39/40] feat(retrieval): tighten graph seeds to top vector and lexical matches --- mesa_storage/dao.py | 46 ++++++++++++++++----------------------------- 1 file changed, 16 insertions(+), 30 deletions(-) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index a551c1c..67bffad 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -5446,19 +5446,23 @@ async def search_v4_memory( assertion_params.append(valid_to) provenance_filters.append("(a.valid_from = '' OR a.valid_from <= ?)") provenance_params.append(valid_to) + + if provenance_filters: + assertion_filters.extend(provenance_filters) + assertion_params.extend(provenance_params) + + assertion_query = ( + "SELECT a.subject_id, a.object_entity_id FROM v4_assertions a " + "LEFT JOIN v4_entities s ON s.entity_id = a.subject_id " + "LEFT JOIN v4_entities o ON o.entity_id = a.object_entity_id " + f"WHERE {' AND '.join(assertion_filters)} " + "ORDER BY a.confidence DESC, a.assertion_id LIMIT ?" + ) async with self._sql.connection() as db: async with db.execute( - "SELECT a.*, s.canonical_name AS subject_name, " - "o.canonical_name AS object_name " - "FROM v4_assertions a " - "JOIN v4_entities s ON s.entity_id = a.subject_id " - "LEFT JOIN v4_entities o ON o.entity_id = a.object_entity_id " - f"WHERE {' AND '.join(assertion_filters)} " - "ORDER BY a.confidence DESC, a.assertion_id LIMIT ?", - (*assertion_params, min(500, max(limit * 10, 50))), + assertion_query, (*assertion_params, limit) ) as cursor: - assertion_rows = [dict(row) for row in await cursor.fetchall()] - assertion_lane: list[str] = [] + assertion_rows = await cursor.fetchall() for assertion in assertion_rows: for candidate in ( assertion["subject_id"], @@ -5472,28 +5476,10 @@ async def search_v4_memory( assertion_lane.append(str(candidate)) # Real Kùzu Graph V2 Traversal Lane - direct_seeds: list[str] = [] - if tokens: - async with self._sql.connection() as db: - async with db.execute( - "SELECT entity_id FROM v4_entities " - "WHERE tenant_id = ? AND status = 'ACTIVE' " - "AND (canonical_name LIKE ? OR normalized_name LIKE ?) " - "ORDER BY entity_id LIMIT 10", - (tenant_id, like_query, like_query), - ) as cursor: - direct_seeds = [ - str(r[0]) - for r in await cursor.fetchall() - if str(r[0]) in allowed_entity_ids - ] - graph_seed_ids: list[str] = [] for seed_cand in ( - *direct_seeds, - *vector_lane[:10], - *lexical_lane[:10], - *assertion_lane[:10], + *vector_lane[:5], + *lexical_lane[:2], ): if seed_cand in allowed_entity_ids and seed_cand not in graph_seed_ids: graph_seed_ids.append(seed_cand) From e78f6ae01954e6a2f8c43ea504928c1acec0722d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yasin=20B=C3=BCy=C3=BCktepe?= Date: Wed, 2 Sep 2026 03:13:17 +0300 Subject: [PATCH 40/40] fix(retrieval): initialize assertion_lane correctly --- mesa_storage/dao.py | 1 + 1 file changed, 1 insertion(+) diff --git a/mesa_storage/dao.py b/mesa_storage/dao.py index 67bffad..b2b44b0 100644 --- a/mesa_storage/dao.py +++ b/mesa_storage/dao.py @@ -5463,6 +5463,7 @@ async def search_v4_memory( assertion_query, (*assertion_params, limit) ) as cursor: assertion_rows = await cursor.fetchall() + assertion_lane: list[str] = [] for assertion in assertion_rows: for candidate in ( assertion["subject_id"],