[Bug]: cascade row stays processing forever when its terminal write (mark_failed / mark_done) fails once

test_replay_5_mark_failed_raises.py:

"""Replay of #542 item 5: when ``mark_failed`` (or ``mark_done``) itself fails once,
the row stays ``processing`` for good.

``_process_one`` calls ``mark_failed`` inside its ``except`` branch
(worker.py:854-892); an exception there propagates out of the task, ``drain_once``'s
``gather`` raises, ``_run_loop`` logs ``cascade_worker_drain_failed`` and carries
on. Nothing retries the terminal write: the scanner skips a ``processing`` row whose
mtime is unchanged (reconciler.py:96-102) and claims take only ``pending``. The row
sits in ``processing`` (shown as ``pending`` by ``cascade status``) until a restart
(``recover_orphan_processing``) or an edit of the md. Same for ``mark_done``, with
the index already correct.

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

from __future__ import annotations

import asyncio
from typing import Any
from unittest import mock

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

M = 1_700_000_000.0


class _Embedder(EmbeddingProvider):
    """Raises an unrecoverable error for texts carrying ``BAD``."""

    dim = DIM

    async def embed(self, text: str) -> list[float]:
        if "BAD" in text:
            raise RuntimeError("unrecoverable: cannot embed this entry")
        return [0.1] * DIM

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


def _raise_once(real: Any, name: str) -> Any:
    calls = {"n": 0}

    async def flaky(*a: Any, **k: Any) -> Any:
        calls["n"] += 1
        if calls["n"] == 1:
            raise RuntimeError(
                f"sqlite3.OperationalError: database is locked (in {name})"
            )
        return await real(*a, **k)

    return flaky


async def _replay() -> None:
    h = await Harness().open(_Embedder())
    try:
        worker = h.worker()
        # bad: the handler fails unrecoverably; good: it succeeds
        await h.write("bad", 1, "BAD content", M)
        await h.write("good", 2, "fine content", M)
        await h.scanner.scan_once()
        assert (await h.counts()) == {"pending": 2}

        with (
            mock.patch.object(
                h.state_repo,
                "mark_failed",
                _raise_once(h.state_repo.mark_failed, "mark_failed"),
            ),
            mock.patch.object(
                h.state_repo,
                "mark_done",
                _raise_once(h.state_repo.mark_done, "mark_done"),
            ),
        ):
            try:
                await worker.drain_once()
                raised = None
            except (
                Exception
            ) as exc:  # what _run_loop catches and logs as cascade_worker_drain_failed
                raised = f"{type(exc).__name__}: {exc}"
            await asyncio.sleep(0.2)  # let the sibling task in the gather finish
        print(f"  drain_once raised: {raised}")
        rb, rg = await h.row("bad"), await h.row("good")
        print(
            f"  after the drain: bad row={rb.status}, good row={rg.status}, good index rows={await h.indexed('good')}"
        )
        assert rb.status == "processing" and rg.status == "processing"
        assert (await h.indexed("good")) == 1, (
            "good's index write landed; only the terminal write was lost"
        )

        # later scans and drains never touch them: mtime unchanged + processing -> scanner skips; claims take only pending
        for _ in range(3):
            decisions = await h.scanner.scan_once()
            assert not decisions
            assert (await worker.drain_once()) == 0
        rb, rg = await h.row("bad"), await h.row("good")
        print(
            f"  after 3 scan+drain passes: bad row={rb.status}, good row={rg.status}, counts={await h.counts()}"
        )
        assert rb.status == "processing" and rg.status == "processing"
        health = worker.health()
        print(f"  worker health reasons: {health.reasons()!r}")

        # control: a restart's orphan recovery frees them (mark_failed / mark_done work from now on)
        n = await h.state_repo.recover_orphan_processing()
        await worker.drain_once()
        rb, rg = await h.row("bad"), await h.row("good")
        print(
            f"  control, after restart recovery ({n} rows) + drain: bad row={rb.status}/retryable={rb.retryable}, good row={rg.status}"
        )
        assert rb.status == "failed" and rb.retryable is False and rg.status == "done"
    finally:
        await h.close()


def test_terminal_write_failing_once_leaves_the_row_processing_for_good() -> 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)