feat(memory): recall and store memory around chat completions

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:

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,

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],

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)),
     )
 
 
@@ -675,6 +683,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", 8000), 8000)
+    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,

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,26 @@ 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.
+    #
+    # 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))
+            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 +417,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 +487,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 +498,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}