fix(*): tell a handed-over memory write from a lost one at shutdown

raven/agent/loop/_shared.py

@@ -86,6 +86,15 @@
 # Teardown's budget for letting outstanding writes finish. See
 # ``drain_backend_stores``: the pipeline cuts retries short first, so this only
 # ever covers a request already on the wire, not a worker asleep in backoff.
+# Deliberately shorter than the write it waits on rather than sized to it: an
+# EverOS turn store is a final flush, and a flush is an extraction -- measured
+# at 16.98s end to end. Sizing this to cover that would park every CLI exit for
+# as long and buy nothing, because the service finishes the request whether or
+# not this process is still listening (one extraction landed 32s after the
+# drain gave up) and nothing needs to recall the turn that just ended. What a
+# ceiling this short costs is certainty, and that is paid where the outcome is
+# reported: a request still running here is reported as unsettled -- neither
+# lost nor written, because from here it is not knowable which.
 _STORE_DRAIN_BUDGET_S: float = 2.0
 # Runtime prose, not the model's: written here and shown to the model so it
 # stops, and carried to the client as a notice rather than as an answer. It

raven/agent/loop/organ_glue.py

@@ -18,6 +18,7 @@
     trace,
     vision_verdict,
 )
+from raven.memory_engine import DrainOutcome
 
 
 class OrganGlueMixin:
@@ -327,18 +328,31 @@ def _dispatch_backend_store(
         """
         self._store_pipeline.enqueue(session_key, messages_slice)
 
-    async def drain_backend_stores(self, timeout: float = _STORE_DRAIN_BUDGET_S) -> int:
-        """Let queued writes finish before the process goes away and return how
-        many turns were lost. The counting is the pipeline's; telling the user
-        is the host's (see ``cli._helpers.report_dropped_memory_writes``), and
-        the return value is what it tells them with."""
-        dropped = await self._store_pipeline.drain(timeout)
-        if dropped:
+    async def drain_backend_stores(self, timeout: float = _STORE_DRAIN_BUDGET_S) -> DrainOutcome:
+        """Let queued writes finish before the process goes away and return what
+        is left behind. The counting is the pipeline's; telling the user is the
+        host's (see ``cli._helpers.report_memory_write_outcome``), and the
+        return value is what it tells them with.
+
+        A write that was inside the backend call is reported apart from one
+        this side watched fail. Cancelling our own request does not cancel the
+        service's work, so calling the first lost is a false alarm -- but
+        entering the call is not delivery either, so calling it written is the
+        opposite false alarm. Both are reported as unsettled.
+        """
+        outcome = await self._store_pipeline.drain(timeout)
+        if outcome.lost:
             logger.warning(
                 "{} turn(s) were not indexed: the memory service never caught up",
-                dropped,
+                outcome.lost,
+            )
+        if outcome.in_flight:
+            logger.info(
+                "{} turn(s) were mid-write when the drain stopped waiting; "
+                "this side cannot see whether the service finished them",
+                outcome.in_flight,
             )
-        return dropped
+        return outcome
 
     def _note_memory_ok(self) -> None:
         """A successful store clears a standing fault, and says so once."""

raven/cli/_helpers.py

@@ -12,22 +12,42 @@
 
 from raven.config.schema import Config
 from raven.core import config_stack
+from raven.memory_engine import DrainOutcome
 
 console = Console()
 
 
-def report_dropped_memory_writes(dropped: int, out: Console | None = None) -> None:
-    """Tell the user how many turns never reached long-term memory at shutdown.
+def report_memory_write_outcome(outcome: DrainOutcome, out: Console | None = None) -> None:
+    """Tell the user what shutdown did to the turns still being indexed.
 
     The loop counts (``AgentLoop.drain_backend_stores``); the host renders, on
-    its own console or stderr, because this is the last moment the loss is
-    still actionable and the loop does not own a terminal.
+    its own console or stderr, because this is the last moment a loss is still
+    actionable and the loop does not own a terminal.
+
+    Two outcomes, two sentences. A turn this side watched fail is gone and is
+    worth a warning. A turn that was still inside the backend call is not:
+    stopping the wait neither cancels the request nor unwrites it, and
+    announcing it as lost sent the user looking for content that was being
+    indexed as they read.
+
+    The second sentence says the outcome is unknown rather than good. Entering
+    the call is not delivery, so a cancelled request may have written nothing,
+    and this side cannot tell the two apart. A reassuring note would send the
+    user away from a turn that never landed, which is the first mistake in a
+    mirror.
     """
-    if dropped:
-        (out or Console(stderr=True)).print(
-            f"[yellow]{dropped} turn(s) were not written to long-term memory "
+    console = out or Console(stderr=True)
+    if outcome.lost:
+        console.print(
+            f"[yellow]{outcome.lost} turn(s) were not written to long-term memory "
             "because the memory service was unavailable.[/yellow]"
         )
+    if outcome.in_flight:
+        console.print(
+            f"[dim]{outcome.in_flight} turn(s) were still being written when shutdown "
+            "stopped waiting; whether the memory service finished them is not known "
+            "here.[/dim]"
+        )
 
 
 def load_runtime_config(config: str | None = None, home: str | None = None) -> Config:

raven/cli/agent_commands.py

@@ -25,7 +25,7 @@
     print_config_migration_notices,
     print_deprecated_allow_destructive_notice,
     print_deprecated_memory_window_notice,
-    report_dropped_memory_writes,
+    report_memory_write_outcome,
 )
 from raven.core.provider_stack import build_model_routing
 from raven.providers.factory import make_provider
@@ -367,8 +367,8 @@ async def run_once():
                     try:
                         # Drain queued writes first: stopping the backend
                         # closes the HTTP client they still need.
-                        dropped = await agent_loop.drain_backend_stores()
-                        report_dropped_memory_writes(dropped, console)
+                        outcome = await agent_loop.drain_backend_stores()
+                        report_memory_write_outcome(outcome, console)
                         await backend.stop()
                     except Exception:
                         logger.exception(

raven/cli/gateway_commands.py

@@ -27,7 +27,7 @@
     print_config_migration_notices,
     print_deprecated_allow_destructive_notice,
     print_deprecated_memory_window_notice,
-    report_dropped_memory_writes,
+    report_memory_write_outcome,
 )
 from raven.core.provider_stack import build_model_routing
 from raven.providers.factory import make_resolving_provider
@@ -1157,8 +1157,8 @@ async def _reload(force: bool) -> dict:
                     _logger.exception("plugin services stop failed; continuing shutdown")
                 if backend is not None:
                     try:
-                        dropped = await agent.drain_backend_stores()
-                        report_dropped_memory_writes(dropped, console)
+                        outcome = await agent.drain_backend_stores()
+                        report_memory_write_outcome(outcome, console)
                         await backend.stop()
                     except Exception:
                         _logger.exception(

raven/cli/tui_commands.py

@@ -27,7 +27,7 @@
 
 import typer
 
-from raven.cli._helpers import report_dropped_memory_writes
+from raven.cli._helpers import report_memory_write_outcome
 from raven.cli._log_file import _strip_tty_stream_handlers, redirect_loguru_to_file
 from raven.i18n import t
 from raven.rpc.cron_events import build_cron_callback_spine, fanout_cron_missed
@@ -616,8 +616,8 @@ async def _start_backend() -> None:
         # Release the embedded index lock so the next process can start.
         if agent_loop is not None and agent_loop.backend is not None:
             try:
-                dropped = await agent_loop.drain_backend_stores()
-                report_dropped_memory_writes(dropped)
+                outcome = await agent_loop.drain_backend_stores()
+                report_memory_write_outcome(outcome)
                 await agent_loop.backend.stop()
             except Exception:
                 from loguru import logger as _logger

raven/memory_engine/__init__.py

@@ -68,11 +68,12 @@
         filter_by_required_tools,
     )
     from raven.memory_engine.skill_local.types import SkillMeta
-    from raven.memory_engine.store_pipeline import StorePipeline
+    from raven.memory_engine.store_pipeline import DrainOutcome, StorePipeline
 
 __all__ = [
     "DAILY_FIRE_PLAN_HEADER",
     "BehaviorsExtractor",
+    "DrainOutcome",
     "BehaviorsOffsets",
     "BackendSkillSource",
     "HubSkillSource",
@@ -115,6 +116,7 @@
     "SkillForgeRouter": "raven.memory_engine.skill_forge",
     "SkillMeta": "raven.memory_engine.skill_local.types",
     "SkillRegistry": "raven.memory_engine.skill_local.registry",
+    "DrainOutcome": "raven.memory_engine.store_pipeline",
     "StorePipeline": "raven.memory_engine.store_pipeline",
     "filter_by_required_tools": "raven.memory_engine.skill_local.registry",
     "DAILY_FIRE_PLAN_HEADER": "raven.memory_engine.consolidate.attention",

raven/memory_engine/store_pipeline.py

@@ -54,6 +54,31 @@ class _Record(NamedTuple):
     trace_ctx: Any
 
 
+class DrainOutcome(NamedTuple):
+    """What a teardown drain leaves behind, split by whether its fate is known.
+
+    ``lost`` will never be indexed: shed at admission, given up on after every
+    retry, or still queued when the budget ran out. Every one of those was
+    refused or never offered, so the loss is observed rather than inferred.
+
+    ``in_flight`` was inside the backend call when the drain cancelled its
+    worker, and how far it got is not observable from here: the request may
+    have been transmitted and be finishing at the service, or the cancellation
+    may have unwound it before anything left. Both happen. An EverOS extraction
+    was measured landing 34-48s after the drain gave up, which is the first
+    case; a backend that persists only after an await, cancelled mid-await, is
+    the second and writes nothing.
+
+    They are reported apart because they are different claims, not because one
+    is success. Counting ``in_flight`` as lost told the user a turn was gone
+    while it was being indexed; counting it as landed tells them it is safe
+    when it may not be. Callers must say neither.
+    """
+
+    lost: int
+    in_flight: int
+
+
 class StorePipeline:
     """Per-session serial delivery of turns to a :class:`MemoryBackend`.
 
@@ -85,6 +110,10 @@ def __init__(
         # teardown that times out under-reports what it lost by one per worker.
         self._active: dict[str, _Record] = {}
         self._slots = asyncio.Semaphore(MAX_CONCURRENCY)
+        # Sessions whose record is inside the backend call right now. ``drain``
+        # reads this before it cancels anything: the cancellation unwinds that
+        # call, and the unwinding clears the very marks it needs to count.
+        self._in_flight: set[str] = set()
         self._dropped = 0
 
     @property
@@ -190,7 +219,11 @@ async def _worker(self, session_key: str) -> None:
                         # other one out of the global budget.
                         async with self._slots:
                             with trace.use_context(record.trace_ctx):
-                                landed = await self._store_once(session_key, record.messages, attempt=attempt)
+                                self._in_flight.add(session_key)
+                                try:
+                                    landed = await self._store_once(session_key, record.messages, attempt=attempt)
+                                finally:
+                                    self._in_flight.discard(session_key)
                     except Exception as e:  # noqa: BLE001 - the turn must survive a failed index
                         reason = str(e)
                         logger.exception(
@@ -244,8 +277,8 @@ async def _store_once(self, session_key: str, messages_slice: list[dict], *, att
         # never claimed the write was lost.
         return await backend.store(session_key, messages_slice, metadata={"attempt": attempt}) is not False
 
-    async def drain(self, timeout: float) -> int:
-        """Settle what is outstanding and report how many turns were lost.
+    async def drain(self, timeout: float) -> DrainOutcome:
+        """Settle what is outstanding and report what is left behind.
 
         Retries are cut short first: a worker parked in a backoff is doing
         nothing, and would otherwise spend the whole budget asleep while the
@@ -260,6 +293,11 @@ async def drain(self, timeout: float) -> int:
         # closes the backend's HTTP client next, and a request still in flight
         # against a closed client fails as a transport error nobody reads.
         leftover = {t for t in self._workers.values() if not t.done()}
+        # Counted before the collect below, not merely before ``cancel``: the
+        # cancellation only unwinds the backend call once this coroutine yields,
+        # and the ``finally`` it unwinds through clears exactly these marks. A
+        # count taken after that await reads zero for every record in flight.
+        in_flight = len(self._in_flight)
         for task in leftover:
             task.cancel()
         if leftover:
@@ -274,9 +312,12 @@ async def drain(self, timeout: float) -> int:
                 if not task.cancelled():
                     task.exception()
         # Anything still queued, plus whatever each cancelled worker had
-        # already taken off its queue, was never going to be written.
+        # already taken off its queue, was never going to be written -- except
+        # the records that were inside the backend call, whose fate this side
+        # cannot see. Those are held out of the count rather than reported as
+        # written: an unobserved outcome is not a loss, and it is not a win.
         abandoned = sum(len(q) for q in self._queues.values()) + len(self._active)
         if abandoned:
-            self._dropped += abandoned
+            self._dropped += max(0, abandoned - in_flight)
             self._active.clear()
-        return self._dropped
+        return DrainOutcome(lost=self._dropped, in_flight=in_flight)