315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""Integration test for ``everos cascade backfill --phase clusters``.
|
|
|
|
Exercises the real Phase 2 scan + synthetic-event-emission path (see
|
|
``memory.cascade._backfill._run_phase_clusters``): seeds real episode /
|
|
agent-case rows in LanceDB, drives ``run_backfill(phase="clusters", ...)``,
|
|
and asserts the phase synthesizes exactly one ``EpisodeExtracted`` /
|
|
``AgentCaseExtracted`` event per row and fans them into the OME engine.
|
|
|
|
The OME engine itself is replaced with an in-memory spy
|
|
(``_FakeClusterEngine``, injected via monkeypatching
|
|
``_backfill._build_cluster_engine``) so this test exercises the scan +
|
|
event-synthesis contract without paying for a real APScheduler + sqlite
|
|
jobstore runtime — that end-to-end wiring is already covered by
|
|
``tests/integration/test_ome_strategies_integration.py``.
|
|
|
|
Covers: empty DB (nothing to backfill, no engine touched), declined
|
|
confirmation (no events emitted, exit 1), a small DB with episodes +
|
|
agent cases (correct event types / counts / fields, exit 0), and a
|
|
missing embedding capability (exit 2, mirroring Phase 1's guard).
|
|
|
|
One test (``test_real_offline_engine_grows_cluster_count``) does NOT
|
|
install the fake spy: it drives Phase 2 through the actual
|
|
``_build_cluster_engine`` path — a real ``OfflineEngine`` + APScheduler
|
|
runtime with ``trigger_profile_clustering`` registered — and asserts
|
|
``cluster_repo.count()`` grows. This is the one check the fake-engine
|
|
tests above cannot make: they prove the scan + event-synthesis contract,
|
|
but never exercise whether the real engine wiring actually drives a
|
|
cluster into existence.
|
|
"""
|
|
|
|
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.ome.events import BaseEvent
|
|
from everos.infra.persistence.lancedb import (
|
|
AgentCase,
|
|
Episode,
|
|
agent_case_repo,
|
|
dispose_connection,
|
|
episode_repo,
|
|
)
|
|
from everos.infra.persistence.sqlite import cluster_repo, sqlite_manager
|
|
from everos.memory.cascade import _backfill as backfill_mod
|
|
from everos.memory.events import AgentCaseExtracted, EpisodeExtracted
|
|
|
|
_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]
|
|
|
|
|
|
class _FakeClusterEngine:
|
|
"""Spy replacing the real OME engine: records emitted events instead of
|
|
dispatching them through APScheduler."""
|
|
|
|
def __init__(self) -> None:
|
|
self.emitted: list[BaseEvent] = []
|
|
self.started = False
|
|
self.stopped = False
|
|
|
|
async def start(self) -> None:
|
|
self.started = True
|
|
|
|
async def emit(self, event: BaseEvent) -> None:
|
|
self.emitted.append(event)
|
|
|
|
async def wait_idle(self, *, timeout: float = 30.0) -> bool: # noqa: ASYNC109
|
|
return True
|
|
|
|
async def stop(self) -> None:
|
|
self.stopped = True
|
|
|
|
|
|
@pytest.fixture
|
|
async def backfill_runtime(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> AsyncIterator[Path]:
|
|
"""Tmp memory root + stub embedder + isolated sqlite engine.
|
|
|
|
Phase 2 touches both LanceDB (episode / agent_case scan) and sqlite
|
|
(``cluster_repo.count()``), so both process-wide singletons are
|
|
disposed and rebuilt against ``tmp_path`` around each test — mirrors
|
|
``backfill_runtime`` in ``test_backfill_phase1.py`` plus the sqlite
|
|
reset from ``test_ome_strategies_integration.py``.
|
|
"""
|
|
monkeypatch.setenv("EVEROS_ROOT", str(tmp_path))
|
|
load_settings.cache_clear()
|
|
await dispose_connection()
|
|
if sqlite_manager._engine is not None:
|
|
await sqlite_manager.dispose_engine()
|
|
monkeypatch.setattr(sqlite_manager, "_engine", None, raising=False)
|
|
monkeypatch.setattr(sqlite_manager, "_session_factory", None, raising=False)
|
|
|
|
import everos.component.embedding.accessor as acc
|
|
|
|
monkeypatch.setattr(
|
|
acc, "_capability", EmbeddingCapability(provider=_StubEmbedder())
|
|
)
|
|
|
|
yield tmp_path
|
|
|
|
await dispose_connection()
|
|
if sqlite_manager._engine is not None:
|
|
await sqlite_manager.dispose_engine()
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_engine(monkeypatch: pytest.MonkeyPatch) -> _FakeClusterEngine:
|
|
"""Install the spy engine in place of ``_build_cluster_engine``."""
|
|
engine = _FakeClusterEngine()
|
|
monkeypatch.setattr(backfill_mod, "_build_cluster_engine", lambda: engine)
|
|
return engine
|
|
|
|
|
|
def _episode(entry_id: str, *, owner_id: str = "u1") -> Episode:
|
|
return Episode(
|
|
id=f"{owner_id}_{entry_id}",
|
|
entry_id=entry_id,
|
|
owner_id=owner_id,
|
|
owner_type="user",
|
|
session_id="s1",
|
|
timestamp=get_utc_now(),
|
|
parent_id=f"mc_{entry_id}",
|
|
sender_ids=[owner_id],
|
|
episode=f"episode body {entry_id}",
|
|
episode_tokens=f"episode body {entry_id}",
|
|
md_path="users/u1/episodes/episode-2026-01-01.md",
|
|
content_sha256=hashlib.sha256(entry_id.encode()).hexdigest(),
|
|
vector=[0.1] * _DIM,
|
|
)
|
|
|
|
|
|
def _agent_case(entry_id: str, *, agent_id: str = "agent1") -> AgentCase:
|
|
return AgentCase(
|
|
id=f"{agent_id}_{entry_id}",
|
|
entry_id=entry_id,
|
|
owner_id=agent_id,
|
|
owner_type="agent",
|
|
session_id="s1",
|
|
timestamp=get_utc_now(),
|
|
parent_id=f"mc_{entry_id}",
|
|
quality_score=0.8,
|
|
task_intent=f"task intent {entry_id}",
|
|
task_intent_tokens=f"task intent {entry_id}",
|
|
approach=f"approach {entry_id}",
|
|
approach_tokens=f"approach {entry_id}",
|
|
md_path="agents/agent1/.cases/agent_case-2026-01-01.md",
|
|
content_sha256=hashlib.sha256(entry_id.encode()).hexdigest(),
|
|
vector=[0.1] * _DIM,
|
|
)
|
|
|
|
|
|
async def test_empty_db_reports_nothing_to_backfill(
|
|
backfill_runtime: Path,
|
|
fake_engine: _FakeClusterEngine,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
code = await run_backfill(phase="clusters", auto_yes=True)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 0
|
|
assert "Nothing to backfill" in out
|
|
assert fake_engine.emitted == []
|
|
assert fake_engine.started is False
|
|
|
|
|
|
async def test_declined_confirmation_emits_nothing_and_exits_one(
|
|
backfill_runtime: Path,
|
|
fake_engine: _FakeClusterEngine,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
await episode_repo.add([_episode("ep1")])
|
|
monkeypatch.setattr("typer.confirm", lambda *a, **k: False)
|
|
|
|
code = await run_backfill(phase="clusters", auto_yes=False)
|
|
|
|
assert code == 1
|
|
assert fake_engine.emitted == []
|
|
assert fake_engine.started is False
|
|
|
|
|
|
async def test_episodes_and_cases_synthesize_matching_events(
|
|
backfill_runtime: Path,
|
|
fake_engine: _FakeClusterEngine,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
await episode_repo.add([_episode("ep1"), _episode("ep2")])
|
|
await agent_case_repo.add([_agent_case("ac1")])
|
|
|
|
code = await run_backfill(phase="clusters", auto_yes=True)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 0
|
|
assert fake_engine.started is True
|
|
assert fake_engine.stopped is True
|
|
assert len(fake_engine.emitted) == 3
|
|
|
|
episode_events = [e for e in fake_engine.emitted if isinstance(e, EpisodeExtracted)]
|
|
case_events = [e for e in fake_engine.emitted if isinstance(e, AgentCaseExtracted)]
|
|
assert len(episode_events) == 2
|
|
assert len(case_events) == 1
|
|
|
|
assert all(e.source == "pipeline" for e in episode_events)
|
|
assert all(e.event_id.startswith("backfill_") for e in fake_engine.emitted)
|
|
assert {e.owner_id for e in episode_events} == {"u1"}
|
|
assert case_events[0].agent_id == "agent1"
|
|
assert case_events[0].task_intent == "task intent ac1"
|
|
assert case_events[0].quality_score == pytest.approx(0.8)
|
|
|
|
assert "phase 2 complete" in out
|
|
|
|
|
|
async def test_missing_embedding_capability_returns_exit_two(
|
|
backfill_runtime: Path,
|
|
fake_engine: _FakeClusterEngine,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Mirrors Phase 1's own guard: no embedding provider configured means
|
|
the strategies can't re-embed, so the phase fails clean with exit 2
|
|
rather than emitting events into strategies that will crash."""
|
|
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="clusters", auto_yes=True)
|
|
|
|
assert code == 2
|
|
assert fake_engine.emitted == []
|
|
assert fake_engine.started is False
|
|
|
|
|
|
async def test_reflection_merged_episodes_excluded_from_synthesis(
|
|
backfill_runtime: Path,
|
|
fake_engine: _FakeClusterEngine,
|
|
) -> None:
|
|
"""Reflection-merged episodes (``parent_type="cluster"``) must not be
|
|
re-synthesized as pipeline events.
|
|
|
|
``trigger_profile_clustering`` excludes Reflection output via
|
|
``applies_to=lambda e: e.source == "pipeline"`` — but that guard is
|
|
defeated if the Episode scan hands it a synthetic event whose
|
|
``memcell_id`` is actually a cluster id (see Important 1 review
|
|
finding on ``_backfill._scan_all_rows`` / ``_run_phase_clusters``).
|
|
Only the pipeline-sourced row should reach the fake engine.
|
|
"""
|
|
pipeline_episode = _episode("ep1")
|
|
merged_episode = _episode("ep_merged").model_copy(
|
|
update={
|
|
"id": "u1_ep_merged",
|
|
"parent_type": "cluster",
|
|
"parent_id": "cluster_abc123",
|
|
}
|
|
)
|
|
await episode_repo.add([pipeline_episode, merged_episode])
|
|
|
|
code = await run_backfill(phase="clusters", auto_yes=True)
|
|
|
|
assert code == 0
|
|
episode_events = [e for e in fake_engine.emitted if isinstance(e, EpisodeExtracted)]
|
|
assert len(episode_events) == 1
|
|
assert episode_events[0].memcell_id == pipeline_episode.parent_id
|
|
|
|
|
|
async def test_real_offline_engine_grows_cluster_count(
|
|
backfill_runtime: Path,
|
|
) -> None:
|
|
"""Phase 2 through the REAL ``_build_cluster_engine`` — no fake spy.
|
|
|
|
Deliberately does not use the ``fake_engine`` fixture: it drives an
|
|
actual ``OfflineEngine`` + APScheduler runtime with
|
|
``trigger_profile_clustering`` registered (see
|
|
``_backfill._build_cluster_engine``), proving the phase's own engine
|
|
wiring — not just the scan/synthesis contract the other tests in
|
|
this module pin — actually grows ``cluster_repo.count()``.
|
|
|
|
``trigger_profile_clustering`` re-embeds each episode's text via
|
|
``get_embedding_capability().require()`` — which ``backfill_runtime``
|
|
already stubs with the deterministic ``_StubEmbedder``, so no
|
|
per-strategy patch is needed here. No agent cases are seeded, so
|
|
``trigger_skill_clustering`` (the engine's other registered
|
|
strategy, whose ``cluster_by_llm`` fallback needs an LLM) never
|
|
fires, keeping this test LLM-free.
|
|
"""
|
|
await episode_repo.add([_episode(f"ep{i}") for i in range(5)])
|
|
|
|
# ``run_backfill`` itself creates the ``cluster`` table lazily (see
|
|
# ``_ensure_cluster_schema``) — call it here too so the "before" count
|
|
# can be taken without racing that lazy create.
|
|
await backfill_mod._ensure_cluster_schema()
|
|
clusters_before = await cluster_repo.count()
|
|
assert clusters_before == 0
|
|
|
|
code = await run_backfill(phase="clusters", auto_yes=True)
|
|
|
|
clusters_after = await cluster_repo.count()
|
|
assert code == 0
|
|
assert clusters_after > clusters_before
|