Repository navigation
Conversation
Adds HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM (default 1, preserves existing sequential behavior) controlling how many LLM batches a single consolidation op fires concurrently via asyncio.gather + Semaphore. Per-bank single-flight at the SQL claim level (ops_postgresql.claim_tasks) is preserved — one worker still owns one consolidation op per bank at a time, so this is intra-op parallelism only. Each parallel batch returns local stat deltas / consolidated_tags which the outer loop merges after gather completes, avoiding lost-update races on shared mutable state. Motivation: on backends without batch inference (telnyx, etc.) where each memory costs a separate ~30-60s LLM call, large re-consolidations (e.g. mission-change cutovers triggering 60k+ source rows to re-derive) are otherwise wall-clock-bound to weeks. With parallelism=8 the throughput becomes provider-bound rather than serial-call-bound. Defaults remain sequential to keep parity with prior releases; opt-in by setting the env var at the API container level.
Addresses review of the intra-op parallelism feature. - Make the tag group the unit of concurrency. Batches within a group share an observation scope and now run serially (the write path executes actions serially per scope); distinct groups run concurrently. Previously batches from one tag group could run in parallel and double-create or lose updates on the same observation rows — the deferred stat merge only guarded the in-memory counters, not the DB writes. - Restore early-exit on cancellation for the sequential default path (it had regressed to processing every remaining batch after the op was cancelled). - Return a _BatchDeltas dataclass instead of a raw dict. - Register consolidation_llm_parallelism in _CONFIGURABLE_FIELDS so it is per-bank overridable like consolidation_llm_batch_size; read it directly off config instead of via getattr. - Document the flag in configuration.md, including the per_tag/all_combinations observation-scope caveat. - Add test_consolidation_parallelism.py proving parallel == sequential (no lost-update / double-count) and that the flag is configurable.
4 of 5 tasks
Collaborator
nicoloboschi
added a commit
that referenced
this pull request
May 29, 2026
) (#1853) * feat(consolidation): scope-locked parallel LLM dispatch Adds opt-in HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM (default 1, sequential). Parallel groups acquire per-scope asyncio.Locks computed from each memory's observation_scopes setting, so two tag groups whose write-scope sets overlap serialise on the overlapping scope rather than racing on the same observation row. Locks acquired in tuple(sorted(scope)) order across all groups for deadlock-freedom. Covers combined / per_tag / all_combinations / explicit-list scopes uniformly with no operator opt-in. Refactor extracts the per-memory observation_scopes resolver into module-level helpers (_resolve_obs_tags_list, _resolve_write_scopes, _parse_observation_scopes, _scope_sort_key) so the dispatcher and the lock layer share one source of truth. Per-batch stats deltas now return as _BatchDeltas and merge serially after dispatch — no lost-update race on shared counters/tag set. * feat(consolidation): per-batch perf log + default parallelism=4 - Per-batch log uses a batch-local ConsolidationPerfLog so timings, llm_calls, and input_tokens reflect only that batch's work — no delta-from-shared-snapshot bleed under parallelism > 1. Local perf merges into the job-level perf at end-of-batch so the final flush still totals everything. - Restore the cumulative processed=N/total progress indicator. The counter increments + snapshots atomically between awaits in single-threaded asyncio, no lock needed. - Bump DEFAULT_CONSOLIDATION_LLM_PARALLELISM from 1 to 4 to match retain_max_concurrent and let combined-mode banks pick up the throughput win out of the box. Lock-on-overlap makes this safe by construction; per_tag / all_combinations banks degrade to serial automatically. - New regression test test_per_batch_log_line_attributes_only_own_work asserts per-batch log fields are isolated (llm calls / memories / created / timing) and cumulative processed indicator is monotonic. * chore: regenerate docs-skill + merge two alembic heads to unblock CI - skills/hindsight-docs/references/developer/configuration.md: regenerated via ./scripts/generate-docs-skill.sh to pick up the new HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM entry from the source configuration.md edited in the previous commit. - alembic/versions/mrgvchgraf01_*: empty merge revision unifying main's two open heads (b5a4c3e2f1d8 add_graph_maintenance_queue and b8c9d0e1f2a3 vchord_cosine_opclass). test_alembic_dag.py::test_single_head catches the divergence and recommends `alembic merge heads`; this is that. Pre-existing on main — only surfaced because this PR touches API code and trips the path-filtered test-api job. * ci: cap every job in test.yml at 30 minutes Adds timeout-minutes: 30 to all 60 jobs. Without it each job inherits GitHub Actions' 6-hour default, so a hung worker or a flaky LLM call can keep the whole suite "running" for hours before someone notices. 30 min is ~2x headroom over the slowest current job (test-api shards ~13 min, test-doc-examples ~14 min, test-python-client-oracle ~13 min). If a specific job legitimately needs more later, bump just that one. * chore: drop redundant alembic merge migration Main shipped its own merge revision c1d2e3f4a5b6 for the same two heads (b5a4c3e2f1d8 and b8c9d0e1f2a3) in #1854/#1857's neighbourhood, so my mrgvchgraf01 became redundant after rebase. Keeping only main's version to avoid a fresh divergent-heads situation. * test: bump pool_max_size from 5 to 30 in memory fixtures The 4 MemoryEngine fixtures in conftest were sized for sequential consolidation; with consolidation_llm_parallelism now defaulting to 4 (and other parallel knobs like retain_max_concurrent=4 already active), a pool of 5 connections can be exhausted when an HTTP integration test triggers multiple async retains that each fan consolidation across several concurrent tag groups. CI surfaced this as test_async_retain_parallel hanging on test-api shard 2 — 5 parallel retains × 4-way intra-op consolidation parallelism + the test's own polling HTTP calls all competed for 5 connections under xdist's worker concurrency. Bumping to 30 keeps tests bounded but matches a more realistic deployment pool size (default prod cap is 100) and removes the head-of-line stall. * test: bump pg0 max_connections to 300, pool to 15, fix configurable counter CI surfaced two real failures from the previous bump: - shard 2: tests/test_hierarchical_config.py::test_hierarchical_fields_categorization hardcoded `assert len(configurable) == 36`. Adding consolidation_llm_parallelism to _CONFIGURABLE_FIELDS made it 37. Bumped and added an explicit membership assertion so a future drop of the flag fails loudly. - shard 3: asyncpg.TooManyConnectionsError. With pool_max_size=30 and 8 xdist workers, peak demand was ~240 connections against postgres's default cap of 100. Two related changes: * EmbeddedPostgres now accepts a ``config: dict[str, str]`` and forwards it to Pg0 (which has been a documented Pg0 kwarg). The pg0_db_url fixture passes ``{"max_connections": "300"}`` so 8 workers × pool=15 fits comfortably. * Pool back to 15 (from 30 in the previous commit). 15 still accommodates default consolidation_llm_parallelism=4 + retain_max_concurrent=4 + the test's own queries without head-of-line stalls, but caps total connections at a sane fraction of the 300 max.
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.
What
Adds an opt-in
HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISMknob that lets one consolidation worker process multiple tag groups concurrently. Default stays1(sequential) — no behavior change unless you set it.Why
Consolidation is LLM-bound and chews through tag groups one at a time, which drags on banks with lots of memories. This fans a single worker out across distinct tag groups while staying HA-safe — batches within a group share an observation scope so they stay serial (no double-create / lost-update on the same rows), distinct groups touch disjoint scopes and run concurrently, per-bank single-flight is still enforced at the SQL claim level, and per-batch stat deltas merge after dispatch to dodge counter races.
Additional Information