Cert/profile b e2e - #83
Open
Yasou13 wants to merge 40 commits into
Open
Conversation
…o support large legal corpus release delivery
…sumer batch limits
…projection batch to 100
…nt record_mutation for V4 chunks
…acted document chunks
… high-throughput batch writes
…cting graph write with event loop lock
… 5000+ ops/sec throughput
…g graph projection
…void locale function parsing issues
… eliminate huge array scans in Cypher
…gainst allowed set
…d legal text content
… triplet count mismatch failure
…sical identifiers
…mum retrieval recall
…ust semantic retrieval
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
It introduces confirmed runtime and correctness issues (notably a broken sqlite Row access in search_v4_memory and lease-fencing removal that can cause duplicate/stuck durable work).
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This PR appears aimed at improving “profile B” end-to-end throughput and robustness by increasing durable work batch sizes, loosening queue limits, and enhancing graph/projection handling (including Kùzu transaction support and broader ID resolution).
Changes:
- Batch and parallelize durable dispatch/projection consumption (higher per-iteration limits; new projection batch completion).
- Expand queue admission limits and concurrency thresholds.
- Adjust graph provider/query behavior (transactions, caching, traversal filtering) and catalog ID resolution.
File summaries
| File | Description |
|---|---|
| mesa_workers/projection_worker.py | Adds graph locking/transaction wrapping, increases claim limit, and introduces batch completion for projections. |
| mesa_workers/ingestion_worker.py | Increases cold-path concurrency and tweaks novelty gate + mutation recording/validation behavior. |
| mesa_storage/repositories/catalog.py | Allows resolving catalog IDs by either external_id or physical_id. |
| mesa_storage/kuzu_provider.py | Adds async transaction context manager and modifies node/edge/assertion write patterns and graph search filtering. |
| mesa_storage/dao.py | Adds projection batch completion, adjusts vector payload text, modifies assertion/link fetching, and changes search lane weighting/selection. |
| mesa_storage/alembic/versions/9a1b2c3d4e5f_add_v4_catalog_provenance.py | Adds an index on v4_assertions(mutation_id). |
| mesa_memory/config.py | Raises durable queue admission limits via defaults and env-configured fields. |
| mesa_memory/api/server.py | Increases dispatch/projection/cleanup batch sizes and changes combined durable consumer loop behavior. |
| mesa_memory/adapter/live.py | Extends OpenAI retry stop-after-delay window. |
Review details
Suppressed comments (1)
mesa_storage/dao.py:5471
- search_v4_memory now keeps assertion_rows as sqlite Row objects (cursor.fetchall()), but still calls assertion.get('object_entity_id'). sqlite3.Row doesn’t implement .get(), so this will raise AttributeError at runtime and break retrieval.
assertion_lane: list[str] = []
for assertion in assertion_rows:
for candidate in (
assertion["subject_id"],
assertion.get("object_entity_id"),
):
- Files reviewed: 9/9 changed files
- Comments generated: 7
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+225
to
+229
| 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, |
Comment on lines
659
to
676
| @@ -642,7 +672,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 | |||
| ) | |||
Comment on lines
160
to
165
| 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) | ||
|
|
Comment on lines
+5081
to
+5088
| 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 |
Comment on lines
615
to
626
| 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, | ||
| { | ||
| "source_id": comp_source_id, | ||
| "target_id": comp_target_id, | ||
| "weight": weight, | ||
| "agent_id": agent_id, | ||
| "updated_at": now, | ||
| "epistemic_uncertainty": epistemic_uncertainty, |
Comment on lines
2102
to
2106
| 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 |
Comment on lines
+227
to
+231
| 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) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.