From 474872d68f22636238df19a64e369f4ae7d866ed Mon Sep 17 00:00:00 2001
From: Dimitris Poulopoulos <dimitris.a.poulopoulos@gmail.com>
Date: Fri, 19 Jun 2026 14:54:16 +0000
Subject: [PATCH 1/4] feat(memory): add the OTARI_MEMORY_ENABLED config flag

A master switch (default off) for persistent memory in platform mode. When off,
the gateway makes no memory calls and behavior is unchanged; per-workspace
enablement is still controlled by the platform.

Refs #189

Signed-off-by: Dimitris Poulopoulos <dimitris.a.poulopoulos@gmail.com>
---
 src/gateway/core/config.py                   | 19 +++++++++++++++++++
 tests/integration/test_config_env_loading.py |  4 ++++
 2 files changed, 23 insertions(+)

diff --git a/src/gateway/core/config.py b/src/gateway/core/config.py
index e032cd33..6e10021d 100644
--- a/src/gateway/core/config.py
+++ b/src/gateway/core/config.py
@@ -227,6 +227,19 @@ class GatewayConfig(BaseSettings):
         default=True,
         description="Enable the /v1/files upload/storage endpoints (standalone mode).",
     )
+    memory_enabled: bool = Field(
+        default=False,
+        description=(
+            "Master switch for persistent memory (platform mode). When on, the gateway "
+            "recalls facts from the platform before a completion and stores facts after. "
+            "Per-workspace enablement is still controlled by the platform; this only gates "
+            "whether the gateway makes the calls at all. Recall runs before dispatch, so a "
+            "slow memory service adds up to its timeout to time-to-first-token; the storing "
+            "call after a completion is fire-and-forget. Both timeouts are tunable via the "
+            "platform settings 'memory_recall_timeout_ms' (default 2000) and "
+            "'memory_remember_timeout_ms' (default 10000), or their PLATFORM_* env aliases."
+        ),
+    )
     files_backend: str = Field(
         default="local",
         description="Blob backend for uploaded file bytes: 'local' (filesystem). Future: 's3', 'gcs'.",
@@ -551,6 +564,12 @@ def _apply_platform_env_overrides(config: dict[str, Any]) -> None:
         "PLATFORM_RESOLVE_TIMEOUT_MS": ("resolve_timeout_ms", int),
         "PLATFORM_USAGE_TIMEOUT_MS": ("usage_timeout_ms", int),
         "PLATFORM_USAGE_MAX_RETRIES": ("usage_max_retries", int),
+        # Best-effort memory timeouts (platform mode). Recall is on the hot path
+        # (added to time-to-first-token); remember is fire-and-forget after the
+        # completion. Both fall back to their defaults on a missing or invalid
+        # value (see _coerce_timeout_ms in routes/_platform.py).
+        "PLATFORM_MEMORY_RECALL_TIMEOUT_MS": ("memory_recall_timeout_ms", int),
+        "PLATFORM_MEMORY_REMEMBER_TIMEOUT_MS": ("memory_remember_timeout_ms", int),
         # Per-attempt budget for streaming fallback: how long to wait for the
         # first chunk from each attempt before treating it as hung and moving
         # to the next entry in the routing policy. Tunable per deployment;
diff --git a/tests/integration/test_config_env_loading.py b/tests/integration/test_config_env_loading.py
index a0e0559a..76f4a347 100644
--- a/tests/integration/test_config_env_loading.py
+++ b/tests/integration/test_config_env_loading.py
@@ -163,6 +163,8 @@ def test_load_config_platform_env_overrides(tmp_path: Path, monkeypatch: pytest.
     monkeypatch.setenv("PLATFORM_RESOLVE_TIMEOUT_MS", "1234")
     monkeypatch.setenv("PLATFORM_USAGE_TIMEOUT_MS", "2345")
     monkeypatch.setenv("PLATFORM_USAGE_MAX_RETRIES", "7")
+    monkeypatch.setenv("PLATFORM_MEMORY_RECALL_TIMEOUT_MS", "1500")
+    monkeypatch.setenv("PLATFORM_MEMORY_REMEMBER_TIMEOUT_MS", "8000")
 
     config = load_config(str(config_file))
 
@@ -171,6 +173,8 @@ def test_load_config_platform_env_overrides(tmp_path: Path, monkeypatch: pytest.
     assert config.platform["resolve_timeout_ms"] == 1234
     assert config.platform["usage_timeout_ms"] == 2345
     assert config.platform["usage_max_retries"] == 7
+    assert config.platform["memory_recall_timeout_ms"] == 1500
+    assert config.platform["memory_remember_timeout_ms"] == 8000
 
 
 def test_load_config_sets_default_platform_base_url_when_token_is_set(

From c2969591f4a0b73de20d86dbfee6ca11de71fd27 Mon Sep 17 00:00:00 2001
From: Dimitris Poulopoulos <dimitris.a.poulopoulos@gmail.com>
Date: Fri, 19 Jun 2026 14:54:28 +0000
Subject: [PATCH 2/4] feat(memory): add a best-effort platform memory client
 and prompt helpers

_recall_platform_memory / _remember_platform_memory call the platform's
/gateway/memory/* endpoints, modeled on the usage reporter: they swallow
timeouts and errors so a slow or unavailable memory service never breaks a chat.
inject_memory_facts prepends recalled facts to the system message (mirroring
inject_purpose_hints); build_remember_messages assembles the minimal new turn to
store. Helpers are unit-tested and not yet wired into the request path.

Refs #189

Signed-off-by: Dimitris Poulopoulos <dimitris.a.poulopoulos@gmail.com>
---
 src/gateway/api/routes/_helpers.py  |  39 +++++++
 src/gateway/api/routes/_platform.py |  79 +++++++++++++
 tests/unit/test_memory_inject.py    |  53 +++++++++
 tests/unit/test_memory_platform.py  | 170 ++++++++++++++++++++++++++++
 4 files changed, 341 insertions(+)
 create mode 100644 tests/unit/test_memory_inject.py
 create mode 100644 tests/unit/test_memory_platform.py

diff --git a/src/gateway/api/routes/_helpers.py b/src/gateway/api/routes/_helpers.py
index 5580b377..f3d2f3dd 100644
--- a/src/gateway/api/routes/_helpers.py
+++ b/src/gateway/api/routes/_helpers.py
@@ -124,6 +124,45 @@ def latest_user_text(messages: Sequence[Any]) -> str:
     return ""
 
 
+MEMORY_FACTS_HEADER = (
+    "Relevant memory about the user (use when helpful; do not mention that you are recalling memory):"
+)
+
+
+def inject_memory_facts(messages: list[dict[str, Any]], facts: list[str]) -> list[dict[str, Any]]:
+    """Prepend or extend the system message with recalled memory facts.
+
+    Mirrors :func:`gateway.services.mcp_loop.inject_purpose_hints`: augments an existing
+    system message, or inserts one when absent. Returns a new list; the input is not
+    mutated. A no-op when ``facts`` is empty.
+    """
+    if not facts:
+        return messages
+    block = "\n".join([MEMORY_FACTS_HEADER, *(f"- {fact}" for fact in facts)])
+    out = list(messages)
+    if out and isinstance(out[0], dict) and out[0].get("role") == "system":
+        existing = out[0].get("content") or ""
+        out[0] = {**out[0], "content": f"{existing}\n\n{block}" if existing else block}
+    else:
+        out.insert(0, {"role": "system", "content": block})
+    return out
+
+
+def build_remember_messages(messages: Sequence[Any], assistant_text: str) -> list[dict[str, str]]:
+    """Build the minimal exchange to store: the latest user turn plus the assistant reply.
+
+    Keeping it to the new turn (rather than the whole history) avoids re-extracting facts
+    the platform has already processed. Returns an empty list when there is nothing to store.
+    """
+    out: list[dict[str, str]] = []
+    user_text = latest_user_text(messages)
+    if user_text:
+        out.append({"role": "user", "content": user_text})
+    if assistant_text:
+        out.append({"role": "assistant", "content": assistant_text})
+    return out
+
+
 async def apply_input_guardrails(
     guardrails: list[GuardrailConfig] | None,
     input_text: str,
diff --git a/src/gateway/api/routes/_platform.py b/src/gateway/api/routes/_platform.py
index fd16ebd5..86ea67ec 100644
--- a/src/gateway/api/routes/_platform.py
+++ b/src/gateway/api/routes/_platform.py
@@ -675,6 +675,85 @@ async def _resolve_platform_code_execution(
     )
 
 
+def _coerce_timeout_ms(value: Any, default_ms: int) -> int:
+    """Parse a configured timeout (milliseconds) into a positive int, falling
+    back to ``default_ms`` for missing, non-numeric, or non-positive values so a
+    misconfigured timeout never breaks a best-effort memory call."""
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return default_ms
+    return parsed if parsed > 0 else default_ms
+
+
+async def _recall_platform_memory(
+    config: GatewayConfig,
+    user_token: str,
+    query: str,
+) -> list[str]:
+    """Recall relevant memory facts for the user-token's workspace.
+
+    Best-effort and on the hot path: any failure (misconfigured platform, timeout,
+    network error, non-200, or a malformed body) returns no facts so a slow or
+    unavailable memory service never breaks the completion. The platform also
+    returns an empty list when the workspace has memory disabled.
+    """
+    platform_base_url = config.platform.get("base_url")
+    if not platform_base_url or not query.strip():
+        return []
+
+    timeout_ms = _coerce_timeout_ms(config.platform.get("memory_recall_timeout_ms", 2000), 2000)
+    url = _platform_url(platform_base_url, "/gateway/memory/recall")
+    headers = {
+        "X-Gateway-Token": config.platform_token or "",
+        "X-User-Token": user_token,
+    }
+
+    try:
+        response = await _post_platform(
+            url=url, headers=headers, body={"query": query}, timeout_seconds=timeout_ms / 1000
+        )
+    except httpx.HTTPError:
+        return []
+
+    if response.status_code != 200:
+        return []
+    try:
+        payload = response.json()
+    except ValueError:
+        return []
+    facts = payload.get("facts") if isinstance(payload, dict) else None
+    if not isinstance(facts, list):
+        return []
+    return [fact for fact in facts if isinstance(fact, str)]
+
+
+async def _remember_platform_memory(
+    config: GatewayConfig,
+    user_token: str,
+    messages: list[dict[str, Any]],
+) -> None:
+    """Store durable facts from a completed exchange. Best-effort and fire-and-forget:
+    failures are swallowed so they never affect the user's response path."""
+    platform_base_url = config.platform.get("base_url")
+    if not platform_base_url or not messages:
+        return
+
+    timeout_ms = _coerce_timeout_ms(config.platform.get("memory_remember_timeout_ms", 10000), 10000)
+    url = _platform_url(platform_base_url, "/gateway/memory/remember")
+    headers = {
+        "X-Gateway-Token": config.platform_token or "",
+        "X-User-Token": user_token,
+    }
+
+    try:
+        await _post_platform(
+            url=url, headers=headers, body={"messages": messages}, timeout_seconds=timeout_ms / 1000
+        )
+    except httpx.HTTPError:
+        return
+
+
 async def _report_platform_usage(
     config: GatewayConfig,
     correlation_id: str,
diff --git a/tests/unit/test_memory_inject.py b/tests/unit/test_memory_inject.py
new file mode 100644
index 00000000..a02c92a4
--- /dev/null
+++ b/tests/unit/test_memory_inject.py
@@ -0,0 +1,53 @@
+"""Unit tests for the memory message helpers (inject_memory_facts, build_remember_messages)."""
+
+from __future__ import annotations
+
+from gateway.api.routes._helpers import (
+    MEMORY_FACTS_HEADER,
+    build_remember_messages,
+    inject_memory_facts,
+)
+
+
+def test_inject_no_facts_is_noop() -> None:
+    messages = [{"role": "user", "content": "hi"}]
+    assert inject_memory_facts(messages, []) is messages
+
+
+def test_inject_creates_system_message_when_absent() -> None:
+    messages = [{"role": "user", "content": "hi"}]
+    out = inject_memory_facts(messages, ["likes metric", "name is Dimitris"])
+    assert out[0]["role"] == "system"
+    assert MEMORY_FACTS_HEADER in out[0]["content"]
+    assert "- likes metric" in out[0]["content"]
+    assert "- name is Dimitris" in out[0]["content"]
+    assert out[1] == {"role": "user", "content": "hi"}
+    # input is not mutated
+    assert messages == [{"role": "user", "content": "hi"}]
+
+
+def test_inject_extends_existing_system_message() -> None:
+    messages = [{"role": "system", "content": "You are helpful."}, {"role": "user", "content": "hi"}]
+    out = inject_memory_facts(messages, ["likes metric"])
+    assert out[0]["content"].startswith("You are helpful.")
+    assert "- likes metric" in out[0]["content"]
+    assert len(out) == 2
+    assert messages[0]["content"] == "You are helpful."  # not mutated
+
+
+def test_build_remember_messages_user_and_assistant() -> None:
+    messages = [{"role": "user", "content": "My name is Dimitris"}]
+    out = build_remember_messages(messages, "Nice to meet you, Dimitris.")
+    assert out == [
+        {"role": "user", "content": "My name is Dimitris"},
+        {"role": "assistant", "content": "Nice to meet you, Dimitris."},
+    ]
+
+
+def test_build_remember_messages_skips_empty_assistant() -> None:
+    messages = [{"role": "user", "content": "hi"}]
+    assert build_remember_messages(messages, "") == [{"role": "user", "content": "hi"}]
+
+
+def test_build_remember_messages_empty_when_nothing() -> None:
+    assert build_remember_messages([], "") == []
diff --git a/tests/unit/test_memory_platform.py b/tests/unit/test_memory_platform.py
new file mode 100644
index 00000000..e2637bb7
--- /dev/null
+++ b/tests/unit/test_memory_platform.py
@@ -0,0 +1,170 @@
+"""Unit tests for the best-effort platform memory client (recall + remember)."""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+
+from gateway.api.routes import _platform as platform_module
+from gateway.api.routes._platform import _recall_platform_memory, _remember_platform_memory
+
+
+def _config(*, base_url: str | None = "https://platform.local", **platform_extra: Any) -> Any:
+    cfg = MagicMock()
+    cfg.platform = {"base_url": base_url} if base_url else {}
+    cfg.platform.update(platform_extra)
+    cfg.platform_token = "gw_test_token"
+    return cfg
+
+
+# ───────────── recall ─────────────
+
+
+@pytest.mark.asyncio
+async def test_recall_returns_facts_and_sends_dual_auth(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_post(
+        *, url: str, headers: dict[str, str], body: dict[str, Any], timeout_seconds: float
+    ) -> httpx.Response:
+        captured.update(url=url, headers=headers, body=body)
+        return httpx.Response(200, json={"facts": ["likes metric", "name is Dimitris"]})
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+
+    out = await _recall_platform_memory(_config(), "tk_user", "what units?")
+
+    assert out == ["likes metric", "name is Dimitris"]
+    assert captured["url"].endswith("/gateway/memory/recall")
+    assert captured["headers"]["X-Gateway-Token"] == "gw_test_token"
+    assert captured["headers"]["X-User-Token"] == "tk_user"
+    assert captured["body"] == {"query": "what units?"}
+
+
+@pytest.mark.asyncio
+async def test_recall_without_base_url_skips_call() -> None:
+    assert await _recall_platform_memory(_config(base_url=None), "tk", "q") == []
+
+
+@pytest.mark.asyncio
+async def test_recall_blank_query_skips_call() -> None:
+    assert await _recall_platform_memory(_config(), "tk", "   ") == []
+
+
+@pytest.mark.asyncio
+async def test_recall_non_200_yields_no_facts(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(**kwargs: Any) -> httpx.Response:
+        return httpx.Response(500, json={"detail": "boom"})
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    assert await _recall_platform_memory(_config(), "tk", "q") == []
+
+
+@pytest.mark.asyncio
+async def test_recall_timeout_is_swallowed(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(**kwargs: Any) -> httpx.Response:
+        raise httpx.TimeoutException("slow")
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    assert await _recall_platform_memory(_config(), "tk", "q") == []
+
+
+@pytest.mark.asyncio
+async def test_recall_malformed_payload_yields_no_facts(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(**kwargs: Any) -> httpx.Response:
+        return httpx.Response(200, json={"facts": "not-a-list"})
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    assert await _recall_platform_memory(_config(), "tk", "q") == []
+
+
+@pytest.mark.asyncio
+async def test_recall_other_httpx_error_is_swallowed(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(**kwargs: Any) -> httpx.Response:
+        raise httpx.RemoteProtocolError("server disconnected")
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    assert await _recall_platform_memory(_config(), "tk", "q") == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("bad_timeout", ["abc", None, "", []])
+async def test_recall_malformed_timeout_falls_back(monkeypatch: pytest.MonkeyPatch, bad_timeout: Any) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_post(*, timeout_seconds: float, **kwargs: Any) -> httpx.Response:
+        captured["timeout_seconds"] = timeout_seconds
+        return httpx.Response(200, json={"facts": ["ok"]})
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    out = await _recall_platform_memory(_config(memory_recall_timeout_ms=bad_timeout), "tk", "q")
+    assert out == ["ok"]
+    assert captured["timeout_seconds"] == 2.0
+
+
+# ───────────── remember ─────────────
+
+
+@pytest.mark.asyncio
+async def test_remember_posts_messages(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_post(
+        *, url: str, headers: dict[str, str], body: dict[str, Any], timeout_seconds: float
+    ) -> httpx.Response:
+        captured.update(url=url, headers=headers, body=body)
+        return httpx.Response(202)
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+
+    messages = [{"role": "user", "content": "My name is Dimitris"}, {"role": "assistant", "content": "Hi"}]
+    await _remember_platform_memory(_config(), "tk_user", messages)
+
+    assert captured["url"].endswith("/gateway/memory/remember")
+    assert captured["headers"]["X-User-Token"] == "tk_user"
+    assert captured["body"] == {"messages": messages}
+
+
+@pytest.mark.asyncio
+async def test_remember_empty_messages_skips_call() -> None:
+    # No base_url guard not needed: empty messages short-circuit before any call.
+    await _remember_platform_memory(_config(), "tk", [])  # must not raise
+
+
+@pytest.mark.asyncio
+async def test_remember_error_is_swallowed(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(**kwargs: Any) -> httpx.Response:
+        raise httpx.NetworkError("down")
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    # Must not raise.
+    await _remember_platform_memory(_config(), "tk", [{"role": "user", "content": "x"}])
+
+
+@pytest.mark.asyncio
+async def test_remember_other_httpx_error_is_swallowed(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(**kwargs: Any) -> httpx.Response:
+        raise httpx.RemoteProtocolError("server disconnected")
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    # Must not raise.
+    await _remember_platform_memory(_config(), "tk", [{"role": "user", "content": "x"}])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("bad_timeout", ["abc", None, "", []])
+async def test_remember_malformed_timeout_falls_back(monkeypatch: pytest.MonkeyPatch, bad_timeout: Any) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_post(*, timeout_seconds: float, **kwargs: Any) -> httpx.Response:
+        captured["timeout_seconds"] = timeout_seconds
+        return httpx.Response(202)
+
+    monkeypatch.setattr(platform_module, "_post_platform", fake_post)
+    await _remember_platform_memory(
+        _config(memory_remember_timeout_ms=bad_timeout), "tk", [{"role": "user", "content": "x"}]
+    )
+    assert captured["timeout_seconds"] == 10.0

From 5a08d6309dc3d430a53cd63ec629c5d237eb5724 Mon Sep 17 00:00:00 2001
From: Dimitris Poulopoulos <dimitris.a.poulopoulos@gmail.com>
Date: Fri, 19 Jun 2026 14:54:43 +0000
Subject: [PATCH 3/4] feat(memory): recall and store memories around chat
 completions

Gated by OTARI_MEMORY_ENABLED in platform mode: recall runs once before dispatch
and injects facts into the system message (covering every path, streaming and
not), and the new turn is stored fire-and-forget afterwards. Non-streaming uses
a background task; streaming uses _stream_with_memory_capture to accumulate the
assistant text and remember only on normal completion (skipped on disconnect).
The capture wrapper is unit-tested. No behavior change when the flag is off.

Closes #189

Signed-off-by: Dimitris Poulopoulos <dimitris.a.poulopoulos@gmail.com>
---
 src/gateway/api/routes/_pipeline.py        |  44 ++++++-
 src/gateway/api/routes/chat.py             |  75 ++++++++++-
 tests/integration/test_hybrid_mode_chat.py | 146 +++++++++++++++++++++
 tests/unit/test_stream_memory_capture.py   | 122 +++++++++++++++++
 4 files changed, 382 insertions(+), 5 deletions(-)
 create mode 100644 tests/unit/test_stream_memory_capture.py

diff --git a/src/gateway/api/routes/_pipeline.py b/src/gateway/api/routes/_pipeline.py
index aa154103..a7966a6e 100644
--- a/src/gateway/api/routes/_pipeline.py
+++ b/src/gateway/api/routes/_pipeline.py
@@ -31,7 +31,7 @@
 import asyncio
 import time
 import uuid
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine
 from contextlib import AsyncExitStack
 from datetime import UTC, datetime
 from enum import Enum, auto
@@ -1484,6 +1484,8 @@ async def run_streaming_with_fallback(
     rate_limit_info: RateLimitInfo | None,
     tool_ctx: ToolContext,
     session_label: str | None = None,
+    extract_stream_text: Callable[[ChunkT], str | None] | None = None,
+    on_memory_settled: Callable[[str], Coroutine[Any, Any, None]] | None = None,
 ) -> StreamingResponse:
     """Multi-attempt streaming for hybrid-mode requests.
 
@@ -1613,6 +1615,9 @@ async def _on_attempt_failed(attempt: ResolvedAttempt, failure: StreamingAttempt
     if pool_for_loop is not None:
         stream_to_return = _stream_with_stack_cleanup(stream, backend_stack)
 
+    if extract_stream_text is not None and on_memory_settled is not None:
+        stream_to_return = _stream_with_memory_capture(stream_to_return, extract_stream_text, on_memory_settled)
+
     return build_streaming_response(
         adapter=adapter,
         stream=stream_to_return,
@@ -1642,6 +1647,43 @@ async def _stream_with_stack_cleanup(
         await backend_stack.aclose()
 
 
+def _swallow_memory_task_exception(task: "asyncio.Task[None]") -> None:
+    """Done-callback that retrieves a detached memory-write task's exception so it
+    is logged rather than leaking as an unretrieved-task-exception warning."""
+    if task.cancelled():
+        return
+    exc = task.exception()
+    if exc is not None:
+        logger.warning("Background memory write failed: %s", exc)
+
+
+async def _stream_with_memory_capture(
+    stream: AsyncIterator[ChunkT],
+    extract_text: Callable[[ChunkT], str | None],
+    on_settled_text: Callable[[str], Coroutine[Any, Any, None]],
+) -> AsyncIterator[ChunkT]:
+    """Accumulate streamed assistant text and, on normal completion, fire a
+    best-effort memory write. The write is skipped on mid-stream cancellation
+    (client disconnect), and scheduled fire-and-forget so it never delays the
+    final bytes, mirroring the streaming usage-report pattern."""
+    parts: list[str] = []
+    completed = False
+    try:
+        async for chunk in stream:
+            piece = extract_text(chunk)
+            if piece:
+                parts.append(piece)
+            yield chunk
+        completed = True
+    finally:
+        if completed and parts:
+            task = asyncio.create_task(on_settled_text("".join(parts)))
+            # Retrieve the result so a raised memory write surfaces as a log line
+            # instead of an "Task exception was never retrieved" warning, keeping
+            # best-effort persistence silent on the response path.
+            task.add_done_callback(_swallow_memory_task_exception)
+
+
 async def run_platform_non_stream(
     *,
     adapter: FormatAdapter[ResultT, Any],
diff --git a/src/gateway/api/routes/chat.py b/src/gateway/api/routes/chat.py
index 9d3db726..2a933be2 100644
--- a/src/gateway/api/routes/chat.py
+++ b/src/gateway/api/routes/chat.py
@@ -1,6 +1,6 @@
 import asyncio
 import uuid
-from collections.abc import AsyncIterator, Callable
+from collections.abc import AsyncIterator, Callable, Coroutine
 from typing import Annotated, Any
 
 import httpx
@@ -17,7 +17,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from gateway.api.deps import get_config, get_db_if_needed, get_log_writer
-from gateway.api.routes._helpers import latest_user_text
+from gateway.api.routes._helpers import build_remember_messages, inject_memory_facts, latest_user_text
 from gateway.api.routes._normalize import normalize_request_messages
 from gateway.api.routes._pipeline import (
     ALL_PROVIDERS_FAILED_DETAIL,
@@ -39,7 +39,11 @@
     run_standalone_non_stream,
     run_streaming_with_fallback,
 )
-from gateway.api.routes._platform import ResolvedAttempt
+from gateway.api.routes._platform import (
+    ResolvedAttempt,
+    _recall_platform_memory,
+    _remember_platform_memory,
+)
 from gateway.api.routes._schema_derive import SESSION_LABEL_DESC, SESSION_LABEL_MAX_LENGTH, derive_request_base
 from gateway.api.routes._tools import _strip_gateway_fields
 from gateway.core.config import GatewayConfig
@@ -233,6 +237,53 @@ def prepare_platform_call_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]
 _USER_FORBIDDEN = "'user' field does not match the authenticated API key's user"
 
 
+def _schedule_memory_remember(
+    background_tasks: BackgroundTasks,
+    config: GatewayConfig,
+    user_token: str | None,
+    messages: list[dict[str, Any]],
+    completion: ChatCompletion,
+) -> None:
+    """Queue a best-effort, fire-and-forget write of the latest turn to memory.
+
+    A no-op when memory is off (``user_token`` is None) or there is nothing to store.
+    """
+    if not user_token:
+        return
+    try:
+        assistant_text = completion.choices[0].message.content or ""
+    except (AttributeError, IndexError, TypeError):
+        assistant_text = ""
+    remember_messages = build_remember_messages(messages, assistant_text)
+    if remember_messages:
+        background_tasks.add_task(_remember_platform_memory, config, user_token, remember_messages)
+
+
+def _chat_chunk_text(chunk: Any) -> str | None:
+    """Extract the assistant content delta from a streamed chat completion chunk."""
+    try:
+        return chunk.choices[0].delta.content  # type: ignore[no-any-return]
+    except (AttributeError, IndexError, TypeError):
+        return None
+
+
+def _streaming_memory_remember(
+    config: GatewayConfig,
+    user_token: str | None,
+    messages: list[dict[str, Any]],
+) -> Callable[[str], Coroutine[Any, Any, None]] | None:
+    """Build the on-stream-complete memory writer, or None when memory is off."""
+    if not user_token:
+        return None
+
+    async def _remember(assistant_text: str) -> None:
+        remember_messages = build_remember_messages(messages, assistant_text)
+        if remember_messages:
+            await _remember_platform_memory(config, user_token, remember_messages)
+
+    return _remember
+
+
 @router.post("/completions", response_model=None)
 async def chat_completions(
     raw_request: Request,
@@ -310,6 +361,18 @@ async def _normalize(
         tools_header=request.tools_header,
     )
 
+    # Memory recall (platform mode, best-effort). Inject recalled facts into the system
+    # message before dispatch so every downstream path (stream/non-stream, tool/no-tool)
+    # sees them. A failure or a disabled workspace yields no facts and never blocks chat.
+    memory_user_token = ctx.user_token if (config.memory_enabled and ctx.hybrid_mode) else None
+    if memory_user_token:
+        try:
+            recalled = await _recall_platform_memory(config, memory_user_token, latest_user_text(request.messages))
+            if recalled:
+                request.messages = inject_memory_facts(request.messages, recalled)
+        except Exception:  # noqa: BLE001 - recall is best-effort and must never break dispatch
+            logger.warning("Memory recall failed; proceeding without recalled facts", exc_info=True)
+
     request_fields = _strip_gateway_fields(
         request.model_dump(exclude_unset=True),
         tools_extracted=tool_ctx.tools_extracted,
@@ -346,6 +409,8 @@ async def _normalize(
                     rate_limit_info=ctx.rate_limit_info,
                     tool_ctx=tool_ctx,
                     session_label=request.session_label,
+                    extract_stream_text=_chat_chunk_text,
+                    on_memory_settled=_streaming_memory_remember(config, memory_user_token, request.messages),
                 )
             except HTTPException:
                 raise
@@ -414,7 +479,7 @@ async def _normalize(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail="Internal error: missing route context",
             )
-        return await run_platform_non_stream(
+        completion = await run_platform_non_stream(
             adapter=_ADAPTER,
             route=route,
             base_request_fields=request_fields,
@@ -425,6 +490,8 @@ async def _normalize(
             rate_limit_info=ctx.rate_limit_info,
             session_label=request.session_label,
         )
+        _schedule_memory_remember(background_tasks, config, memory_user_token, request.messages, completion)
+        return completion
 
     resolved = resolve_dispatch_provider(ctx, config, request.model)
     call_kwargs = {**resolved.kwargs, **request_fields, "model": resolved.dispatch_model}
diff --git a/tests/integration/test_hybrid_mode_chat.py b/tests/integration/test_hybrid_mode_chat.py
index a4cf4a7e..4b4be7b6 100644
--- a/tests/integration/test_hybrid_mode_chat.py
+++ b/tests/integration/test_hybrid_mode_chat.py
@@ -1622,3 +1622,149 @@ async def fake_mcp_tool_loop(**kwargs: Any) -> ChatCompletion:
 
     assert response.status_code == 200
     assert captured["max_iterations"] == 2
+
+
+def _memory_resolve_response() -> httpx.Response:
+    return httpx.Response(
+        200,
+        json={
+            "request_id": "7af2c39d-4eb8-4b3f-8242-46a97f7d5e68",
+            "fallback_enabled": False,
+            "attempts": [
+                {
+                    "attempt_id": "7af2c39d-4eb8-4b3f-8242-46a97f7d5e68",
+                    "position": 0,
+                    "provider": "openai",
+                    "model": "gpt-4o-mini",
+                    "api_key": "sk-platform-key",
+                    "api_base": "https://api.openai.com/v1",
+                    "managed": True,
+                }
+            ],
+        },
+    )
+
+
+def _memory_completion(content: str) -> ChatCompletion:
+    return ChatCompletion(
+        id="chatcmpl-platform",
+        object="chat.completion",
+        created=1700000000,
+        model="gpt-4o-mini",
+        choices=[
+            Choice(
+                index=0,
+                message=ChatCompletionMessage(role="assistant", content=content),
+                finish_reason="stop",
+            )
+        ],
+        usage=CompletionUsage(prompt_tokens=10, completion_tokens=7, total_tokens=17),
+    )
+
+
+def test_platform_mode_memory_disabled_makes_no_memory_calls(
+    platform_client: TestClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    memory_calls: list[str] = []
+
+    async def fake_post_platform(
+        url: str,
+        headers: dict[str, str],
+        body: dict[str, Any],
+        timeout_seconds: float,
+    ) -> httpx.Response:
+        if url.endswith("/gateway/provider-keys/resolve"):
+            return _memory_resolve_response()
+        if "/gateway/memory/" in url:
+            memory_calls.append(url)
+            return httpx.Response(200, json={"facts": []})
+        return httpx.Response(204)
+
+    async def fake_acompletion(**kwargs: Any) -> ChatCompletion:
+        return _memory_completion("hello")
+
+    monkeypatch.setattr("gateway.api.routes._platform._post_platform", fake_post_platform)
+    monkeypatch.setattr("gateway.api.routes.chat.acompletion", fake_acompletion)
+
+    response = platform_client.post(
+        "/v1/chat/completions",
+        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]},
+        headers={"Authorization": "Bearer user_test_token"},
+    )
+
+    assert response.status_code == 200
+    # memory_enabled defaults to False, so neither recall nor remember is called.
+    assert memory_calls == []
+
+
+def test_platform_mode_memory_recall_injects_and_remembers(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("OTARI_AI_TOKEN", "gw_test_token")
+    app = create_app(
+        GatewayConfig(
+            mode="platform",
+            platform={"base_url": "http://platform.test/api/v1"},
+            memory_enabled=True,
+        )
+    )
+
+    recall_bodies: list[dict[str, Any]] = []
+    remember_bodies: list[dict[str, Any]] = []
+    dispatched_messages: dict[str, Any] = {}
+
+    async def fake_post_platform(
+        url: str,
+        headers: dict[str, str],
+        body: dict[str, Any],
+        timeout_seconds: float,
+    ) -> httpx.Response:
+        if url.endswith("/gateway/provider-keys/resolve"):
+            return _memory_resolve_response()
+        if url.endswith("/gateway/memory/recall"):
+            recall_bodies.append(body)
+            return httpx.Response(200, json={"facts": ["the user prefers concise answers"]})
+        if url.endswith("/gateway/memory/remember"):
+            remember_bodies.append(body)
+            return httpx.Response(204)
+        return httpx.Response(204)
+
+    async def fake_acompletion(**kwargs: Any) -> ChatCompletion:
+        dispatched_messages["messages"] = kwargs.get("messages")
+        return _memory_completion("ok, noted")
+
+    monkeypatch.setattr("gateway.api.routes._platform._post_platform", fake_post_platform)
+    monkeypatch.setattr("gateway.api.routes.chat.acompletion", fake_acompletion)
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/v1/chat/completions",
+            json={
+                "model": "gpt-4o-mini",
+                "messages": [{"role": "user", "content": "remember I like short answers"}],
+            },
+            headers={"Authorization": "Bearer user_test_token"},
+        )
+
+    reset_config()
+    reset_db()
+
+    assert response.status_code == 200
+    # Recall is queried with the latest user text before dispatch.
+    assert recall_bodies == [{"query": "remember I like short answers"}]
+    # The recalled fact is injected as a system message the provider sees.
+    system_messages = [
+        m for m in (dispatched_messages["messages"] or []) if isinstance(m, dict) and m.get("role") == "system"
+    ]
+    assert system_messages
+    assert "the user prefers concise answers" in system_messages[0]["content"]
+    # After the completion the new turn is stored fire-and-forget.
+    assert remember_bodies == [
+        {
+            "messages": [
+                {"role": "user", "content": "remember I like short answers"},
+                {"role": "assistant", "content": "ok, noted"},
+            ]
+        }
+    ]
diff --git a/tests/unit/test_stream_memory_capture.py b/tests/unit/test_stream_memory_capture.py
new file mode 100644
index 00000000..4b2a6a3d
--- /dev/null
+++ b/tests/unit/test_stream_memory_capture.py
@@ -0,0 +1,122 @@
+"""Unit tests for streaming memory capture: the wrapper accumulates assistant text and
+fires a best-effort remember on normal completion (but not on mid-stream disconnect)."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncGenerator, AsyncIterator
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import MagicMock
+
+import pytest
+
+from gateway.api.routes import chat as chat_module
+from gateway.api.routes._pipeline import _stream_with_memory_capture
+from gateway.api.routes.chat import _chat_chunk_text, _streaming_memory_remember
+
+
+def _identity(chunk: Any) -> Any:
+    return chunk
+
+
+async def _gen(pieces: list[Any]) -> AsyncIterator[Any]:
+    for piece in pieces:
+        yield piece
+
+
+@pytest.mark.asyncio
+async def test_capture_remembers_accumulated_text_on_complete() -> None:
+    captured: list[str] = []
+
+    async def on_settled(text: str) -> None:
+        captured.append(text)
+
+    out = [chunk async for chunk in _stream_with_memory_capture(_gen(["Hel", "", "lo"]), _identity, on_settled)]
+    await asyncio.sleep(0)  # let the fire-and-forget task run
+
+    assert out == ["Hel", "", "lo"]  # chunks pass through unchanged
+    assert captured == ["Hello"]
+
+
+@pytest.mark.asyncio
+async def test_capture_skips_remember_on_disconnect() -> None:
+    captured: list[str] = []
+
+    async def on_settled(text: str) -> None:
+        captured.append(text)
+
+    gen = cast(
+        "AsyncGenerator[Any, None]",
+        _stream_with_memory_capture(_gen(["Hi", " there"]), _identity, on_settled),
+    )
+    assert await gen.__anext__() == "Hi"
+    await gen.aclose()  # client disconnect before the stream completes
+    await asyncio.sleep(0)
+
+    assert captured == []
+
+
+@pytest.mark.asyncio
+async def test_capture_swallows_settled_callback_error() -> None:
+    async def on_settled(text: str) -> None:
+        raise RuntimeError("memory backend exploded")
+
+    # A raised settled callback must not surface as an unretrieved-task-exception
+    # or otherwise disrupt the (already drained) stream.
+    out = [chunk async for chunk in _stream_with_memory_capture(_gen(["Hi"]), _identity, on_settled)]
+    await asyncio.sleep(0)  # let the detached task run and its done-callback fire
+
+    assert out == ["Hi"]
+
+
+@pytest.mark.asyncio
+async def test_capture_no_remember_when_no_text() -> None:
+    captured: list[str] = []
+
+    async def on_settled(text: str) -> None:
+        captured.append(text)
+
+    async for _ in _stream_with_memory_capture(_gen([None, None]), _identity, on_settled):
+        pass
+    await asyncio.sleep(0)
+
+    assert captured == []
+
+
+def _chunk(content: str | None) -> Any:
+    return SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace(content=content))])
+
+
+def test_chat_chunk_text_extracts_delta_content() -> None:
+    assert _chat_chunk_text(_chunk("hi")) == "hi"
+    assert _chat_chunk_text(_chunk(None)) is None
+    assert _chat_chunk_text(SimpleNamespace(choices=[])) is None  # no choices
+
+
+def test_streaming_remember_none_when_no_token() -> None:
+    assert _streaming_memory_remember(MagicMock(), None, []) is None
+
+
+@pytest.mark.asyncio
+async def test_streaming_remember_closure_calls_platform(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[tuple[str, list[dict[str, str]]]] = []
+
+    async def fake_remember(config: Any, token: str, messages: list[dict[str, str]]) -> None:
+        calls.append((token, messages))
+
+    monkeypatch.setattr(chat_module, "_remember_platform_memory", fake_remember)
+
+    callback = _streaming_memory_remember(MagicMock(), "tk", [{"role": "user", "content": "My name is X"}])
+    assert callback is not None
+    await callback("Hello X")
+
+    assert calls == [
+        (
+            "tk",
+            [
+                {"role": "user", "content": "My name is X"},
+                {"role": "assistant", "content": "Hello X"},
+            ],
+        )
+    ]

From 2c80c54e7d9e874c408b8e0994e178d8400386cc Mon Sep 17 00:00:00 2001
From: Dimitris Poulopoulos <dimitris@mozilla.ai>
Date: Wed, 15 Jul 2026 13:01:01 +0300
Subject: [PATCH 4/4] feat(memory): follow the platform for memory enablement

Read the memory_enabled flag off the resolve response and gate
recall/remember on it, so the platform is authoritative for
per-workspace enablement. The gateway's own switch becomes a self-hosted
opt-out that defaults on, removing the silent-off footgun where memory
was enabled on the platform but the gateway never called it. The flag
defaults to true when a platform omits it, so recall never silently
stops. Also raise the default recall timeout from 2000ms to 8000ms.

Signed-off-by: Dimitris Poulopoulos <dimitris@mozilla.ai>
---
 docs/hybrid-mode-protocol.md               | 12 +++-
 src/gateway/api/routes/_platform.py        | 10 ++-
 src/gateway/api/routes/chat.py             | 10 ++-
 src/gateway/core/config.py                 | 13 ++--
 tests/integration/test_hybrid_mode_chat.py | 58 +++++++++++++++++-
 tests/unit/test_memory_platform.py         |  2 +-
 tests/unit/test_resolve_payload_parsing.py | 71 ++++++++++++++++++++++
 7 files changed, 163 insertions(+), 13 deletions(-)
 create mode 100644 tests/unit/test_resolve_payload_parsing.py

diff --git a/docs/hybrid-mode-protocol.md b/docs/hybrid-mode-protocol.md
index b8f7fc03..e96326f5 100644
--- a/docs/hybrid-mode-protocol.md
+++ b/docs/hybrid-mode-protocol.md
@@ -90,7 +90,8 @@ Content-Type: application/json
       "api_base": "https://api.openai.com/v1",
       "managed": false
     }
-  ]
+  ],
+  "memory_enabled": false          // optional; whether the workspace has memory enabled
 }
 ```
 
@@ -111,6 +112,15 @@ own behaviour.
 `attempts` MUST contain at least one entry. An empty list is treated as a
 platform bug and surfaced as `502 Bad Gateway`.
 
+`memory_enabled` tells Otari whether the workspace has persistent memory turned
+on. The platform is authoritative, so Otari makes the hot-path recall call (and
+the fire-and-forget remember call) only when this is `true`, avoiding a wasted
+round-trip for workspaces that do not use memory. It is optional: when a platform
+omits it, Otari defaults to `true` and relies on the memory endpoints' own
+empty-when-disabled behaviour, so no gateway ever silently stops recalling. A
+gateway operator can still hard-disable all memory calls locally with
+`OTARI_MEMORY_ENABLED=false` regardless of this flag.
+
 ### Response — single-attempt shape
 
 Otari also accepts a flat payload:
diff --git a/src/gateway/api/routes/_platform.py b/src/gateway/api/routes/_platform.py
index 86ea67ec..de7baacc 100644
--- a/src/gateway/api/routes/_platform.py
+++ b/src/gateway/api/routes/_platform.py
@@ -97,6 +97,11 @@ class ResolvedRoute(BaseModel):
     request_id: str
     fallback_enabled: bool
     attempts: list[ResolvedAttempt]
+    # Whether the workspace has persistent memory enabled. The platform is authoritative;
+    # the gateway uses this to decide whether to make the hot-path recall call at all.
+    # Defaults to True so a gateway pointed at an older platform (which omits the field)
+    # still attempts recall, relying on the platform's empty-when-off short-circuit.
+    memory_enabled: bool = True
 
 
 class _AttemptFailure(NamedTuple):
@@ -439,6 +444,8 @@ def _parse_resolve_payload(payload: dict[str, Any]) -> ResolvedRoute:
             request_id=str(payload["request_id"]),
             fallback_enabled=bool(payload.get("fallback_enabled", False)),
             attempts=attempts,
+            # Absent on older platforms; default True so recall still runs (see ResolvedRoute).
+            memory_enabled=bool(payload.get("memory_enabled", True)),
         )
 
     correlation_id = str(payload["correlation_id"])
@@ -456,6 +463,7 @@ def _parse_resolve_payload(payload: dict[str, Any]) -> ResolvedRoute:
                 managed=bool(payload.get("managed", False)),
             )
         ],
+        memory_enabled=bool(payload.get("memory_enabled", True)),
     )
 
 
@@ -702,7 +710,7 @@ async def _recall_platform_memory(
     if not platform_base_url or not query.strip():
         return []
 
-    timeout_ms = _coerce_timeout_ms(config.platform.get("memory_recall_timeout_ms", 2000), 2000)
+    timeout_ms = _coerce_timeout_ms(config.platform.get("memory_recall_timeout_ms", 8000), 8000)
     url = _platform_url(platform_base_url, "/gateway/memory/recall")
     headers = {
         "X-Gateway-Token": config.platform_token or "",
diff --git a/src/gateway/api/routes/chat.py b/src/gateway/api/routes/chat.py
index 2a933be2..6996df27 100644
--- a/src/gateway/api/routes/chat.py
+++ b/src/gateway/api/routes/chat.py
@@ -364,7 +364,15 @@ async def _normalize(
     # Memory recall (platform mode, best-effort). Inject recalled facts into the system
     # message before dispatch so every downstream path (stream/non-stream, tool/no-tool)
     # sees them. A failure or a disabled workspace yields no facts and never blocks chat.
-    memory_user_token = ctx.user_token if (config.memory_enabled and ctx.hybrid_mode) else None
+    #
+    # Enablement is the AND of: this gateway not opting out locally (config.memory_enabled,
+    # default on), the platform reporting the workspace has memory on (route.memory_enabled),
+    # and hybrid mode. The platform flag is authoritative, so a workspace with memory off
+    # costs no recall round-trip; the local flag is only a self-hosted kill switch.
+    platform_memory_enabled = ctx.route is not None and ctx.route.memory_enabled
+    memory_user_token = (
+        ctx.user_token if (config.memory_enabled and platform_memory_enabled and ctx.hybrid_mode) else None
+    )
     if memory_user_token:
         try:
             recalled = await _recall_platform_memory(config, memory_user_token, latest_user_text(request.messages))
diff --git a/src/gateway/core/config.py b/src/gateway/core/config.py
index 6e10021d..bcbe7f77 100644
--- a/src/gateway/core/config.py
+++ b/src/gateway/core/config.py
@@ -228,15 +228,16 @@ class GatewayConfig(BaseSettings):
         description="Enable the /v1/files upload/storage endpoints (standalone mode).",
     )
     memory_enabled: bool = Field(
-        default=False,
+        default=True,
         description=(
-            "Master switch for persistent memory (platform mode). When on, the gateway "
-            "recalls facts from the platform before a completion and stores facts after. "
-            "Per-workspace enablement is still controlled by the platform; this only gates "
-            "whether the gateway makes the calls at all. Recall runs before dispatch, so a "
+            "Self-hosted opt-out for persistent memory (platform mode). The platform is "
+            "authoritative for per-workspace enablement (carried on the resolve response), "
+            "so leave this on (the default) to follow the platform: memory is used only for "
+            "workspaces that enabled it. Set it false to hard-disable all memory calls from "
+            "this gateway regardless of platform config. Recall runs before dispatch, so a "
             "slow memory service adds up to its timeout to time-to-first-token; the storing "
             "call after a completion is fire-and-forget. Both timeouts are tunable via the "
-            "platform settings 'memory_recall_timeout_ms' (default 2000) and "
+            "platform settings 'memory_recall_timeout_ms' (default 8000) and "
             "'memory_remember_timeout_ms' (default 10000), or their PLATFORM_* env aliases."
         ),
     )
diff --git a/tests/integration/test_hybrid_mode_chat.py b/tests/integration/test_hybrid_mode_chat.py
index 4b4be7b6..df0ce981 100644
--- a/tests/integration/test_hybrid_mode_chat.py
+++ b/tests/integration/test_hybrid_mode_chat.py
@@ -1624,7 +1624,7 @@ async def fake_mcp_tool_loop(**kwargs: Any) -> ChatCompletion:
     assert captured["max_iterations"] == 2
 
 
-def _memory_resolve_response() -> httpx.Response:
+def _memory_resolve_response(*, memory_enabled: bool = True) -> httpx.Response:
     return httpx.Response(
         200,
         json={
@@ -1641,6 +1641,8 @@ def _memory_resolve_response() -> httpx.Response:
                     "managed": True,
                 }
             ],
+            # Platform-authoritative: the gateway makes memory calls only when this is true.
+            "memory_enabled": memory_enabled,
         },
     )
 
@@ -1675,7 +1677,7 @@ async def fake_post_platform(
         timeout_seconds: float,
     ) -> httpx.Response:
         if url.endswith("/gateway/provider-keys/resolve"):
-            return _memory_resolve_response()
+            return _memory_resolve_response(memory_enabled=False)
         if "/gateway/memory/" in url:
             memory_calls.append(url)
             return httpx.Response(200, json={"facts": []})
@@ -1694,7 +1696,57 @@ async def fake_acompletion(**kwargs: Any) -> ChatCompletion:
     )
 
     assert response.status_code == 200
-    # memory_enabled defaults to False, so neither recall nor remember is called.
+    # The platform reports the workspace has memory off, so neither recall nor remember runs
+    # even though the gateway's local opt-out is on (default).
+    assert memory_calls == []
+
+
+def test_platform_mode_local_opt_out_suppresses_memory_calls(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # A self-hosted gateway can hard-disable memory locally (memory_enabled=False) even when
+    # the platform reports the workspace has memory on.
+    monkeypatch.setenv("OTARI_AI_TOKEN", "gw_test_token")
+    app = create_app(
+        GatewayConfig(
+            mode="platform",
+            platform={"base_url": "http://platform.test/api/v1"},
+            memory_enabled=False,
+        )
+    )
+
+    memory_calls: list[str] = []
+
+    async def fake_post_platform(
+        url: str,
+        headers: dict[str, str],
+        body: dict[str, Any],
+        timeout_seconds: float,
+    ) -> httpx.Response:
+        if url.endswith("/gateway/provider-keys/resolve"):
+            return _memory_resolve_response(memory_enabled=True)
+        if "/gateway/memory/" in url:
+            memory_calls.append(url)
+            return httpx.Response(200, json={"facts": []})
+        return httpx.Response(204)
+
+    async def fake_acompletion(**kwargs: Any) -> ChatCompletion:
+        return _memory_completion("hello")
+
+    monkeypatch.setattr("gateway.api.routes._platform._post_platform", fake_post_platform)
+    monkeypatch.setattr("gateway.api.routes.chat.acompletion", fake_acompletion)
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/v1/chat/completions",
+            json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]},
+            headers={"Authorization": "Bearer user_test_token"},
+        )
+
+    reset_config()
+    reset_db()
+
+    assert response.status_code == 200
     assert memory_calls == []
 
 
diff --git a/tests/unit/test_memory_platform.py b/tests/unit/test_memory_platform.py
index e2637bb7..f58ad5a8 100644
--- a/tests/unit/test_memory_platform.py
+++ b/tests/unit/test_memory_platform.py
@@ -102,7 +102,7 @@ async def fake_post(*, timeout_seconds: float, **kwargs: Any) -> httpx.Response:
     monkeypatch.setattr(platform_module, "_post_platform", fake_post)
     out = await _recall_platform_memory(_config(memory_recall_timeout_ms=bad_timeout), "tk", "q")
     assert out == ["ok"]
-    assert captured["timeout_seconds"] == 2.0
+    assert captured["timeout_seconds"] == 8.0
 
 
 # ───────────── remember ─────────────
diff --git a/tests/unit/test_resolve_payload_parsing.py b/tests/unit/test_resolve_payload_parsing.py
new file mode 100644
index 00000000..1f5cceac
--- /dev/null
+++ b/tests/unit/test_resolve_payload_parsing.py
@@ -0,0 +1,71 @@
+"""Unit tests for `_parse_resolve_payload`, focused on the `memory_enabled` flag.
+
+The platform is authoritative for per-workspace memory enablement and carries the
+signal on the resolve response. The gateway must read it, and must default to
+`True` when an older platform omits it so recall never silently stops.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from gateway.api.routes._platform import _parse_resolve_payload
+
+
+def _multi_attempt(**extra: Any) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "request_id": "01HXY",
+        "fallback_enabled": False,
+        "attempts": [
+            {
+                "attempt_id": "01HX1",
+                "position": 0,
+                "provider": "openai",
+                "model": "gpt-4o",
+                "api_key": "sk-test",
+                "api_base": None,
+                "managed": False,
+            }
+        ],
+    }
+    payload.update(extra)
+    return payload
+
+
+def _single_attempt(**extra: Any) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "provider": "openai",
+        "model": "gpt-4o",
+        "api_key": "sk-test",
+        "api_base": None,
+        "managed": False,
+        "correlation_id": "01HXC",
+    }
+    payload.update(extra)
+    return payload
+
+
+@pytest.mark.parametrize("value", [True, False])
+def test_multi_attempt_reads_memory_enabled(value: bool) -> None:
+    route = _parse_resolve_payload(_multi_attempt(memory_enabled=value))
+    assert route.memory_enabled is value
+
+
+def test_multi_attempt_defaults_memory_enabled_true_when_absent() -> None:
+    # Older platform omits the field: default to True so recall still runs and relies
+    # on the memory endpoints' own empty-when-disabled behaviour (never silently off).
+    route = _parse_resolve_payload(_multi_attempt())
+    assert route.memory_enabled is True
+
+
+@pytest.mark.parametrize("value", [True, False])
+def test_single_attempt_reads_memory_enabled(value: bool) -> None:
+    route = _parse_resolve_payload(_single_attempt(memory_enabled=value))
+    assert route.memory_enabled is value
+
+
+def test_single_attempt_defaults_memory_enabled_true_when_absent() -> None:
+    route = _parse_resolve_payload(_single_attempt())
+    assert route.memory_enabled is True