xiaoxue_admin
/
EverOS
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(lancedb): reclaim stale versions via write-locked prune (#379) 

* fix(lancedb): reclaim stale versions via write-locked prune

The storage soak (48h, sustained churn + fuzz) proved the bundled
lock-free `optimize(cleanup_older_than=...)` loses its commit-conflict
race against concurrent writes — cleanup ran only 16 of ~250 times, so
old dataset versions / FTS orphans piled up and the index dir grew to
the 40G disk guardrail and never reclaimed under load. main still had
that bundled call.

Split the maintenance path:
- `LanceRepoBase.optimize()` is now compact-only and lock-free (a commit
  conflict here is benign — the next beat retries, so it must not stall
  writers).
- `LanceRepoBase.prune(older_than)` runs `cleanup_older_than +
  delete_unverified=True` **under the per-table write lock**, so no
  writer is in flight: the Rewrite has the manifest to itself (cleanup
  completes every beat) and aggressive deletion is safe. It also removes
  the empty `_indices/<uuid>/` husks cleanup leaves behind (soak: 13061
  dirs, 98% empty), offloaded to a thread.
- The cascade worker's heavy beat calls `prune()`; the light beat calls
  `optimize()`. A benign light-beat commit conflict is logged at debug
  and does not count toward the failure streak or trigger a rebuild.
- Prune's retention window (`cleanup_older_than`) is decoupled from the
  prune cadence and defaulted short (60s). It runs under the write lock,
  so the window only needs to outlive an in-flight read; keeping it =
  cadence (300s) left ~2 cadences of superseded full-table copies on
  disk between beats (soak: transient ~15G/table peaks). 60s reclaims
  all but the last minute each beat — same live floor (~625MB/table),
  far lower transient footprint.

Result on the re-run soak: disk sawtooths and reclaims to live-data size
(~1.3G total) under active load — vs run1 stuck at 40G until writes
stopped — with 0 crashes / 0 OOM / 0 stuck cleanups over 48h.

Cascade projection health is now observable:
- `CascadeOrchestrator.health()` -> `CascadeHealth`, combining the
  worker's in-memory signals (drain-loop failures, unrecoverable count,
  optimize streak, prune staleness) with the SQLite queue summary.
- `GET /health` gains a typed `cascade` readiness block. `healthy`
  reflects operational health only (drain / optimize / prune);
  `failed_permanent` (files awaiting `cascade fix`) is a data-quality
  backlog reported as an informational count that does NOT flip
  `healthy` — otherwise the signal would sit red permanently.

The scanner-side retry cap and the `_MAX_TOTAL_RETRIES` budget already
on main handle re-enqueue storms, so no duplicate is added here.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(deps): pin lancedb to >=0.34.0,<0.35.0

The previous open-ended `>=0.13.0` let any environment float to an
untested release, including 0.32-0.34 which carry a compaction
offset-overflow regression (lance-format/lance#7653) that stalls
version cleanup and grows the index dir without bound.

- Floor 0.34.0: the current resolved version; runs safely thanks to
  the with_position=False FTS workaround shipped in #336. Verified that
  data written by lancedb 0.32.0 (lance v6) reads correctly under
  0.34.0 (lance v8), so existing deployments upgrade cleanly. Never
  widen the floor below 0.34 -- older lance cannot read v8-format data.
- Ceiling <0.35.0: 0.35 embeds lance-rust v9 (large encoding jump, not
  yet stable-released); it must pass the soak harness before we allow
  it.

Resolved version is unchanged (still 0.34.0); this only tightens the
declared constraint.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
(cherry picked from commit 94f9aa67d11a07c93c7d59d6b134e80858b60316)

* fix(search): bridge agent-kind metadata into the agentic doc contract

The agent AGENTIC path (`agent_case` / `agent_skill`) fed recall
candidates straight into `aagentic_retrieve`, whose `_format_docs`
(LLM sufficiency / multi-query prompt) reads `metadata["episode"]` as a
`{subject, content}` dict plus a ms-epoch `timestamp`. Agent-kind rows
carry their body in the recaller's `text_field` and time as a datetime,
so `_format_docs` raised `TypeError: Candidate ... has no episode dict`
and `POST /api/v*/memory/search` returned 500 for any
`owner_type=agent` + `method=agentic` request.

Mirror the episode path's bridge: reshape agent candidate metadata into
the everalgo doc contract before `aagentic_retrieve`, and revert it
before DTO shaping so the agent shapers still see a datetime timestamp.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
(cherry picked from commit bb9a3a585e044643d3852b9d2810b42c05a7a47c)

* test(search): regenerate search seed to current linkage + migrate e2e

The committed `search_seed` fixture and several search e2e tests were
written for the pre-1.5 memcell fact-linkage model. Current extraction
links atomic_facts to episodes via `parent_id == episode.entry_id`
(parent_type="episode"), and user_memory clusters store episode
entry_id members — so the stale fixture made VECTOR/AGENTIC recall and
the cluster-narrowing path find nothing, and stale assertions checked
an old error code.

- Regenerate `search_seed/*` from a fresh corpus in the current
  entry_id format; facts now bridge across multiple episodes (richer
  agentic / hierarchical-eviction coverage).
- Fix `_dump_search_seed.py` sampling: pick episodes that host facts
  first and keep facts by episode entry_id, so re-dumps stay coherent.
- Migrate e2e tests to the entry_id model (hierarchical-eviction,
  session/timestamp filters, cluster seeding helper) and update the
  filter-error assertion to the current `INVALID_INPUT` code.
- Provision `ome.toml` in the full-app pipeline fixture (the OME config
  reloader requires it; strategies are code-registered so the packaged
  default suffices), unblocking corpus regeneration.

Full search e2e suite now green (49/49).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
(cherry picked from commit 44cd9cecc07775cb9270104a5804f36e32cd788d)

* fix(lancedb): detect schema type drift and add cascade rebuild recovery

verify_business_schemas only compared column names, so a column whose
on-disk Arrow type had drifted (name unchanged) slipped through and
detonated later inside merge_insert as an opaque LanceError(IO)
"Spill has sent an error" (#337). Now compare each shared column's Arrow
type against schema.to_arrow_schema() — the exact schema get_table
builds tables from, so a healthy table never false-positives.

Reproduced #337 byte-identically: an episode.subject_vector column left
as string or null by an older build, plus a real 1024-d vector on
upsert, yields the exact crash. No lancedb version (0.13-0.34) renders
Optional[Vector] as a non-vector type, so the startup guard is what
should catch it — not the runtime.

Add `everos cascade rebuild` as the safe recovery: it drops the business
LanceDB tables and re-indexes from markdown, skipping the verify guard
(which the drift would otherwise trip on startup). Unlike removing only
.index/lancedb it re-populates already-done entries (reset_all clears
the cascade queue); unlike removing all of .index it preserves
unprocessed_buffer (messages not yet extracted).

Fixes #337.

* docs(cascade): document cascade rebuild and correct recovery guidance

Add the `everos cascade rebuild` command to the runbook, CLI, and
how-memory-works docs. Correct the old recovery guidance: a bare
`rm -rf .index/lancedb` leaves md_change_state marked `done`, so the
scanner skips those files and the index comes back empty — the runbook
previously claimed a full repopulation that does not happen. `cascade
rebuild` is the safe path (re-populates done entries, preserves
unprocessed_buffer). Also document that verify now checks column types.

* chore(rebase): adapt #354 integration test to soft-embedding main

CascadeOrchestrator dropped the embedder param when embedding became a
soft dependency (main); fold the schema-drift integration test onto it.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(cascade): freeze monotonic clock in prune-staleness health tests

Both prune-staleness health tests fabricated
`_started_at = time.monotonic() - (ALERT + 100)`, assuming monotonic()
is a large value. On a fresh CI runner monotonic() is only ~100-180s, so
the subtraction went negative, the source clamped the baseline to 0, and
staleness read back as ~130s < 900s — failing on CI while passing on
long-lived dev boxes where monotonic() is huge.

Freeze the monotonic clock via monkeypatch so staleness is deterministic
regardless of the runner's boot uptime. Source logic is unchanged; only
the tests are made hermetic.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(cascade): wait for stable terminal state in rename scenarios

`_wait_path_done` reached a terminal status, slept a 0.1s settle window,
then asserted the status was still terminal — which contradicted its own
docstring ("absorb any last-second re-enqueue"). A rename's delete event
or an atomic-replace echo can flip a done row back to `processing` inside
that window, so on a slow CI runner the assert fired
("flipped back to processing after reaching done"), failing
test_rename_cross_owner_keeps_frontmatter_owner intermittently (seen on
the 3.13 job). `make integration` runs without `--reruns`, so a single
flake fails the whole job.

Wait for a terminal state that *survives* the settle window instead:
absorb a transient re-enqueue by waiting for terminal again, still bounded
by `deadline` so a row that never settles surfaces as a timeout. Pre-
existing flake on main, unrelated to the prune change; the scenario's real
assertions (row counts, frontmatter owner) are untouched.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(cascade): cross-process prune safety + backfill reclaim fix

Review of #379 found two P0s plus P1/P2s, all verified against the code:

- P0-1: cascade backfill still called the removed
  optimize(cleanup_older_than=…) kwarg → TypeError swallowed by the
  best-effort try/except → backfill silently skipped all compaction +
  reclaim (the exact bloat this PR fixes). CI stayed green because the test
  double kept the stale signature. Fix: call optimize() + prune(0) at the
  call site; make the fake mirror the real signature so the drift can't hide
  again; pin prune in the backfill tests.

- P0-2: prune ran delete_unverified=True guarded only by an in-process
  asyncio lock, but the runbook promises `cascade sync` is safe alongside a
  live server — and the CLI's first optimize beat does prune, in a separate
  process. It could delete files the daemon is mid-commit on. Fix: switch
  prune to delete_unverified=False. Measured to reclaim identically on
  churned tables (both collapse superseded versions ~97%); True only
  additionally deletes in-flight/dangling files — exactly the corruption
  vector. No cross-process lock needed; the write-lock/commit fix (the real
  reclaim win) is unchanged.

- P1-3: /health called orch.health() (6 SQLite aggregates) with no guard →
  a locked/full/migrating DB would 500 the liveness probe and restart the
  container. Wrap it: unhealthy readiness + reason, HTTP stays 200.

- P1-4: rebuild drops + recreates tables; a live daemon holds cached handles
  pointing at the dropped dataset. Runbook now says stop the server first —
  the one cascade command unsafe alongside a live server.

- P2: narrow _is_benign_commit_conflict to the "commit conflict" phrase (a
  bare "retryable" swallowed unrelated recoverable errors); add a timeout
  around the prune cleanup so a hung lance call can't wedge the write lock;
  correct two stale docstrings.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs: fix schema-recovery guidance + drop dead internal doc refs

Follow-up to the review; two doc issues neither the review nor the fix
commit caught:

- The schema-drift startup error's docstrings (verify_business_schemas
  and LanceDBLifespanProvider) still described the recovery as
  `rm -rf ~/.everos/.index/lancedb` — which the runbook explicitly calls
  the WRONG recovery (it leaves the cascade queue `done`, so nothing
  re-indexes and the index comes back empty). The raised error already
  points to `everos cascade rebuild`; align the docstrings to match.

- 15 dangling references to an internal numbered design-doc set
  (12_/13_/16_/17_*.md) that was never shipped to this repo. Point the
  schema-recovery ones at docs/cascade_runbook.md; drop the rest (pure
  provenance in table/component docstrings) while keeping the substance.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs: align maintenance docstrings with split optimize/prune API

Two docstring residuals from the #379 review's P2 list:

- _run_optimize_once still described the pre-split bundled heavy beat
  ("same work plus cleanup_older_than ... older than one cadence");
  the heavy beat now calls prune() under the write lock and the
  retention window is decoupled from the cadence
  (DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS).

- _restore_shaper_metadata converts any numeric timestamp, wider than
  an exact inverse of the bridge; document that this is deliberate
  (the shaper contract requires a datetime either way).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* style(e2e): hoist fixture-body imports to conftest module top

shutil / importlib.resources.files were imported inside the
core_pipeline_runtime fixture body; move them to the module top to
match the repo import style (#379 review P2).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(cascade): back off a hung prune via a separate attempt clock (N1)

The write-lock timeout on prune (review P2) didn't achieve its goal: it
was 300s — equal to the prune cadence — and `last_prune_at` only advanced
on success. So a hung lance cleanup timed out after 300s, `should_prune`
was still true (clock never moved), and the next beat re-pruned ~10s
later — pinning the per-table write lock ~97% of the time, the exact
write-starvation the timeout was meant to prevent.

A real cleanup is milliseconds even on a heavily churned table (measured
~40ms at 320k writes / 100 versions), so the timeout is a pure hang-catcher:
lower it to 60s (~1500x headroom, never fires normally, well below the 300s
cadence).

Split the prune clock so a failed prune backs off without masking the
health signal:
- last_prune_attempt_at (new) gates scheduling, advanced before the call
  whether it succeeds or times out → a hung prune waits a full cadence
  before retrying (light lock-free compaction runs meanwhile), so the lock
  is held at most ~timeout/cadence ≈ 17% in the worst case.
- last_prune_at advances only on success and still drives the
  prune-staleness health signal, so a persistently failing prune surfaces
  as degraded instead of being hidden by an advanced schedule clock.

Regression test: a raising prune advances the attempt clock (next beat is
light, no immediate re-prune) but not the success clock.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: zhanghui <zhanghui@shanda.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
zhanghui 2026-08-03 15:41:13 +08:00 committed by GitHub
parent 6d62ecbd6f
commit e5118c52a8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
47 changed files with 76739 additions and 48156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
docs/cascade_runbook.md
View File

@ -79,8 +79,50 @@ everos cascade sync users/u1/episodes/X.md # re-enqueue + drain
```
The CLI builds the same `CascadeOrchestrator` as the daemon but only
calls `sync_once` / `drain_once` — no watcher / scanner background
task. So it's safe to run in parallel with a live `everos server`.
calls `sync_once` / `drain_once` — no watcher / scanner background task.
Its drain still runs the same compaction + version-cleanup (`prune`) as
the daemon, but `prune` uses `delete_unverified=False`, so it never
deletes a file another process may be mid-commit on. Safe to run in
parallel with a live `everos server`.
## Rebuild the index: `everos cascade rebuild`
The safe recovery from a drifted or corrupt LanceDB index. It rebuilds
the whole index from markdown (the source of truth) in one shot:
```bash
everos cascade rebuild # prompts for confirmation
everos cascade rebuild --yes # non-interactive
```
> **Stop the `everos server` first.** Unlike `cascade sync`, rebuild
> **drops and recreates** the LanceDB tables. A running daemon holds
> cached table handles that would keep pointing at (and writing to) the
> dropped dataset, corrupting the rebuild. This is the one cascade
> command that is **not** safe to run alongside a live server.
What it does, in order:
1. **Drops** every business LanceDB table (`drop_business_tables`) and
evicts them from the connection cache.
2. **Recreates** them empty from the current schema + FTS indexes
(`ensure_business_indexes`).
3. **Clears** the cascade queue (`md_change_state.reset_all`) so every
md file re-enqueues as `added` on the next scan.
4. **Re-scans + drains** (`sync_once`): re-embeds and re-inserts every
md entry.
It deliberately **skips `verify_business_schemas`** — the drift it
recovers from would otherwise trip that guard on startup before the
rebuild could run (chicken-and-egg).
Why not a bare `rm`:
| Recovery | Re-populates `done` entries | Preserves `unprocessed_buffer` |
|---|---|---|
| `rm -rf .index/lancedb` | ❌ scanner skips `done` rows → empty index | ✅ |
| `rm -rf .index` | ✅ | ❌ deletes un-extracted messages |
| `everos cascade rebuild` | ✅ | ✅ |
## Recovery paths
@ -91,15 +133,28 @@ an on-disk table has columns the current Pydantic schema does not
declare (or vice versa), the boot fails with:
```
LanceDB table 'episode' schema drift: missing=[...], extra=[...].
The index is rebuildable from md — recover with
`rm -rf ~/.everos/.index/lancedb` and restart.
LanceDB table 'episode' schema drift: missing=[...], extra=[...],
type_drift=[...]. The index is rebuildable from md — recover with
`everos cascade rebuild`.
```
This is the documented recovery: delete the index, restart the
server, the scanner will pick up every md file on its first sweep and
the worker repopulates LanceDB. Markdown is the source of truth, so
no data is lost.
`verify_business_schemas` compares both the column **names** and their
**Arrow types** against the current schema. Catching type drift matters:
an `episode.subject_vector` column left as `string` (or `null`) by an
older build, while the schema now declares a 1024-d `fixed_size_list`,
has the same column *name* — so a name-only check would wave it through
and it would detonate later inside `merge_insert` as an opaque
`LanceError(IO): Spill has sent an error` (EverOS #337). The type check
turns that into this clean startup error.
Recover with **`everos cascade rebuild`** (documented above). Do **not** just
`rm -rf ~/.everos/.index/lancedb`: that clears the vectors but leaves
`md_change_state` marked `done`, so the scanner skips every already-
indexed file and the index comes back **empty**. And do **not**
`rm -rf ~/.everos/.index`: that also deletes `unprocessed_buffer`
(messages received but not yet extracted — not rebuildable from md).
`cascade rebuild` is correct on both counts. Markdown is the source of
truth, so no memory content is lost.
### inotify watch-limit exhaustion (Linux)
@ -262,7 +317,9 @@ is a deployment-side change with no schema work.
## What cascade does NOT do (yet)
- **Schema migration**: LanceDB column changes require `rm -rf`.
- **Schema migration**: LanceDB has no in-place column migration; a
schema change is recovered by rebuilding from md (`everos cascade
rebuild`), not an automatic `ALTER`.
- **Parent-id back-link**: Episode rows currently carry
`parent_id=None`; the writer doesn't preserve the source memcell id
in the entry inline. Tracked separately.

6
docs/cli.md
View File

@ -43,8 +43,10 @@ Each subcommand lives in its own module under
registered in `cli/main.py`. The CLI is intentionally small — hot-path
business (`/add` `/flush` `/search` `/get`) is the **HTTP API**, not the
CLI; the CLI covers setup (`init`), running the server, and index ops
(`cascade`). There is no `reindex` command — rebuild by deleting
`<root>/.index/lancedb` and restarting, or run `everos cascade sync`.
(`cascade`). There is no `reindex` command — for an incremental
catch-up run `everos cascade sync`; to rebuild the whole index from
markdown (recovery from drift / corruption) run `everos cascade
rebuild`.
## `everos server start`

16
docs/how-memory-works.md
View File

@ -90,7 +90,8 @@ visually distinct from a user-named one).
table `md_change_state`) — there is no `.cascade.log` / `.manifest.json`
file in the current implementation. The `<app>/<project>` nesting is
real and always present (`default_app/default_project` for the default
scope). There is **no `everos reindex` command** (see
scope). There is no command literally named `reindex`, but
`everos cascade rebuild` rebuilds the index from markdown (see
[Operating it](#operating-it)).
The path manager is
@ -296,12 +297,17 @@ The CLI ([cli.md](cli.md)) is intentionally small:
| `everos cascade status` | queue / LSN summary |
| `everos cascade sync` | drain the cascade queue now (force md → LanceDB) |
| `everos cascade fix` | list failed rows / re-enqueue retryable ones |
| `everos cascade rebuild` | rebuild the whole index from markdown (drift / corruption recovery) |
!!! warning "There is no `everos reindex` or `everos flush`"
- **Reindex** = the index is rebuildable: stop the server,
`rm -rf <memory-root>/.index/lancedb`, restart — the cascade
rebuilds from markdown. For an incremental catch-up, use
`everos cascade sync`.
- **Reindex** = the index is rebuildable from markdown. To rebuild
the whole index, run `everos cascade rebuild` — it drops the
LanceDB tables and re-indexes from md, re-populating even entries
the queue already marked `done` and preserving un-extracted
buffered messages. (A bare `rm -rf <memory-root>/.index/lancedb`
is **not** enough: the cascade queue still shows those files
`done`, so the scanner skips them and the index comes back empty.)
For an incremental catch-up, use `everos cascade sync`.
- **Flush** is an HTTP endpoint (`POST /api/v2/memory/flush`), not a
CLI command — it forces *extraction* of the session buffer, which is
a different thing from forcing *index sync* (`cascade sync`).

71
docs/openapi.json
View File

@ -12,7 +12,7 @@
"health"
],
"summary": "Health",
"description": "Liveness probe with capabilities and disabled features.",
"description": "Liveness + capabilities + cascade readiness probe.\n\n``status`` stays ``\"ok\"`` whenever the process is up — the HTTP code\nis a *liveness* signal and a degraded cascade must not trigger a\nrestart (a crash-loop fixes neither a bad md file nor disk bloat).\nThe ``cascade`` block is the *readiness* signal: ``healthy=false``\nwith human-readable ``reasons`` **only** when the projection pipeline\nitself is stuck (drain failing, optimize stuck, version cleanup\nstalled). ``failed_permanent`` — files awaiting ``cascade fix`` — is\na data-quality backlog reported as an informational count that does\nnot flip ``healthy``. Alert on ``cascade.healthy``.",
"operationId": "health_health_get",
"responses": {
"200": {
@ -1805,6 +1805,63 @@
],
"title": "Body_replace_document_route_api_v2_knowledge_documents__doc_id__put"
},
"CascadeHealthBlock": {
"properties": {
"healthy": {
"type": "boolean",
"title": "Healthy"
},
"reasons": {
"items": {
"type": "string"
},
"type": "array",
"title": "Reasons"
},
"pending": {
"type": "integer",
"title": "Pending"
},
"failed_permanent": {
"type": "integer",
"title": "Failed Permanent"
},
"failed_retryable": {
"type": "integer",
"title": "Failed Retryable"
},
"drain_consecutive_failures": {
"type": "integer",
"title": "Drain Consecutive Failures"
},
"unrecoverable_total": {
"type": "integer",
"title": "Unrecoverable Total"
},
"optimize_failure_streak": {
"type": "integer",
"title": "Optimize Failure Streak"
},
"prune_stale_seconds": {
"type": "number",
"title": "Prune Stale Seconds"
}
},
"type": "object",
"required": [
"healthy",
"reasons",
"pending",
"failed_permanent",
"failed_retryable",
"drain_consecutive_failures",
"unrecoverable_total",
"optimize_failure_streak",
"prune_stale_seconds"
],
"title": "CascadeHealthBlock",
"description": "Readiness of the md → LanceDB projection (cascade) subsystem.\n\n``healthy`` reflects **operational** health only — drain loop alive,\noptimize not stuck, version cleanup (prune) not stalled — and is what\nalerting should watch. ``failed_permanent`` (md files awaiting\n``cascade fix``) is a normal data-quality backlog reported as an\ninformational count; it does **not** flip ``healthy``, otherwise the\nsignal would sit red forever."
},
"CategoryDTO": {
"properties": {
"category_id": {
@ -2768,6 +2825,16 @@
},
"type": "array",
"title": "Disabled Features"
},
"cascade": {
"anyOf": [
{
"$ref": "#/components/schemas/CascadeHealthBlock"
},
{
"type": "null"
}
]
}
},
"type": "object",
@ -2778,7 +2845,7 @@
"disabled_features"
],
"title": "HealthResponse",
"description": "Response schema for ``GET /health``.\n\nDeclared as a Pydantic model (not ``dict``) so the generated\nOpenAPI schema carries the full field shape — ``capabilities`` and\n``disabled_features`` are typed. A bare ``-> dict`` return type\ndegrades the OpenAPI response to ``additionalProperties: true``,\nwhich robs clients (and codegen) of any structure to lean on."
"description": "Response schema for ``GET /health``.\n\nDeclared as a Pydantic model (not ``dict``) so the generated\nOpenAPI schema carries the full field shape — ``capabilities``,\n``disabled_features`` and ``cascade`` are typed. A bare ``-> dict``\nreturn type degrades the OpenAPI response to\n``additionalProperties: true``, which robs clients (and codegen) of\nany structure to lean on."
},
"KnowledgeSearchRequest": {
"properties": {

7
pyproject.toml
View File

@ -26,7 +26,12 @@ dependencies = [
"pydantic-settings>=2.0.0",
# Storage stack (md-first three-piece set)
"lancedb>=0.13.0", # Vector + BM25 + scalar filter (Arrow-based)
# Upper bound: 0.35.0 embeds lance-rust v9 (large storage/encoding jump, not
# yet stable-released). 0.32-0.34 carry a compaction offset-overflow regression
# (lance-format/lance#7653); we run 0.34.0 safely via the with_position=False
# FTS workaround. Do not float past 0.34.x until 0.35 is validated by the soak
# harness. Never widen this floor back below 0.34 (older lance can't read v8 data).
"lancedb>=0.34.0,<0.35.0", # Vector + BM25 + scalar filter (Arrow-based)
"aiosqlite>=0.20.0", # Async SQLite driver (used by SA async engine)
"sqlmodel>=0.0.22", # ORM (Pydantic + SQLAlchemy 2.0 async)
"alembic>=1.13.0", # SQLite schema migrations

2
src/everos/component/embedding/factory.py
View File

@ -8,7 +8,7 @@ from everos.config import EmbeddingSettings
from .openai_provider import OpenAIEmbeddingProvider
from .protocol import EmbeddingProvider
# Vector dim for the LanceDB index column — see ``17_lancedb_tables_design.md``.
# Vector dim for the LanceDB index column.
_DEFAULT_DIM = 1024

2
src/everos/component/tokenizer/factory.py
View File

@ -3,7 +3,7 @@
Single implementation today (``JiebaTokenizer``). Lifting this into a
factory keeps callers (cascade handler) decoupled from the concrete
choice, so swapping to char-bigram / hf tokenizer later is a one-file
change — see ``17_lancedb_tables_design.md`` §2.4.1.
change.
"""
from __future__ import annotations

4
src/everos/component/tokenizer/protocol.py
View File

@ -1,7 +1,7 @@
"""Tokenizer protocol.
App-layer tokenisation gates every BM25-indexed field in LanceDB
(``17_lancedb_tables_design.md`` §2.4.1): the source surface form lives
App-layer tokenisation gates every BM25-indexed field in LanceDB:
the source surface form lives
in ``<field>`` while the space-joined token stream lives in
``<field>_tokens``, and the FTS index reads only the latter using a
whitespace tokenizer. Keeping the tokenizer decision in the app layer

3
src/everos/core/persistence/lancedb/base.py
View File

@ -68,8 +68,7 @@ class BaseLanceTable(LanceModel):
is OFF. FTS *does* keep lightweight English-aware normalisation
(``lower_case`` / ``stem`` / ``ascii_folding``) as a belt-and-
braces layer on the same English tokens that survive jieba.
See ``17_lancedb_tables_design.md`` §2.4.1 and
:meth:`ensure_fts_indexes` below for the exact knobs."""
See :meth:`ensure_fts_indexes` below for the exact knobs."""
created_at: dt.datetime = Field(default_factory=get_utc_now)
updated_at: dt.datetime = Field(default_factory=get_utc_now)

135
src/everos/core/persistence/lancedb/repository.py
View File

@ -13,6 +13,7 @@ from __future__ import annotations
import asyncio
import datetime as dt
from collections.abc import Sequence
from pathlib import Path
from typing import Any, ClassVar
from lancedb import AsyncTable
@ -23,6 +24,55 @@ from .base import BaseLanceTable
logger = get_logger(__name__)
# Safety cap on a single prune's ``optimize(cleanup_older_than=…)`` call — a
# pure hang-catcher, not a bound on normal runtime. A real cleanup is
# milliseconds even on a heavily churned table (measured ~40ms at 320k writes /
# 100 versions), so 60s is ~1500× headroom and never fires in normal operation.
# Cleanup only deletes files already unreferenced by the current manifest, so a
# timeout that cancels it mid-scan just reclaims less this beat — it cannot
# corrupt the table — but it releases the per-table write lock instead of
# wedging every writer behind a hung lance cleanup. Kept well below the prune
# cadence (worker ``DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS``) so a hung beat
# leaves a real write window before the next attempt (review P2 / N1).
_PRUNE_TIMEOUT_SECONDS = 60.0
def _remove_empty_index_dirs(table_uri: str) -> int:
"""Delete empty ``_indices/<uuid>/`` husks under a table dir; return count.
``optimize(cleanup_older_than=…)`` deletes the *files* belonging to
superseded index versions but leaves the now-empty per-UUID
directory behind. Over a long-lived churned table these husks
accumulate into tens of thousands of empty dirs (soak: 13061 dirs,
98% empty), which bloats inode usage and slows directory scans even
though they hold no data.
Pure filesystem bookkeeping, no LanceDB manifest involvement — an
empty ``_indices/<uuid>/`` means its files were already cleaned
because no live manifest references that index version, so removing
the directory cannot orphan live data. Must only run while the
table's write lock is held (no concurrent index build could be
mid-populating a freshly-created, still-empty UUID dir). Best-effort:
a dir that becomes non-empty or vanishes between the scan and the
``rmdir`` is skipped, not an error.
"""
indices = Path(table_uri) / "_indices"
if not indices.is_dir():
return 0
removed = 0
for child in indices.iterdir():
if not child.is_dir():
continue
try:
next(child.iterdir()) # has an entry → not empty, keep
except StopIteration:
try:
child.rmdir()
removed += 1
except OSError:
pass # raced (repopulated / already gone) — skip
return removed
def _q(value: str) -> str:
"""Escape single quotes for a LanceDB SQL-like ``where`` predicate.
@ -173,7 +223,7 @@ class LanceRepoBase[T: BaseLanceTable]:
# ── Maintenance ────────────────────────────────────────────────────────
async def optimize(self, *, cleanup_older_than: dt.timedelta | None = None) -> None:
async def optimize(self) -> None:
"""Compact fragments + merge new data into the FTS / vector indexes.
``optimize()`` is a **performance + storage-hygiene** operation,
@ -198,9 +248,9 @@ class LanceRepoBase[T: BaseLanceTable]:
- **Query speed** — the unindexed tail is flat-scanned on every
query; merging it into the index keeps that scan bounded as
ingest accumulates.
- **Storage hygiene** — with ``cleanup_older_than`` it prunes
replaced fragments / stale manifests / dead index files,
bounding the on-disk file count (and FD usage at scan time).
- **Storage hygiene** is *not* done here — physical reclamation
of replaced fragments / stale manifests / dead index files is
:meth:`prune`, a separate write-locked call.
Cascade triggers this through a per-kind throttle + trailing
edge scheduler (``CascadeWorker._schedule_optimize``): at most
@ -210,22 +260,71 @@ class LanceRepoBase[T: BaseLanceTable]:
we cap it under sustained write pressure. Because visibility no
longer depends on it, the throttle window can be generous.
Args:
cleanup_older_than: When set, also prune (physically delete)
files belonging to dataset versions older than this
interval. ``None`` (default) compacts only — historical
manifests, replaced data fragments, and stale index
UUID files are kept on disk forever, which inflates the
file count (and FD usage at scan time) without bound.
Cascade passes a non-None value on a slower beat
(``CascadeWorker._optimize_prune_interval``) so the
hot drain path stays cheap. Note: this does *not*
shrink **active** index internals (FTS ``part_N`` count
or vector index UUID count) — those only collapse via
``drop_index + create_index``, which is not done here.
This is **compaction only** — physical reclamation of superseded
files is :meth:`prune`, a separate write-locked call. Kept lock-free
on purpose: a ``Retryable commit conflict`` against a concurrent
writer is benign here (compaction is not urgent — the next scheduled
beat retries), so it must not stall writers.
"""
table = await self._table()
await table.optimize(cleanup_older_than=cleanup_older_than)
await table.optimize()
async def prune(self, older_than: dt.timedelta) -> None:
"""Physically reclaim files from versions older than ``older_than``.
LanceDB's ``AsyncTable`` cannot clean up independently of compaction —
the only handle is ``optimize(cleanup_older_than=..., delete_unverified=...)``,
which bundles compact + cleanup into one manifest commit. Under
sustained churn that commit is a Rewrite that concurrent Delete /
Update writes preempt, so the bundled cleanup loses the race and
never runs (observed in the storage soak: 16 successes / 547 conflicts
over 21h → the index dir grew unbounded to the disk guardrail).
Fix: run it **under the per-table write lock** so no write is in
flight for its duration. That does two things at once:
1. **No commit conflict** — the Rewrite has the manifest to itself,
so cleanup actually completes every beat.
2. **Cross-process safe** — ``delete_unverified=False`` keeps lance
from deleting any file it cannot tie to a removed version, i.e. a
file a writer in *another process* (a CLI ``cascade sync`` /
``backfill``) may be mid-commit on. The per-table write lock is
in-process only, so it cannot fence a second process; the flag is
what makes concurrent processes safe. Measured to reclaim
identically to ``delete_unverified=True`` on churned tables (both
collapse superseded versions ~97%), because ordinary churn
orphans are all version-referenced and therefore verifiable —
``True`` only additionally deletes in-flight / dangling files,
which is exactly the corruption vector. Reclaiming *during* active
load comes from running under the write lock so the cleanup commit
never loses the manifest race, not from the flag.
After the cleanup commit, the now-empty ``_indices/<uuid>/`` husks
that ``cleanup_older_than`` leaves behind are removed (still under
the lock, so no index build can be mid-populating one); the sweep is
offloaded to a thread so a large dir count does not block the loop.
The trade-off is a brief write stall (~seconds on a churned table,
dominated by the cleanup's file scan/delete — flat, not proportional
to the backlog). Cascade runs it on a slow beat
(``CascadeWorker._optimize_prune_interval``, default 300s), so the
stall is rare. Does *not* shrink **active** index internals (FTS
``part_N`` / index UUID count) — that is ``rebuild_indexes``'s job.
"""
table = await self._table()
async with self._write_lock(self.table_name):
async with asyncio.timeout(_PRUNE_TIMEOUT_SECONDS):
await table.optimize(
cleanup_older_than=older_than, delete_unverified=False
)
table_uri = await table.uri()
removed = await asyncio.to_thread(_remove_empty_index_dirs, table_uri)
if removed:
logger.debug(
"lancedb_pruned_empty_index_dirs",
table=self.table_name,
removed=removed,
)
async def rebuild_indexes(self) -> None:
"""Drop and re-create every index on this table.

2
src/everos/entrypoints/api/lifespans/lancedb.py
View File

@ -91,7 +91,7 @@ class LanceDBLifespanProvider(LifespanProvider):
2. ``verify_business_schemas`` — fail loud if an on-disk table's
columns drift from the current Pydantic schema. LanceDB has no
online migration; cascade is rebuildable from md so the recovery
is documented as ``rm -rf ~/.everos/.index/lancedb``.
is ``everos cascade rebuild`` (see ``docs/cascade_runbook.md``).
3. ``ensure_business_indexes`` — idempotent FTS index creation.
4. ``_log_unbackfilled_hint`` — warn if unbackfilled rows exist.
"""

91
src/everos/entrypoints/api/routes/health.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations
from fastapi import APIRouter
from fastapi import APIRouter, Request
from pydantic import BaseModel
from everos import __version__
@ -11,6 +11,10 @@ from everos.component.embedding import get_embedding_capability
from everos.component.multimodal import get_multimodal_llm_capability
from everos.component.parser import parser_available
from everos.component.rerank import get_rerank_capability
from everos.core.observability.logging import get_logger
from everos.entrypoints.api.utils import cascade_orchestrator
logger = get_logger(__name__)
router = APIRouter(tags=["health"])
@ -29,25 +33,62 @@ class HealthCapabilities(BaseModel):
parser: bool
class CascadeHealthBlock(BaseModel):
"""Readiness of the md → LanceDB projection (cascade) subsystem.
``healthy`` reflects **operational** health only — drain loop alive,
optimize not stuck, version cleanup (prune) not stalled — and is what
alerting should watch. ``failed_permanent`` (md files awaiting
``cascade fix``) is a normal data-quality backlog reported as an
informational count; it does **not** flip ``healthy``, otherwise the
signal would sit red forever.
"""
healthy: bool
reasons: list[str]
pending: int
failed_permanent: int
failed_retryable: int
drain_consecutive_failures: int
unrecoverable_total: int
optimize_failure_streak: int
prune_stale_seconds: float
class HealthResponse(BaseModel):
"""Response schema for ``GET /health``.
Declared as a Pydantic model (not ``dict``) so the generated
OpenAPI schema carries the full field shape — ``capabilities`` and
``disabled_features`` are typed. A bare ``-> dict`` return type
degrades the OpenAPI response to ``additionalProperties: true``,
which robs clients (and codegen) of any structure to lean on.
OpenAPI schema carries the full field shape — ``capabilities``,
``disabled_features`` and ``cascade`` are typed. A bare ``-> dict``
return type degrades the OpenAPI response to
``additionalProperties: true``, which robs clients (and codegen) of
any structure to lean on.
"""
status: str
version: str
capabilities: HealthCapabilities
disabled_features: list[str]
cascade: CascadeHealthBlock | None = None
"""Present when the cascade lifespan is running; ``None`` for a
minimal app built without it."""
@router.get("/health", response_model=HealthResponse)
async def health() -> HealthResponse:
"""Liveness probe with capabilities and disabled features."""
async def health(request: Request) -> HealthResponse:
"""Liveness + capabilities + cascade readiness probe.
``status`` stays ``"ok"`` whenever the process is up — the HTTP code
is a *liveness* signal and a degraded cascade must not trigger a
restart (a crash-loop fixes neither a bad md file nor disk bloat).
The ``cascade`` block is the *readiness* signal: ``healthy=false``
with human-readable ``reasons`` **only** when the projection pipeline
itself is stuck (drain failing, optimize stuck, version cleanup
stalled). ``failed_permanent`` — files awaiting ``cascade fix`` — is
a data-quality backlog reported as an informational count that does
not flip ``healthy``. Alert on ``cascade.healthy``.
"""
# ``llm`` is hardcoded ``True`` — kept for symmetry with the caps
# dict rather than probed live. Rationale: LLM is a Tier-1 hard
# requirement enforced at startup by ``LLMLifespanProvider``
@ -65,9 +106,45 @@ async def health() -> HealthResponse:
multimodal_llm=get_multimodal_llm_capability().available,
parser=parser_available(),
)
cascade: CascadeHealthBlock | None = None
orch = cascade_orchestrator(request)
if orch is not None:
try:
ch = await orch.health()
cascade = CascadeHealthBlock(
healthy=ch.healthy,
reasons=ch.reasons,
pending=ch.pending,
failed_permanent=ch.failed_permanent,
failed_retryable=ch.failed_retryable,
drain_consecutive_failures=ch.drain_consecutive_failures,
unrecoverable_total=ch.unrecoverable_total,
optimize_failure_streak=ch.optimize_failure_streak,
prune_stale_seconds=round(ch.prune_stale_seconds, 1),
)
except Exception as exc:
# The probe reads SQLite (queue_summary runs aggregate counts).
# A locked / full / mid-migration DB must NOT turn /health into a
# 500 — that flips the liveness signal and makes k8s restart the
# container, which fixes neither a stuck DB nor disk bloat (see
# the handler docstring). Surface it as unhealthy *readiness*
# with a reason and keep HTTP 200.
logger.warning("cascade_health_probe_failed", error=repr(exc))
cascade = CascadeHealthBlock(
healthy=False,
reasons=[f"cascade health probe failed: {exc!r}"],
pending=0,
failed_permanent=0,
failed_retryable=0,
drain_consecutive_failures=0,
unrecoverable_total=0,
optimize_failure_streak=0,
prune_stale_seconds=0.0,
)
return HealthResponse(
status="ok",
version=__version__,
capabilities=caps,
disabled_features=compute_disabled_features(caps.model_dump()),
cascade=cascade,
)

14
src/everos/entrypoints/api/utils.py
View File

@ -5,9 +5,23 @@ from __future__ import annotations
from fastapi import Request
from everos.core.observability.tracing import gen_request_id
from everos.memory.cascade import CascadeOrchestrator
def extract_request_id(request: Request) -> str:
"""Return the request_id set by middleware, or mint a fresh fallback."""
rid = getattr(request.state, "request_id", None)
return str(rid) if rid else gen_request_id()
def cascade_orchestrator(request: Request) -> CascadeOrchestrator | None:
"""Return the running cascade orchestrator, or ``None``.
The cascade lifespan stashes the orchestrator at
``app.state.lifespan_data["cascade"]``. An app built without that
lifespan (e.g. a minimal test app) has no entry, so callers get
``None`` and degrade gracefully instead of erroring.
"""
data = getattr(request.app.state, "lifespan_data", None) or {}
orch = data.get("cascade")
return orch if isinstance(orch, CascadeOrchestrator) else None

71
src/everos/entrypoints/cli/commands/cascade.py
View File

@ -1,6 +1,6 @@
"""``everos cascade`` subcommand group.
Three one-shot operations on the cascade subsystem, all run in-process
One-shot operations on the cascade subsystem, all run in-process
without standing up the FastAPI app:
- ``cascade sync [PATH]`` — flush the work queue. With ``PATH`` the
@ -15,6 +15,11 @@ without standing up the FastAPI app:
vectors, build clusters, extract skills. See
:func:`everos.entrypoints.cli.commands._backfill_cmd.run_backfill`
for the phase orchestration.
- ``cascade rebuild`` — drop every business LanceDB table and re-index
all md from scratch. Recovery for a drifted / corrupt index; safe
because md is the source of truth and un-extracted buffered messages
are preserved. Skips the schema-verify guard (which the drift would
otherwise trip on startup).
CLI is in-process (12 doc §7.1 + 16 doc §9.2): it constructs the same
:class:`CascadeOrchestrator` as the daemon but only calls
@ -43,6 +48,7 @@ from everos.entrypoints.cli._log_setup import configure_cli_logging
from everos.entrypoints.cli.commands._backfill_cmd import run_backfill
from everos.infra.persistence.lancedb import (
dispose_connection,
drop_business_tables,
ensure_business_indexes,
get_connection,
verify_business_schemas,
@ -125,18 +131,24 @@ _VERBOSE_OPTION_HELP = (
@asynccontextmanager
async def _runtime(): # type: ignore[no-untyped-def]
async def _runtime(*, verify: bool = True): # type: ignore[no-untyped-def]
"""Stand up sqlite + lancedb the same way the API lifespan would.
The CLI piggybacks on the same singletons as the running daemon
(lazy + process-wide), so if a server happens to be running on
the same memory root, both share state correctly.
``verify=False`` skips :func:`verify_business_schemas` — required by
``cascade rebuild``, whose whole purpose is to recover from a table
whose schema *has* drifted; running the guard there would abort
startup before the rebuild could fix it (chicken-and-egg).
"""
engine = get_engine()
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
await get_connection()
await verify_business_schemas()
if verify:
await verify_business_schemas()
await ensure_business_indexes()
try:
yield
@ -411,6 +423,59 @@ def backfill(
raise typer.Exit(code=code)
# ── rebuild ────────────────────────────────────────────────────────────────
@app.command("rebuild")
def rebuild(
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip the confirmation prompt."),
] = False,
) -> None:
"""Rebuild the LanceDB index from markdown (recover from schema drift).
Drops every business LanceDB table and re-indexes all md from
scratch. Markdown is the source of truth, so no memory content is
lost, and this is the **safe** recovery from a drifted / corrupt
index (e.g. the ``verify_business_schemas`` startup failure):
- unlike ``rm -rf ~/.everos/.index/lancedb``, it re-populates
already-indexed entries (that command leaves the cascade queue
marked ``done``, so nothing re-indexes and the index comes back
empty);
- unlike ``rm -rf ~/.everos/.index``, it preserves SQLite state that
is NOT rebuildable from md — notably ``unprocessed_buffer``
(messages received but not yet extracted).
"""
if not yes:
typer.confirm(
"Drop all LanceDB business tables and re-index from markdown?",
abort=True,
)
async def _run() -> None:
# verify=False: the on-disk schema may be exactly what we're here
# to fix; the startup guard would abort before we could rebuild.
async with _runtime(verify=False):
dropped = await drop_business_tables()
typer.echo(
f"dropped {len(dropped)} LanceDB table(s): "
f"{', '.join(dropped) or '(none)'}"
)
# Recreate the tables (current schema) + FTS indexes.
await ensure_business_indexes()
# Clear the work queue so every md file re-enqueues as `added`.
cleared = await md_change_state_repo.reset_all()
typer.echo(f"reset {cleared} cascade queue row(s)")
# Re-scan + drain: re-embed and re-insert every md entry.
orchestrator = _build_orchestrator()
processed = await orchestrator.sync_once()
typer.echo(f"rebuild complete — re-indexed {processed} md file(s)")
asyncio.run(_run())
# ── helpers ──────────────────────────────────────────────────────────────

84
src/everos/infra/persistence/lancedb/__init__.py
View File

@ -21,7 +21,7 @@ External usage::
Three index kinds: scalar / BM25 / vector. Tables are created lazily on
first access; row population is the cascade daemon's job (see
``12_cascade_design.md``).
``docs/cascade_runbook.md``).
"""
import contextlib
@ -34,6 +34,7 @@ from everos.core.persistence import BaseLanceTable, MemoryRoot, memory_root_lock
# schema so callers can rely on the package alone to surface every schema.
from . import tables as tables
from .lancedb_manager import dispose_connection as dispose_connection
from .lancedb_manager import drop_tables as _drop_tables
from .lancedb_manager import get_connection as get_connection
from .lancedb_manager import get_table as get_table
from .repos import agent_case_repo as agent_case_repo
@ -68,9 +69,10 @@ class LanceDBSchemaMismatchError(RuntimeError):
from the corresponding Pydantic schema.
Cascade re-builds LanceDB from md (the SoT), so the recovery is
deterministic: delete the index directory and let it reindex.
The lifespan surfaces the explicit ``rm -rf ~/.everos/.index/
lancedb`` instruction in the error message; see
deterministic: ``everos cascade rebuild`` drops the business tables
and re-indexes from md, preserving SQLite state that is *not*
rebuildable from md (notably ``unprocessed_buffer`` — messages not
yet extracted). The error message surfaces that command; see
``docs/cascade_runbook.md`` for the wider context.
"""
@ -292,39 +294,74 @@ async def ensure_business_indexes() -> None:
async def verify_business_schemas() -> None:
"""Fail loud at startup if an existing LanceDB table's columns don't
match its current Pydantic schema.
match its current Pydantic schema — in **name or type**.
LanceDB doesn't migrate columns automatically; an older index dir
(e.g. with the pre-``content_sha256`` shape) would fail
unpredictably on upsert. Checking column names up-front turns that
into a clean startup error pointing the user at the recovery path
(``rm -rf ~/.everos/.index/lancedb`` — the index is rebuildable
from md, see ``12_cascade_design.md``).
would fail unpredictably on upsert. Checking the schema up-front
turns that into a clean startup error pointing the user at the
recovery path (``everos cascade rebuild`` — re-indexes from md,
preserving un-extracted buffered messages; see
``docs/cascade_runbook.md``). A bare ``rm -rf`` of the index dir is
*not* the recovery — it leaves the cascade queue marked ``done`` so
nothing re-indexes and the index comes back empty.
Both dimensions are checked against ``schema.to_arrow_schema()`` —
the exact schema ``get_table`` builds the table from, so a healthy
table never false-positives:
* **Column set** — a missing / extra column (e.g. a pre-``content_sha256``
table) is caught by name.
* **Column type** — a column whose on-disk Arrow type drifted from
the current schema. This is the class of drift behind EverOS #337:
an ``episode.subject_vector`` column left as ``string`` (or ``null``)
by an older build, while the current schema declares a 1024-d
``fixed_size_list``. The name matches, so a name-only check waves it
through and it detonates deep inside ``merge_insert`` as an opaque
``LanceError(IO): Spill has sent an error``. Comparing types surfaces
it here instead.
"""
for schema in _BUSINESS_SCHEMAS:
table = await get_table(schema.TABLE_NAME, schema)
arrow_schema = await table.schema()
actual = set(arrow_schema.names)
expected = set(schema.model_fields.keys())
missing = expected - actual
extra = actual - expected
if missing or extra:
on_disk = await table.schema()
expected = schema.to_arrow_schema()
on_disk_names = set(on_disk.names)
expected_names = set(expected.names)
missing = expected_names - on_disk_names
extra = on_disk_names - expected_names
# Type drift on columns present in both, compared against the
# authoritative to_arrow_schema() Arrow types.
type_drift = [
f"{name}: on-disk {on_disk.field(name).type} "
f"!= expected {expected.field(name).type}"
for name in sorted(on_disk_names & expected_names)
if not on_disk.field(name).type.equals(expected.field(name).type)
]
if missing or extra or type_drift:
raise LanceDBSchemaMismatchError(
f"LanceDB table {schema.TABLE_NAME!r} schema drift: "
f"missing={sorted(missing)}, extra={sorted(extra)}.\n"
f"missing={sorted(missing)}, extra={sorted(extra)}, "
f"type_drift={type_drift}.\n"
"Recovery, escalating:\n"
" 1. Restart the server — an in-flight migration may "
"still be finishing (harmless if this is your first "
"restart after upgrading EverOS).\n"
" 2. If restart doesn't clear it, the on-disk index is "
"genuinely out of sync with the code's schema. Because "
"the LanceDB index is fully rebuildable from md, wipe "
"it: `rm -rf ~/.everos/.index/lancedb` and restart — "
"the cascade daemon will re-index from the source-of-"
"truth markdown."
" 2. If restart doesn't clear it, recover with "
"`everos cascade rebuild` (drops + re-indexes from md, "
"preserving un-extracted buffered messages)."
)
async def drop_business_tables() -> list[str]:
"""Drop every business LanceDB table; return the names dropped.
The tables are a rebuildable projection of markdown, so dropping is
non-destructive to memory content — ``cascade rebuild`` recreates and
re-populates them from md. Evicts the dropped tables from the manager
cache so a later :func:`get_table` reopens the fresh table.
"""
return await _drop_tables([schema.TABLE_NAME for schema in _BUSINESS_SCHEMAS])
__all__ = [
"BUSINESS_SCHEMAS_WITH_VECTOR",
"AgentCase",
@ -341,6 +378,7 @@ __all__ = [
"agent_skill_repo",
"atomic_fact_repo",
"dispose_connection",
"drop_business_tables",
"ensure_business_indexes",
"episode_repo",
"foresight_repo",

22
src/everos/infra/persistence/lancedb/lancedb_manager.py
View File

@ -11,6 +11,7 @@ manually.
from __future__ import annotations
import asyncio
from collections.abc import Sequence
from lancedb import AsyncConnection, AsyncTable
@ -53,6 +54,27 @@ async def get_table(
return _tables[name]
async def drop_tables(names: Sequence[str]) -> list[str]:
"""Drop the named tables if present; return the names actually dropped.
Each dropped table is also evicted from the cache so a later
:func:`get_table` recreates it fresh from the current schema. Used by
``cascade rebuild`` to reset a corrupt / drifted index — the tables
are a rebuildable projection of markdown, so dropping is safe.
"""
async with _lock:
conn = await _ensure_connection_locked()
existing = set((await conn.list_tables()).tables)
dropped: list[str] = []
for name in names:
if name in existing:
await conn.drop_table(name)
_tables.pop(name, None)
dropped.append(name)
logger.info("lancedb_table_dropped", name=name)
return dropped
async def dispose_connection() -> None:
"""Close the connection + clear table cache. Idempotent."""
global _conn

2
src/everos/infra/persistence/lancedb/tables/agent_case.py
View File

@ -1,6 +1,6 @@
"""LanceDB ``agent_case`` table schema.
Field set per 17_lancedb_tables_design.md §3.4. Each row records one
Field set for the agent-case LanceDB row. Each row records one
task an agent worked on: intent, approach, optional pivotal insight,
and a quality score. A MemCell extracted on the agent's own execution
log yields at most one AgentCase.

2
src/everos/infra/persistence/lancedb/tables/agent_skill.py
View File

@ -1,6 +1,6 @@
"""LanceDB ``agent_skill`` table schema.
Field set per 17_lancedb_tables_design.md §3.5. AgentSkill is a *named
Field set for the agent-skill LanceDB row. AgentSkill is a *named
entity* rather than a daily-log entry — PK is ``<owner_id>_<skill_name>``
(no date / seq), and same agent + same name is the same row (upsert).

2
src/everos/infra/persistence/lancedb/tables/atomic_fact.py
View File

@ -1,6 +1,6 @@
"""LanceDB ``atomic_fact`` table schema.
Field set per 17_lancedb_tables_design.md §3.2. Each row carries one
Field set for the atomic-fact LanceDB row. Each row carries one
atomic fact extracted by the algo layer; the parent is always the source
MemCell — recorded via ``parent_type`` / ``parent_id``.
"""

3
src/everos/infra/persistence/lancedb/tables/episode.py
View File

@ -72,8 +72,7 @@ class Episode(BaseLanceTable):
embedding-relevant fields changed — the entry is skipped (no
re-upsert, no re-embed). Inline audit fields (owner_id /
session_id / timestamp / parent_id / sender_ids) are intentionally
NOT in the hash so editing them doesn't waste an embedding call.
See ``16_cascade_impl_design.md`` §3.3."""
NOT in the hash so editing them doesn't waste an embedding call."""
deprecated_by: str | None = None
"""Soft-delete marker set by Reflection when this episode is

2
src/everos/infra/persistence/lancedb/tables/foresight.py
View File

@ -1,6 +1,6 @@
"""LanceDB ``foresight`` table schema.
Field set per 17_lancedb_tables_design.md §3.3. Each row carries a
Field set for the foresight LanceDB row. Each row carries a
forward-looking inference about the user (intent window, planned
action, projected need); ``start_time`` / ``end_time`` describe the
window the foresight applies to.

20
src/everos/infra/persistence/sqlite/repos/md_change_state.py
View File

@ -29,7 +29,7 @@ from __future__ import annotations
import dataclasses
from sqlalchemy import func, select, text, update
from sqlalchemy import delete, func, select, text, update
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
@ -418,6 +418,24 @@ class _MdChangeStateRepo(RepoBase[MdChangeState]):
await s.commit()
return int(result.rowcount or 0)
async def reset_all(self) -> int:
"""`cascade rebuild` engine: clear the entire work-queue table.
This table is pure sync bookkeeping — a projection of (md ×
index) — so deleting every row forces the next scan to treat
every md file as newly ``added`` and re-index it from scratch.
Touches **only** ``md_change_state``; other SQLite tables
(``memcell``, ``unprocessed_buffer``, …) are left intact, which
is what makes ``cascade rebuild`` non-destructive to un-extracted
buffered messages.
Returns the number of rows deleted.
"""
async with session_scope(self._factory) as s:
result = await s.execute(delete(MdChangeState))
await s.commit()
return int(result.rowcount or 0)
async def queue_summary(self) -> QueueSummary:
"""Aggregate the table for the ``cascade status`` CLI."""
async with session_scope(self._factory) as s:

5
src/everos/infra/persistence/sqlite/tables/md_change_state.py
View File

@ -6,9 +6,8 @@ scanner (periodic sweep) UPSERT into this table; the worker consumes
internal ``processing`` claim state, and lands them in ``done`` or
``failed`` (with a ``retryable`` flag).
Schema sourced from ``12_cascade_design.md`` §4.1 + decisions DD-3 …
DD-12; the four indexes below are required by ``13_cascade_design.md``
§7 status / fix queries.
The schema + the four indexes below back the cascade queue and its
status / fix queries (see ``docs/cascade_runbook.md``).
"""
from __future__ import annotations

2
src/everos/memory/cascade/__init__.py
View File

@ -22,6 +22,7 @@ from ._backfill import BackfillPhase as BackfillPhase
from ._backfill import BackfillPresenter as BackfillPresenter
from ._backfill import NullBackfillPresenter as NullBackfillPresenter
from .orchestrator import CascadeConfig as CascadeConfig
from .orchestrator import CascadeHealth as CascadeHealth
from .orchestrator import CascadeOrchestrator as CascadeOrchestrator
from .registry import KIND_REGISTRY as KIND_REGISTRY
from .registry import KindSpec as KindSpec
@ -32,6 +33,7 @@ __all__ = [
"BackfillPhase",
"BackfillPresenter",
"CascadeConfig",
"CascadeHealth",
"CascadeOrchestrator",
"KindSpec",
"NullBackfillPresenter",

9
src/everos/memory/cascade/_backfill.py
View File

@ -744,7 +744,14 @@ async def _backfill_table(
# path (see ``lancedb/__init__.py:140``).
if result.rows_processed > 0:
try:
await backlog.spec.repo.optimize(cleanup_older_than=dt.timedelta(0))
# optimize() compacts the per-row-update fragments; prune()
# physically reclaims the superseded manifest versions. Split
# after the repo API separated them (compact is lock-free, prune
# runs under the write lock). prune(0) reclaims everything now;
# it is cross-process safe (delete_unverified=False), so running
# ``backfill`` alongside a live daemon cannot corrupt the table.
await backlog.spec.repo.optimize()
await backlog.spec.repo.prune(dt.timedelta(0))
logger.info(
"cascade_backfill_table_optimized",
table=backlog.table_name,

57
src/everos/memory/cascade/orchestrator.py
View File

@ -31,6 +31,37 @@ from .worker import CascadeWorker
logger = get_logger(__name__)
@dataclasses.dataclass(frozen=True)
class CascadeHealth:
"""Cascade health verdict for ``/health``.
``healthy`` reflects **operational** health only — is the md →
LanceDB projection pipeline itself working: drain loop alive,
optimize not stuck, version cleanup (prune) not stalled. It is the
boolean ops/alerting acts on.
``failed_permanent`` is deliberately **not** part of that verdict:
a handful of md files failing to index is a normal data-quality
backlog (almost always non-zero in a real deployment), so folding
it into ``healthy`` would pin the signal red forever. It is reported
as an informational count; per-file triage lives in the
``cascade status`` / ``cascade fix`` CLI. ``reasons`` is the
operational "why not" list (empty when healthy).
"""
healthy: bool
reasons: list[str]
pending: int
failed_permanent: int
"""Informational: md files awaiting ``cascade fix``. Does NOT affect
:attr:`healthy` (see class docstring)."""
failed_retryable: int
drain_consecutive_failures: int
unrecoverable_total: int
optimize_failure_streak: int
prune_stale_seconds: float
@dataclasses.dataclass(frozen=True)
class CascadeConfig:
"""Construction-time knobs for the orchestrator.
@ -145,3 +176,29 @@ class CascadeOrchestrator:
async def queue_summary(self) -> QueueSummary:
"""Forward to the repo so callers don't reach past this class."""
return await md_change_state_repo.queue_summary()
async def health(self) -> CascadeHealth:
"""Verdict for ``/health``: operational health + informational counts.
One query (:meth:`queue_summary`) plus the worker's in-memory
counters. ``healthy`` is driven **only** by operational signals
(:meth:`CascadeWorkerHealth.reasons` — drain / optimize / prune),
never by ``failed_permanent``: a per-file triage backlog is normal
steady state and must not pin the health signal red (see
:class:`CascadeHealth`). ``failed_permanent`` is still reported as
an informational count.
"""
wh = self._worker.health()
summary = await self.queue_summary()
reasons = wh.reasons() # operational only
return CascadeHealth(
healthy=not reasons,
reasons=reasons,
pending=summary.pending,
failed_permanent=summary.failed_permanent,
failed_retryable=summary.failed_retryable,
drain_consecutive_failures=wh.drain_consecutive_failures,
unrecoverable_total=wh.unrecoverable_total,
optimize_failure_streak=wh.optimize_failure_streak,
prune_stale_seconds=wh.prune_stale_seconds,
)

3
src/everos/memory/cascade/registry.py
View File

@ -9,8 +9,7 @@ same path — :func:`match_kind` returns the first match.
Path matching uses :class:`pathlib.PurePosixPath.match` (not bare
``fnmatch``) so that ``*`` matches a single path component, never the
``/`` separator — see ``17_lancedb_tables_design.md`` §2.4.2 and
``12_cascade_design.md`` §5.1 (path filter is a single whitelist layer).
``/`` separator (the path filter is a single whitelist layer).
"""
from __future__ import annotations

273
src/everos/memory/cascade/worker.py
View File

@ -75,12 +75,23 @@ DEFAULT_RETRY_BACKOFF_SECONDS = 2.0
DEFAULT_OPTIMIZE_MIN_INTERVAL_SECONDS = 10.0
DEFAULT_OPTIMIZE_HEARTBEAT_SECONDS = 60.0
_OPTIMIZE_FAILURE_ALERT_THRESHOLD = 5
"""Consecutive ``optimize()`` failures (per kind) before the log is
escalated from ``warning`` to ``error``. A one-off failure is benign
(next tick retries); a sustained streak means compaction + version
cleanup are stuck and the index dir will grow unbounded — that must
surface to health checks / alerting rather than rot as a warning nobody
reads (the failure mode behind lance-format/lance#7653)."""
"""Consecutive **non-benign** ``optimize()`` failures (per kind) before
the log escalates ``warning``→``error``, a fallback rebuild is triggered
(:meth:`_run_rebuild_once`), and :meth:`CascadeWorker.health` reports the
kind degraded. A benign light-beat commit conflict (lost concurrency
race) is expected under churn, logged at ``debug``, and does **not**
count — otherwise the streak pins high on a busy table and drowns the
real signal (the disk-bloat failure mode behind lance-format/lance#7653)."""
_DRAIN_FAILURE_ALERT_THRESHOLD = 3
"""Consecutive :meth:`drain_once` exceptions at or above which
:meth:`CascadeWorker.health` reports degraded — the md → LanceDB
projection is not making progress, so accepted writes are not indexed."""
_PRUNE_STALE_FACTOR = 3.0
"""Multiple of the prune interval without any successful prune before
:meth:`CascadeWorker.health` reports degraded. The primary disk-bloat
signal — fires whether the cause is lost commit-conflict races, a stuck
heavy beat, or a wedged worker, without depending on a particular
exception showing up."""
_MAX_TOTAL_RETRIES = 12
"""Total retry budget across scanner re-enqueue cycles.
@ -118,30 +129,46 @@ in the regular hot path will do the same job for ~free.
"""
DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS = 300.0
"""How often (per kind) to add ``cleanup_older_than`` to ``optimize()``.
"""**Cadence** — how often (per kind) the heavy write-locked
:meth:`LanceRepoBase.prune` beat runs, vs the light lock-free
:meth:`~LanceRepoBase.optimize` compaction on every other tick.
``optimize()`` without ``cleanup_older_than`` compacts fragments and
merges new data into indexes, but **leaves stale physical files on disk
forever** (replaced data fragments, historical manifests, stale index
UUID files). On a lightweight (single-user / small-team) deployment
with steady-state cascade ingest, that file count grows without bound
and eventually
exhausts file descriptors at index-scan time (observed: macOS / Linux
default ``ulimit -n`` of 1024 — the ``os error 24`` reported in CI).
The prune itself is cheap when scoped to recent versions; we just don't
want to pay it on every optimize throttle tick. 5 minutes is the
shortest interval that comfortably outlives any in-flight query / index
build, while keeping the on-disk footprint bounded. It is also passed
as ``cleanup_older_than`` itself (semantically: "the retention window
equals the prune cadence") — every file replaced more than one cadence
ago becomes eligible.
prune holds the per-table write lock (brief same-table write stall,
~seconds on a churned table), so it runs on this slow beat rather than
every throttle tick. This is the *frequency* of reclamation; how much
history each prune keeps is a separate knob — see
:data:`DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS`.
Does **not** shrink active index internals (FTS ``part_N`` count or
vector index UUID count): those only collapse via ``drop_index +
create_index``, which is intentionally out of scope here.
"""
DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS = 60.0
"""**Retention window** — ``cleanup_older_than`` passed to prune: files
belonging to dataset versions replaced more than this long ago become
eligible for physical deletion.
Decoupled from (and much shorter than) the prune *cadence*: prune runs
under the per-table write lock, so no concurrent writer can be
referencing an old version, and the only thing the window must outlive
is an **in-flight read** (a ``/search`` holding a version reference) —
sub-second to a few seconds. 60s is comfortably safe.
Why short matters: under sustained churn each compaction leaves a
full-table-sized *superseded* fragment behind; the retention window is
how long those pile up before reclamation. The storage soak showed a
300s window × the churn rate retaining ~24 full-table copies (a
transient ~15G/table peak that reclaimed to ~625MB live once churn
eased). Shrinking the window to 60s cuts that transient footprint ~5×
without changing the reclaimed floor. Tune via the constructor."""
_PRUNE_STALE_SECONDS_ALERT = (
_PRUNE_STALE_FACTOR * DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS
)
"""Absolute prune-staleness alert threshold (seconds) = 900s under the
shipped 300s prune cadence (three missed heavy beats)."""
@dataclass
class _KindOptimizerState:
@ -156,14 +183,26 @@ class _KindOptimizerState:
per kind so concurrent LanceDB writes never collide on the same
table's manifest.
``last_prune_at`` is the monotonic timestamp of the last
``optimize()`` call that passed ``cleanup_older_than``; the runner
consults it to decide whether the next call should also prune. ``0``
means "never pruned" — the first run after worker startup always
prunes, which is what we want for catching up from a prior session.
Two prune clocks, deliberately split:
- ``last_prune_attempt_at`` gates **scheduling** — the monotonic time
of the last heavy (prune) beat *attempt*, advanced whether it
succeeds or times out. So a prune that hangs and is killed by the
write-lock timeout backs off a full cadence before the next attempt
instead of retrying every ~10s and holding the lock ~97% of the
time (review N1).
- ``last_prune_at`` records the last *successful* prune, advanced only
after the call returns. It drives the prune-staleness **health**
signal (:meth:`CascadeWorker._prune_stale_seconds`), so a
persistently failing/hanging prune still surfaces as degraded
instead of being masked by advancing the schedule clock.
Both default ``0`` ("never") — the first run after worker startup
always prunes, catching up from a prior session.
"""
last_run_at: float = 0.0
last_prune_attempt_at: float = 0.0
last_prune_at: float = 0.0
dirty: bool = False
optimize_failures: int = 0
@ -182,6 +221,72 @@ class _KindOptimizerState:
"""
@dataclass(frozen=True)
class CascadeWorkerHealth:
"""Snapshot of the worker's in-memory health signals.
Cheap to produce (no IO — pure in-process counters). The orchestrator
combines it with the SQLite queue summary for the full verdict.
"""
drain_consecutive_failures: int
"""Consecutive :meth:`drain_once` exceptions; 0 when the last drain
completed cleanly."""
unrecoverable_total: int
"""Cumulative unrecoverable handler failures since worker start —
each is a md file whose projection to LanceDB permanently failed
and now needs a user edit (surfaced by ``cascade fix``)."""
optimize_failure_streak: int
"""Max consecutive non-benign optimize/prune failure count across
kinds; benign light-beat commit conflicts do not count."""
prune_stale_seconds: float
"""Seconds since the most recent successful prune (version cleanup)
across all active kinds, measured from worker start if none has run
yet. ``0`` when there has been no write activity to prune."""
def reasons(self) -> list[str]:
"""Operational degradation reasons; empty when healthy.
Combined with the SQLite ``failed_permanent`` count in
:meth:`CascadeOrchestrator.health` to produce the ``/health``
verdict. These in-memory signals do not include the permanent-
failure backlog (that needs a query)."""
out: list[str] = []
if self.drain_consecutive_failures >= _DRAIN_FAILURE_ALERT_THRESHOLD:
out.append(
f"drain loop failing ({self.drain_consecutive_failures} in a row)"
)
if self.optimize_failure_streak >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD:
out.append(f"optimize stuck ({self.optimize_failure_streak} in a row)")
if self.prune_stale_seconds >= _PRUNE_STALE_SECONDS_ALERT:
out.append(
f"version cleanup stalled ({int(self.prune_stale_seconds)}s "
"since last prune — disk may grow)"
)
return out
def _is_benign_commit_conflict(exc: BaseException) -> bool:
"""True for a LanceDB optimistic-concurrency retry error.
LanceDB surfaces the Rust ``Retryable commit conflict`` as a plain
exception whose message carries the phrase — there is no dedicated
class to catch. On the lock-free light beat this is expected under
concurrent writes and benign (the next beat retries), so it is
logged at ``debug`` and does not count toward the failure streak.
Match only the specific ``commit conflict`` phrase (a substring of the
Rust message), not a bare ``retryable`` — the latter appears in the
repr of unrelated recoverable errors (e.g. ``ExternalServiceError``
carrying ``retryable=True``) and would silently swallow real failures
(review P2).
"""
return "commit conflict" in str(exc).lower()
class CascadeWorker:
"""Owns the claim → dispatch → mark cycle.
@ -203,6 +308,9 @@ class CascadeWorker:
optimize_prune_interval_seconds: float = (
DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS
),
optimize_prune_retention_seconds: float = (
DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS
),
optimize_rebuild_interval_seconds: float = (
DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS
),
@ -215,17 +323,23 @@ class CascadeWorker:
self._optimize_min_interval = optimize_min_interval_seconds
self._optimize_heartbeat = optimize_heartbeat_seconds
self._optimize_prune_interval = optimize_prune_interval_seconds
self._optimize_prune_retention = optimize_prune_retention_seconds
self._optimize_rebuild_interval = optimize_rebuild_interval_seconds
self._task: asyncio.Task[None] | None = None
self._heartbeat_task: asyncio.Task[None] | None = None
self._rebuild_task: asyncio.Task[None] | None = None
self._stop = asyncio.Event()
self._optimizer_states: dict[str, _KindOptimizerState] = {}
# ── in-memory health signals (see :meth:`health`) ──────────────
self._started_at: float = 0.0
self._drain_consecutive_failures: int = 0
self._unrecoverable_total: int = 0
async def start(self) -> None:
if self._task is not None:
return
self._stop.clear()
self._started_at = time.monotonic()
self._task = asyncio.create_task(self._run_loop(), name="cascade-worker")
self._heartbeat_task = asyncio.create_task(
self._heartbeat_loop(), name="cascade-worker-heartbeat"
@ -329,14 +443,52 @@ class CascadeWorker:
await self._flush_optimizers()
return total
def health(self) -> CascadeWorkerHealth:
"""Snapshot the worker's in-memory health signals (no IO).
Combines the drain-loop and unrecoverable counters with the worst
per-kind optimize streak and the prune-staleness clock.
"""
states = self._optimizer_states.values()
optimize_failure_streak = max((s.optimize_failures for s in states), default=0)
return CascadeWorkerHealth(
drain_consecutive_failures=self._drain_consecutive_failures,
unrecoverable_total=self._unrecoverable_total,
optimize_failure_streak=optimize_failure_streak,
prune_stale_seconds=self._prune_stale_seconds(),
)
def _prune_stale_seconds(self) -> float:
"""Seconds since the most recent successful prune across kinds,
measured from worker start when nothing has pruned yet.
Returns ``0`` before the worker has started (``_started_at == 0``)
or before any kind has registered an optimizer state — no prune
beat has run yet, so there is nothing to be stale about. Once a
beat registers state, staleness is the time since the newest
``last_prune_at`` (or since start, whichever is later).
"""
states = list(self._optimizer_states.values())
if not states or self._started_at == 0.0:
return 0.0
latest_prune = max(s.last_prune_at for s in states)
baseline = max(latest_prune, self._started_at)
return max(0.0, time.monotonic() - baseline)
# ── internals ──────────────────────────────────────────────────────────
async def _run_loop(self) -> None:
while not self._stop.is_set():
try:
processed = await self.drain_once()
self._drain_consecutive_failures = 0
except Exception as exc:
logger.exception("cascade_worker_drain_failed", error=str(exc))
self._drain_consecutive_failures += 1
logger.exception(
"cascade_worker_drain_failed",
error=str(exc),
consecutive_failures=self._drain_consecutive_failures,
)
processed = 0
if processed == 0:
try:
@ -422,10 +574,12 @@ class CascadeWorker:
return None
except Exception as exc:
last_error = f"{type(exc).__name__}: {exc}"
self._unrecoverable_total += 1
logger.exception(
"cascade_worker_unrecoverable",
md_path=row.md_path,
kind=row.kind,
unrecoverable_total=self._unrecoverable_total,
)
await md_change_state_repo.mark_failed(
row.md_path,
@ -538,11 +692,14 @@ class CascadeWorker:
async def _run_optimize_once(self, kind: str) -> None:
"""Run one ``optimize()`` for ``kind``, opportunistically pruning.
Most calls take the **light** path — pure compaction + index
merge, fast. Every ``_optimize_prune_interval`` seconds the
next call takes the **heavy** path: same work plus
``cleanup_older_than`` so the storage layer physically deletes
files belonging to versions older than one cadence.
Most calls take the **light** path — lock-free ``optimize()``,
pure compaction + index merge, fast. Every
``_optimize_prune_interval`` seconds the next call takes the
**heavy** path — ``prune()`` under the per-table write lock,
which compacts *and* physically deletes files belonging to
versions older than ``_optimize_prune_retention`` (a short
window decoupled from the beat cadence; see
:data:`DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS`).
Pruning is opt-in per call rather than a separate task so the
existing per-kind serialisation (one in-flight runner per kind)
@ -557,17 +714,34 @@ class CascadeWorker:
now = time.monotonic()
should_prune = (
state is None
or (now - state.last_prune_at) >= self._optimize_prune_interval
)
cleanup = (
dt.timedelta(seconds=self._optimize_prune_interval)
if should_prune
else None
or (now - state.last_prune_attempt_at) >= self._optimize_prune_interval
)
try:
await repo.optimize(cleanup_older_than=cleanup)
if should_prune and state is not None:
state.last_prune_at = now
if should_prune:
# Heavy beat: physically reclaim old versions under the
# per-table write lock (inside ``repo.prune``) so churn
# can't preempt the cleanup commit. ``prune`` uses
# ``delete_unverified=False`` so it stays safe even against a
# second process (CLI ``cascade sync``). The retention window
# is short + decoupled from the cadence (see
# DEFAULT_OPTIMIZE_PRUNE_RETENTION_SECONDS) so superseded
# full-table copies don't pile up between beats.
#
# Advance the *attempt* clock before the call: if prune hangs
# and the write-lock timeout kills it, the next beat still
# waits a full cadence instead of retrying immediately and
# pinning the write lock (review N1). The *success* clock
# (last_prune_at, for the staleness health signal) advances
# only after the call returns.
if state is not None:
state.last_prune_attempt_at = now
await repo.prune(dt.timedelta(seconds=self._optimize_prune_retention))
if state is not None:
state.last_prune_at = now
else:
# Light beat: lock-free compaction. A commit conflict here
# is benign — handled below.
await repo.optimize()
if state is not None:
state.optimize_failures = 0
logger.debug(
@ -576,6 +750,19 @@ class CascadeWorker:
pruned=should_prune,
)
except Exception as exc:
# Benign light-beat commit conflict: the lock-free compaction
# lost the optimistic-concurrency race against a live writer.
# Expected under churn, self-heals next beat — log at debug and
# do NOT count it toward the streak (which would otherwise pin
# high on a busy table) or trigger a fallback rebuild. The heavy
# beat runs under the write lock, so it can't hit this benignly.
if not should_prune and _is_benign_commit_conflict(exc):
logger.debug(
"cascade_lancedb_optimize_conflict",
kind=kind,
error=f"{type(exc).__name__}: {exc}",
)
return
failures = 0
if state is not None:
state.optimize_failures += 1

73
src/everos/memory/search/agentic_agent.py
View File

@ -20,13 +20,15 @@ Hyperparameters are aligned to the memsys_opensource ``AgenticConfig`` defaults
from __future__ import annotations
import datetime as _dt
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from everalgo.rank.agentic import aagentic_retrieve
from everalgo.rank.hybrid import ahybrid_retrieve
from everalgo.types import Candidate
from everos.component.utils.datetime import from_timestamp, to_timestamp_ms
from everos.core.observability.tracing import memory_span
from everos.memory.search.callbacks import build_rerank_fn
from everos.memory.search.shaper import (
@ -56,6 +58,53 @@ _MULTI_QUERY_COUNT: int = 3 # num_queries
_REFINEMENT_STRATEGY: str = "multi_query"
def _to_everalgo_doc_metadata(
metadata: dict[str, Any], *, text_field: str
) -> dict[str, Any]:
"""Bridge agent recall metadata to the everalgo ``_format_docs`` contract.
``aagentic_retrieve`` renders round-1 candidates into the sufficiency /
multi-query LLM prompt via ``everalgo.rank.agentic._format_docs``, which
reads ``metadata["episode"]`` as a dict with ``subject`` + ``content`` and
a ms-epoch ``metadata["timestamp"]``. Agent-kind rows carry their body in
``text_field`` (``task_intent`` / ``skill``) and the time in ``timestamp``
(datetime); without this bridge ``_format_docs`` raises ``TypeError``.
Mirrors the episode path's bridge in ``agentic.py``.
``_restore_shaper_metadata`` reverts it before DTO shaping.
"""
bridged = dict(metadata)
content = metadata.get(text_field)
if isinstance(content, str):
bridged["episode"] = {
"subject": metadata.get("subject", ""),
"content": content,
}
timestamp = metadata.get("timestamp")
if isinstance(timestamp, _dt.datetime):
bridged["timestamp"] = to_timestamp_ms(timestamp)
return bridged
def _restore_shaper_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
"""Revert ``_to_everalgo_doc_metadata`` before agent-DTO shaping.
The shaper reads ``timestamp`` as a ``datetime``; the bridged ms-epoch
int must be reverted. The injected ``episode`` dict is inert for the
agent shapers (they read case/skill fields), so it is simply dropped.
Deliberately wider than an exact inverse of the bridge: a row whose
``timestamp`` was *natively* numeric (agent tables store datetimes
today, so this does not occur) would also be converted — correct
either way, because the shaper contract requires a ``datetime``.
"""
reverted = dict(metadata)
timestamp = metadata.get("timestamp")
if isinstance(timestamp, (int, float)):
reverted["timestamp"] = from_timestamp(timestamp)
reverted.pop("episode", None)
return reverted
async def search_agent_cases_agentic(
query: str,
*,
@ -171,7 +220,7 @@ async def _run_agentic_retrieve(
observation_type="retriever",
metadata={"phase": "agentic_hybrid"},
):
return await ahybrid_retrieve(
hits = await ahybrid_retrieve(
q,
dense_retrieve=_dense,
sparse_retrieve=_sparse,
@ -180,6 +229,19 @@ async def _run_agentic_retrieve(
sparse_candidates=_SPARSE_CANDIDATES,
rrf_k=_HYBRID_RRF_K,
)
# Bridge to the everalgo doc contract so ``_format_docs`` (the LLM
# sufficiency / multi-query prompt) sees an episode dict + ms
# timestamp; agent-kind rows otherwise lack it and _format_docs raises.
return [
c.model_copy(
update={
"metadata": _to_everalgo_doc_metadata(
c.metadata, text_field=recaller.text_field
)
}
)
for c in hits
]
rerank_fn = build_rerank_fn(reranker, text_field=recaller.text_field)
@ -197,4 +259,9 @@ async def _run_agentic_retrieve(
multi_query_count=_MULTI_QUERY_COUNT,
rrf_k=_HYBRID_RRF_K,
)
return candidates
# Revert the doc-contract bridge so the agent DTO shapers see the
# original metadata shape (timestamp as datetime, no ``episode`` dict).
return [
c.model_copy(update={"metadata": _restore_shaper_metadata(c.metadata)})
for c in candidates
]

10
tests/e2e/conftest.py
View File

@ -34,7 +34,9 @@ from __future__ import annotations
import asyncio
import importlib
import json
import shutil
from collections.abc import AsyncIterator, Awaitable, Callable
from importlib.resources import files
from pathlib import Path
import httpx
@ -136,6 +138,14 @@ async def core_pipeline_runtime(
monkeypatch.setattr(client_mod, "_llm_client", None, raising=False)
_reset_strategy_singletons(monkeypatch)
# The full-app lifespan starts OME, whose config reloader requires an
# ``ome.toml`` in the memory root (normally created by ``everos init``).
# Provision the packaged default so the pipeline e2e can boot; OME
# strategies are code-registered (see api/lifespans/ome.py), so the
# default config is sufficient for extraction.
default_ome = files("everos.config") / "default_ome.toml"
shutil.copyfile(str(default_ome), str(tmp_path / "ome.toml"))
yield tmp_path

50
tests/e2e/test_search_endpoint_e2e.py
View File

@ -156,35 +156,38 @@ async def _seed_user_profiles(rows: list[dict[str, Any]]) -> list[UserProfile]:
async def _seed_user_memory_cluster(eps: list[dict], *, owner_id: str) -> None:
"""Seed one ``user_memory`` cluster covering every memcell in ``eps``.
"""Seed one ``user_memory`` cluster covering every episode in ``eps``.
The AGENTIC episode path goes through ``acluster_retrieve`` (see
``memory/search/agentic.py``), which narrows hybrid candidates to the
union of cluster member memcell ids. Tests that exercise the AGENTIC
union of cluster member ids. Production user_memory clusters store
episode ``entry_id`` members (``member_type="episode"``; see
``strategies/trigger_profile_clustering.py``), matching the entry_id
keying of ``fetch_all_for_owner``. Tests that exercise the AGENTIC
method therefore need at least one cluster whose members cover the
seeded episodes' ``parent_id``s — otherwise ``cluster_scoped`` yields
seeded episodes' ``entry_id``s — otherwise ``cluster_scoped`` yields
nothing and the agentic pipeline returns ``[]``.
Centroid is embedded from one of the episode bodies via the live
embedder; with a single cluster the cosine ranking against the query
is trivial (only one candidate), so any reasonable anchor works.
"""
memcell_ids = list({ep["parent_id"] for ep in eps})
entry_ids = list({ep["entry_id"] for ep in eps})
centroid_text = eps[0]["episode"]
centroid_vec = await get_embedding_capability().require().embed(centroid_text)
await cluster_repo.upsert_with_members(
AlgoCluster(
id=mint_cluster_id(),
centroid=np.asarray(centroid_vec, dtype=np.float32),
count=len(memcell_ids),
count=len(entry_ids),
last_ts=int(time.time() * 1000),
preview=[ep["episode"][:80] for ep in eps[:3]],
members=memcell_ids,
members=entry_ids,
),
owner_id=owner_id,
owner_type="user",
kind="user_memory",
member_type="memcell",
member_type="episode",
)
@ -1070,27 +1073,28 @@ async def test_search_hybrid_hierarchical_eviction_with_memcell_facts(
"""
eps = _eps_for_owner(search_seed, "caroline")
await _seed_episodes(eps)
ep_parent_ids = {r["parent_id"] for r in eps}
ep_entry_ids = {r["entry_id"] for r in eps}
matching_facts = [
r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_parent_ids
r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_entry_ids
]
assert matching_facts, "seed should have at least one fact sharing a memcell"
assert matching_facts, "seed should have at least one fact bridging an episode"
await _seed_atomic_facts(matching_facts)
resp = await _post(client, query="counseling", method="hybrid", top_k=5)
assert resp.status_code == 200
data = resp.json()["data"]
assert data["episodes"], "hybrid should return at least one episode"
# Whichever facts *do* get embedded must share parent_id with their
# host episode (the memcell-bridge invariant).
# Whichever facts *do* get embedded must bridge to their host episode
# via ``parent_id == episode.entry_id`` (the current fact-linkage
# invariant; see extract_atomic_facts).
for ep in data["episodes"]:
if not ep["atomic_facts"]:
continue
host_parent = next((e["parent_id"] for e in eps if e["id"] == ep["id"]), None)
host_entry = next((e["entry_id"] for e in eps if e["id"] == ep["id"]), None)
for fact in ep["atomic_facts"]:
seed_fact = next((r for r in matching_facts if r["id"] == fact["id"]), None)
if seed_fact is not None:
assert seed_fact["parent_id"] == host_parent
assert seed_fact["parent_id"] == host_entry
@pytest.mark.slow
@ -1122,11 +1126,11 @@ async def test_hybrid_hierarchical_eviction_injects_facts_with_alpha_zero(
eps = _eps_for_owner(search_seed, "caroline")
await _seed_episodes(eps)
ep_parent_ids = {r["parent_id"] for r in eps}
ep_entry_ids = {r["entry_id"] for r in eps}
matching_facts = [
r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_parent_ids
r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_entry_ids
]
assert matching_facts, "seed should have at least one fact sharing a memcell"
assert matching_facts, "seed should have at least one fact bridging an episode"
await _seed_atomic_facts(matching_facts)
resp = await _post(client, query="counseling", method="hybrid", top_k=10)
@ -1139,8 +1143,8 @@ async def test_hybrid_hierarchical_eviction_injects_facts_with_alpha_zero(
"alpha=0 should let hierarchical eviction promote >=1 fact"
)
# Memcell-bridge invariant — every attached fact's parent_id must
# match its host episode's parent_id.
# Fact-linkage invariant — every attached fact's parent_id must match
# its host episode's entry_id (current fact→episode bridge).
eps_by_id = {e["id"]: e for e in eps}
for ep in data["episodes"]:
host = eps_by_id.get(ep["id"])
@ -1149,7 +1153,7 @@ async def test_hybrid_hierarchical_eviction_injects_facts_with_alpha_zero(
for fact in ep["atomic_facts"]:
seed_fact = next((r for r in matching_facts if r["id"] == fact["id"]), None)
if seed_fact is not None:
assert seed_fact["parent_id"] == host["parent_id"]
assert seed_fact["parent_id"] == host["entry_id"]
# ═══════════════════════════════════════════════════════════════════════
@ -1284,7 +1288,7 @@ async def test_search_filter_error_returns_422(
# FastAPI's default ``{"detail": ...}``). The FilterError text
# lands in ``error.message``.
body = resp.json()
assert body["error"]["code"] == "HTTP_ERROR"
assert body["error"]["code"] == "INVALID_INPUT"
assert "this_field_does_not_exist" in body["error"]["message"]
@ -1312,7 +1316,7 @@ async def test_vector_search_with_session_filter(
base = _eps_for_owner(search_seed, "caroline")
facts = _facts_for_owner(search_seed, "caroline")
half = len(base) // 2
target_parent_ids = {r["parent_id"] for r in base[:half]}
target_parent_ids = {r["entry_id"] for r in base[:half]}
await _seed_episodes(
[{**r, "session_id": "sess_target"} for r in base[:half]]
@ -1376,7 +1380,7 @@ async def test_agentic_search_with_timestamp_filter(
[
{
**f,
"parent_id": eps_post[i % len(eps_post)]["parent_id"],
"parent_id": eps_post[i % len(eps_post)]["entry_id"],
"timestamp": eps_post[i % len(eps_post)]["timestamp"],
}
for i, f in enumerate(facts)

76
tests/fixtures/_dump_search_seed.py vendored
View File

@ -8,13 +8,14 @@ under ``tests/fixtures/search_seed/``.
Sampling rules:
- **episode**: first 8 rows per owner (caroline + melanie). Captures
the parent_id (= memcell_id) set so downstream tables can be
bridge-consistent.
- **atomic_fact**: every row whose ``parent_id`` is in the episode-
parent set above, capped at 50 to keep the seed compact. This
guarantees hierarchical-eviction testing can verify "facts sharing a
memcell with the matched episode get embedded".
- **episode + atomic_fact**: sampled together for fact↔episode
coherence. Facts link to their host episode via
``atomic_fact.parent_id == episode.entry_id`` (parent_type="episode";
see ``memory/strategies/extract_atomic_facts.py``). Episodes that host
facts are picked first (up to 8/owner) so hierarchical-eviction and the
fact-first paths (vector MaxSim, agentic) have a non-trivial,
multi-episode corpus; facts are then kept iff their host episode made
the cut (≤15/owner), guaranteeing every kept fact bridges back.
- **foresight**: 5 per owner. Archived for future use; current
``/search`` does not query foresight, so the seed only exists so
downstream tests can opt in without re-cutting the corpus.
@ -74,35 +75,43 @@ def main() -> None:
OUT_DIR.mkdir(parents=True, exist_ok=True)
db = lancedb.connect(str(CORPUS))
# 1) episodes — first 8 per owner.
# 1) episodes + 2) atomic_facts — sampled together so the slice is
# fact↔episode coherent AND rich (facts spread across several
# episodes, not piled on one).
#
# Current extraction links a fact to its host episode via
# ``atomic_fact.parent_id == episode.entry_id`` (parent_type="episode";
# see ``memory/strategies/extract_atomic_facts.py``). The fact-first
# search paths (vector MaxSim, agentic) fetch episodes by that
# entry_id, so the seed must preserve that linkage. Sample the
# episodes that actually HAVE facts first (up to 8/owner), so the
# agentic LLM-sufficiency step and hierarchical eviction have a
# non-trivial, multi-episode corpus to work with; only then fall back
# to fact-less episodes to top up owner coverage.
eps_all = _read(db, "episode")
eps: list[dict[str, Any]] = []
parent_memcells: set[str] = set()
for owner in ALL_OWNERS:
owned = [r for r in eps_all if r["owner_id"] == owner][:8]
eps.extend(owned)
for r in owned:
parent_memcells.add(r["parent_id"])
# 2) atomic_facts — every fact whose parent_id is in the episode
# parent set, capped to keep the seed compact (and so hierarchical
# ``facts_for_episodes`` has a useful but bounded pool to
# bucket back into episodes).
afs_all = _read(db, "atomic_fact")
# Atomic facts fan out per-owner (a single fact about a memcell that
# mentions two users gets two rows, one for each owner) — sampling
# naively can leave one owner with zero facts. Take per-owner caps
# so both caroline and melanie have facts whose parent_id matches
# their own episodes' parent_id (memcell bridge).
eps: list[dict[str, Any]] = []
afs: list[dict[str, Any]] = []
for owner in ALL_OWNERS:
afs.extend(
[
r
for r in afs_all
if r["owner_id"] == owner and r["parent_id"] in parent_memcells
][:10]
)
owner_eps = [r for r in eps_all if r["owner_id"] == owner]
owner_facts = [r for r in afs_all if r["owner_id"] == owner]
facts_by_ep: dict[str, list[dict[str, Any]]] = {}
for f in owner_facts:
facts_by_ep.setdefault(f["parent_id"], []).append(f)
# Episodes that host facts come first (richest bridges), then the
# rest — capped at 8 to keep the seed compact.
with_facts = [e for e in owner_eps if e["entry_id"] in facts_by_ep]
without_facts = [e for e in owner_eps if e["entry_id"] not in facts_by_ep]
chosen = (with_facts + without_facts)[:8]
eps.extend(chosen)
# Spread facts across episodes (<=3 per host) so several episodes are
# bridged, not one — richer corpus for agentic + hierarchical
# eviction. Every kept fact bridges back via its host entry_id.
owner_afs: list[dict[str, Any]] = []
for e in chosen:
owner_afs.extend(facts_by_ep.get(e["entry_id"], [])[:3])
afs.extend(owner_afs[:15])
# 3) foresights — 5 per owner, archived for future use.
fss_all = _read(db, "foresight")
@ -128,7 +137,8 @@ def main() -> None:
for name, count, size in written:
print(f" {name:14s}: {count:3d} rows ({size // 1024} KB)")
print(f" parent_memcells captured: {len(parent_memcells)}")
bridged = len({f["parent_id"] for f in afs})
print(f" fact-bridged episodes: {bridged}")
if __name__ == "__main__":

52040
tests/fixtures/search_seed/atomic_fact.json vendored
View File

File diff suppressed because it is too large Load Diff

49838
tests/fixtures/search_seed/episode.json vendored
View File

File diff suppressed because it is too large Load Diff

20866
tests/fixtures/search_seed/foresight.json vendored
View File

File diff suppressed because it is too large Load Diff

48
tests/fixtures/search_seed/user_profile.json vendored
View File

File diff suppressed because one or more lines are too long

111
tests/integration/test_cascade_cli_integration.py
View File

@ -175,6 +175,117 @@ def test_status_handles_pending_rows(cli_runtime: Path) -> None:
assert "pending: 1" in result.stdout
def _fake_orchestrator_factory(): # type: ignore[no-untyped-def]
from everos.component.tokenizer import build_tokenizer
from everos.core.persistence import MemoryRoot
from everos.memory.cascade import CascadeOrchestrator
def _build() -> CascadeOrchestrator:
root = MemoryRoot.resolve()
root.ensure()
return CascadeOrchestrator(
memory_root=root,
tokenizer=build_tokenizer(),
)
return _build
def test_rebuild_recovers_drifted_index_and_reindexes(
cli_runtime: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""``cascade rebuild`` recovers a type-drifted index.
Proves the three properties bare ``rm`` can't offer together:
(1) it runs despite a schema-type drift that trips the normal
startup guard (``verify_business_schemas``); (2) it drops + recreates
the drifted table with the current (correct) type; (3) it re-indexes
md from scratch even for entries the queue already marked ``done``.
"""
import datetime as _dt
import lancedb
import pyarrow as pa
from everos.core.persistence import MemoryRoot
from everos.infra.persistence.lancedb import get_table
from everos.infra.persistence.lancedb.tables.atomic_fact import AtomicFact
from everos.infra.persistence.lancedb.tables.episode import Episode
from everos.infra.persistence.markdown import AtomicFactWriter
root = MemoryRoot.resolve()
root.ensure()
owner_id = "u_rebuild"
bucket = _dt.date(2026, 5, 18)
md_path = (
f"default_app/default_project/users/{owner_id}/.atomic_facts/"
f"atomic_fact-{bucket.isoformat()}.md"
)
async def _seed_md() -> None:
writer = AtomicFactWriter(root=root)
items = [
(
{
"owner_id": owner_id,
"session_id": f"s_{j}",
"timestamp": "2026-05-18T07:04:26+00:00",
"parent_id": f"mc_{j}",
"sender_ids": [owner_id],
},
{"Fact": f"seed fact {j}"},
)
for j in range(2)
]
await writer.append_entries(owner_id, items, date=bucket)
async def _drift_episode_table() -> None:
conn = await lancedb.connect_async(str(root.lancedb_dir))
drifted = pa.schema(
[
pa.field("subject_vector", pa.string(), nullable=True)
if f.name == "subject_vector"
else f
for f in Episode.to_arrow_schema()
]
)
await conn.create_table("episode", schema=drifted)
conn.close()
async def _episode_subject_vector_type(): # type: ignore[no-untyped-def]
tbl = await get_table("episode", Episode)
return (await tbl.schema()).field("subject_vector").type
async def _atomic_fact_row_count() -> int:
tbl = await get_table(AtomicFact.TABLE_NAME, AtomicFact)
return await tbl.count_rows(filter=f"md_path = '{md_path}'")
asyncio.run(_seed_md())
asyncio.run(_drift_episode_table())
asyncio.run(_dispose_all())
monkeypatch.setattr(
cascade_mod, "_build_orchestrator", _fake_orchestrator_factory()
)
# Contrast: a normal command boots via _runtime() → verify trips on the drift.
status_result = CliRunner().invoke(cascade_mod.app, ["status"])
assert status_result.exit_code != 0
asyncio.run(_dispose_all())
# rebuild skips verify, recreates the table, and re-indexes md.
result = CliRunner().invoke(cascade_mod.app, ["rebuild", "--yes"])
assert result.exit_code == 0, result.stdout
assert "rebuild complete" in result.stdout
asyncio.run(_dispose_all())
# Table recreated with the correct vector type; md re-indexed.
assert asyncio.run(_episode_subject_vector_type()).equals(
Episode.to_arrow_schema().field("subject_vector").type
)
assert asyncio.run(_atomic_fact_row_count()) == 2
# Reduce false negatives on date drift.
def test_resolve_relative_via_command_arg(cli_runtime: Path) -> None:
"""An absolute path under the root works through ``cascade sync <path>``."""

33
tests/integration/test_cascade_scenarios.py
View File

@ -132,32 +132,33 @@ def _build_orchestrator(
async def _wait_path_done(md_path: str, *, deadline: float = 15.0) -> None:
"""Wait until ``md_path`` lands in state AND reaches ``status='done'``.
"""Wait until ``md_path`` lands in state AND *stably* reaches a terminal
status (``done``/``failed``).
Bare ``_wait_drain`` returns immediately when the queue is empty,
which is exactly the case right after a single ``append_entries``
fires once but the watcher hasn't yet enqueued anything. This helper
polls for the row first (i.e. watcher has noticed), then waits for
terminal state, then re-checks after a short settle to absorb any
last-second re-enqueue (e.g. atomic-replace echo).
Bare ``_wait_drain`` returns immediately when the queue is empty, which
is exactly the case right after a single ``append_entries`` fires once
but the watcher hasn't yet enqueued anything. This helper polls for the
row first (i.e. watcher has noticed), then waits for a terminal state
that *survives* a short settle window: a last-second re-enqueue (an
atomic-replace echo, or a rename's delete event) flips the row back to
``processing``, so we absorb it by waiting for terminal again rather
than treating the transient flip as a failure. Bounded by ``deadline``,
so a row that never settles still surfaces as a timeout.
"""
async with asyncio.timeout(deadline):
while True:
row = await md_change_state_repo.get_by_id(md_path)
if row is not None:
if await md_change_state_repo.get_by_id(md_path) is not None:
break
await asyncio.sleep(0.05)
while True:
row = await md_change_state_repo.get_by_id(md_path)
if row is not None and row.status in ("done", "failed"):
break
await asyncio.sleep(0.1) # settle
row = await md_change_state_repo.get_by_id(md_path)
if row is not None and row.status in ("done", "failed"):
return # stably terminal
# flipped back to processing (re-enqueue) — keep waiting
await asyncio.sleep(0.05)
await asyncio.sleep(0.1)
row = await md_change_state_repo.get_by_id(md_path)
assert row is not None and row.status in ("done", "failed"), (
f"path {md_path} flipped back to {row.status if row else 'NONE'} "
f"after reaching done"
)
async def _wait_paths_done(*md_paths: str, deadline: float = 15.0) -> None:

81
tests/unit/test_core/test_persistence/test_lancedb/test_repository.py
View File

@ -16,6 +16,7 @@ specific business schema (episode / atomic_fact / …).
from __future__ import annotations
import asyncio
import datetime as dt
from pathlib import Path
from typing import ClassVar
@ -703,3 +704,83 @@ async def test_migrate_fts_indexes_runs_once_and_rebuilds(
assert not list(await table.list_indices())
finally:
await dispose_connection()
async def test_prune_holds_write_lock_and_is_cross_process_safe(
tmp_path: Path,
) -> None:
"""``prune`` runs the underlying optimize **under the per-table write
lock** (so concurrent churn in this process can't preempt its Rewrite)
and passes ``delete_unverified=False``.
The write lock is the fix for the bundled optimize+prune starving
cleanup under churn (soak: 16 prune successes / 547 commit conflicts
over 21h → unbounded index dir). ``delete_unverified=False`` is the
cross-process guard: the in-process lock cannot fence a *second* process
(a CLI ``cascade sync`` / ``backfill``), and lance warns that
``delete_unverified=True`` can corrupt the dataset if any other process
is writing. ``False`` reclaims identically on churned tables (both
collapse superseded versions ~97% — measured) because ordinary orphans
are version-referenced and thus verifiable; ``True`` only additionally
deletes in-flight/dangling files — exactly the corruption vector.
"""
captured: dict = {}
state = {"held": False}
class _MockTable:
async def optimize(self, *, cleanup_older_than=None, delete_unverified=False):
state["held"] = repo._write_lock(repo.table_name).locked()
captured["cleanup_older_than"] = cleanup_older_than
captured["delete_unverified"] = delete_unverified
async def uri(self) -> str:
return str(tmp_path)
repo = _NoteRepo(table=_MockTable()) # type: ignore[arg-type]
await repo.prune(dt.timedelta(seconds=42))
assert state["held"], "prune must hold the write lock while optimizing"
assert captured["delete_unverified"] is False
assert captured["cleanup_older_than"] == dt.timedelta(seconds=42)
async def test_prune_removes_empty_index_dir_husks(tmp_path: Path) -> None:
"""After cleanup, ``prune`` removes the empty ``_indices/<uuid>/`` dirs
that ``cleanup_older_than`` leaves behind, but keeps non-empty ones."""
indices = tmp_path / "_indices"
(indices / "empty_uuid").mkdir(parents=True)
populated = indices / "live_uuid"
populated.mkdir()
(populated / "index.idx").write_text("data")
class _MockTable:
async def optimize(self, **_kw):
return None
async def uri(self) -> str:
return str(tmp_path)
repo = _NoteRepo(table=_MockTable()) # type: ignore[arg-type]
await repo.prune(dt.timedelta(seconds=1))
assert not (indices / "empty_uuid").exists(), "empty husk must be removed"
assert populated.exists(), "non-empty index dir must be kept"
async def test_optimize_is_lock_free_compaction_only() -> None:
"""The light beat ``optimize`` compacts only — no ``delete_unverified`` —
and must NOT hold the write lock, so it never stalls writers (a commit
conflict against a concurrent write is benign and retried next beat)."""
captured: dict = {}
state = {"held": True}
class _MockTable:
async def optimize(self, **kwargs):
state["held"] = repo._write_lock(repo.table_name).locked()
captured.update(kwargs)
repo = _NoteRepo(table=_MockTable()) # type: ignore[arg-type]
await repo.optimize()
assert state["held"] is False, "light optimize must be lock-free"
assert captured == {}, "light optimize passes no cleanup/delete_unverified args"

113
tests/unit/test_entrypoints/test_api/test_routes/test_cascade_health_routes.py Normal file
View File

@ -0,0 +1,113 @@
"""Cascade-health surfacing on ``GET /health``.
The route reads the running :class:`CascadeOrchestrator` off
``app.state.lifespan_data["cascade"]`` (stashed by the cascade
lifespan). These tests inject an autospec orchestrator so the route
logic is exercised without booting sqlite / lancedb / the worker.
Cascade degradation is surfaced *only* here (a readiness signal), never
on the write DTOs — a degraded projection does not change what ``/add``
returns.
"""
from __future__ import annotations
import unittest.mock as mock
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from everos.entrypoints.api.routes.health import router as health_router
from everos.memory.cascade import CascadeHealth, CascadeOrchestrator
def _healthy() -> CascadeHealth:
return CascadeHealth(
healthy=True,
reasons=[],
pending=0,
failed_permanent=0,
failed_retryable=0,
drain_consecutive_failures=0,
unrecoverable_total=0,
optimize_failure_streak=0,
prune_stale_seconds=0.0,
)
def _degraded() -> CascadeHealth:
return CascadeHealth(
healthy=False,
reasons=["version cleanup stalled (1200s since last prune — disk may grow)"],
pending=42,
failed_permanent=3,
failed_retryable=0,
drain_consecutive_failures=0,
unrecoverable_total=3,
optimize_failure_streak=0,
prune_stale_seconds=1200.0,
)
def _orch(health: CascadeHealth) -> mock.MagicMock:
"""An autospec orchestrator — ``isinstance(_, CascadeOrchestrator)`` holds."""
orch = mock.create_autospec(CascadeOrchestrator, instance=True)
orch.health.return_value = health # create_autospec makes this an AsyncMock
return orch
def _client(orch: object | None) -> AsyncClient:
app = FastAPI()
app.include_router(health_router)
app.state.lifespan_data = {"cascade": orch} if orch is not None else {}
return AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
async def test_health_without_cascade_is_plain_liveness() -> None:
"""No cascade lifespan → plain 200 liveness, no cascade block."""
async with _client(None) as c:
resp = await c.get("/health")
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["cascade"] is None # no cascade lifespan → block omitted
async def test_health_healthy_cascade() -> None:
async with _client(_orch(_healthy())) as c:
resp = await c.get("/health")
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["cascade"]["healthy"] is True
assert body["cascade"]["reasons"] == []
async def test_health_degraded_stays_200_but_flags_cascade() -> None:
"""Degraded cascade must NOT fail liveness (no crash-loop) — the
readiness signal lives in the cascade block."""
async with _client(_orch(_degraded())) as c:
resp = await c.get("/health")
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["cascade"]["healthy"] is False
assert body["cascade"]["failed_permanent"] == 3
assert body["cascade"]["prune_stale_seconds"] == 1200.0
assert any("cleanup stalled" in r for r in body["cascade"]["reasons"])
async def test_health_probe_exception_stays_200_and_flags_unhealthy() -> None:
"""The probe reads SQLite; a locked / full / mid-migration DB makes
``orch.health()`` raise. That must NOT turn /health into a 500 (which
would flip liveness and restart the container) — the endpoint reports
unhealthy readiness with a reason and keeps HTTP 200 (review P1-3)."""
orch = mock.create_autospec(CascadeOrchestrator, instance=True)
orch.health.side_effect = RuntimeError("database is locked")
async with _client(orch) as c:
resp = await c.get("/health")
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["cascade"]["healthy"] is False
assert any("probe failed" in r for r in body["cascade"]["reasons"])
assert any("database is locked" in r for r in body["cascade"]["reasons"])

7
tests/unit/test_entrypoints/test_cli/test_cascade_command.py
View File

@ -3,7 +3,7 @@
The orchestrator paths require live sqlite + lancedb singletons; those
are exercised by integration tests. Here we cover:
- subcommand registration (sync / status / fix)
- subcommand registration (sync / status / fix / rebuild)
- ``--help`` exit codes
- ``_resolve_relative`` (path arithmetic vs. memory root)
- ``_print_failed_table`` (formatting of failed rows)
@ -25,9 +25,9 @@ from everos.entrypoints.cli.commands import cascade as cascade_mod
from everos.infra.persistence.sqlite import dispose_engine, get_engine
def test_app_registers_four_commands() -> None:
def test_app_registers_expected_commands() -> None:
names = {cmd.name for cmd in cascade_mod.app.registered_commands}
assert names == {"sync", "status", "fix", "backfill"}
assert names == {"sync", "status", "fix", "backfill", "rebuild"}
def test_help_exits_zero() -> None:
@ -36,6 +36,7 @@ def test_help_exits_zero() -> None:
assert "sync" in result.stdout
assert "status" in result.stdout
assert "fix" in result.stdout
assert "rebuild" in result.stdout
def test_resolve_relative_under_root(

139
tests/unit/test_infra/test_lancedb/test_verify_schemas.py Normal file
View File

@ -0,0 +1,139 @@
"""``verify_business_schemas`` startup guard.
Regression coverage for EverOS #337: the guard must catch a column
whose on-disk Arrow *type* drifted from the current schema (not only a
missing/extra column name), and must NOT false-positive on a healthy
table freshly built from the current schema.
White-box surfaces: builds LanceDB tables directly on disk under an
isolated ``EVEROS_ROOT`` and drives ``verify_business_schemas`` /
``get_table`` against them.
"""
from __future__ import annotations
from pathlib import Path
import lancedb
import pyarrow as pa
import pytest
from everos.core.persistence import MemoryRoot
from everos.infra.persistence.lancedb import (
LanceDBSchemaMismatchError,
drop_business_tables,
ensure_business_indexes,
get_connection,
get_table,
lancedb_manager,
verify_business_schemas,
)
from everos.infra.persistence.lancedb.tables.episode import Episode
@pytest.fixture(autouse=True)
async def _isolated_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
"""Point the manager singleton at an isolated memory root."""
monkeypatch.setenv("EVEROS_ROOT", str(tmp_path))
lancedb_manager._conn = None
lancedb_manager._tables.clear()
yield
await lancedb_manager.dispose_connection()
async def _create_episode_table_on_disk(arrow_schema: pa.Schema) -> None:
"""Create the ``episode`` table straight on disk, then drop the
connection so the manager reopens it from disk on next use."""
root = MemoryRoot.resolve()
root.ensure()
conn = await lancedb.connect_async(str(root.lancedb_dir))
await conn.create_table("episode", schema=arrow_schema)
conn.close()
lancedb_manager._conn = None
lancedb_manager._tables.clear()
def _episode_schema_with_subject_vector_as(dtype: pa.DataType) -> pa.Schema:
"""Current episode Arrow schema, but subject_vector forced to ``dtype``."""
return pa.schema(
[
pa.field("subject_vector", dtype, nullable=True)
if f.name == "subject_vector"
else f
for f in Episode.to_arrow_schema()
]
)
async def test_verify_passes_on_healthy_tables() -> None:
"""A table built from the current schema must NOT trip the guard.
Guards against a type comparison that false-positives (e.g. on the
fixed_size_list item name or the datetime tz rewrite)."""
# verify_business_schemas creates every business table fresh via
# get_table, then re-reads and compares — so a clean run proves no
# false positive across ALL business schemas.
await verify_business_schemas()
# A second run over the now-existing tables must also pass.
await verify_business_schemas()
async def test_verify_raises_on_subject_vector_string_drift() -> None:
"""#337: subject_vector left as `string` must be caught by type."""
await _create_episode_table_on_disk(
_episode_schema_with_subject_vector_as(pa.string())
)
with pytest.raises(LanceDBSchemaMismatchError) as exc:
await verify_business_schemas()
msg = str(exc.value)
assert "episode" in msg
assert "type_drift" in msg
assert "subject_vector" in msg
async def test_verify_raises_on_subject_vector_null_drift() -> None:
"""A `null`-typed subject_vector (data-inferred all-None) is also drift."""
await _create_episode_table_on_disk(
_episode_schema_with_subject_vector_as(pa.null())
)
with pytest.raises(LanceDBSchemaMismatchError) as exc:
await verify_business_schemas()
assert "subject_vector" in str(exc.value)
async def test_verify_raises_on_missing_column() -> None:
"""A table missing a current column is caught by name (unchanged)."""
reduced = pa.schema(
[f for f in Episode.to_arrow_schema() if f.name != "subject_vector"]
)
await _create_episode_table_on_disk(reduced)
with pytest.raises(LanceDBSchemaMismatchError) as exc:
await verify_business_schemas()
assert "missing" in str(exc.value)
assert "subject_vector" in str(exc.value)
async def test_drop_business_tables_removes_then_recreatable() -> None:
"""drop_business_tables drops existing business tables + clears the cache;
the next get_table recreates a fresh, current-schema table."""
# Materialise a drifted episode table + the rest of the business set.
await _create_episode_table_on_disk(
_episode_schema_with_subject_vector_as(pa.string())
)
await ensure_business_indexes() # create the remaining business tables
conn = await get_connection()
assert "episode" in set((await conn.list_tables()).tables)
dropped = await drop_business_tables()
assert "episode" in dropped
conn = await get_connection()
assert "episode" not in set((await conn.list_tables()).tables)
assert "episode" not in lancedb_manager._tables # cache evicted
# Recreated fresh from the current schema → correct vector type, not string.
tbl = await get_table("episode", Episode)
assert (
(await tbl.schema())
.field("subject_vector")
.type.equals(Episode.to_arrow_schema().field("subject_vector").type)
)

27
tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py
View File

@ -394,6 +394,33 @@ async def test_reset_retryable_to_pending_zero_when_none_eligible(
assert await repo.reset_retryable_to_pending() == 0
# ── reset_all ───────────────────────────────────────────────────────────
async def test_reset_all_clears_every_row(repo: _MdChangeStateRepo) -> None:
"""`cascade rebuild` engine: every row is deleted regardless of status."""
await repo.upsert("a.md", kind="episode", change_type="added", mtime=0.0)
await repo.claim_one("a.md")
await repo.mark_done("a.md") # a: done
await repo.upsert("b.md", kind="episode", change_type="added", mtime=0.0)
await repo.claim_one("b.md")
await repo.mark_failed("b.md", retryable=False, error="x", new_retry_count=0)
await repo.upsert("c.md", kind="episode", change_type="added", mtime=0.0) # pending
deleted = await repo.reset_all()
assert deleted == 3
assert await repo.get_by_id("a.md") is None
assert await repo.get_by_id("b.md") is None
assert await repo.get_by_id("c.md") is None
summary = await repo.queue_summary()
assert summary.pending == 0 and summary.done == 0
async def test_reset_all_zero_on_empty_table(repo: _MdChangeStateRepo) -> None:
assert await repo.reset_all() == 0
# ── list_failed ─────────────────────────────────────────────────────────

61
tests/unit/test_memory/test_cascade/test_backfill_optimize.py
View File

@ -28,6 +28,7 @@ This file pins:
from __future__ import annotations
import datetime as dt
from typing import Any
import pytest
@ -58,8 +59,15 @@ class _FakeSchema:
class _FakeRepo:
"""Records ``update`` and ``optimize`` calls; can be told to fail
either operation to model poison writes / failing optimize.
"""Records ``update`` / ``optimize`` / ``prune`` calls; can be told to
fail an operation to model poison writes / failing maintenance.
The ``optimize`` / ``prune`` signatures MUST mirror the real
``LanceRepoBase`` exactly. A stale double here previously kept a removed
``optimize(cleanup_older_than=…)`` kwarg after the repo API split
compact (``optimize()``) from reclaim (``prune()``); the fake happily
accepted the old call while the real ``optimize`` raised ``TypeError``,
hiding a silent backfill regression from CI (review P0-1).
"""
def __init__(
@ -67,27 +75,32 @@ class _FakeRepo:
*,
update_fails: bool = False,
optimize_fails: bool = False,
prune_fails: bool = False,
) -> None:
self.update_fails = update_fails
self.optimize_fails = optimize_fails
self.prune_fails = prune_fails
self.update_calls: list[tuple[dict[str, Any], str]] = []
self.optimize_calls = 0
self.prune_calls = 0
self.last_prune_older_than: dt.timedelta | None = None
async def update(self, values: dict[str, Any], *, where: str) -> None:
self.update_calls.append((values, where))
if self.update_fails:
raise RuntimeError("simulated per-row write failure")
async def optimize(self, *, cleanup_older_than=None) -> None:
# ``cleanup_older_than`` mirrors the real ``LanceRepoBase.optimize``
# signature — round-4 review M2 added the kwarg at the backfill
# call site to physically prune old manifest versions, and this
# test double must accept it without breaking prior coverage.
async def optimize(self) -> None:
self.optimize_calls += 1
self.last_cleanup_older_than = cleanup_older_than
if self.optimize_fails:
raise RuntimeError("simulated optimize failure (e.g. lock contention)")
async def prune(self, older_than: dt.timedelta) -> None:
self.prune_calls += 1
self.last_prune_older_than = older_than
if self.prune_fails:
raise RuntimeError("simulated prune failure (e.g. lock contention)")
class _HappyProvider:
"""``embed_batch`` returns deterministic vectors. Never falls back."""
@ -154,6 +167,11 @@ async def test_backfill_table_calls_optimize_when_rows_processed(
assert result.rows_failed == 0
assert len(repo.update_calls) == 3
assert repo.optimize_calls == 1
# Compact then reclaim: prune fires once with a zero retention (reclaim
# everything now) — this is what the removed ``cleanup_older_than`` kwarg
# used to do inline (review P0-1).
assert repo.prune_calls == 1
assert repo.last_prune_older_than == dt.timedelta(0)
# Happy path must not fire the failure log.
assert "cascade_backfill_table_optimize_failed" not in caplog.text
@ -172,6 +190,7 @@ async def test_backfill_table_skips_optimize_when_no_rows_written() -> None:
assert result.rows_failed == 2
assert len(repo.update_calls) == 0
assert repo.optimize_calls == 0
assert repo.prune_calls == 0
async def test_backfill_table_skips_optimize_when_backlog_is_empty() -> None:
@ -187,6 +206,7 @@ async def test_backfill_table_skips_optimize_when_backlog_is_empty() -> None:
assert result.rows_processed == 0
assert repo.optimize_calls == 0
assert repo.prune_calls == 0
async def test_backfill_table_optimize_failure_does_not_abort(
@ -209,5 +229,30 @@ async def test_backfill_table_optimize_failure_does_not_abort(
assert len(repo.update_calls) == 3
# Optimize was attempted and raised — the failure log names the table.
assert repo.optimize_calls == 1
# optimize() raised before prune() could run.
assert repo.prune_calls == 0
assert "cascade_backfill_table_optimize_failed" in caplog.text
assert "fake_table" in caplog.text
async def test_backfill_table_prune_failure_does_not_abort(
caplog: pytest.LogCaptureFixture,
) -> None:
"""prune() is best-effort maintenance too — a raising ``prune`` (after a
clean ``optimize``) must not invalidate the writes or lose the counters;
it logs the same failure warning naming the table."""
repo = _FakeRepo(prune_fails=True)
backlog = _backlog(repo, [_row(f"r{i}", f"text {i}") for i in range(3)])
with caplog.at_level("WARNING", logger="everos.memory.cascade._backfill"):
result = await _backfill_table( # type: ignore[arg-type]
backlog, _HappyProvider(), presenter=NullBackfillPresenter()
)
assert result.rows_processed == 3
assert result.rows_failed == 0
assert len(repo.update_calls) == 3
assert repo.optimize_calls == 1
assert repo.prune_calls == 1
assert "cascade_backfill_table_optimize_failed" in caplog.text
assert "fake_table" in caplog.text

65
tests/unit/test_memory/test_cascade/test_orchestrator.py
View File

@ -92,3 +92,68 @@ async def test_drain_once_returns_zero_on_empty_queue(
) -> None:
orch = _make_orchestrator(runtime)
assert await orch.drain_once() == 0
async def test_health_is_healthy_on_fresh_runtime(runtime: MemoryRoot) -> None:
"""A quiet, freshly-booted cascade reports healthy with no reasons."""
orch = _make_orchestrator(runtime)
health = await orch.health()
assert health.healthy is True
assert health.reasons == []
assert health.failed_permanent == 0
assert health.prune_stale_seconds == 0.0
async def test_permanent_failures_are_informational_not_unhealthy(
runtime: MemoryRoot,
) -> None:
"""A permanently-failed md row is reported but must NOT flip ``healthy``.
A per-file triage backlog is normal steady state; folding it into the
verdict would pin the signal red forever. ``failed_permanent`` is
surfaced as an informational count while ``healthy`` stays true so
long as the pipeline itself (drain / optimize / prune) is fine.
"""
from everos.component.utils.datetime import get_utc_now
from everos.infra.persistence.sqlite import md_change_state_repo
orch = _make_orchestrator(runtime)
await md_change_state_repo.upsert(
"users/u1/episodes/ep_1.md",
kind="episode",
change_type="added",
mtime=get_utc_now().timestamp(),
)
await md_change_state_repo.claim_pending_batch(10)
await md_change_state_repo.mark_failed(
"users/u1/episodes/ep_1.md",
retryable=False,
error="boom",
new_retry_count=0,
)
health = await orch.health()
assert health.failed_permanent == 1 # reported…
assert health.healthy is True # …but pipeline is operationally fine
assert health.reasons == []
async def test_operational_signal_flips_healthy(
runtime: MemoryRoot, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An operational reason (prune stalled) — not a data-quality backlog —
is what flips ``healthy`` false."""
import everos.memory.cascade.worker as wmod
orch = _make_orchestrator(runtime)
# Freeze the monotonic clock so staleness is deterministic regardless of
# the runner's boot uptime (see test_worker prune-staleness test).
now = 10_000.0
monkeypatch.setattr(wmod.time, "monotonic", lambda: now)
# simulate a running worker whose version cleanup has gone stale
orch._worker._started_at = now - (wmod._PRUNE_STALE_SECONDS_ALERT + 100)
orch._worker._optimizer_states["episode"] = wmod._KindOptimizerState()
health = await orch.health()
assert health.healthy is False
assert any("cleanup stalled" in r for r in health.reasons)

268
tests/unit/test_memory/test_cascade/test_worker.py
View File

@ -282,12 +282,17 @@ def test_worker_handler_deps_construct_with_real_classes() -> None:
class _FakeLanceRepo:
"""Records every optimize() / rebuild_indexes() call.
"""Records every optimize() / prune() / rebuild_indexes() call.
``optimize_delay`` / ``rebuild_delay`` simulate slow operations.
``rebuild_raises`` makes ``rebuild_indexes`` raise (for crash-safety tests).
Each ``optimize`` call's ``cleanup_older_than`` is preserved so
prune-cadence tests can assert which calls took the heavy path.
The optimize path is split: ``optimize()`` is the light lock-free
compaction (no args); ``prune(older_than)`` is the heavy write-locked
reclaim. ``beats`` combines both in call order — most scheduler tests
only care that *a maintenance beat* ran, not which. The first beat per
kind is always a prune (``last_prune_at`` starts at 0).
``optimize_delay`` / ``rebuild_delay`` simulate slow operations (the
delay applies to both maintenance beats). ``rebuild_raises`` makes
``rebuild_indexes`` raise (crash-safety tests).
"""
def __init__(
@ -298,17 +303,28 @@ class _FakeLanceRepo:
rebuild_raises: bool = False,
) -> None:
self.optimize_calls: list[float] = []
self.optimize_cleanup_args: list[dt.timedelta | None] = []
self.prune_calls: list[float] = []
self.prune_args: list[dt.timedelta] = []
self.rebuild_calls: list[float] = []
self.optimize_delay = optimize_delay
self.rebuild_delay = rebuild_delay
self.rebuild_raises = rebuild_raises
async def optimize(self, *, cleanup_older_than: dt.timedelta | None = None) -> None:
@property
def beats(self) -> list[float]:
"""All maintenance beats (optimize + prune) in call order."""
return sorted(self.optimize_calls + self.prune_calls)
async def optimize(self) -> None:
if self.optimize_delay > 0:
await asyncio.sleep(self.optimize_delay)
self.optimize_calls.append(time.monotonic())
self.optimize_cleanup_args.append(cleanup_older_than)
async def prune(self, older_than: dt.timedelta) -> None:
if self.optimize_delay > 0:
await asyncio.sleep(self.optimize_delay)
self.prune_calls.append(time.monotonic())
self.prune_args.append(older_than)
async def rebuild_indexes(self) -> None:
if self.rebuild_delay > 0:
@ -358,10 +374,8 @@ async def test_schedule_optimize_collapses_burst_within_throttle_window(
for _ in range(10):
w._schedule_optimize("episode")
await w._flush_optimizers()
assert fake.optimize_calls, "expected at least one optimize"
assert len(fake.optimize_calls) == 1, (
f"burst should collapse, got {len(fake.optimize_calls)} calls"
)
assert fake.beats, "expected at least one optimize"
assert len(fake.beats) == 1, f"burst should collapse, got {len(fake.beats)} calls"
async def test_schedule_optimize_reruns_when_dirty_set_during_optimize(
@ -383,7 +397,7 @@ async def test_schedule_optimize_reruns_when_dirty_set_during_optimize(
await asyncio.sleep(0.01) # ensure first task is mid-optimize
w._schedule_optimize("episode")
await w._flush_optimizers()
assert len(fake.optimize_calls) == 2
assert len(fake.beats) == 2
async def test_concurrent_schedules_keep_one_task_per_kind(
@ -419,7 +433,7 @@ async def test_flush_optimizers_awaits_pending_task(
w._schedule_optimize("episode")
assert w._optimizer_states["episode"].task is not None
await w._flush_optimizers()
assert fake.optimize_calls, "flush should not return before optimize ran"
assert fake.beats, "flush should not return before optimize ran"
assert w._optimizer_states["episode"].task is None
@ -436,7 +450,7 @@ async def test_drain_until_empty_flushes_optimizers_before_returning(
)
await w.drain_until_empty()
assert patched_repo.done == ["a.md"]
assert len(fake.optimize_calls) == 1
assert len(fake.beats) == 1
assert w._optimizer_states["episode"].task is None
@ -456,9 +470,9 @@ async def test_drain_once_does_not_block_on_optimize(
drain_elapsed = time.monotonic() - started
# drain returned long before the 0.2s optimize would finish
assert drain_elapsed < 0.1, f"drain blocked on optimize: {drain_elapsed:.3f}s"
assert not fake.optimize_calls, "optimize should still be in flight"
assert not fake.beats, "optimize should still be in flight"
await w._flush_optimizers()
assert len(fake.optimize_calls) == 1
assert len(fake.beats) == 1
async def test_stop_waits_for_in_flight_optimize(
@ -483,7 +497,7 @@ async def test_stop_waits_for_in_flight_optimize(
w._schedule_optimize("episode")
await asyncio.sleep(0.01) # let optimize start
await w.stop()
assert len(fake.optimize_calls) == 1
assert len(fake.beats) == 1
async def test_optimize_failure_does_not_crash_drain_loop(
@ -495,6 +509,9 @@ async def test_optimize_failure_does_not_crash_drain_loop(
async def optimize(self) -> None:
raise RuntimeError("simulated lancedb manifest conflict")
async def prune(self, older_than: dt.timedelta) -> None:
raise RuntimeError("simulated lancedb manifest conflict")
class _HandlerWithFailingRepo(_OkHandler):
def __init__(self) -> None:
super().__init__()
@ -535,46 +552,89 @@ async def test_heartbeat_schedules_every_handler_kind(
# Let at least one heartbeat tick happen.
await asyncio.sleep(0.12)
await w.stop()
assert fake_a.optimize_calls, "heartbeat should have scheduled episode"
assert fake_b.optimize_calls, "heartbeat should have scheduled atomic_fact"
assert fake_a.beats, "heartbeat should have scheduled episode"
assert fake_b.beats, "heartbeat should have scheduled atomic_fact"
async def test_optimize_prunes_on_first_call_then_throttles(
patched_repo: _FakeRepo,
) -> None:
"""First optimize() per kind passes ``cleanup_older_than``; subsequent
calls within ``optimize_prune_interval_seconds`` do not.
"""First maintenance beat per kind is a heavy ``prune()``; subsequent
beats within ``optimize_prune_interval_seconds`` take the light
lock-free ``optimize()`` path.
Rationale lives in ``DEFAULT_OPTIMIZE_PRUNE_INTERVAL_SECONDS``:
LanceDB ``optimize()`` without ``cleanup_older_than`` leaves stale
physical files on disk; passing it on every 1-second optimize tick
is wasteful, but never passing it leaks files until FDs exhaust.
A separate cadence — prune ≪ optimize — balances the two.
``prune`` (write-locked, ``delete_unverified``) physically reclaims
stale files but briefly stalls writes; running it on every 1-second
tick is wasteful, but never pruning leaks files until FDs / disk
exhaust. A separate cadence — prune ≪ optimize — balances the two.
"""
fake = _FakeLanceRepo()
w = CascadeWorker(
{"episode": _OkHandlerWithRepo(fake)},
retry_backoff_seconds=0,
optimize_min_interval_seconds=0.01,
optimize_prune_interval_seconds=10.0, # long — second call should NOT prune
optimize_prune_interval_seconds=10.0, # cadence: long — 2nd beat is light
optimize_prune_retention_seconds=45.0, # retention: decoupled from cadence
)
# First call: state has never pruned, must include cleanup_older_than.
# First beat: state has never pruned, must take the heavy prune path.
w._schedule_optimize("episode")
await w._flush_optimizers()
assert len(fake.optimize_calls) == 1
assert fake.optimize_cleanup_args[0] is not None, (
"first optimize must prune to catch up from prior session"
)
assert fake.optimize_cleanup_args[0] == dt.timedelta(seconds=10.0)
assert len(fake.prune_calls) == 1, "first beat must prune to catch up"
assert not fake.optimize_calls
# prune is passed the RETENTION window, not the cadence.
assert fake.prune_args[0] == dt.timedelta(seconds=45.0)
# Second call within the prune window: light path (no cleanup).
# Second beat within the prune window: light lock-free optimize.
await asyncio.sleep(0.02) # exceed optimize throttle (0.01), not prune (10)
w._schedule_optimize("episode")
await w._flush_optimizers()
assert len(fake.optimize_calls) == 2
assert fake.optimize_cleanup_args[1] is None, (
"second optimize within prune window should skip cleanup_older_than"
assert len(fake.prune_calls) == 1, "second beat within window must not re-prune"
assert len(fake.optimize_calls) == 1, "second beat is the light path"
async def test_failed_prune_backs_off_a_cadence_and_keeps_health_signal(
patched_repo: _FakeRepo,
) -> None:
"""A prune that fails (e.g. killed by the write-lock timeout on a hung
lance cleanup) advances the *attempt* clock but not the *success* clock:
- attempt clock advances → the next beat waits a full cadence and takes
the light lock-free path instead of immediately re-pruning, so a hung
prune can't pin the write lock ~97% of the time (review N1);
- success clock (``last_prune_at``) does NOT advance → the prune-staleness
health signal still climbs, so a persistently failing prune surfaces as
degraded rather than being masked.
"""
class _PruneRaisesRepo(_FakeLanceRepo):
async def prune(self, older_than: dt.timedelta) -> None:
self.prune_calls.append(time.monotonic())
self.prune_args.append(older_than)
raise TimeoutError("simulated hung cleanup killed by write-lock timeout")
fake = _PruneRaisesRepo()
w = CascadeWorker(
{"episode": _OkHandlerWithRepo(fake)},
retry_backoff_seconds=0,
optimize_min_interval_seconds=0.01,
optimize_prune_interval_seconds=10.0, # long cadence: 2nd beat is light
optimize_prune_retention_seconds=45.0,
)
# First beat: prune is attempted and raises.
w._schedule_optimize("episode")
await w._flush_optimizers()
st = w._optimizer_states["episode"]
assert len(fake.prune_calls) == 1, "first beat attempts a prune"
assert st.last_prune_attempt_at > 0, "attempt clock advances even on failure"
assert st.last_prune_at == 0.0, "success clock must NOT advance on a failed prune"
# Second beat within the cadence: must fall to the light path, not re-prune.
await asyncio.sleep(0.02) # exceeds optimize throttle (0.01), not cadence (10)
w._schedule_optimize("episode")
await w._flush_optimizers()
assert len(fake.prune_calls) == 1, "failed prune must not immediately retry (N1)"
assert len(fake.optimize_calls) == 1, "second beat backs off to the light path"
# ── Rebuild scheduler tests ────────────────────────────────────────────────
@ -646,23 +706,33 @@ async def test_rebuild_failure_does_not_crash_daemon(
# Give startup rebuild a chance to throw, then heartbeat to keep optimizing.
await asyncio.sleep(0.12)
# Optimize should still progress despite rebuild errors.
assert fake.optimize_calls, "heartbeat optimize should run even when rebuild fails"
assert fake.beats, "heartbeat optimize should run even when rebuild fails"
await w.stop()
# Worker is still alive (stop() returned cleanly).
assert w._task is None
class _OptimizeFailingRepo(_FakeLanceRepo):
"""Fake repo whose ``optimize()`` raises until ``fail`` is cleared."""
"""Fake repo whose ``optimize()`` AND ``prune()`` raise until ``fail``
is cleared. ``error`` selects the exception so a test can distinguish a
genuine failure from a benign commit conflict."""
def __init__(self, **kw) -> None: # type: ignore[no-untyped-def]
def __init__(self, *, error: Exception | None = None, **kw) -> None: # type: ignore[no-untyped-def]
super().__init__(**kw)
self.fail = True
self._error = error or RuntimeError(
"Max offset of 9 exceeds length of values 3"
)
async def optimize(self, *, cleanup_older_than: dt.timedelta | None = None) -> None:
async def optimize(self) -> None:
if self.fail:
raise RuntimeError("Max offset of 9 exceeds length of values 3")
await super().optimize(cleanup_older_than=cleanup_older_than)
raise self._error
await super().optimize()
async def prune(self, older_than: dt.timedelta) -> None:
if self.fail:
raise self._error
await super().prune(older_than)
async def test_optimize_failures_counted_escalated_and_reset(
@ -749,3 +819,115 @@ async def test_optimize_fallback_rebuild_on_sustained_failure(
state = w._optimizer_states["episode"]
assert state.optimize_failures == 0, "rebuild should reset failure counter"
assert len(repo.rebuild_calls) == 1, "exactly one fallback rebuild expected"
# ── health signals ───────────────────────────────────────────────────────────
def test_worker_health_dataclass_thresholds() -> None:
"""reasons() fires exactly on each threshold, one entry per crossed signal."""
from everos.memory.cascade import worker as wmod
mk = wmod.CascadeWorkerHealth
assert mk(0, 0, 0, 0.0).reasons() == []
assert mk(wmod._DRAIN_FAILURE_ALERT_THRESHOLD, 0, 0, 0.0).reasons()
assert mk(0, 0, wmod._OPTIMIZE_FAILURE_ALERT_THRESHOLD, 0.0).reasons()
assert mk(0, 0, 0, wmod._PRUNE_STALE_SECONDS_ALERT).reasons()
reasons = mk(
wmod._DRAIN_FAILURE_ALERT_THRESHOLD, 0, 0, wmod._PRUNE_STALE_SECONDS_ALERT
).reasons()
assert len(reasons) == 2 # drain + prune, not the sub-threshold optimize
def test_worker_health_idle_is_not_stale() -> None:
"""A worker with no optimize activity is never prune-stale (nothing to
reclaim), even if it started long ago."""
from everos.memory.cascade import worker as wmod
w = CascadeWorker({"episode": _OkHandlerWithRepo(_FakeLanceRepo())})
w._started_at = time.monotonic() - (wmod._PRUNE_STALE_SECONDS_ALERT + 1000)
h = w.health()
assert h.prune_stale_seconds == 0.0
assert h.reasons() == []
def test_worker_health_reports_prune_staleness(monkeypatch: pytest.MonkeyPatch) -> None:
"""An active kind that has never successfully pruned since start goes
stale once past the alert threshold."""
from everos.memory.cascade import worker as wmod
# Freeze the monotonic clock so staleness is deterministic. Real
# ``time.monotonic()`` returns process/boot uptime, which is huge on a
# long-lived dev box but only ~100s on a fresh CI runner — a bare
# ``monotonic() - 1000`` would go negative there and read as not-stale.
now = 10_000.0
monkeypatch.setattr(wmod.time, "monotonic", lambda: now)
w = CascadeWorker({"episode": _OkHandlerWithRepo(_FakeLanceRepo())})
w._started_at = now - (wmod._PRUNE_STALE_SECONDS_ALERT + 100)
w._optimizer_states["episode"] = wmod._KindOptimizerState() # last_prune_at=0
h = w.health()
assert h.prune_stale_seconds >= wmod._PRUNE_STALE_SECONDS_ALERT
assert any("cleanup stalled" in r for r in h.reasons())
def test_worker_health_forwards_counters() -> None:
"""drain / unrecoverable / optimize-streak counters surface verbatim."""
from everos.memory.cascade import worker as wmod
w = CascadeWorker({"episode": _OkHandlerWithRepo(_FakeLanceRepo())})
w._drain_consecutive_failures = 2
w._unrecoverable_total = 7
st = wmod._KindOptimizerState()
st.optimize_failures = 4
w._optimizer_states["episode"] = st
h = w.health()
assert h.drain_consecutive_failures == 2
assert h.unrecoverable_total == 7
assert h.optimize_failure_streak == 4
async def test_light_beat_commit_conflict_is_debug_and_uncounted(
patched_repo: _FakeRepo,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A benign light-beat commit conflict must not pollute the signal.
The lock-free compaction can lose the optimistic-concurrency race
against a live writer; that is expected under churn and self-heals
next beat. It is logged at ``debug``, does NOT increment the failure
streak, and does NOT trigger a fallback rebuild.
"""
from everos.memory.cascade import worker as wmod
calls: list[tuple[str, str]] = []
class _SpyLogger:
def __getattr__(self, level: str): # type: ignore[no-untyped-def]
def rec(event: str, **_kw) -> None: # type: ignore[no-untyped-def]
calls.append((level, event))
return rec
monkeypatch.setattr(wmod, "logger", _SpyLogger())
repo = _OptimizeFailingRepo(error=RuntimeError("Retryable commit conflict"))
w = CascadeWorker(
{"episode": _OkHandlerWithRepo(repo)},
retry_backoff_seconds=0,
)
state = wmod._KindOptimizerState()
# Force the LIGHT beat: pretend we just attempted a prune so should_prune=False
# (scheduling reads the attempt clock, not the success clock — see N1 split).
state.last_prune_attempt_at = time.monotonic()
w._optimizer_states["episode"] = state
await w._run_optimize_once("episode")
assert state.optimize_failures == 0, "benign conflict must not count"
events = [ev for _, ev in calls]
levels = {lvl for lvl, _ in calls}
assert "cascade_lancedb_optimize_conflict" in events
assert "cascade_lancedb_optimize_failed" not in events
assert "cascade_lancedb_optimize_fallback_rebuild" not in events
assert "error" not in levels and "warning" not in levels

2
uv.lock
View File

@ -633,7 +633,7 @@ requires-dist = [
{ name = "fastapi", specifier = ">=0.104.0" },
{ name = "greenlet", specifier = ">=3.0" },
{ name = "jieba", specifier = ">=0.42.1,<1.0" },
{ name = "lancedb", specifier = ">=0.13.0" },
{ name = "lancedb", specifier = ">=0.34.0,<0.35.0" },
{ name = "openai", specifier = ">=1.0.0" },
{ name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'otel'", specifier = ">=1.27.0" },
{ name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.27.0" },