[Bug]: one hung cascade handler stalls the whole projection while /health stays healthy

test_replay_4_stuck_handler_health_green.py:

"""Replay of #542 item 4: one handler that never returns stalls the whole
projection, and ``/health`` stays healthy.

``drain_once`` gathers the whole claimed batch (worker.py:687-701) and
``_run_loop`` only claims the next batch when the gather returns, so a single
handler that hangs (an embedding call that never completes) stops every later
row from being processed. The health verdict (``CascadeWorkerHealth.reasons``,
worker.py:421-442; ``CascadeOrchestrator.health``, orchestrator.py:429-446)
looks at drain exceptions, optimize failures and prune staleness only; a stuck
drain raises nothing, so ``healthy`` stays ``True`` while ``pending`` grows.

Real worker loop (``start()``: drain, heartbeat and rebuild loops under the
supervisor), real scanner / handler / repos; the embedder blocks forever on one
file's text. Control: once that call is released, the backlog drains.

Run: uv run pytest .work_context/replay_542/test_replay_4_stuck_handler_health_green.py -q -s
"""

from __future__ import annotations

import asyncio
from typing import Any

from _harness import DIM, Harness
from everos.component.embedding import EmbeddingProvider

M = 1_700_000_000.0


class _Embedder(EmbeddingProvider):
    """Blocks forever on texts carrying ``HANG`` until ``release`` is set."""

    dim = DIM

    def __init__(self) -> None:
        self.release = asyncio.Event()
        self.hung = asyncio.Event()

    async def embed(self, text: str) -> list[float]:
        if "HANG" in text:
            self.hung.set()
            await self.release.wait()
        return [0.1] * DIM

    async def embed_batch(self, texts: Any) -> list[list[float]]:
        return [await self.embed(t) for t in texts]


async def _replay() -> None:
    emb = _Embedder()
    h = await Harness().open(emb)
    worker = h.worker(poll_interval_seconds=0.05, handler_concurrency=2)
    try:
        await worker.start()
        await h.write("stuck", 1, "HANG content", M)
        await h.scanner.scan_once()
        await asyncio.wait_for(emb.hung.wait(), timeout=10)
        r = await h.row("stuck")
        print(f"  the worker loop is inside the hung handler: stuck row={r.status}")

        # the users keep writing; the scanner keeps enqueuing; nothing drains
        for o in ("a", "b", "c"):
            await h.write(o, 2, "fine content", M)
        await h.scanner.scan_once()
        await asyncio.sleep(1.0)  # 20 poll intervals
        counts = await h.counts()
        health = worker.health()
        reasons = health.reasons()
        pending_public = counts.get("pending", 0) + counts.get(
            "processing", 0
        )  # QueueSummary.pending
        print(
            f"  1 s later: counts={counts}; /health would say healthy={not reasons} reasons={reasons} pending={pending_public}"
        )
        assert counts.get("pending") == 3 and counts.get("processing") == 1
        assert not reasons, "health reports no problem while nothing has drained"
        for o in ("a", "b", "c"):
            assert (await h.indexed(o)) == 0

        # control: release the hung call -> the batch completes and the loop drains the rest
        emb.release.set()
        for _ in range(100):
            await asyncio.sleep(0.05)
            if (await h.counts()) == {"done": 4}:
                break
        print(f"  control, after releasing the hung embed: counts={await h.counts()}")
        assert (await h.counts()) == {"done": 4}
    finally:
        emb.release.set()
        await worker.stop()
        await h.close()


def test_one_hung_handler_stalls_everything_and_health_stays_green() -> None:
    asyncio.run(_replay())

_harness.py:

"""Shared harness for the #542 replays: real scanner / worker / EpisodeHandler on a
temp memory root, real ``md_change_state`` repository (temp SQLite), real LanceDB
table; the embedder is injected per replay, the tokenizer is a stand-in."""

from __future__ import annotations

import datetime as dt
import shutil
import tempfile
from pathlib import Path
from typing import Any
from unittest import mock

from sqlmodel import SQLModel

import everos.component.embedding.accessor as acc
from everos.component.embedding import EmbeddingCapability, EmbeddingProvider
from everos.component.tokenizer import Tokenizer
from everos.config import LanceDBSettings, SqliteSettings
from everos.core.persistence import (
    MemoryRoot,
    create_session_factory,
    create_system_engine,
)
from everos.core.persistence.lancedb import LanceRepoBase, open_lancedb_connection
from everos.infra.persistence.index import Episode
from everos.infra.persistence.lancedb.repos.episode import _EpisodeRepo
from everos.infra.persistence.markdown import EpisodeWriter
from everos.infra.persistence.predicate import eq
from everos.infra.persistence.sqlite.repos.md_change_state import _MdChangeStateRepo
from everos.memory.cascade import scanner as scanner_mod
from everos.memory.cascade import worker as worker_mod
from everos.memory.cascade.handlers import HandlerDeps
from everos.memory.cascade.handlers.episode import EpisodeHandler
from everos.memory.cascade.scanner import CascadeScanner
from everos.memory.cascade.worker import CascadeWorker

DATE = dt.date(2026, 5, 14)
DIM = 1024


class _Tokenizer(Tokenizer):
    def tokenize(self, text: str) -> list[str]:
        return [t for t in text.split() if t]

    def tokenize_batch(self, texts: Any) -> list[list[str]]:
        return [self.tokenize(t) for t in texts]


def entry(owner: str, n: int, content: str) -> tuple[dict[str, Any], dict[str, Any]]:
    return (
        {
            "owner_id": owner,
            "session_id": "s1",
            "timestamp": "2026-05-14T10:00:00+00:00",
            "parent_type": "memcell",
            "parent_id": f"mc_{n}",
            "sender_ids": [owner],
        },
        {"Subject": "T", "Summary": "S", "Content": content},
    )


def rel(owner: str) -> str:
    return f"default_app/default_project/users/{owner}/episodes/episode-{DATE.isoformat()}.md"


class Harness:
    def __init__(self) -> None:
        self.tmp = Path(tempfile.mkdtemp(prefix="replay542_"))
        self.root = MemoryRoot(self.tmp / "root")
        self.root.ensure()
        self.engine = create_system_engine(self.root.system_db, SqliteSettings())
        self.factory = create_session_factory(self.engine)
        self.state_repo = _MdChangeStateRepo(session_factory=self.factory)
        self.index: Any = None
        self.patches: list[Any] = []
        self.writer = EpisodeWriter(self.root)
        self.scanner = CascadeScanner(self.root, scan_interval_seconds=3600.0)

    async def open(self, embedder: EmbeddingProvider) -> Harness:
        async with self.engine.begin() as conn:
            await conn.run_sync(SQLModel.metadata.create_all)
        LanceRepoBase._reset_locks_for_tests()
        conn = await open_lancedb_connection(self.root.lancedb_dir, LanceDBSettings())
        table = await conn.create_table(Episode.TABLE_NAME, schema=Episode)
        self.index = _EpisodeRepo(table=table)
        acc._capability = EmbeddingCapability(provider=embedder)
        self.patches = [
            mock.patch.object(scanner_mod, "md_change_state_repo", self.state_repo),
            mock.patch.object(scanner_mod, "get_session_factory", lambda: self.factory),
            mock.patch.object(worker_mod, "md_change_state_repo", self.state_repo),
            mock.patch.object(EpisodeHandler, "index_repo", self.index),
        ]
        for p in self.patches:
            p.start()
        return self

    def handler(self) -> EpisodeHandler:
        return EpisodeHandler(
            HandlerDeps(memory_root=self.root, tokenizer=_Tokenizer())
        )

    def worker(self, **overrides: Any) -> CascadeWorker:
        kw: dict[str, Any] = dict(
            batch_size=10,
            handler_concurrency=2,
            max_retry=0,
            retry_backoff_seconds=0.0,
            poll_interval_seconds=3600.0,
            optimize_min_interval_seconds=0.0,
            optimize_heartbeat_seconds=3600.0,
            optimize_rebuild_interval_seconds=3600.0,
        )
        kw.update(overrides)
        return CascadeWorker({"episode": self.handler()}, **kw)

    async def write(self, owner: str, n: int, content: str, mtime: float) -> Path:
        await self.writer.append_entries(owner, [entry(owner, n, content)], date=DATE)
        path = self.root.root / rel(owner)
        import os

        os.utime(path, (mtime, mtime))
        return path

    async def row(self, owner: str) -> Any:
        return next(
            (r for r in await self.state_repo.list_all() if r.md_path == rel(owner)),
            None,
        )

    async def counts(self) -> dict[str, int]:
        out: dict[str, int] = {}
        for r in await self.state_repo.list_all():
            out[r.status] = out.get(r.status, 0) + 1
        return out

    async def indexed(self, owner: str) -> int:
        return len(await self.index.find_where(eq("md_path", rel(owner))))

    async def close(self) -> None:
        for p in self.patches:
            p.stop()
        await self.engine.dispose()
        shutil.rmtree(self.tmp, ignore_errors=True)