feat(playbook): refresh incremental cluster centroids

[yyiilluu]

Summary

• Keep hourly aggregation bounded as the user-playbook corpus grows: discover at most the newest 20,000 unclustered playbooks and never fall back to a full-corpus recluster.
• Refresh an existing cluster from its current agent playbook plus only the newly attached playbooks from the current run, then use the replacement agent playbook embedding as the centroid for future matching.
• Rebuild invalidated clusters from at most the newest 100 retained members when a source is edited, archived, or deleted.
• Preserve durable, retryable progress across scheduler runs while separating invalidation work from model-generation work.

Changes

Aggregation pipeline

• Adds explicit delta-refresh and bounded-repair inputs to the incremental aggregator.
• Supersedes the previous agent playbook atomically after generating and embedding its replacement.
• Keeps partial LLM failures isolated to their retryable members instead of discarding healthy outcomes.
• Bounds residual discovery, prompt grouping, and shared prompt context to avoid repeated quadratic or full-corpus work.

Scheduling and storage

• Throttles idle repair scans and retains pending work when vector search is unavailable.
• Adds durable cluster retry timing, member selection, cluster replacement, and lifecycle-retirement contracts.
• Implements the contracts and centroid index maintenance for SQLite, including create scheduling without redundant invalidation rows.

CodeRabbit follow-up

• Makes each cluster mutation its own atomic scope so a stale fence cannot roll back healthy peer clusters.
• Recovers missing or invalid legacy agent centroids by re-embedding the current agent playbook.
• Hardens SQLite centroid migration validation and adds bounded orphan-vector repair plus the cluster-to-agent partial index.
• Makes rebuild disposition transitions atomic when called outside an existing transaction.
• Removes the obsolete legacy vector_sum accumulation and expands both refresh and new-cluster prompt-bound coverage.

Documentation and tests

• Updates the server and playbook code maps with the hourly delta-refresh and bounded-repair invariants.
• Expands scheduler, aggregator, and SQLite integration coverage for retries, discovery limits, invalidation, cluster retirement, stale fences, legacy centroid recovery, and migration repair.

Diagrams

The hourly run advances clusters without reading their complete membership history.

flowchart LR
    A["Newest unclustered playbooks<br/>up to 20,000"] --> B{"Nearest active centroid?"}
    B -->|"match"| C["Current agent playbook<br/>+ this run's new members"]
    B -->|"no match"| D["Create bounded new clusters"]
    C --> E["Generate replacement agent playbook"]
    E --> F["Embed replacement text"]
    F --> G["Atomically replace agent playbook<br/>and centroid"]
    H["Edited, archived, or deleted source"] --> I["Newest retained members<br/>up to 100"]
    I --> E

Loading

Test Plan

• uv run ruff check and uv run ruff format on every changed Python file.
• uv run pyright on every changed Python file: 0 errors.
• Focused aggregation/scheduler/SQLite integration suite after review fixes: 149 passed.
• OSS unit and integration marker suite after review fixes: 1,287 passed, 64 skipped.
• Full OSS suite before the review follow-up: 5,677 passed, 73 skipped, 6 subtests passed.
• OSS end-to-end suite before the review follow-up: 47 passed, 51 skipped.

Summary by CodeRabbit

• Improvements

  • Playbook aggregation now processes updates incrementally, reducing unnecessary reprocessing.
  • Large backlogs are handled in bounded batches, with continued processing scheduled automatically.
  • Invalidated or changed playbooks are repaired more reliably, including retry handling for temporary failures.
  • Aggregated playbooks preserve valid existing guidance while incorporating newly matched information.
  • Clustering and similarity results are more consistent after updates, replacements, and rebuilds.

• Documentation

  • Added guidance covering aggregation scheduling, incremental refreshes, retries, invalidation handling, and processing limits.

[coderabbitai]

📝 Walkthrough

Walkthrough

The playbook aggregation pipeline now drains invalidations in fixed batches, performs bounded incremental refreshes and rebuilds, stores canonical agent centroids, and tracks retry and intake state. Tests and documentation cover the updated scheduling, persistence, prompt, and lifecycle behavior.

Changes

Playbook aggregation lifecycle

Layer / File(s)	Summary
Aggregation contracts and durable state reflexio/server/services/storage/storage_base/playbook/*, reflexio/server/services/storage/sqlite_storage/playbook/*, reflexio/server/services/storage/sqlite_storage/_base.py, reflexio/server/services/storage/sqlite_storage/_lineage.py	Storage contracts and SQLite state now track intake floors, rebuild retries, invalidation targets, bounded samples, and canonical embeddings.
Bounded aggregation and refresh flow reflexio/server/services/playbook/aggregation_scheduler.py, reflexio/server/services/playbook/components/aggregator.py, reflexio/server/prompt/prompt_bank/playbook_aggregation/*	The scheduler drains invalidations before LLM work. PlaybookAggregator performs bounded refresh, rebuild, residual, and new-cluster processing with canonical-playbook context.
Cluster rebuild and centroid persistence reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py	Cluster rebuilds now support sampling, retry deferral, completion, discard, orphan cleanup, guarded agent replacement, and canonical centroid updates.
Documentation and behavioral validation reflexio/server/README.md, reflexio/server/services/playbook/README.md, tests/server/services/playbook/*, tests/server/services/storage/test_playbook_aggregation_state_integration.py	Documentation and tests cover hourly coalescing, bounded inputs, invalidation draining, incremental replacement, rebuild retries, and cluster cleanup.

Estimated code review effort: 5 (Critical) | ~120 minutes

Sequence Diagram(s)

sequenceDiagram
  participant AggregationScheduler
  participant SQLiteStorage
  participant PlaybookAggregator
  participant LLM
  AggregationScheduler->>SQLiteStorage: Fetch up to 101 invalidations
  AggregationScheduler->>SQLiteStorage: Apply up to 100 invalidations
  AggregationScheduler->>PlaybookAggregator: Run with residual_batch_limit
  PlaybookAggregator->>SQLiteStorage: Load bounded aggregation work
  PlaybookAggregator->>LLM: Generate refresh and rebuild outcomes
  PlaybookAggregator->>SQLiteStorage: Persist agents, centroids, attachments, and residuals

Loading

Possibly related PRs

• ReflexioAI/reflexio#405: Extends the durable incremental aggregation flow introduced by this PR.
• ReflexioAI/reflexio#393: Provides earlier pagination and clustering foundations used by the updated aggregation flow.
• ReflexioAI/reflexio#241: Relates to incremental generation, replacement, and null-handling behavior.

Suggested reviewers: guangyu-reflexio

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 28.38% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly describes the primary change: refreshing cluster centroids during incremental playbook aggregation.

✨ Finishing Touches 💡 1

📝 Generate docstrings 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch codex/agent-playbook-centroid-refresh

---

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 5

🧹 Nitpick comments (5)

reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py (2)

144-161: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

Add an index on playbook_aggregation_cluster(agent_playbook_id).

Each new trigger runs four statements that filter playbook_aggregation_cluster by agent_playbook_id. The declared indexes on that table cover cluster_id (primary key), index_rowid (unique), (agent_version, state, cluster_id), and (agent_version, rebuild_next_attempt_at, cluster_id). None of them serve agent_playbook_id, so every statement performs a full table scan.

The delete trigger has no WHEN guard, so it fires per row. delete_all_agent_playbooks and delete_agent_playbooks_by_ids delete in one statement but pay the scan cost per deleted id, which makes the cost O(deleted agent playbooks × clusters). The update trigger adds the same cost to routine content, trigger, rationale, and embedding edits.

⚡ Proposed index addition in AGGREGATION_DDL

 CREATE INDEX IF NOT EXISTS idx_playbook_aggregation_cluster_version
     ON playbook_aggregation_cluster(agent_version, state, cluster_id);
+CREATE INDEX IF NOT EXISTS idx_playbook_aggregation_cluster_agent
+    ON playbook_aggregation_cluster(agent_playbook_id)
+    WHERE agent_playbook_id IS NOT NULL;

🤖 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 144 - 161, Add an index on
playbook_aggregation_cluster(agent_playbook_id) in the AGGREGATION_DDL schema
definition. Ensure the index is declared alongside the table’s existing indexes
so all trigger statements filtering by agent_playbook_id, including the delete
and update trigger paths, can use it.

---

656-681: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

Drop the now-dead vector_sum accumulation in legacy adoption.

The centroid now comes from centroid_embedding, and completion writes vector_sum=NULL. Nothing else reads vector_sum for this cluster: the only reader is this same method resuming its own partial sum on the next page. The per-member vector add and the per-page JSON encode of a full-dimension list therefore produce a value that is always discarded.

Keep member_count, which is still persisted and used. Remove the sum to cut allocation and JSON work from the bounded adoption loop.

♻️ Proposed simplification

-                if inserted.rowcount == 1:
-                    vector_sum = [
-                        left + right
-                        for left, right in zip(vector_sum, embedding, strict=True)
-                    ]
-                    member_count += 1
+                if inserted.rowcount == 1:
+                    member_count += 1
             centroid = centroid_embedding if complete and member_count else None
             self.conn.execute(
-                "UPDATE playbook_aggregation_cluster SET vector_sum=?, centroid=?, "
+                "UPDATE playbook_aggregation_cluster SET vector_sum=NULL, centroid=?, "
                 "member_count=?, rebuild_cursor=?, state=? WHERE cluster_id=?",
                 (
-                    None if complete else json.dumps(vector_sum),
                     json.dumps(centroid) if centroid is not None else None,

The member_embeddings dimension validation at Line 609 stays useful, because it still rejects a provenance change. The vector_sum read at Line 650 and its initialization can be removed with the same change.

🤖 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 656 - 681, Remove the obsolete vector_sum read, initialization,
per-member accumulation, and per-page JSON persistence from the legacy adoption
method surrounding member_embeddings. Preserve the member_count increment and
persistence, retain the existing dimension validation, and continue writing
vector_sum as NULL for the cluster update.

reflexio/server/services/storage/storage_base/playbook/_aggregation.py (1)

269-302: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Document the fenced failure contract for rebuild transitions.

The SQLite implementations of defer_playbook_aggregation_cluster_rebuild, complete_playbook_aggregation_cluster_rebuild, and discard_playbook_aggregation_cluster_rebuild raise RuntimeError when the expected agent no longer owns a rebuilding cluster, and complete_... also raises ValueError on an embedding-dimension change and RuntimeError when no residual members remain. The aggregator relies on those exceptions to abort a run. Other methods in this package document such failure modes in a Raises section. Add the same for these three methods so an alternative backend implements the identical fencing behavior.

🤖 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/storage_base/playbook/_aggregation.py`
around lines 269 - 302, Update the docstrings for
defer_playbook_aggregation_cluster_rebuild,
complete_playbook_aggregation_cluster_rebuild, and
discard_playbook_aggregation_cluster_rebuild to add Raises sections documenting
RuntimeError when the expected agent no longer owns a rebuilding cluster;
additionally document ValueError for embedding-dimension changes and
RuntimeError when no residual members remain in
complete_playbook_aggregation_cluster_rebuild. Preserve the existing method
behavior and signatures.

tests/server/services/playbook/test_playbook_aggregator.py (1)

1898-1918: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Consider covering the non-refresh prompt bound too.

This test passes current_agent_playbooks, so it exercises only the incremental-refresh branch of _select_generation_prompt_sources. reflexio/server/services/playbook/README.md lines 126-128 also state that new-cluster generation uses at most 100 sources while retaining the complete discovered membership. That branch stays uncovered here.

Parametrize on the presence of current_agent_playbooks to pin both bounds.

🤖 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 1898
- 1918, Extend test_generation_prompt_is_bounded_without_truncating_membership
to run with and without current_agent_playbooks, covering both
incremental-refresh and new-cluster paths in _select_generation_prompt_sources.
Assert each invocation captures at most 100 sources while preserving the
expected selected IDs and the complete cluster in outcomes[0].source_cluster.

tests/server/services/storage/test_playbook_aggregation_state_integration.py (1)

917-1072: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoff

Consider splitting this end-to-end test into staged tests.

This test now asserts five sequential lifecycle stages in one function: first generation, delta refresh, revision rebuild, single-member rebuild, and empty-cluster retirement. Each stage depends on the state left by the previous stage. A failure in an early stage hides every later assertion, and the failure message does not identify the stage.

The assertions themselves are correct. Extract the later stages into separate tests that build their own starting state, or add stage markers so a failure names the stage.

🤖 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 917 - 1072, Split
test_incremental_run_refreshes_agent_and_centroid_after_match into staged tests
covering initial generation, delta refresh, revision rebuild, single-member
rebuild, and empty-cluster retirement; have each test establish its own required
state so failures identify the affected lifecycle stage without depending on
earlier 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/playbook/components/aggregator.py`:
- Around line 1010-1017: Update the legacy centroid resolution in the
aggregation method to recover when legacy_agents[0].embedding is missing: reuse
the already-resolved embed_fn to re-embed the legacy agent’s content and use
that result as centroid_embedding. If re-embedding fails, log a warning
containing agent_playbook_id, skip only the affected fingerprint, and allow
bootstrap to continue processing remaining fingerprints without returning the
permanent “legacy cluster adoption pending” state.
- Around line 823-873: The aggregation run currently aborts its shared
transaction when one refresh or rebuild loses its cluster fence, discarding
unrelated generated results. Update the orchestration around the refresh loop
and _apply_rebuild_outcomes to isolate each cluster mutation with its own
savepoint or equivalent error boundary; on attach, replacement, or rebuild fence
loss, mark only that cluster’s members residual with reason
cluster_agent_unavailable, skip that cluster, and allow other refreshes,
rebuilds, and new clusters to commit.

In `@reflexio/server/services/storage/sqlite_storage/_base.py`:
- Around line 931-956: Update the migration query in the surrounding migrate()
logic to require c.index_rowid IS NOT NULL, and skip embeddings whose length
differs from self.embedding_dimensions before any database writes. Preserve
valid-row migration while preventing int(row[1]) and vector-table insertion
failures from aborting initialization.

In `@reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py`:
- Around line 1206-1301: Update defer_playbook_aggregation_cluster_rebuild and
complete_playbook_aggregation_cluster_rebuild to roll back the connection when
an exception occurs and _own_transaction() is true, covering all mutations and
fence-check failures. Preserve existing commit behavior on success, and mirror
the explicit rollback pattern used by attach_playbook_aggregation_items and
replace_playbook_aggregation_cluster_agent.
- Around line 124-190: Add guarded cleanup for playbook_aggregation_clusters_vec
in the retirement paths of retire_playbook_aggregation_cluster_on_agent_update
and retire_playbook_aggregation_cluster_on_agent_delete. Remove vector rows
whose rowids no longer have a matching playbook_aggregation_cluster, either
before cluster deletion or through the existing periodic/application maintenance
path, while preserving the current residual-item and aggregation-state updates.

---

Nitpick comments:
In `@reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py`:
- Around line 144-161: Add an index on
playbook_aggregation_cluster(agent_playbook_id) in the AGGREGATION_DDL schema
definition. Ensure the index is declared alongside the table’s existing indexes
so all trigger statements filtering by agent_playbook_id, including the delete
and update trigger paths, can use it.
- Around line 656-681: Remove the obsolete vector_sum read, initialization,
per-member accumulation, and per-page JSON persistence from the legacy adoption
method surrounding member_embeddings. Preserve the member_count increment and
persistence, retain the existing dimension validation, and continue writing
vector_sum as NULL for the cluster update.

In `@reflexio/server/services/storage/storage_base/playbook/_aggregation.py`:
- Around line 269-302: Update the docstrings for
defer_playbook_aggregation_cluster_rebuild,
complete_playbook_aggregation_cluster_rebuild, and
discard_playbook_aggregation_cluster_rebuild to add Raises sections documenting
RuntimeError when the expected agent no longer owns a rebuilding cluster;
additionally document ValueError for embedding-dimension changes and
RuntimeError when no residual members remain in
complete_playbook_aggregation_cluster_rebuild. Preserve the existing method
behavior and signatures.

In `@tests/server/services/playbook/test_playbook_aggregator.py`:
- Around line 1898-1918: Extend
test_generation_prompt_is_bounded_without_truncating_membership to run with and
without current_agent_playbooks, covering both incremental-refresh and
new-cluster paths in _select_generation_prompt_sources. Assert each invocation
captures at most 100 sources while preserving the expected selected IDs and the
complete cluster in outcomes[0].source_cluster.

In
`@tests/server/services/storage/test_playbook_aggregation_state_integration.py`:
- Around line 917-1072: Split
test_incremental_run_refreshes_agent_and_centroid_after_match into staged tests
covering initial generation, delta refresh, revision rebuild, single-member
rebuild, and empty-cluster retirement; have each test establish its own required
state so failures identify the affected lifecycle stage without depending on
earlier 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: 8a7ed081-6b28-4214-b7e2-e6dd109c39b7

📥 Commits

Reviewing files that changed from the base of the PR and between eb88f44 and 79226bc.

📒 Files selected for processing (15)

• reflexio/server/README.md
• reflexio/server/prompt/prompt_bank/playbook_aggregation/v2.4.0.prompt.md
• reflexio/server/services/playbook/README.md
• reflexio/server/services/playbook/aggregation_scheduler.py
• reflexio/server/services/playbook/components/aggregator.py
• reflexio/server/services/storage/sqlite_storage/_base.py
• reflexio/server/services/storage/sqlite_storage/_lineage.py
• reflexio/server/services/storage/sqlite_storage/playbook/_agent.py
• reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py
• reflexio/server/services/storage/storage_base/playbook/__init__.py
• reflexio/server/services/storage/storage_base/playbook/_agent.py
• reflexio/server/services/storage/storage_base/playbook/_aggregation.py
• tests/server/services/playbook/test_aggregation_scheduler.py
• tests/server/services/playbook/test_playbook_aggregator.py
• tests/server/services/storage/test_playbook_aggregation_state_integration.py

[yyiilluu]

CodeRabbit review follow-up (commit 1d4dc49):\n\nImplemented:\n- Added a partial cluster-to-agent index for the current-agent lookup.\n- Removed obsolete legacy vector_sum accumulation; the canonical agent embedding is now the centroid.\n- Documented transaction/fence RuntimeError behavior on rebuild disposition methods.\n- Parameterized the prompt-bound test to cover both refresh and new-cluster generation paths.\n\nDeclined:\n- I kept the sequential lifecycle E2E test intact. Its purpose is to prove that one persistent cluster evolves correctly through attach, refresh, invalidation, and rebuild transitions. Splitting it would remove that cross-transition invariant and duplicate substantial setup; focused unit/regression tests cover the individual branches separately.\n\nAll five inline findings were also fixed and replied to on their exact threads.