feat(playbook): schedule bounded incremental aggregation - #405
Merged
Conversation
📝 WalkthroughWalkthroughThe change replaces direct background aggregation with durable, lease-based scheduling. It adds SQLite aggregation state, bounded incremental processing, invalidation handling, fenced reruns, application lifecycle wiring, configuration, documentation, and integration coverage. ChangesPlaybook aggregation pipeline
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
reflexio/server/services/playbook/components/aggregator.py (1)
957-974: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winEmit a failure event for the safety-cap abort.
This branch now raises instead of returning skip stats, but it still records
outcome="should_skip"onaggregation_gate_evaluated. The raise also happens before thetryblock at line 1084, so noaggregation_failedevent is emitted. An oversized rerun therefore reports as a skip in telemetry while the caller sees an exception, and the scheduler or admin route marks the claim failed.Record the abort as a failure so telemetry matches the control flow.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/playbook/components/aggregator.py` around lines 957 - 974, The safety-cap branch in the aggregation gate currently records outcome="should_skip" even though it raises and fails the claim. Update the record_usage_event call in this branch to emit the appropriate failure event/outcome, ensuring oversized reruns are represented as failures despite occurring before the try block.
🧹 Nitpick comments (15)
tests/server/services/storage/test_playbook_aggregation_state_integration.py (5)
380-383: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueExtract the sqlite-vec skip guard into a fixture.
The pattern
if not store._has_sqlite_vec: pytest.skip(...)is repeated at lines 205, 382, 426, 479, 592, 684, 768, and 838. It also reaches into a private attribute from test code. Replace it with one autouse-free helper fixture that builds the store and skips once.♻️ Proposed helper
`@pytest.fixture` def vec_store(tmp_path) -> SQLiteStorage: store = _store(tmp_path) if not store._has_sqlite_vec: pytest.skip("sqlite-vec is unavailable") return storeThen each vector test takes
vec_storeinstead oftmp_pathplus the inline guard.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/storage/test_playbook_aggregation_state_integration.py` around lines 380 - 383, Introduce a shared `vec_store` fixture that creates the storage via `_store(tmp_path)`, skips when sqlite-vec is unavailable, and returns the validated store. Update all affected vector tests, including `test_sqlite_vec_centroid_lookup_and_attachment`, to depend on `vec_store` and remove their `tmp_path` setup and inline `_has_sqlite_vec` guards.
342-377: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse keyword arguments for
PlaybookAggregationBacklogat line 188.Line 370 constructs
PlaybookAggregationBacklogwith keywords, but line 188 uses the positional formPlaybookAggregationBacklog(0, 0, 0). The positional call binds to declaration order. If a field is reordered or inserted beforeinvalidations, line 188 still compiles and silently means a different backlog. That test relies on the backlog being empty sopendingisFalse; a wrong binding would flip its meaning without an obvious failure.♻️ Proposed change at line 188
- backlog=PlaybookAggregationBacklog(0, 0, 0), + backlog=PlaybookAggregationBacklog( + undisposed=0, residual=0, invalidations=0 + ),🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/storage/test_playbook_aggregation_state_integration.py` around lines 342 - 377, Update the positional PlaybookAggregationBacklog construction in the affected test to use explicit keyword arguments for each zero-valued field, matching the keyword-based construction in test_dirty_cluster_keeps_aggregation_pending. Preserve the assertion that an empty backlog has pending set to False.
888-928: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument that line 903 relies on a database trigger.
Line 903 deletes the row with raw SQL and bypasses
delete_user_playbook. Line 927 still expects a("hard_delete", 1)invalidation. That row can only be produced by a SQL trigger, not by the Python method. The dependency is load-bearing but implicit.Add a short comment at line 903 stating that the raw delete exercises the trigger path. Without it, a later change to
delete_user_playbooklooks safe while this assertion quietly proves something different from what a reader expects.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/storage/test_playbook_aggregation_state_integration.py` around lines 888 - 928, Add a concise inline comment immediately before the raw SQL DELETE in test_rerun_snapshot_finishes_only_materialized_work, explicitly noting that the direct deletion intentionally exercises the database trigger path that creates the hard_delete invalidation. Leave the test behavior and assertions unchanged.
199-249: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove the repeated function-local imports to module scope.
from unittest.mock import MagicMockandimport jsonare imported inside the test body here and repeated intest_incremental_run_creates_once_then_attaches_without_replacement(lines 589, 596),test_incremental_run_commits_healthy_clusters_when_one_llm_outcome_fails(lines 681, 688), andtest_fenced_full_rerun_rebuilds_typed_cluster_state(lines 835, 842). Neither import is optional or cycle-breaking. Hoist both to the module header with the other imports.The fencing assertion itself is correct:
replacementbumps the fence paststale, so the run must abort before anyagent_playbooksrow is written.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/storage/test_playbook_aggregation_state_integration.py` around lines 199 - 249, Move the function-local imports of MagicMock and json used by test_stale_claim_cannot_commit_incremental_effect and the other named tests to the module-level import section, alongside the existing imports. Remove the duplicated in-function import statements while preserving all test behavior and the fencing assertions.
60-78: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMake the AST scan robust to module-scope calls and to
rglobordering.Two fragility points:
- Line 63 reads
self.function_stack[-1]. If a matching call ever appears at module scope, in a class body, or in a lambda, the visitor raisesIndexErrorinstead of failing the assertion with a readable diff.- Line 78 compares
cleanup_statuseswith an ordered list.Path.rglobdoes not guarantee a stable order across platforms. The assertion passes today only because exactly one element is collected. Sort the list or compare aCounterso a second call site does not produce order-dependent failures.♻️ Proposed hardening
if isinstance(node.func, ast.Attribute): if node.func.attr == "delete_all_user_playbooks_by_status": - call_sites.append((self.relative_path, self.function_stack[-1])) + enclosing = ( + self.function_stack[-1] if self.function_stack else "<module>" + ) + call_sites.append((self.relative_path, enclosing))- assert cleanup_statuses == ["ARCHIVED"] + assert sorted(cleanup_statuses) == ["ARCHIVED"]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/storage/test_playbook_aggregation_state_integration.py` around lines 60 - 78, Harden the AST visitor around CallVisitor.visit_Call: record a safe module/class/lambda context when function_stack is empty instead of indexing self.function_stack[-1], so unexpected matching calls produce an assertion diff rather than IndexError. Make the cleanup_statuses assertion order-independent by sorting the collected values before comparison, while preserving the existing expected statuses.reflexio/server/services/playbook/components/aggregator.py (2)
434-441: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winDecouple the dedup-context read from the remaining work budget.
limit=min(500, budget)shrinks the existing-playbook context as the budget shrinks. A near-exhausted budget (for examplebudget=4after invalidations and legacy adoption) gives the LLM only 4 existing playbooks for deduplication, so the aggregator can regenerate near-duplicates. The prompt is already bounded downstream by_select_relevant_existing_playbooksat 20 entries.Use a fixed context bound instead of the work budget.
♻️ Proposed change
existing_playbooks = self.storage.get_agent_playbooks( # type: ignore[attr-defined] - limit=min(500, budget), + limit=500, agent_version=self.agent_version,🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/playbook/components/aggregator.py` around lines 434 - 441, Update the existing-playbook read in the aggregator method to use a fixed limit of 500 rather than min(500, budget), keeping deduplication context independent of the remaining work budget and preserving the downstream 20-entry selection bound.
487-498: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueKey the replacement map by index instead of
id(outcome).
id(outcome)is only valid while every outcome object stays alive and unreplaced. The mapping breaks silently if a later change rebuilds or copies an outcome between the two loops. Index the outcomes list instead.♻️ Proposed change
- replacement_agent_ids_by_outcome = { - id(outcome): self.storage.get_playbook_aggregation_replacement_agent_ids( # type: ignore[attr-defined] + replacement_agent_ids_by_outcome = { + index: self.storage.get_playbook_aggregation_replacement_agent_ids( # type: ignore[attr-defined] self.agent_version, [ int(item.user_playbook_id) for item in outcome.source_cluster if item.user_playbook_id is not None ], ) - for outcome in outcomes + for index, outcome in enumerate(outcomes) if outcome.status == "generated" }Then iterate with
for index, outcome in enumerate(outcomes):and usereplacement_agent_ids_by_outcome[index].Also applies to: 580-580
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/playbook/components/aggregator.py` around lines 487 - 498, Update the replacement map comprehension in the outcome aggregation flow to key entries by the outcome’s list index rather than id(outcome), using enumerate(outcomes) and preserving the generated-status filter. Update the later lookup to use the corresponding index while iterating outcomes, including the logic around the replacement map access.reflexio/server/services/playbook/README.md (1)
38-41: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFix the data-flow indentation for
AgentPlaybook.
AgentPlaybookis produced byPlaybookAggregator, but line 41 is indented at the level ofPlaybookAggregationScheduler. The diagram now reads as if the durable signal writesAgentPlaybookdirectly.📝 Proposed diagram fix
-> Durable aggregation signal -> PlaybookAggregationScheduler (hourly cadence, bounded work) -> PlaybookAggregator (incremental clustering) - -> AgentPlaybook (aggregated insights) -> Storage + -> AgentPlaybook (aggregated insights) -> Storage🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/playbook/README.md` around lines 38 - 41, Adjust the README data-flow diagram indentation so AgentPlaybook (aggregated insights) -> Storage is nested under PlaybookAggregator, reflecting that PlaybookAggregator produces AgentPlaybook rather than the durable signal or scheduler.tests/server/services/playbook/test_playbook_aggregator.py (1)
1840-1860: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert that each outcome keeps its original source cluster.
The test checks only the statuses. The important invariant in
_generate_playbook_outcomes_with_source_clustersis thatsource_clusterholds the originalUserPlaybookobjects, not the prompt-preprocessed copies._run_incrementalderives item dispositions fromsource_cluster[*].user_playbook_id, so a regression there writes dispositions for the wrong rows.💚 Proposed assertion
+ cluster_a = [_raw(1)] + cluster_b = [_raw(2)] outcomes = agg._generate_playbook_outcomes_with_source_clusters( - {0: [_raw(1)], 1: [_raw(2)]}, [] + {0: cluster_a, 1: cluster_b}, [] ) assert [item.status for item in outcomes] == [ "semantic_null", "retryable_failure", ] + assert [item.source_cluster for item in outcomes] == [cluster_a, cluster_b]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/playbook/test_playbook_aggregator.py` around lines 1840 - 1860, Update test_batch_preserves_one_tagged_outcome_per_cluster to also assert that each returned outcome’s source_cluster contains the original UserPlaybook objects for its corresponding cluster, including their original user_playbook_id values, rather than prompt-preprocessed copies. Keep the existing status assertions and verify both cluster entries preserve their input objects.reflexio/lib/_generation.py (1)
95-110: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winImport the retry constants instead of duplicating them.
60and1repeat_RETRY_SECONDSand_BACKLOG_RETRY_SECONDSfromaggregation_scheduler, and they repeat again at both call sites in this function. The module already importsaggregation_min_interval_secondsfrom that module. Export and import the two retry constants so the retry policy has one definition.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/lib/_generation.py` around lines 95 - 110, Replace the duplicated retry values in the playbook aggregation claim handling with _RETRY_SECONDS and _BACKLOG_RETRY_SECONDS from aggregation_scheduler. Export those constants there, import them alongside aggregation_min_interval_seconds, and use them at both finish_playbook_aggregation_claim call sites while preserving the existing retry behavior.reflexio/server/services/playbook/aggregation_trigger.py (1)
81-89: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReuse the scheduler module logger instead of the string literal.
The module already imports from
reflexio.server.services.playbook.aggregation_schedulerat line 8. The literal logger name duplicates that path and goes stale without any failure if the module is renamed. Import theloggerobject from the scheduler module, or emit this event through the module-levelloggeron line 12.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/playbook/aggregation_trigger.py` around lines 81 - 89, Update the playbook aggregation progress logging to reuse the existing module-level logger from aggregation_scheduler instead of calling logging.getLogger with a duplicated string literal. Adjust the relevant import and emit the scheduled event through logger, preserving the current message and arguments.tests/server/services/playbook/test_aggregation_scheduler.py (2)
118-132: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd the positive case for the repair interval.
The test proves the second scan is suppressed. It does not prove the scan resumes after
_REPAIR_INTERVAL_SECONDS. Advance the clock past the interval and assert a secondrepair_playbook_aggregation_pending_statecall. Without it, a change that disables repair permanently still passes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/playbook/test_aggregation_scheduler.py` around lines 118 - 132, The test_scheduler_throttles_idle_repair_scans test only verifies suppression, not resumption. Advance the mocked aggregation_scheduler.time.monotonic value beyond _REPAIR_INTERVAL_SECONDS before a subsequent scheduler._run_context(context) call, then assert repair_playbook_aggregation_pending_state was called twice while preserving the existing claim call assertions.
122-122: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePatching
time.monotonicmutates the stdlib module globally.
aggregation_scheduler.timeis the stdlibtimemodule, so this replacestime.monotonicfor every thread in the process until the test ends. Background threads started by other code in the same process observe the frozen value. Drive the throttle through the scheduler state instead, for example by seedingscheduler._last_repair_at, or by injecting a clock intoPlaybookAggregationScheduler.reflexio/server/services/playbook/aggregation_scheduler.py (1)
199-217: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConsider backoff for repeated failures.
retry_after_secondsis the constant_RETRY_SECONDS(60). An agent version that fails persistently retries every 60 seconds without limit. Each attempt constructs an LLM client and runs aggregation work. Add exponential backoff based on a consecutive-failure count, or cap the retry rate for a version that keeps failing.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/playbook/aggregation_scheduler.py` around lines 199 - 217, The failure path in the aggregation scheduler retries persistent agent-version failures every fixed 60 seconds. Update the claim completion flow around `finish_playbook_aggregation_claim` to track consecutive failures and apply exponential backoff or an equivalent capped retry rate, while preserving the existing retry behavior for successful or non-repeated failures.reflexio/server/api.py (1)
518-541: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConstruction of
RequestContextblocks the startup event loop.
RequestContext(org_id=bootstrap_org_id)builds a configurator, a prompt manager, and a storage connection synchronously inside the asynclifespan. The neighbouring schedulers avoid this by passinglambda org_id: RequestContext(org_id=org_id)and letting the scheduler thread construct the context. Consider deferring construction the same way so startup does not perform storage I/O on the event loop.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/api.py` around lines 518 - 541, Defer RequestContext construction in the OSS scheduler startup path instead of instantiating it synchronously in lifespan. Update ensure_local_playbook_aggregation_scheduler usage to pass a callable such as the neighboring scheduler patterns, allowing the scheduler thread to create RequestContext for each org while preserving bootstrap_org_id behavior and existing exception logging.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@reflexio/lib/_generation.py`:
- Around line 71-90: The admin rerun claim created in the playbook aggregation
flow has no lease renewal and can expire during long runs. Reuse _LeaseHeartbeat
from aggregation_scheduler around PlaybookAggregator execution, passing the
claimed aggregation_claim and ensuring the heartbeat starts after claim
acquisition and stops when the rerun completes or fails, while preserving the
existing claim and fence handling.
- Around line 91-102: Protect the failure-path call to
storage.finish_playbook_aggregation_claim within the except block so any
exception from claim release cannot replace the original playbook_aggregator.run
error. Preserve the original exception by handling or suppressing cleanup
failures, then execute the existing bare raise to re-raise the aggregation
error.
In `@reflexio/server/services/playbook/aggregation_scheduler.py`:
- Around line 65-73: Update the lease-renewal loop in _run to catch any
exception from storage.renew_playbook_aggregation_claim, set _lost when renewal
fails or returns None, and return immediately so require_live() no longer treats
the lease as healthy after a renewal error.
- Around line 273-292: Update ensure_local_playbook_aggregation_scheduler and
the scheduler shutdown path so a stopped scheduler is removed from
_LOCAL_SCHEDULERS, or replaced when the incoming RequestContext differs. Ensure
a later app lifespan creates or uses a scheduler bound to the current context
rather than restarting one that retains the stale lambda-captured context.
In `@reflexio/server/services/playbook/components/aggregator.py`:
- Around line 2008-2020: The aggregation flow around the retryable-failure
outcome must enforce a configured maximum attempt count instead of retrying
indefinitely. Track and evaluate each member’s attempt count, transition members
to terminal_noop once the limit is reached, and record the terminal reason;
preserve retryable behavior and backoff for attempts below the limit. Use the
existing retryable-failure handling and member state symbols rather than
introducing a separate retry path.
In `@reflexio/server/services/storage/sqlite_storage/_base.py`:
- Around line 846-848: Ensure playbook aggregation does not access
playbook_aggregation_clusters_vec when _has_sqlite_vec is false: either make
supports_incremental_playbook_aggregation reflect extension availability or
guard every aggregation mixin vector path, including invalidation, rerun,
cleanup, legacy adoption, creation, and attachment. Preserve existing behavior
when sqlite-vec is available and prevent no-such-table errors otherwise.
In `@reflexio/server/services/storage/sqlite_storage/_lineage.py`:
- Around line 101-119: Update the candidate ID construction and invalidation
insert in the lineage flow to retain the parsed numeric value for entity_id,
rather than calling int(entity_id) again. Ensure the invalidation record uses
that parsed value when entity_id is numeric, while preserving source-ID-based
version lookup and avoiding ValueError when entity_id is non-numeric.
In `@reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py`:
- Around line 762-789: Replace the unbounded retry_rows fetch and Python min
scan in the residual cooldown calculation with a single SQL aggregate over
residual rows for the agent_version, computing each row’s capped exponential
retry deadline and returning the minimum remaining cooldown. Preserve the
current zero result when no residual rows exist or when attempt_count is
zero/last_attempt_at is null, and reuse the existing
idx_playbook_aggregation_item_residual-covered predicate.
- Around line 344-358: In the CAS failure branch of
finish_playbook_aggregation_claim, roll back self.conn before returning False
when the caller owns the transaction. Preserve the existing rowcount check and
successful-update behavior, and ensure the rollback is limited to caller-owned
transaction handling.
- Around line 989-1008: Update the vector search query in the surrounding
aggregation method to explicitly declare sqlite-vec’s cosine distance metric for
playbook_aggregation_clusters_vec, using the supported distance-metric
configuration for the MATCH parameter. Keep the existing similarity conversion
and _run_incremental cosine-threshold comparison consistent with that metric.
In `@tests/server/services/playbook/test_playbook_generation_service.py`:
- Around line 86-92: Rename the test function
test_inline_aggregation_default_path_does_not_inject_processor to reflect its
actual assertion that _trigger_playbook_aggregation schedules durable
aggregation with version "v1", such as
test_inline_aggregation_default_path_schedules_durably. Leave the test body
unchanged.
In
`@tests/server/services/storage/test_playbook_aggregation_state_integration.py`:
- Around line 122-132: Update test_org_claim_is_fenced_and_fence_is_monotonic to
make both scheduled playbook aggregation rows use the same next_attempt_at value
before claiming. Preserve the existing v2-then-v1 scheduling and assertions so
the test deterministically exercises the agent_version tie-break and fencing
behavior.
---
Outside diff comments:
In `@reflexio/server/services/playbook/components/aggregator.py`:
- Around line 957-974: The safety-cap branch in the aggregation gate currently
records outcome="should_skip" even though it raises and fails the claim. Update
the record_usage_event call in this branch to emit the appropriate failure
event/outcome, ensuring oversized reruns are represented as failures despite
occurring before the try block.
---
Nitpick comments:
In `@reflexio/lib/_generation.py`:
- Around line 95-110: Replace the duplicated retry values in the playbook
aggregation claim handling with _RETRY_SECONDS and _BACKLOG_RETRY_SECONDS from
aggregation_scheduler. Export those constants there, import them alongside
aggregation_min_interval_seconds, and use them at both
finish_playbook_aggregation_claim call sites while preserving the existing retry
behavior.
In `@reflexio/server/api.py`:
- Around line 518-541: Defer RequestContext construction in the OSS scheduler
startup path instead of instantiating it synchronously in lifespan. Update
ensure_local_playbook_aggregation_scheduler usage to pass a callable such as the
neighboring scheduler patterns, allowing the scheduler thread to create
RequestContext for each org while preserving bootstrap_org_id behavior and
existing exception logging.
In `@reflexio/server/services/playbook/aggregation_scheduler.py`:
- Around line 199-217: The failure path in the aggregation scheduler retries
persistent agent-version failures every fixed 60 seconds. Update the claim
completion flow around `finish_playbook_aggregation_claim` to track consecutive
failures and apply exponential backoff or an equivalent capped retry rate, while
preserving the existing retry behavior for successful or non-repeated failures.
In `@reflexio/server/services/playbook/aggregation_trigger.py`:
- Around line 81-89: Update the playbook aggregation progress logging to reuse
the existing module-level logger from aggregation_scheduler instead of calling
logging.getLogger with a duplicated string literal. Adjust the relevant import
and emit the scheduled event through logger, preserving the current message and
arguments.
In `@reflexio/server/services/playbook/components/aggregator.py`:
- Around line 434-441: Update the existing-playbook read in the aggregator
method to use a fixed limit of 500 rather than min(500, budget), keeping
deduplication context independent of the remaining work budget and preserving
the downstream 20-entry selection bound.
- Around line 487-498: Update the replacement map comprehension in the outcome
aggregation flow to key entries by the outcome’s list index rather than
id(outcome), using enumerate(outcomes) and preserving the generated-status
filter. Update the later lookup to use the corresponding index while iterating
outcomes, including the logic around the replacement map access.
In `@reflexio/server/services/playbook/README.md`:
- Around line 38-41: Adjust the README data-flow diagram indentation so
AgentPlaybook (aggregated insights) -> Storage is nested under
PlaybookAggregator, reflecting that PlaybookAggregator produces AgentPlaybook
rather than the durable signal or scheduler.
In `@tests/server/services/playbook/test_aggregation_scheduler.py`:
- Around line 118-132: The test_scheduler_throttles_idle_repair_scans test only
verifies suppression, not resumption. Advance the mocked
aggregation_scheduler.time.monotonic value beyond _REPAIR_INTERVAL_SECONDS
before a subsequent scheduler._run_context(context) call, then assert
repair_playbook_aggregation_pending_state was called twice while preserving the
existing claim call assertions.
In `@tests/server/services/playbook/test_playbook_aggregator.py`:
- Around line 1840-1860: Update
test_batch_preserves_one_tagged_outcome_per_cluster to also assert that each
returned outcome’s source_cluster contains the original UserPlaybook objects for
its corresponding cluster, including their original user_playbook_id values,
rather than prompt-preprocessed copies. Keep the existing status assertions and
verify both cluster entries preserve their input objects.
In
`@tests/server/services/storage/test_playbook_aggregation_state_integration.py`:
- Around line 380-383: Introduce a shared `vec_store` fixture that creates the
storage via `_store(tmp_path)`, skips when sqlite-vec is unavailable, and
returns the validated store. Update all affected vector tests, including
`test_sqlite_vec_centroid_lookup_and_attachment`, to depend on `vec_store` and
remove their `tmp_path` setup and inline `_has_sqlite_vec` guards.
- Around line 342-377: Update the positional PlaybookAggregationBacklog
construction in the affected test to use explicit keyword arguments for each
zero-valued field, matching the keyword-based construction in
test_dirty_cluster_keeps_aggregation_pending. Preserve the assertion that an
empty backlog has pending set to False.
- Around line 888-928: Add a concise inline comment immediately before the raw
SQL DELETE in test_rerun_snapshot_finishes_only_materialized_work, explicitly
noting that the direct deletion intentionally exercises the database trigger
path that creates the hard_delete invalidation. Leave the test behavior and
assertions unchanged.
- Around line 199-249: Move the function-local imports of MagicMock and json
used by test_stale_claim_cannot_commit_incremental_effect and the other named
tests to the module-level import section, alongside the existing imports. Remove
the duplicated in-function import statements while preserving all test behavior
and the fencing assertions.
- Around line 60-78: Harden the AST visitor around CallVisitor.visit_Call:
record a safe module/class/lambda context when function_stack is empty instead
of indexing self.function_stack[-1], so unexpected matching calls produce an
assertion diff rather than IndexError. Make the cleanup_statuses assertion
order-independent by sorting the collected values before comparison, while
preserving the existing expected statuses.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 39adcd79-aae5-4ca7-b4f3-a212b037f2a7
📒 Files selected for processing (29)
.env.examplereflexio/lib/_generation.pyreflexio/server/README.mdreflexio/server/api.pyreflexio/server/services/playbook/README.mdreflexio/server/services/playbook/aggregation_scheduler.pyreflexio/server/services/playbook/aggregation_trigger.pyreflexio/server/services/playbook/components/aggregator.pyreflexio/server/services/playbook/components/aggregator_clustering.pyreflexio/server/services/playbook/components/aggregator_prompt_formatting.pyreflexio/server/services/storage/sqlite_storage/__init__.pyreflexio/server/services/storage/sqlite_storage/_base.pyreflexio/server/services/storage/sqlite_storage/_lineage.pyreflexio/server/services/storage/sqlite_storage/playbook/__init__.pyreflexio/server/services/storage/sqlite_storage/playbook/_aggregation.pyreflexio/server/services/storage/sqlite_storage/playbook/_user.pyreflexio/server/services/storage/storage_base/__init__.pyreflexio/server/services/storage/storage_base/playbook/__init__.pyreflexio/server/services/storage/storage_base/playbook/_aggregation.pyreflexio/server/services/storage/storage_base/playbook/_user.pytests/lib/test_generation_unit.pytests/server/services/playbook/test_aggregation_lineage_integration.pytests/server/services/playbook/test_aggregation_scheduler.pytests/server/services/playbook/test_aggregation_soft_delete_integration.pytests/server/services/playbook/test_playbook_aggregator.pytests/server/services/playbook/test_playbook_generation_service.pytests/server/services/playbook_optimizer/test_gepa_user_playbook_publication.pytests/server/services/playbook_optimizer/test_judge_frozen_plan.pytests/server/services/storage/test_playbook_aggregation_state_integration.py
Contributor
Author
|
CodeRabbit follow-up is in Addressed: renewable admin leases; cleanup-error preservation; heartbeat failure fencing; stale local scheduler eviction; sqlite-vec degraded capability; numeric lineage handling; SQLite CAS rollback; SQL-only residual cooldown; cosine vec0 distance; oversized-rerun failure telemetry; deterministic/renamed/hardened tests; logger/README/index-key maintenance. Declined after verification:
Also verified the proposed SQLite hard-delete lineage change is unnecessary: the existing Validation after the follow-up:
A broad enterprise integration run was attempted but was not counted as green: it accumulated unrelated integration failures and then stalled without progress at 44%, so it was terminated. The aggregation-focused enterprise suite above completed cleanly. |
There was a problem hiding this comment.
🧹 Nitpick comments (1)
reflexio/server/services/playbook/components/aggregator.py (1)
483-498: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winBatch the replacement-agent-id lookup instead of querying per outcome.
replacement_agent_ids_by_outcomecallsget_playbook_aggregation_replacement_agent_idsonce per "generated" outcome. The only consumer,replaced_agent_ids.update(replacement_agent_ids_by_outcome[outcome_index])at line 580, flattens all results into one set.get_playbook_aggregation_replacement_agent_idsruns aDISTINCT ... WHERE user_playbook_id IN (...)query, so passing the union of member IDs from every generated outcome in a single call returns the same final set. Replace the per-outcome dict with one batched call before the loop.♻️ Proposed batching
- replacement_agent_ids_by_outcome = { - index: self.storage.get_playbook_aggregation_replacement_agent_ids( # type: ignore[attr-defined] - self.agent_version, - [ - int(item.user_playbook_id) - for item in outcome.source_cluster - if item.user_playbook_id is not None - ], - ) - for index, outcome in enumerate(outcomes) - if outcome.status == "generated" - } + generated_member_ids = sorted( + { + int(item.user_playbook_id) + for outcome in outcomes + if outcome.status == "generated" + for item in outcome.source_cluster + if item.user_playbook_id is not None + } + ) + replacement_agent_ids = set( + self.storage.get_playbook_aggregation_replacement_agent_ids( # type: ignore[attr-defined] + self.agent_version, generated_member_ids + ) + )- replaced_agent_ids.update( - replacement_agent_ids_by_outcome[outcome_index] - ) + replaced_agent_ids.update(replacement_agent_ids)Also applies to: 580-582
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/playbook/components/aggregator.py` around lines 483 - 498, Replace the per-outcome replacement_agent_ids_by_outcome dictionary in the aggregation flow with one call to get_playbook_aggregation_replacement_agent_ids using the union of user_playbook_id values from all generated outcomes. Update the consumer that currently calls replaced_agent_ids.update(replacement_agent_ids_by_outcome[outcome_index]) to use the single batched result while preserving the final set contents.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@reflexio/server/services/playbook/components/aggregator.py`:
- Around line 483-498: Replace the per-outcome replacement_agent_ids_by_outcome
dictionary in the aggregation flow with one call to
get_playbook_aggregation_replacement_agent_ids using the union of
user_playbook_id values from all generated outcomes. Update the consumer that
currently calls
replaced_agent_ids.update(replacement_agent_ids_by_outcome[outcome_index]) to
use the single batched result while preserving the final set contents.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: a7455b93-3cf9-4ae0-98fd-3cded4f6d36b
📒 Files selected for processing (14)
reflexio/lib/_generation.pyreflexio/server/services/playbook/README.mdreflexio/server/services/playbook/aggregation_scheduler.pyreflexio/server/services/playbook/aggregation_trigger.pyreflexio/server/services/playbook/components/aggregator.pyreflexio/server/services/storage/sqlite_storage/_base.pyreflexio/server/services/storage/sqlite_storage/_lineage.pyreflexio/server/services/storage/sqlite_storage/playbook/_aggregation.pyreflexio/server/services/storage/sqlite_storage/playbook/_user.pytests/lib/test_generation_unit.pytests/server/services/playbook/test_aggregation_scheduler.pytests/server/services/playbook/test_playbook_aggregator.pytests/server/services/playbook/test_playbook_generation_service.pytests/server/services/storage/test_playbook_aggregation_state_integration.py
🚧 Files skipped from review as they are similar to previous changes (7)
- reflexio/server/services/storage/sqlite_storage/_lineage.py
- reflexio/server/services/storage/sqlite_storage/_base.py
- reflexio/server/services/playbook/README.md
- reflexio/server/services/playbook/aggregation_trigger.py
- reflexio/server/services/storage/sqlite_storage/playbook/_user.py
- reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py
- tests/server/services/playbook/test_playbook_aggregator.py
yyiilluu
force-pushed
the
codex/review-deduplication-repair-plan
branch
from
August 3, 2026 07:22
e60bc7b to
6a780fb
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
tests/server/services/playbook/test_aggregation_scheduler.py (1)
154-181: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winClear the global scheduler registry after the test too.
The test clears
aggregation_scheduler._LOCAL_SCHEDULERSat line 156 but leaves two schedulers registered when it finishes. Later tests in the same session then observe module-level state created here, which makes failures order-dependent.Register a cleanup that clears the registry again.
♻️ Proposed change
def test_stopped_local_scheduler_drops_captured_context(monkeypatch) -> None: monkeypatch.delenv("DEPLOYMENT_MODE", raising=False) aggregation_scheduler._LOCAL_SCHEDULERS.clear() + monkeypatch.setattr( + aggregation_scheduler, + "_LOCAL_SCHEDULERS", + type(aggregation_scheduler._LOCAL_SCHEDULERS)(), + )An
addfinalizerthat callsaggregation_scheduler._LOCAL_SCHEDULERS.clear()works as well.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/playbook/test_aggregation_scheduler.py` around lines 154 - 181, Update test_stopped_local_scheduler_drops_captured_context to register teardown cleanup that clears aggregation_scheduler._LOCAL_SCHEDULERS after the test, while preserving the existing pre-test clear and assertions.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@reflexio/server/services/storage/sqlite_storage/_lineage.py`:
- Around line 112-132: Update the merge invalidation logic around the
agent_version lookup to handle all candidate versions consistently: either
validate and reject mixed agent_version values before writing, or iterate over
each distinct non-empty version and create corresponding invalidation and
aggregation-state records. Do not retain the current LIMIT 1 behavior that
processes only one version.
In `@reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py`:
- Around line 93-96: Update the AGGREGATION_DDL handled by
init_playbook_aggregation_tables so the pending-invalidation index is created
with IF NOT EXISTS and is not dropped on every connection. Move the existing
DROP INDEX into the one-time schema-migration path only if required to replace
the prior definition, preserving the partial-index predicate and columns.
- Around line 991-1011: Update the candidate lookup around the ANN subquery in
the aggregation matching flow so compatibility filters for agent_version,
embedding_model, embedding_dimension, and active state are applied before
exhausting candidate_limit. Over-fetch ANN results or retry with an expanded
limit until compatible clusters can be considered, while preserving the existing
nearest-match ordering and result mapping in PlaybookAggregationClusterMatch.
---
Nitpick comments:
In `@tests/server/services/playbook/test_aggregation_scheduler.py`:
- Around line 154-181: Update
test_stopped_local_scheduler_drops_captured_context to register teardown cleanup
that clears aggregation_scheduler._LOCAL_SCHEDULERS after the test, while
preserving the existing pre-test clear and assertions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 22a064e2-72f5-46f3-b1bb-8b88800e47ab
📒 Files selected for processing (31)
.env.examplereflexio/lib/_generation.pyreflexio/server/README.mdreflexio/server/api.pyreflexio/server/services/README.mdreflexio/server/services/playbook/README.mdreflexio/server/services/playbook/aggregation_scheduler.pyreflexio/server/services/playbook/aggregation_trigger.pyreflexio/server/services/playbook/components/aggregator.pyreflexio/server/services/playbook/components/aggregator_clustering.pyreflexio/server/services/playbook/components/aggregator_prompt_formatting.pyreflexio/server/services/storage/sqlite_storage/__init__.pyreflexio/server/services/storage/sqlite_storage/_base.pyreflexio/server/services/storage/sqlite_storage/_lineage.pyreflexio/server/services/storage/sqlite_storage/playbook/__init__.pyreflexio/server/services/storage/sqlite_storage/playbook/_aggregation.pyreflexio/server/services/storage/sqlite_storage/playbook/_user.pyreflexio/server/services/storage/storage_base/__init__.pyreflexio/server/services/storage/storage_base/playbook/__init__.pyreflexio/server/services/storage/storage_base/playbook/_aggregation.pyreflexio/server/services/storage/storage_base/playbook/_user.pytests/lib/test_generation_unit.pytests/server/services/durable_learning/test_compute_persist_split.pytests/server/services/playbook/test_aggregation_lineage_integration.pytests/server/services/playbook/test_aggregation_scheduler.pytests/server/services/playbook/test_aggregation_soft_delete_integration.pytests/server/services/playbook/test_playbook_aggregator.pytests/server/services/playbook/test_playbook_generation_service.pytests/server/services/playbook_optimizer/test_gepa_user_playbook_publication.pytests/server/services/playbook_optimizer/test_judge_frozen_plan.pytests/server/services/storage/test_playbook_aggregation_state_integration.py
🚧 Files skipped from review as they are similar to previous changes (22)
- reflexio/server/services/storage/storage_base/playbook/_user.py
- reflexio/server/services/storage/sqlite_storage/playbook/init.py
- reflexio/server/services/storage/sqlite_storage/init.py
- reflexio/server/services/storage/storage_base/init.py
- reflexio/server/services/playbook/components/aggregator_clustering.py
- tests/server/services/playbook/test_aggregation_lineage_integration.py
- tests/server/services/playbook_optimizer/test_gepa_user_playbook_publication.py
- reflexio/server/services/storage/sqlite_storage/_base.py
- reflexio/lib/_generation.py
- tests/server/services/playbook/test_aggregation_soft_delete_integration.py
- tests/lib/test_generation_unit.py
- reflexio/server/services/playbook/components/aggregator_prompt_formatting.py
- reflexio/server/services/storage/storage_base/playbook/init.py
- tests/server/services/playbook_optimizer/test_judge_frozen_plan.py
- reflexio/server/services/storage/sqlite_storage/playbook/_user.py
- tests/server/services/playbook/test_playbook_generation_service.py
- reflexio/server/api.py
- reflexio/server/services/playbook/aggregation_scheduler.py
- reflexio/server/services/playbook/aggregation_trigger.py
- reflexio/server/services/storage/storage_base/playbook/_aggregation.py
- tests/server/services/playbook/test_playbook_aggregator.py
- reflexio/server/services/playbook/components/aggregator.py
Persist aggregation work, attach new playbooks to same-version centroids, and cluster residuals in bounded hourly runs.
Partition lifecycle invalidations by agent version, keep SQLite aggregation index initialization idempotent, and filter vec0 candidates before the ANN limit.
Avoid local daylight-saving transitions changing fixed profile expiration intervals.
yyiilluu
force-pushed
the
codex/review-deduplication-repair-plan
branch
from
August 3, 2026 17:19
6a780fb to
9061ee0
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
tests/server/services/playbook/test_aggregation_scheduler.py (2)
141-151: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBound the heartbeat loop so a regression fails instead of hanging.
Line 146 makes
heartbeat._stop.waitalways returnFalse._runtherefore exits only becauserenew_playbook_aggregation_claimraises and the loop breaks. If a future change makes_runcatch the renewal error and continue,waitnever returnsTrueand this test blocks forever rather than failing.Return
Trueafter the first iteration so the loop always terminates.♻️ Proposed fix
- heartbeat._stop.wait = MagicMock(return_value=False) + heartbeat._stop.wait = MagicMock(side_effect=[False, True])🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/playbook/test_aggregation_scheduler.py` around lines 141 - 151, Update test_lease_heartbeat_marks_renewal_exception_as_lost so the mocked heartbeat._stop.wait returns False for the first call and True thereafter, ensuring AggregationLeaseHeartbeat._run terminates even if renewal errors are caught and retried.
154-181: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRestore
_LOCAL_SCHEDULERSafter the test.Line 156 clears the module-level
_LOCAL_SCHEDULERSregistry, and the test then inserts two entries throughensure_local_playbook_aggregation_scheduler. The test never removes them.monkeypatchreverts the patchedstartandis_runningmethods, but it does not revert the dictionary. A later test that callsensure_local_playbook_aggregation_schedulercan therefore receive this test's leftover scheduler instead of creating its own.Clear the registry in a fixture so cleanup runs even when the test fails.
♻️ Proposed fix
+@pytest.fixture +def clean_local_schedulers(): + aggregation_scheduler._LOCAL_SCHEDULERS.clear() + yield + aggregation_scheduler._LOCAL_SCHEDULERS.clear() + + -def test_stopped_local_scheduler_drops_captured_context(monkeypatch) -> None: +def test_stopped_local_scheduler_drops_captured_context( + monkeypatch, clean_local_schedulers +) -> None: monkeypatch.delenv("DEPLOYMENT_MODE", raising=False) - aggregation_scheduler._LOCAL_SCHEDULERS.clear() monkeypatch.setattr(🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/playbook/test_aggregation_scheduler.py` around lines 154 - 181, Ensure test_local_playbook_aggregation_scheduler cleanup restores the module-level _LOCAL_SCHEDULERS registry after the test, using a fixture with teardown so cleanup runs even on failure. Keep the existing setup and assertions unchanged while preventing ensure_local_playbook_aggregation_scheduler state from leaking into later tests.tests/server/services/storage/test_playbook_aggregation_state_integration.py (1)
455-462: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAvoid asserting exact SQL text.
Line 462 matches the literal fragment
"COALESCE(MAX(0, MIN(CASE". Any whitespace or expression reordering insideget_playbook_aggregation_backlogbreaks this test without changing behavior. The intent is to prove the cooldown stays a single SQL aggregate.Assert the observable property instead: no per-item row read occurs. For example, assert that no traced statement selects
attempt_countorlast_attempt_atas a bare column list fromplaybook_aggregation_item, or assert the traced statement count stays constant as the residual set grows.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/storage/test_playbook_aggregation_state_integration.py` around lines 455 - 462, Update the test around get_playbook_aggregation_backlog to remove the exact SQL fragment assertion and verify the observable single-aggregate behavior instead. Use the captured statements to assert no per-item row read selects bare attempt_count or last_attempt_at columns from playbook_aggregation_item, or otherwise confirm statement count does not grow with the residual set; retain the backlog range assertion.reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py (1)
989-1013: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConsider reducing per-candidate KNN round trips.
Lines 990-1008 issue one KNN statement for each candidate. The caller's budget can reach
REFLEXIO_MAX_CLUSTERING_PLAYBOOKS(default 20,000) rows per scheduled unit, so one unit can execute up to 20,000 separate vector queries while holding_lock. sqlite-vec cannot accept several query vectors in oneMATCH, so batching is not possible directly. Two options reduce the cost:
- Prepare the statement once and reuse it, instead of re-parsing the same SQL for each candidate.
- Cap the number of centroid-match probes per unit below the full row budget, and leave the remainder residual for a later unit.
Neither change is required for correctness.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py` around lines 989 - 1013, Reduce per-candidate KNN overhead in the candidate-matching loop by reusing a prepared SQLite statement for the repeated query instead of calling self.conn.execute with the SQL text on every iteration. Preserve the existing parameters, ordering, limit, lock scope, and matches behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py`:
- Around line 519-545: Update the completion and activation logic in the
aggregation update flow so clusters with no member IDs cannot be persisted as
active. When the cluster is empty, force complete to false (or skip activation),
keeping its state rebuilding and avoiding active records with a NULL centroid
and missing vector-index entry.
In `@tests/server/services/profile/test_profile_generation_service_utils.py`:
- Around line 268-286: Update
test_calculate_expiration_timestamp_ignores_local_dst so the final tzset()
restoration runs in a finally block surrounding the temporary TZ context and
calculate_expiration_timestamp call. Preserve the existing environment cleanup
and assertion while ensuring the process timezone is restored even if
calculation raises.
---
Nitpick comments:
In `@reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py`:
- Around line 989-1013: Reduce per-candidate KNN overhead in the
candidate-matching loop by reusing a prepared SQLite statement for the repeated
query instead of calling self.conn.execute with the SQL text on every iteration.
Preserve the existing parameters, ordering, limit, lock scope, and matches
behavior.
In `@tests/server/services/playbook/test_aggregation_scheduler.py`:
- Around line 141-151: Update
test_lease_heartbeat_marks_renewal_exception_as_lost so the mocked
heartbeat._stop.wait returns False for the first call and True thereafter,
ensuring AggregationLeaseHeartbeat._run terminates even if renewal errors are
caught and retried.
- Around line 154-181: Ensure test_local_playbook_aggregation_scheduler cleanup
restores the module-level _LOCAL_SCHEDULERS registry after the test, using a
fixture with teardown so cleanup runs even on failure. Keep the existing setup
and assertions unchanged while preventing
ensure_local_playbook_aggregation_scheduler state from leaking into later tests.
In
`@tests/server/services/storage/test_playbook_aggregation_state_integration.py`:
- Around line 455-462: Update the test around get_playbook_aggregation_backlog
to remove the exact SQL fragment assertion and verify the observable
single-aggregate behavior instead. Use the captured statements to assert no
per-item row read selects bare attempt_count or last_attempt_at columns from
playbook_aggregation_item, or otherwise confirm statement count does not grow
with the residual set; retain the backlog range assertion.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 385e147e-8a16-4e3b-b0f1-8bfa77d0b589
📒 Files selected for processing (33)
.env.examplereflexio/lib/_generation.pyreflexio/server/README.mdreflexio/server/api.pyreflexio/server/services/README.mdreflexio/server/services/playbook/README.mdreflexio/server/services/playbook/aggregation_scheduler.pyreflexio/server/services/playbook/aggregation_trigger.pyreflexio/server/services/playbook/components/aggregator.pyreflexio/server/services/playbook/components/aggregator_clustering.pyreflexio/server/services/playbook/components/aggregator_prompt_formatting.pyreflexio/server/services/profile/profile_generation_service_utils.pyreflexio/server/services/storage/sqlite_storage/__init__.pyreflexio/server/services/storage/sqlite_storage/_base.pyreflexio/server/services/storage/sqlite_storage/_lineage.pyreflexio/server/services/storage/sqlite_storage/playbook/__init__.pyreflexio/server/services/storage/sqlite_storage/playbook/_aggregation.pyreflexio/server/services/storage/sqlite_storage/playbook/_user.pyreflexio/server/services/storage/storage_base/__init__.pyreflexio/server/services/storage/storage_base/playbook/__init__.pyreflexio/server/services/storage/storage_base/playbook/_aggregation.pyreflexio/server/services/storage/storage_base/playbook/_user.pytests/lib/test_generation_unit.pytests/server/services/durable_learning/test_compute_persist_split.pytests/server/services/playbook/test_aggregation_lineage_integration.pytests/server/services/playbook/test_aggregation_scheduler.pytests/server/services/playbook/test_aggregation_soft_delete_integration.pytests/server/services/playbook/test_playbook_aggregator.pytests/server/services/playbook/test_playbook_generation_service.pytests/server/services/playbook_optimizer/test_gepa_user_playbook_publication.pytests/server/services/playbook_optimizer/test_judge_frozen_plan.pytests/server/services/profile/test_profile_generation_service_utils.pytests/server/services/storage/test_playbook_aggregation_state_integration.py
🚧 Files skipped from review as they are similar to previous changes (25)
- reflexio/server/services/storage/storage_base/playbook/_user.py
- tests/server/services/playbook/test_playbook_generation_service.py
- tests/server/services/playbook/test_aggregation_lineage_integration.py
- reflexio/server/services/storage/sqlite_storage/playbook/init.py
- tests/server/services/playbook/test_aggregation_soft_delete_integration.py
- reflexio/server/services/storage/storage_base/init.py
- reflexio/server/services/storage/sqlite_storage/init.py
- tests/server/services/playbook_optimizer/test_judge_frozen_plan.py
- tests/server/services/durable_learning/test_compute_persist_split.py
- reflexio/server/services/playbook/components/aggregator_clustering.py
- reflexio/server/services/storage/storage_base/playbook/init.py
- tests/server/services/playbook_optimizer/test_gepa_user_playbook_publication.py
- reflexio/server/api.py
- reflexio/server/services/playbook/aggregation_trigger.py
- reflexio/server/services/playbook/components/aggregator_prompt_formatting.py
- reflexio/server/services/storage/sqlite_storage/playbook/_user.py
- reflexio/server/services/README.md
- tests/server/services/playbook/test_playbook_aggregator.py
- reflexio/server/services/storage/storage_base/playbook/_aggregation.py
- reflexio/lib/_generation.py
- reflexio/server/services/storage/sqlite_storage/_base.py
- reflexio/server/README.md
- tests/lib/test_generation_unit.py
- reflexio/server/services/playbook/aggregation_scheduler.py
- reflexio/server/services/playbook/components/aggregator.py
yyiilluu
added a commit
that referenced
this pull request
Aug 4, 2026
## Summary - Revert #407 and restore session outcomes, governance erasure, billing, and search behavior to the pre-open-world-evidence contracts. - Also revert #408 and #409 because their finalization-receipt and exposure-retention changes depend entirely on APIs introduced by #407. - Preserve the independent incremental aggregation work from #405 and #410. - Address every valid CodeRabbit finding, including SQLite downgrade compatibility and retry-safe metering. - Fix callback drop-rate anomaly emission on hosts with less than one hour of monotonic uptime, discovered by the full validation run. ## Changes ### Evidence foundation rollback - Remove search-exposure recording and session-outcome identity helpers. - Restore the prior session outcome schemas, client surface, and SQLite/storage contracts. - Restore the prior governance erase/claim flow and retention behavior. - Restore the prior resumable extraction and learning-billing behavior. ### Dependent follow-ups - Remove receipt-winner finalization behavior from #408. - Remove the exposure ownership and protected-retention behavior from #409. ### Review follow-ups - Rebuild #407-era SQLite `session_outcomes` tables into the restored schema, preserving `success`/`failure` rows, backfilling governance subject references, and explicitly dropping unrepresentable `unknown` outcomes with a warning. - Restore the SQLite 3.35 minimum required by existing `RETURNING` and `DROP COLUMN` usage. - Make outcome erasure resilient to governance-secret rotation and return a stable `session_outcomes` deletion count. - Acquire SQLite governance write locks before state checks, serialize idempotent purge begin/prepare flows across connections, and roll back failed target writes so SQLite cannot retain a stale writer transaction.\n- Reject legacy session-outcome schemas with empty governance-subject defaults and rebuild them with derived subject references.\n- Make synchronous playbook/profile persistence atomic while keeping scheduler dispatch strictly post-commit. - Meter resumable extraction from persisted survivors only, use retry-stable fallback keys, and emit learning billing from incremental aggregation. - Treat post-persist optimization and aggregation scheduling failures as best-effort side effects. ### Validation follow-up - Represent the callback executor's last anomaly time with an explicit unset sentinel so the first threshold crossing is never suppressed by low system uptime. ## Test Plan - `uv run ruff check reflexio tests` - `uv run ruff format --check reflexio tests` - Pyright on all 23 staged Python files: 0 errors, 0 warnings - Latest affected review files: 301 passed - OSS non-E2E suite: 5,535 passed, 73 skipped, 6 subtests passed - OSS E2E suite: 47 passed, 51 skipped - `npm --prefix docs run lint`: 0 errors (3 existing warnings) - `cd docs && npx tsc --noEmit` - `python -c "import reflexio"` Reverts `85a4b2255a96ef2a5b50f4cbe7c10758439e76b3`, plus dependent follow-ups `785a9e053ff771f40704bb7b0b5bbbe36048806a` and `eb88f44fd3b53457b76e8500ac1a30ba7d4ab16e`. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **Changes** - Session outcomes now support only success or failure, with simpler responses and retry behavior. - Governance data erasure workflows have streamlined retry and completion handling, including session-outcome removal. - Search exposure event recording has been removed; search results and metering remain available. - Learning-generation billing supports durable per-record tracking, retry-stable keys, and count-based fallback. - Scheduler failures during playbook processing are logged without preventing other scheduled actions. - **Documentation** - Quick Start prerequisites now list Node.js without the previous SQLite verification step. - Billing and extraction guidance has been updated. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
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
1 participant
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.
Summary
REFLEXIO_MAX_CLUSTERING_PLAYBOOKSby processing a bounded row budget rather than treating it as a corpus ceiling.Changes
Scheduling and execution
REFLEXIO_AGGREGATION_MIN_INTERVAL_SECONDS(default: one hour).Incremental aggregation
REFLEXIO_MAX_CLUSTERING_PLAYBOOKS(default: 20,000) as the maximum rows admitted to one scheduled unit of work, including intake and invalidation repair; it is also the fail-before-mutation cap for an administrative full rerun.Durable storage
Documentation and compatibility
Flow
flowchart LR A["User playbook committed"] --> B["Durable state due now"] B --> C["Scheduler claim"] C --> D["Bounded intake for one agent version"] D --> E{"Compatible centroid match?"} E -->|Yes| F["Attach and update centroid"] E -->|No| G["Durable residual"] G --> H["Bounded residual clustering"] H --> I["Generate agent playbook"] F --> J["Atomic fenced commit"] I --> J J --> K{"Backlog remains?"} K -->|Yes| C K -->|No| L["One-hour minimum before next drained run"]Test Plan
uv run ruff checkanduv run ruff format --checkpassed across the complete OSS source and test tree (879 files).Summary by CodeRabbit
New Features
Documentation
Bug Fixes
Tests