264 lines
9.1 KiB
Python
264 lines
9.1 KiB
Python
"""Integration test for ``everos cascade backfill --phase vectors``.
|
|
|
|
Exercises the real Phase 1 re-embed path against actual LanceDB tables
|
|
under a tmp memory root: seeds rows with ``vector IS NULL``, drives
|
|
``run_backfill(phase="vectors", ...)``, and asserts the rows now carry
|
|
a vector. A stub embedder replaces the process-wide embedding
|
|
capability singleton so no real network call ever happens (mirrors the
|
|
pattern in ``tests/integration/test_cascade_all_kinds_consistency.py``).
|
|
|
|
Covers: empty DB (nothing to backfill), declined confirmation (nothing
|
|
written, exit 1), a small DB with rows that need backfilling (rows
|
|
updated, pre-existing vectors left untouched, token estimate rendered
|
|
in K/M notation), a batch-embed failure that is tallied without
|
|
aborting the phase, an episode's ``subject_vector`` secondary embed,
|
|
and a missing embedding capability returning exit 2.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from everos.component.embedding import EmbeddingCapability, EmbeddingProvider
|
|
from everos.component.utils.datetime import get_utc_now
|
|
from everos.config import load_settings
|
|
from everos.entrypoints.cli.commands._backfill_cmd import run_backfill
|
|
from everos.infra.persistence.lancedb import (
|
|
AtomicFact,
|
|
Episode,
|
|
atomic_fact_repo,
|
|
dispose_connection,
|
|
episode_repo,
|
|
)
|
|
|
|
_DIM = 1024
|
|
|
|
|
|
class _StubEmbedder(EmbeddingProvider):
|
|
dim = _DIM
|
|
|
|
async def embed(self, text: str) -> list[float]:
|
|
return [float(len(text) % 7)] * self.dim
|
|
|
|
async def embed_batch(self, texts): # type: ignore[no-untyped-def]
|
|
return [[float(len(t) % 7)] * self.dim for t in texts]
|
|
|
|
|
|
@pytest.fixture
|
|
async def backfill_runtime(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> AsyncIterator[Path]:
|
|
"""Tmp memory root + stub embedder; disposes the lancedb singleton
|
|
around the test so no stale connection from a neighbouring test
|
|
leaks in (mirrors ``cascade_runtime`` in the cascade test suite)."""
|
|
monkeypatch.setenv("EVEROS_ROOT", str(tmp_path))
|
|
load_settings.cache_clear()
|
|
await dispose_connection()
|
|
|
|
import everos.component.embedding.accessor as acc
|
|
|
|
monkeypatch.setattr(
|
|
acc, "_capability", EmbeddingCapability(provider=_StubEmbedder())
|
|
)
|
|
|
|
yield tmp_path
|
|
await dispose_connection()
|
|
|
|
|
|
def _episode(
|
|
entry_id: str,
|
|
*,
|
|
text: str = "",
|
|
vector: list[float] | None = None,
|
|
subject: str | None = None,
|
|
subject_vector: list[float] | None = None,
|
|
) -> Episode:
|
|
body = text or f"episode body {entry_id}"
|
|
return Episode(
|
|
id=f"u1_{entry_id}",
|
|
entry_id=entry_id,
|
|
owner_id="u1",
|
|
owner_type="user",
|
|
timestamp=get_utc_now(),
|
|
parent_id="mc1",
|
|
sender_ids=["u1"],
|
|
subject=subject,
|
|
episode=body,
|
|
episode_tokens=body,
|
|
md_path="users/u1/episodes/episode-2026-01-01.md",
|
|
content_sha256=hashlib.sha256(entry_id.encode()).hexdigest(),
|
|
vector=vector,
|
|
subject_vector=subject_vector,
|
|
)
|
|
|
|
|
|
def _fact(entry_id: str, *, vector: list[float] | None = None) -> AtomicFact:
|
|
return AtomicFact(
|
|
id=f"u1_{entry_id}",
|
|
entry_id=entry_id,
|
|
owner_id="u1",
|
|
owner_type="user",
|
|
timestamp=get_utc_now(),
|
|
parent_id="mc1",
|
|
sender_ids=["u1"],
|
|
fact=f"fact body {entry_id}",
|
|
fact_tokens=f"fact body {entry_id}",
|
|
md_path="users/u1/.atomic_facts/atomic_fact-2026-01-01.md",
|
|
content_sha256=hashlib.sha256(entry_id.encode()).hexdigest(),
|
|
vector=vector,
|
|
)
|
|
|
|
|
|
async def test_empty_db_reports_nothing_to_backfill(
|
|
backfill_runtime: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
code = await run_backfill(phase="vectors", auto_yes=True)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 0
|
|
assert "Nothing to backfill" in out
|
|
|
|
|
|
async def test_declined_confirmation_writes_nothing_and_exits_one(
|
|
backfill_runtime: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
await episode_repo.add([_episode("ep1")])
|
|
monkeypatch.setattr("typer.confirm", lambda *a, **k: False)
|
|
|
|
code = await run_backfill(phase="vectors", auto_yes=False)
|
|
|
|
assert code == 1
|
|
row = await episode_repo.get_by_id("u1_ep1")
|
|
assert row is not None
|
|
assert row.vector is None
|
|
|
|
|
|
async def test_small_db_backfills_null_vector_rows_only(
|
|
backfill_runtime: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
pre_embedded = [0.5] * _DIM
|
|
await episode_repo.add([_episode("ep1"), _episode("ep2", vector=pre_embedded)])
|
|
await atomic_fact_repo.add([_fact("af1")])
|
|
|
|
code = await run_backfill(phase="vectors", auto_yes=True)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 0
|
|
ep1 = await episode_repo.get_by_id("u1_ep1")
|
|
ep2 = await episode_repo.get_by_id("u1_ep2")
|
|
af1 = await atomic_fact_repo.get_by_id("u1_af1")
|
|
assert ep1 is not None and ep1.vector is not None
|
|
assert ep2 is not None and ep2.vector == pytest.approx(pre_embedded)
|
|
assert af1 is not None and af1.vector is not None
|
|
assert "memories to process: 2" in out
|
|
assert "phase 1 complete" in out
|
|
|
|
|
|
async def test_token_estimate_uses_kilo_notation(
|
|
backfill_runtime: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
long_text = "word " * 3000
|
|
await episode_repo.add([_episode("epbig", text=long_text)])
|
|
|
|
code = await run_backfill(phase="vectors", auto_yes=True)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 0
|
|
assert "input tokens:" in out
|
|
assert "K" in out
|
|
|
|
|
|
async def test_failed_row_is_tallied_and_does_not_abort_the_phase(
|
|
backfill_runtime: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Round-2 finding M8: a batch failure now falls back to per-row
|
|
``embed(...)`` calls (see ``_embed_primary_batch``). A poison row —
|
|
one whose per-row retry also fails — must (a) not abort the phase,
|
|
(b) not touch the healthy rows sharing its batch, and (c) drive the
|
|
final exit code to ``4`` (COMPLETED_WITH_FAILURES) rather than ``0``
|
|
so automation can distinguish clean success from partial success.
|
|
|
|
``_FlakyEmbedder`` fails BOTH ``embed_batch`` and ``embed`` on the
|
|
poison text ("boom"), so the row remains unwritable after fallback.
|
|
Healthy rows still succeed via the per-row retry path.
|
|
"""
|
|
import everos.component.embedding.accessor as acc
|
|
from everos.component.embedding import EmbeddingCapability
|
|
|
|
class _FlakyEmbedder(_StubEmbedder):
|
|
async def embed_batch(self, texts: list[str]) -> list[list[float]]:
|
|
if any("boom" in t for t in texts):
|
|
raise RuntimeError("simulated batch embed failure")
|
|
return await super().embed_batch(texts)
|
|
|
|
async def embed(self, text: str) -> list[float]:
|
|
if "boom" in text:
|
|
raise RuntimeError("simulated per-row embed failure")
|
|
return await super().embed(text)
|
|
|
|
monkeypatch.setattr(
|
|
acc, "_capability", EmbeddingCapability(provider=_FlakyEmbedder())
|
|
)
|
|
await episode_repo.add(
|
|
[_episode("epok", text="fine content"), _episode("epbad", text="boom content")]
|
|
)
|
|
|
|
code = await run_backfill(phase="vectors", auto_yes=True)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 4
|
|
ok_row = await episode_repo.get_by_id("u1_epok")
|
|
bad_row = await episode_repo.get_by_id("u1_epbad")
|
|
assert ok_row is not None and ok_row.vector is not None
|
|
assert bad_row is not None and bad_row.vector is None
|
|
assert "phase 1 complete" in out
|
|
assert "COMPLETED_WITH_FAILURES" in out
|
|
assert "1 rows failed embedding" in out
|
|
|
|
|
|
async def test_episode_subject_vector_also_backfilled(
|
|
backfill_runtime: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""Episode is the only table with a second, independent embed
|
|
(``subject_vector`` — see ``EpisodeHandler._build_row``). A row that
|
|
has a ``subject`` must come out of Phase 1 with *both* vectors set,
|
|
exercising ``_embed_subject_batch``'s secondary ``embed_batch`` call.
|
|
"""
|
|
await episode_repo.add(
|
|
[_episode("epsub", text="episode body", subject="What is the meaning of X?")]
|
|
)
|
|
|
|
code = await run_backfill(phase="vectors", auto_yes=True)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 0
|
|
row = await episode_repo.get_by_id("u1_epsub")
|
|
assert row is not None
|
|
assert row.vector is not None and len(row.vector) == _DIM
|
|
assert row.subject_vector is not None and len(row.subject_vector) == _DIM
|
|
assert "phase 1 complete" in out
|
|
|
|
|
|
async def test_missing_embedding_capability_returns_exit_two(
|
|
backfill_runtime: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""When the embedding capability is unavailable (no provider
|
|
configured), Phase 1 must fail cleanly with exit 2 rather than
|
|
silently skipping rows or crashing uncaught."""
|
|
import everos.component.embedding.accessor as acc
|
|
|
|
await episode_repo.add([_episode("ep1")])
|
|
monkeypatch.setattr(acc, "_capability", EmbeddingCapability(provider=None))
|
|
|
|
code = await run_backfill(phase="vectors", auto_yes=True)
|
|
|
|
assert code == 2
|
|
row = await episode_repo.get_by_id("u1_ep1")
|
|
assert row is not None and row.vector is None
|