perf(daemon): bound catch-up chunk convergence to the chunk's own input - #4694
Conversation
|
Important
This repository does not receive automatic reviews because it has fewer than 10 stars. ⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Team Run ID: Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d9ae9e0393
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if chunk_index != len(chunks): | ||
| ingest_kwargs["whole_archive_convergence"] = False | ||
| metrics = await self._ingest_files(chunk_paths, **ingest_kwargs) |
There was a problem hiding this comment.
In the catch-up path, a backlog with one changed Claude Workflow artifact and more than one chunk of Codex files can place the Claude artifact in an early round-robin chunk and only Codex paths in the last chunk. This marks the Claude stage SKIPPED without debt in the early chunk, while the final pass receives only its own paths; make_claude_workflow_stage.check_many() then finds no relevant Claude candidate and never rebuilds the archive-wide workflow graph. Preserve relevant triggers from prior chunks or run the final whole-archive stages independently of the last chunk's path set.
AGENTS.md reference: AGENTS.md:L109-L115
Useful? React with 👍 / 👎.
| from polylogue.sources.live.hook_paste_enrichment import enrich_paste_from_hooks | ||
|
|
||
| enrich_paste_from_hooks(self._cursor._db_path) | ||
| enrich_paste_from_hooks(self._cursor._db_path, session_ids=tuple(dict.fromkeys(session_ids))) |
There was a problem hiding this comment.
When a UserPromptSubmit journal entry arrives after session A's transcript batch, and the next live batch changes only session B, this call scans only B's journal. No hook-event route invokes this enrichment directly, so if A receives no later transcript update its message remains observably has_paste = 0 indefinitely; the previous archive-wide scan would process A during B's batch. Track pending hook sessions or otherwise consume the durable hook evidence without restricting it solely to the current ingest's session IDs.
AGENTS.md reference: AGENTS.md:L109-L115
Useful? React with 👍 / 👎.
| "watcher.live_ingest.full_convergence", | ||
| self._converge_paths, | ||
| full_result.succeeded, | ||
| whole_archive=whole_archive_convergence, | ||
| session_ids=full_result.changed_session_ids, |
There was a problem hiding this comment.
When the final catch-up chunk contains changed files from multiple full-ingest source groups, or an append group plus full-ingest groups, ingest_files() invokes _converge_paths() separately for each group and forwards whole_archive_convergence=True every time. Consequently the exact FTS audit, delegation projection, and raw-authority scan can each traverse the entire archive several times in the nominally final chunk, so its cost is multiplied by the number of internal groups rather than being the promised single pass. Hoist the whole-archive pass after all groups or otherwise permit it only once.
AGENTS.md reference: AGENTS.md:L109-L113
Useful? React with 👍 / 👎.
| state.stages[stage_name] = StageState.FAILED | ||
| state.error_count += 1 | ||
| else: | ||
| _record_stage_times(batch_stage_times, f"{stage_name}.check", time.perf_counter() - t_check, {}) |
There was a problem hiding this comment.
For a batch stage such as raw_authority_verdict_cache, returning False while bounded work remains causes check_many() to run a second time during the post-execution recheck, but this records only the initial probe; the same timing is also omitted when the initial probe raises because the recorder is confined to the else branch. On a large archive either omitted scan can dominate convergence time, so the returned <stage>.check ledger does not provide the fully attributed cost promised by converge_batch(). Time each check invocation, including rechecks and exceptional exits.
Useful? React with 👍 / 👎.
Partition inspection and publish selected a session's identity-ledger rows
with substr(block_id, 1, length(key) + 1) = key || ':', which plans as a
full scan of messages_fts_identity for every key on every pass. block_id is
session_id || ':' || ..., so the half-open range [key || ':', key || ';')
selects exactly the same rows through the block_id UNIQUE index (';' is the
code point after ':'). Measured on a 359-session archive: 10.5 ms -> 0.1 ms
per key.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid
find_raw_authority_verdict_cache_work probed every logical_source_key with two point queries, so each daemon convergence check cost O(cohorts) statements. It now reads the cohort rows and the cached fingerprints in one pass each and applies the same staleness rule per key. write_raw_authority_verdict_cache deleted only the cohort's own rows; a raw already cached under a previous logical_source_key collided with the raw_id primary key (54 warmup failures in one rehearsal, each re-running the whole-archive check on the next chunk). Stale rows for the cohort's raw_ids are removed before the insert. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid
The catch-up loop broke out of its priority groups on the first empty group, so a backlog with no file modified in the last hour was never planned: a fresh daemon over a quiet source tree logged "watching" and ingested nothing. An empty group is skipped; only a stop request ends the loop. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid
… input Per-chunk convergence paid archive-wide costs on every chunk of a catch-up: the exact FTS readiness audit (whole-table aggregates, run twice per chunk: after FTS repair and after insights), the raw-authority verdict warmer (one probe per cohort), the Claude workflow and delegation graphs (rebuilt from every raw artifact), a scan of every hook sidecar journal (1.65 GB on the rehearsal spool: 4.5 s warm, 73 s once the page cache was evicted), and the periodic whole-archive FTS sweep interleaving with chunks. Stage check time was outside the ledger, so a 102 s chunk showed 2.5 s of stages. - ConvergenceStage.whole_archive marks stages whose work is a function of the archive; converge_batch(whole_archive=False) records them SKIPPED. The watcher runs every catch-up chunk but the last in that scope, so the archive-wide stages run once per catch-up. - The readiness audit is its own whole-archive stage (fts_readiness), run after fts and derived; partition-scoped FTS passes and insight rebuilds no longer publish it. The periodic whole-archive FTS sweep waits while a chunked catch-up is active. - Hook paste enrichment reads only the batch's sessions' sidecar journals (<provider>-<native_id>.jsonl). - The convergence ledger charges <stage>.check and hook_paste_enrichment. Scratch run, 400 files / 325 MB, 100 chunks, same host: convergence 200.4 s -> see PR body for the after figures. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid
d9ae9e0 to
1a0ac23
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1a0ac2368f
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| # The rebuild commits its own rows; the exact archive-wide FTS audit is | ||
| # published once per whole-archive pass by ``make_fts_readiness_stage``. | ||
| conn.commit() |
There was a problem hiding this comment.
When a hot session's stage-specific derived debt later retries, _drain_convergence_debt_once() constructs a converger containing only make_derived_stage, so the new fts_readiness stage never runs. This rebuild can change session_work_events and its trigger-maintained FTS rows while leaving the previously recorded exact counters/readiness unchanged; during an active catch-up the immediately following periodic FTS audit is explicitly suppressed, so status and health surfaces can report the old snapshot until a later sweep. Keep the audit on this standalone derived route or include fts_readiness after a successful derived debt retry.
AGENTS.md reference: AGENTS.md:L111-L115
Useful? React with 👍 / 👎.
| if result.outcome is FtsOutcome.DONE and partition_keys is None: | ||
| self._publish_readiness_projection(conn) | ||
| state = { | ||
| FtsOutcome.DONE: FtsOwnerState.READY_EXACT, |
There was a problem hiding this comment.
With partition_keys=("session-a",) and an unrelated stale partition still present, this branch correctly skips the archive-wide readiness projection, but the subsequent mapping still returns FtsOwnerState.READY_EXACT with exact=True. Consequently callers of FtsConvergenceOwner.run_once_sync() observe an exact-ready archive even though only the requested partition was inspected and repaired. Return a distinct partition-complete result, or otherwise keep the archive-level ready/exact fields false while allowing make_fts_stage() to recognize successful partition convergence.
Useful? React with 👍 / 👎.
1 participant
Summary
During a chunked catch-up every chunk paid archive-wide convergence costs, so per-chunk convergence grew with the archive. Whole-archive stages now run once per catch-up (on the last chunk), the archive-wide FTS readiness audit is its own deferred stage, hook-paste enrichment reads only the batch's sessions' sidecars, FTS partition inspection uses the
block_idindex instead of asubstrscan, the raw-authority warmer finds its work in two statements (itsraw_idcollision is fixed by #4693, on which this branch is rebased), the periodic whole-archive FTS sweep waits while catch-up is active, and the convergence ledger now attributes stagechecktime. A separate defect found on the way: a catch-up with no source file modified in the last hour ingested nothing.Problem
Rehearsal (fresh daemon, real sources,
/realm/tmp/work/rehearsal-4): chunk 1 parse 1.1 s / convergence 5.6 s; chunk 150 parse 23.6 s / convergence 62.9 s for 1.9 MB; average over 150 chunks parse 10.6 s, convergence 20.2 s, both growing with archive size (305 sessions, 45,657 messages). Chunk 149 held the writer 103.9 s while its stage ledger summed to 2.5 s.Profile (in-process sampler on a scratch daemon over 400 real files; SQL timed against a copy of the rehearsal's 359-session index):
enrich_paste_from_hooksafter every chunk scanned every sidecar journal inhooks/(371 files, 1.65 GB): 4.5 s warm, 72.9 s once the page cache was evicted. Untimed (outside the stage ledger); this is the size of chunk 149's unattributed time.fts_invariant_snapshot_sync(five whole-table aggregates overblocks/messages_fts_docsize/identity) ran twice per chunk, fromFtsConvergenceOwner._publish_readiness_projectionand_record_fts_freshness_after_insights: 0.6 s per call at 90k blocks, 49 % of scratch convergence time (98 s of 200 s).inspect/publishselected a session's identity rows withsubstr(block_id, 1, length(?)+1) = ? || ':':SCAN messages_fts_identityper key, 10.5 ms at 85k rows vs 0.1 ms for the index range.raw_authority_verdict_cache.check_manyprobed every cohort with two statements (532 cohorts at chunk 150), and its warmup failed 54 times withUNIQUE constraint failed: raw_authority_verdicts.raw_id, re-running the whole-archive check on every following chunk.claude_workflowanddelegation_work_evidencere-read every raw artifact / every delegation row per chunk.maintenance.fts_convergence(whole-archiveinspect_all) held the writer 1061 s across 40 runs (max 132 s); chunks waited 1977 s on the writer in total._catch_upbroke out of its (hot, cold) group loop on the first empty group, so a backlog with no file modified in the last hour was never planned. The after-run reproduced it (0 attempts, 0 cursors,catch_up_completeset after 40 s).Solution
ConvergenceStage.whole_archive;DaemonConverger.converge_batch(whole_archive=False)records such stagesSKIPPED(converged, no debt: their staleness is re-derived from content). The watcher runs every catch-up chunk but the last in that scope. Stages flagged:raw_authority_verdict_cache,claude_workflow,delegation_work_evidence, and the newfts_readiness.make_fts_readiness_stagepublishes the exact archive-wide FTS audit once per pass, afterftsandderived; partition-scoped FTS passes and insight rebuilds no longer publish it._periodic_convergence_checkskips the whole-archive FTS sweep whilewatcher.catch_up_active.enrich_paste_from_hooks(db, session_ids=...)reads only*-<native_id>.jsonlfor the batch's sessions (session_ids_by_pathon append,changed_session_idson full ingest).[key || ':', key || ';')range on theblock_idUNIQUE index ininspect(excess, duplicates) andpublish.find_raw_authority_verdict_cache_work: one pass overraw_sessionsand one overraw_authority_verdicts(master's fix(storage): a re-keyed raw replaces its stale verdict-cache row #4693 writer, which removes a re-keyed raw's stale row, is kept as is).<stage>.checkandhook_paste_enrichmenttimings.Verification
.venv/bin/python -m devtools verify --quick-> all 13 gatesok(format, lint, mypy, generated-surfaces, layering, patterns, doc-commands, schema-manifest, oracle-integrity, testmon-selection, consumer-reachability, timestamp-doctrine, schema-privacy)..venv/bin/python -m devtools test tests/unit/daemon/test_catch_up_chunk_cost.py tests/unit/daemon/test_daemon_convergence.py tests/unit/sources/test_hook_paste_enrichment.py tests/unit/storage/test_fts_derivation.py tests/unit/storage/test_raw_authority_verdict_cache.py tests/unit/daemon/test_convergence_stages.py tests/unit/daemon/test_fts_readiness_fallback.py tests/unit/daemon/test_fts_identity_convergence.py tests/unit/sources/test_live_batch_convergence.py tests/unit/pipeline/test_ingest_batch_fts_repair.py tests/unit/sources/test_live_catchup_planning.py tests/unit/sources/test_live_watcher.py tests/unit/sources/test_live_hot_session_convergence.py tests/unit/daemon/test_daemon_cli.py->1 failed, 378 passed in 252.37son head 1a0ac23 (pueue 1655,.cache/verify/pytest-slot-1672251.log); the one failure,test_daemon_cli.py::test_spool_pending_check_ignores_terminal_cursor_states, is inherited at base 754553b and fixed by test: follow the merged runner rename, corpus width, and read-only cursor probe #4695 (the test lambda lacked**_kwargs)test_chunk_convergence_cost_does_not_grow_with_archive_size(real live-ingest route, archives of 2 and 14 sessions, one 2-file chunk: product SQL statements inside_converge_pathsequal, 2 sidecars read, 0 snapshot calls, no deferred stage in the ledger; reverting any deferral adds two statements per cohort / one per artifact row),test_final_catch_up_chunk_runs_the_whole_archive_stages, converger chunk-scope and check-ledger, sidecar scoping, FTSEXPLAIN QUERY PLANhas noSCANof the identity ledger, raw-authority rebinding and two-statement work finder, cold-backlog catch-up, last-chunkwhole_archive_convergenceflag.Scratch run (same host, same 400 files / 325 MB from
~/.claude/projectsand~/.codex/sessions, 100 catch-up chunks,polylogued runwith an in-process sampler):fts_invariant_snapshot_syncinside the chunk pathmessages/blocksinserts; a full-corpus pytest ran on the host during the after run and was OOM-killed, pueue 1636 exit 137)Per-chunk convergence no longer grows with the archive. End-to-end throughput on this scratch is parse-bound (0.61–0.80 MB/s parse-only), so the 2x-of-parse criterion from polylogue-t4iy5.11.1 still needs the rehearsal driver on a quiet host for its operational proof.
Residuals
_check_fts_readiness_medium) recomputes the exact archive-wide FTS audit on every status payload: 51 s of the after-run's profile, all from/api/statusand the status collector. The rehearsal driver probes status every 60 s; at 5M blocks each probe would take tens of seconds. It should read the freshness ledger thefts_readinessstage publishes.index_parsed_write;messages/blocksinserts with 18 secondary indexes) plus page-cache-miss stalls under host memory pressure; no whole-archive statement was found on the write path.claude_workflowanddelegation_work_evidencestill rebuild from the whole archive whenever they run (once per catch-up now, then per live batch).Beads: polylogue-t4iy5.11.1, polylogue-623q
🤖 Generated with Claude Code
https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid