fix(*): make everos ownership explicit and stop memory failures from stalling a session

raven/agent/loop/main.py

@@ -53,6 +53,15 @@
 from raven.tracing import semconv, trace
 from raven.utils.helpers import estimate_prompt_tokens, is_image_part, is_inline_image
 
+# How long a turn is willing to wait on plugin-side indexing before letting the
+# write finish on its own. A budget, not a deadline: the task keeps running.
+_STORE_TURN_BUDGET_S: float = 5.0
+# Outstanding detached writes past which a turn waits for one to land, so a slow
+# memory service cannot grow an unbounded queue behind a fast typist.
+_STORE_MAX_INFLIGHT: int = 4
+# Teardown's total budget for letting those writes finish.
+_STORE_DRAIN_BUDGET_S: float = 15.0
+
 _ABORTED_ACTION_REPLY = (
     "The operation was not completed, and no alternative method will be attempted. "
     "Would you like me to continue with the remaining parts of the task that do not "
@@ -427,6 +436,9 @@ def __init__(
         # pipeline unchanged. See ``_dispatch_backend_store`` for the call
         # site that consumes it.
         self.backend: "MemoryBackend | None" = backend
+        # Writes that outran their turn budget and are still running. Held so
+        # teardown can drain them instead of dropping whatever was slowest.
+        self._store_inflight: set[asyncio.Task] = set()
 
         # Tools contributed by activated plugins; registered into the
         # ToolRegistry by ``_register_default_tools``.
@@ -1299,13 +1311,41 @@ async def _dispatch_backend_store(
             return
         if not messages_slice:
             return
-        try:
-            await self.backend.store(session_key, messages_slice)
-        except Exception:
-            logger.exception(
-                "backend.store failed for session {}; turn data preserved in session log, plugin-side indexing skipped",
-                session_key,
-            )
+
+        async def _store() -> None:
+            try:
+                await self.backend.store(session_key, messages_slice)  # type: ignore[union-attr]
+            except Exception:
+                logger.exception(
+                    "backend.store failed for session {}; turn data preserved in session log, "
+                    "plugin-side indexing skipped",
+                    session_key,
+                )
+
+        task = asyncio.create_task(_store())
+        self._store_inflight.add(task)
+        task.add_done_callback(self._store_inflight.discard)
+        if len(self._store_inflight) > _STORE_MAX_INFLIGHT:
+            # Backpressure rather than unbounded growth: a service slow enough
+            # to accumulate this many outstanding writes is one whose queue
+            # should stop growing, not one to keep feeding.
+            await asyncio.wait(set(self._store_inflight), return_when=asyncio.FIRST_COMPLETED)
+        # Deliberately not cancelled on timeout: the point is to stop *waiting*,
+        # not to abandon the write. A turn that indexes quickly still does so
+        # inline, which keeps ordering intact in the common case.
+        await asyncio.wait({task}, timeout=_STORE_TURN_BUDGET_S)
+
+    async def drain_backend_stores(self, timeout: float = _STORE_DRAIN_BUDGET_S) -> None:
+        """Let detached writes finish before the process goes away.
+
+        Writes that outran their turn budget are still in flight. Exiting on top
+        of them loses exactly the turns that were slowest to index, which is a
+        silent and biased kind of data loss.
+        """
+        pending = {t for t in self._store_inflight if not t.done()}
+        if not pending:
+            return
+        await asyncio.wait(pending, timeout=timeout)
 
     def _collect_injected_skill_ids(
         self,

raven/cli/agent_commands.py

@@ -338,6 +338,9 @@ async def run_once():
             finally:
                 if backend is not None:
                     try:
+                        # Detached indexing writes first: stopping the backend
+                        # closes the HTTP client they still need.
+                        await agent_loop.drain_backend_stores()
                         await backend.stop()
                     except Exception:
                         logger.exception(

raven/cli/doctor_commands.py

@@ -72,6 +72,9 @@ class MemoryInfo:
     """
 
     backend: Optional[str] = None
+    root: Optional[str] = None
+    owned: bool = True
+    address: Optional[str] = None
     server_running: bool = False
     reports_capabilities: bool = False
     configured: list[str] = field(default_factory=list)
@@ -223,14 +226,22 @@ def _probe_memory(config: "RavenConfig") -> MemoryInfo:
     info = MemoryInfo(backend=backend)
     if backend != "everos":
         return info
-    from raven.config.update_everos import everos_role_configured
+    from raven.config.update_everos import everos_owned, everos_role_configured, everos_root
     from raven.plugin.memory.everos._health import (
         DEGRADING_SECTIONS,
         REQUIRED_SECTIONS,
         configured_base_url,
         probe_capabilities,
     )
 
+    # Which memories, and whose. Neither was reachable from any command before:
+    # the wizard printed the path once while converging and nothing showed it
+    # again, so "where are my memories" had no answer short of reading
+    # config.json by hand. This is the place that question gets asked.
+    info.root = str(everos_root())
+    info.owned = everos_owned()
+    info.address = configured_base_url(config)
+
     info.configured = [s for s in (*REQUIRED_SECTIONS, *DEGRADING_SECTIONS) if everos_role_configured(s)]
     # Recall quality is decided by the embedding role in the user-level
     # everos.toml: with it recall matches meaning, without it only keywords.
@@ -261,6 +272,10 @@ def _render_memory_capabilities(memory: MemoryInfo) -> None:
 
     if memory.backend != "everos":
         return
+    console.print(f"  Memories:   {memory.root}")
+    if not memory.owned:
+        console.print("  [dim]Managed by you -- Raven reads this one and never writes or restarts it.[/dim]")
+    console.print(f"  Address:    {memory.address}")
     if not memory.server_running:
         console.print("  Server:     [dim]not running  (starts on demand)[/dim]")
         if memory.configured:

raven/cli/gateway_commands.py

@@ -592,6 +592,7 @@ async def _do_restart() -> None:
                 # spawned during AgentLoop teardown can complete.
                 if backend is not None:
                     try:
+                        await agent.drain_backend_stores()
                         await backend.stop()
                     except Exception:
                         _logger.exception(

raven/cli/import_commands.py

@@ -97,6 +97,32 @@ class ImportRunResult:
     skill_error: str = ""
 
 
+def _require_memory_service_ready(backend: object) -> None:
+    """Refuse to import when the memory service is not actually there.
+
+    ``backend.start()`` no longer raises: a session that cannot reach EverOS
+    degrades and keeps probing, which is right for a session and wrong here.
+    An import is one deliberate batch, and running it against nothing writes
+    nothing while consuming the source list.
+
+    Backends that do not report a state -- anything other than the everos one
+    -- are left alone rather than locked out.
+    """
+    state = getattr(backend, "_state", None)
+    if state is None:
+        return
+    from raven.plugin.memory.everos.backend import ServiceState
+
+    if state is ServiceState.READY:
+        return
+    from raven.plugin.memory.everos._server import server_log_path
+
+    console.print(f"[red]Memory service is not available ({state.value}); nothing would be imported.[/red]")
+    console.print(f"[dim]Check the server log: {server_log_path()}[/dim]")
+    console.print("[dim]Retry: raven import run[/dim]")
+    raise typer.Exit(1)
+
+
 async def _build_and_run(
     items: list[tuple[Scanner, ScanResult]],
     state: ImportState,
@@ -118,15 +144,10 @@ async def _build_and_run(
         )
         raise typer.Exit(1)
 
-    try:
-        await backend.start()
-    except Exception as e:
-        from raven.plugin.memory.everos._server import server_log_path
-
-        console.print(f"[red]Failed to start EverOS memory server: {e}[/red]")
-        console.print(f"[dim]Check the server log: {server_log_path()}[/dim]")
-        console.print("[dim]Retry: raven import run[/dim]")
-        raise typer.Exit(1)
+    await backend.start()
+    # Asked after start rather than caught around it: start reports through the
+    # backend's state now, so an except here would never fire.
+    _require_memory_service_ready(backend)
     try:
         summary = await run_import(items, backend, state, on_progress=on_progress, cancel_path=cancel_path)
         # Both phases below are additive and run after the EverOS pass, so a