Cert/profile b e2e

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(1800)
 
 
 class OpenAICompatibleAdapter(BaseUniversalLLMAdapter):

mesa_memory/api/server.py

@@ -220,36 +220,20 @@ 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)
-    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(
-            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(
@@ -259,7 +243,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=1)
+
+    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 +258,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=100)
+        cleanup = await process_artifact_cleanup_once(dao, worker_id=worker_id, limit=100)
     return {
         "dispatches": len(claimed),
         "finalizations": len(finalizations),
@@ -288,16 +276,22 @@ 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
 async def _runtime_lifespan(app: FastAPI, runtime: RuntimeProfileConfig):
+    refresh_config_from_environment()
     state.is_ready = False
 
     state.obs_layer = ObservabilityLayer()

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"

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)"