[Bug]: cascade retry budget is forgotten by a restart during the inline backoff; a row can retry forever without becoming failed

test_replay_6_restart_resets_retry_budget.py:

"""Replay of #542 item 6: a restart inside the inline retry ladder resets the retry
budget.

``_process_one`` counts inline attempts in memory and only the final ``mark_failed``
persists ``retry_count`` (worker.py:838-876). A process that dies during the
backoff sleep leaves the row ``processing`` with the old ``retry_count``; the
restart's ``recover_orphan_processing`` resets it to ``pending`` without touching
``retry_count``, and the next drain starts the ladder from attempt 0 again. With
restarts recurring, the row never reaches ``failed``, so ``_MAX_TOTAL_RETRIES`` (12)
never bounds the attempts and ``cascade fix`` never lists the file. Control: with
no restarts the budget is spent in 4 drain cycles.

The process death is simulated by cancelling the drain task during the backoff
sleep, the same way the OME replay (#540) simulated it; the SQLite row looks the
same either way.

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

from __future__ import annotations

import asyncio
import contextlib
from typing import Any

from _harness import DIM, Harness
from everos.component.embedding import EmbeddingProvider
from everos.core.errors import EmbeddingServiceError

M = 1_700_000_000.0


class _Embedder(EmbeddingProvider):
    """The embedding service is down for good: every call fails (recoverable class)."""

    dim = DIM

    def __init__(self) -> None:
        self.calls = 0

    async def embed(self, text: str) -> list[float]:
        self.calls += 1
        raise EmbeddingServiceError("embedding 503")

    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)
    try:
        # 3 inline attempts with 0.5 s backoff between them (defaults are 2 s / 4 s / 6 s)
        worker = h.worker(max_retry=3, retry_backoff_seconds=0.5)
        await h.write("u1", 1, "content", M)
        await h.scanner.scan_once()

        for i in range(1, 6):
            task = asyncio.create_task(worker.drain_once())
            await asyncio.sleep(0.25)  # attempt 0 failed, worker sleeps in the backoff
            task.cancel()  # the process dies here
            with contextlib.suppress(asyncio.CancelledError):
                await task
            r = await h.row("u1")
            assert r.status == "processing"
            n = await h.state_repo.recover_orphan_processing()  # the restart
            r = await h.row("u1")
            print(
                f"  restart {i}: recovered {n} row, status={r.status}, persisted retry_count={r.retry_count}, handler attempts so far={emb.calls}"
            )
            assert r.status == "pending" and r.retry_count == 0
        print(
            f"  five restarts: {emb.calls} attempts made, the row is still pending with retry_count 0; cascade fix would list nothing"
        )

        # control: no restarts -> the budget is spent and the row becomes permanently failed
        cycles = 0
        while True:
            await worker.drain_once()
            cycles += 1
            r = await h.row("u1")
            if r.status == "failed" and r.retryable is False:
                break
            assert cycles < 20, (r.status, r.retryable, r.retry_count)
            await (
                h.scanner.scan_once()
            )  # scanner auto-retry keeps retry_count (failed + mtime unchanged)
        print(
            f"  control: permanently failed after {cycles} uninterrupted drain cycles, retry_count={r.retry_count}"
        )
        assert r.retry_count == 12 and cycles == 4
    finally:
        await h.close()


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