diff --git a/raven/agent/loop/_shared.py b/raven/agent/loop/_shared.py index aed40ba4f..f83185a13 100644 --- a/raven/agent/loop/_shared.py +++ b/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 diff --git a/raven/agent/loop/organ_glue.py b/raven/agent/loop/organ_glue.py index 3ba31779f..212b51f30 100644 --- a/raven/agent/loop/organ_glue.py +++ b/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.""" diff --git a/raven/cli/_helpers.py b/raven/cli/_helpers.py index 18ed56dd9..90f2b9610 100644 --- a/raven/cli/_helpers.py +++ b/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: diff --git a/raven/cli/agent_commands.py b/raven/cli/agent_commands.py index 834f8e20e..cc4f97486 100644 --- a/raven/cli/agent_commands.py +++ b/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( diff --git a/raven/cli/gateway_commands.py b/raven/cli/gateway_commands.py index 59eb98e87..3d87b095a 100644 --- a/raven/cli/gateway_commands.py +++ b/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( diff --git a/raven/cli/tui_commands.py b/raven/cli/tui_commands.py index 521dad052..85f6077c6 100644 --- a/raven/cli/tui_commands.py +++ b/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 diff --git a/raven/memory_engine/__init__.py b/raven/memory_engine/__init__.py index 382dcbc56..299e51e0f 100644 --- a/raven/memory_engine/__init__.py +++ b/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", diff --git a/raven/memory_engine/store_pipeline.py b/raven/memory_engine/store_pipeline.py index 29263ec43..310516cce 100644 --- a/raven/memory_engine/store_pipeline.py +++ b/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) diff --git a/tests/test_agent_loop_backend_dispatch.py b/tests/test_agent_loop_backend_dispatch.py index 8c31bc2ed..73049eb2e 100644 --- a/tests/test_agent_loop_backend_dispatch.py +++ b/tests/test_agent_loop_backend_dispatch.py @@ -567,8 +567,9 @@ async def test_drain_counts_records_still_queued_behind_a_wedged_write(self, tmp """A wedged first write parks the worker mid-attempt, so it never gets back to the nine records behind it in the queue. Those nine were neither indexed nor reported: ``_store_dropped`` stayed at 0 while the - queue silently held onto them. The tenth is the one the worker had - already taken off the queue, which counting the queues alone missed. + queue silently held onto them, and they are a genuine loss. The tenth + is the one the worker had already handed to the backend, which counting + the queues alone missed and which is not lost -- it is in flight. """ import asyncio import time @@ -583,10 +584,10 @@ async def store(self, session_id, messages, **kw): agent._dispatch_backend_store("s", [{"role": "user", "content": str(i)}]) t0 = time.monotonic() - await agent.drain_backend_stores(timeout=0.2) + outcome = await agent.drain_backend_stores(timeout=0.2) assert time.monotonic() - t0 < 1.0 - assert agent._store_pipeline.dropped == 10 + assert (outcome.lost, outcome.in_flight) == (9, 1) class TestTheGiveUpMessageIsHonest: @@ -610,10 +611,12 @@ async def store(self, session_id, messages, **kw): agent = _make_loop(tmp_path, backend=_Never()) agent._dispatch_backend_store("s", [{"role": "user", "content": "x"}]) - dropped = await agent.drain_backend_stores(timeout=5.0) + outcome = await agent.drain_backend_stores(timeout=5.0) assert agent._store_pipeline.dropped == 1 - assert dropped == 1 # the host renders the notice from this count; the loop prints nothing + # The host renders the notice from this outcome; the loop prints nothing. + # A give-up is a real loss: every retry was answered, so nothing is in flight. + assert (outcome.lost, outcome.in_flight) == (1, 0) assert "were not written" not in capsys.readouterr().err async def test_a_turn_that_succeeds_on_retry_is_not_counted( @@ -736,6 +739,10 @@ class TestShutdownAccountsForEveryTurnItLoses: """A worker has already taken its current record off the queue, so counting only the queues under-reports by one per worker -- and leaving that worker running lets it race the HTTP client the caller closes next. + + The accounting is split between ``lost`` and ``in_flight`` now, but the + invariant is the same one: every admitted turn lands in exactly one of the + two columns, and none of them goes missing. """ async def test_every_enqueued_turn_is_accounted_for(self, tmp_path: Path) -> None: @@ -750,8 +757,8 @@ async def store(self, session_id, messages, **kw): agent._dispatch_backend_store("s", [{"role": "user", "content": str(i)}]) await asyncio.sleep(0.05) - await agent.drain_backend_stores(timeout=0.2) - assert agent._store_pipeline.dropped == 10, f"reported {agent._store_pipeline.dropped} of 10 lost turns" + outcome = await agent.drain_backend_stores(timeout=0.2) + assert outcome.lost + outcome.in_flight == 10, f"reported {outcome} of 10 admitted turns" async def test_no_worker_outlives_the_drain(self, tmp_path: Path) -> None: import asyncio @@ -950,3 +957,128 @@ async def store(self, session_id, messages, **kw): for task in list(pipe._workers.values()): task.cancel() + + +class TestTheDrainTellsAnUnsettledWriteFromALostOne: + """A record still inside the backend call is not a lost turn. + + Cancelling this client does not cancel a request the service already has: + on a one-shot run an EverOS extraction was measured landing 34-48s after + the drain gave up. Counting that with the records that never left the queue + told the user their turn was gone while it was being written. + + It is not a written turn either. Entering the call is not delivery, and a + backend that persists only after an await writes nothing when the drain + cancels it mid-await. Both outcomes wear the same mark here, which is why + the mark means unsettled and the tests below check what is *claimed* about + it, not only which bucket it lands in. + """ + + async def test_a_handed_over_write_is_in_flight_not_lost(self, tmp_path: Path) -> None: + import asyncio + + class _Wedged: + async def store(self, session_id, messages, **kw): + await asyncio.sleep(30) + return True + + agent = _make_loop(tmp_path, backend=_Wedged()) + agent._dispatch_backend_store("s", [{"role": "user", "content": "x"}]) + await asyncio.sleep(0.05) + + outcome = await agent.drain_backend_stores(timeout=0.2) + + assert (outcome.in_flight, outcome.lost) == (1, 0) + + for task in list(agent._store_pipeline._workers.values()): + task.cancel() + + async def test_records_still_queued_behind_it_are_lost(self, tmp_path: Path) -> None: + """Only the record the worker took off the queue reached the service. + The nine behind it never did, and stay a genuine loss.""" + import asyncio + + class _Wedged: + async def store(self, session_id, messages, **kw): + await asyncio.sleep(30) + return True + + agent = _make_loop(tmp_path, backend=_Wedged()) + for i in range(10): + agent._dispatch_backend_store("s", [{"role": "user", "content": str(i)}]) + await asyncio.sleep(0.05) + + outcome = await agent.drain_backend_stores(timeout=0.2) + + assert (outcome.in_flight, outcome.lost) == (1, 9) + + for task in list(agent._store_pipeline._workers.values()): + task.cancel() + + async def test_a_write_cancelled_before_the_service_had_it_claims_no_delivery(self, tmp_path: Path) -> None: + """The premise the split rests on, driven rather than assumed. + + A backend that persists only after an await writes nothing when the + drain cancels it mid-await. From this side that is indistinguishable + from a request the service already holds, so neither the log nor the + notice the host renders from this outcome may say the turn landed. + """ + import asyncio + import io + + from loguru import logger + from rich.console import Console + + from raven.cli._helpers import report_memory_write_outcome + + # A sink of our own: loguru binds the real ``sys.stderr`` when its + # handler is added, so ``capsys`` replacing that object captures none + # of this and every assertion against it would pass unread. + logged = io.StringIO() + sink = logger.add(logged, format="{message}", level="INFO") + + persisted: list[str] = [] + + class _PersistsAfterAnAwait: + async def store(self, session_id, messages, **kw): + await asyncio.sleep(30) + persisted.append(session_id) + return True + + agent = _make_loop(tmp_path, backend=_PersistsAfterAnAwait()) + agent._dispatch_backend_store("s", [{"role": "user", "content": "x"}]) + await asyncio.sleep(0.05) + + try: + outcome = await agent.drain_backend_stores(timeout=0.2) + finally: + logger.remove(sink) + + assert persisted == [] + assert (outcome.in_flight, outcome.lost) == (1, 0) + + buf = io.StringIO() + report_memory_write_outcome(outcome, Console(file=buf, force_terminal=False, width=200)) + told = logged.getvalue() + buf.getvalue() + assert "1 turn(s)" in told + assert "turn(s)" in logged.getvalue() + assert "reached" not in told + + for task in list(agent._store_pipeline._workers.values()): + task.cancel() + + async def test_a_write_that_never_reached_the_service_is_not_in_flight(self, tmp_path: Path) -> None: + """A backend that keeps answering False exhausts its retries and is + given up on. Nothing is in flight at that point -- the give-up is the + old, honest meaning of a lost turn, and it must survive the split.""" + + class _Never: + async def store(self, session_id, messages, **kw): + return False + + agent = _make_loop(tmp_path, backend=_Never()) + agent._dispatch_backend_store("s", [{"role": "user", "content": "x"}]) + + outcome = await agent.drain_backend_stores(timeout=5.0) + + assert (outcome.in_flight, outcome.lost) == (0, 1) diff --git a/tests/test_cli_agent_commands.py b/tests/test_cli_agent_commands.py index e1209e84a..ab09cd482 100644 --- a/tests/test_cli_agent_commands.py +++ b/tests/test_cli_agent_commands.py @@ -125,7 +125,12 @@ async def _skip_background_grace(*_args, **_kwargs) -> None: def _invoke_agent_capturing_session( - monkeypatch: pytest.MonkeyPatch, home: Path, extra_args: list[str] + monkeypatch: pytest.MonkeyPatch, + home: Path, + extra_args: list[str], + *, + backend: object | None = None, + drain_outcome: object | None = None, ) -> tuple[object, dict[str, str]]: """Run ``agent -m`` with the provider and AgentLoop stubbed out, capturing the session_id that reaches the spine turn (req.conversation is the session @@ -169,6 +174,11 @@ async def await_pending_extractions(self, **_kw) -> None: async def close_mcp(self) -> None: pass + async def drain_backend_stores(self, *_a, **_kw): + from raven.memory_engine import DrainOutcome + + return drain_outcome if drain_outcome is not None else DrainOutcome(lost=0, in_flight=0) + monkeypatch.setattr("raven.cli.agent_commands.make_provider", lambda _: object()) monkeypatch.setattr("raven.agent.loop.AgentLoop", _StubAgentLoop) monkeypatch.setattr("raven.cli.agent_commands._wait_for_background_work", _skip_background_grace) @@ -177,7 +187,7 @@ async def close_mcp(self) -> None: # embedded everos runtime is heavy and not under test here). monkeypatch.setattr( "raven.core.plugin_stack.maybe_build_memory_backend", - lambda *a, **k: None, + lambda *a, **k: backend, ) monkeypatch.setattr( "raven.core.plugin_stack.build_plugin_tools", @@ -205,6 +215,47 @@ def test_agent_default_mints_fresh_session(tmp_config: Path, tmp_path: Path, mon assert cap1["session_id"] != cap2["session_id"], "each bare invocation must mint a NEW session" +def test_a_one_shot_shutdown_tells_the_user_what_the_drain_left( + tmp_config: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """``agent -m`` renders the drain outcome; it does not merely compute it. + + The one-shot exit is the only place a CLI user learns a turn did not + settle -- there is no next screen to carry the notice. The same teardown + in ``tui_commands`` is driven by its own test and this copy was driven by + none, so dropping the render here would have cost nothing red. + """ + from raven.memory_engine import DrainOutcome + + class _Backend: + def __init__(self) -> None: + self.stopped = False + + async def start(self) -> None: + pass + + async def stop(self) -> None: + self.stopped = True + + ws = tmp_path / "chanwork" + ws.mkdir() + backend = _Backend() + + r, _ = _invoke_agent_capturing_session( + monkeypatch, + ws, + [], + backend=backend, + drain_outcome=DrainOutcome(lost=2, in_flight=0), + ) + + assert r.exit_code == 0, r.stdout + assert "2 turn(s) were not written" in r.stdout + # The drain has to precede the stop: stopping closes the HTTP client the + # queued writes still need. + assert backend.stopped + + def test_agent_continue_binds_most_recent_cli_session( tmp_config: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_providers_factory.py b/tests/test_providers_factory.py index 4a796c39e..06081147b 100644 --- a/tests/test_providers_factory.py +++ b/tests/test_providers_factory.py @@ -635,19 +635,59 @@ def test_no_notice_prints_nothing_at_all(capsys, monkeypatch) -> None: assert (captured.out, captured.err) == ("", "") -def test_report_dropped_memory_writes_renders_on_the_given_console_only_when_something_was_lost() -> None: +def _notice(outcome) -> str: import io from rich.console import Console - from raven.cli._helpers import report_dropped_memory_writes + from raven.cli._helpers import report_memory_write_outcome buf = io.StringIO() - out = Console(file=buf, force_terminal=False, width=200) - report_dropped_memory_writes(0, out) - assert buf.getvalue() == "" - report_dropped_memory_writes(2, out) - assert "2 turn(s) were not written to long-term memory" in buf.getvalue() + report_memory_write_outcome(outcome, Console(file=buf, force_terminal=False, width=200)) + return buf.getvalue() + + +def test_report_memory_write_outcome_says_nothing_when_every_write_landed() -> None: + from raven.memory_engine import DrainOutcome + + assert _notice(DrainOutcome(lost=0, in_flight=0)) == "" + + +def test_report_memory_write_outcome_names_the_turns_that_never_reached_the_service() -> None: + from raven.memory_engine import DrainOutcome + + assert "2 turn(s) were not written to long-term memory" in _notice(DrainOutcome(lost=2, in_flight=0)) + + +def test_an_unsettled_turn_is_not_announced_as_unwritten() -> None: + """The drain stopped waiting; it did not watch a write fail. + + Saying the turn was not written -- and blaming a service that may well + have answered -- asserts the failure this side never observed. The other + direction is ``test_an_unsettled_turn_is_not_announced_as_landed``; between + them the notice is pinned to the only honest claim, which is neither. + """ + from raven.memory_engine import DrainOutcome + + text = _notice(DrainOutcome(lost=0, in_flight=1)) + assert "1 turn(s)" in text + assert "were not written" not in text + assert "unavailable" not in text + assert "not known" in text + + +def test_an_unsettled_turn_is_not_announced_as_landed() -> None: + """Being inside ``backend.store()`` when the drain cancelled is not delivery. + + A backend that persists after an await is cancelled mid-await and writes + nothing. From this side that is indistinguishable from a request the + service already holds, so the notice may claim neither outcome. + """ + from raven.memory_engine import DrainOutcome + + text = _notice(DrainOutcome(lost=0, in_flight=1)) + assert "1 turn(s)" in text + assert "reached long-term memory" not in text # --- generation settings reach the provider the ordinary assembly paths build ---