修复 Add/Search 短暂缺表失败，增加默认 5 秒的可配置恢复窗口

agent_memory_leaderboard/README.md

@@ -164,10 +164,33 @@ the `idempotent_v1` protocol in SQLite **before** sending HTTP. On a later Add
 retry, an attempt without an acknowledged memory ID can safely resend the same
 key and body: the backend returns the existing receipt or creates the record
 once. This recovers both pre-commit server failures and a crash after journaling
-but before sending. The adapter still makes one write attempt per incoming Add
-attempt; it does not add an internal retry loop. Once the memory ID is known,
+but before sending. Ordinary errors still end the current Add attempt; only the
+specific catalog-visibility errors described below are retried internally.
+Once the memory ID is known,
 retries use point reads and retrieval probes instead of reposting.
 
+When Memoria returns HTTP 500 with a recognized MatrixOne missing `mem_*` table
+error (1064/42000 with the missing-table parser message, or 1146/42S02 with
+`no such table`), the adapter uses a bounded recovery window configured by
+`MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS` (default 5 seconds, finite in `[0, 120]`;
+0 disables retries). Helm exposes the same value as `catalogRetryTimeoutSeconds`.
+The window starts on the first recognized error and is shared by every backend
+call and chunk in one Add/Search request; it never resets per retry or chunk.
+Backoff starts at 0.5 seconds and is capped at 1 second and the remaining window.
+Retry admission waits and HTTP I/O limits are capped by that same deadline.
+As with the request budget, synchronous I/O may unwind after expiry; a late
+success is rejected, and the window is not a hard cancellation timer.
+This covers atomic keyed writes, point reads,
+recovery queries and Search; a write retry preserves its exact payload and
+idempotency key. Backoff releases the backend slot and consumes the existing
+request budget and, during verification, the original visibility budget. Other
+server errors, SQL syntax errors, authorization/validation failures and
+non-atomic writes do not enter this retry loop. Exhaustion keeps the existing
+error mapping and ambiguous-write journal state; it never claims the write was
+not committed. Retry warnings contain only attempt numbers and delay durations.
+This is bounded recovery for transient visibility, not a guarantee that a
+permanently missing schema will recover or a replacement for backend consistency.
+
 Atomic writes, all point reads/recovery queries/visibility probes, and public
 Search explicitly use branch `main`, keeping the receipt scope independent of
 a backend checkout.

agent_memory_leaderboard/src/budget.py

@@ -7,12 +7,40 @@
 import time
 from contextlib import contextmanager
 from contextvars import ContextVar
+from dataclasses import dataclass
 
 from .errors import AdapterError
 
 DEADLINE: ContextVar[float | None] = ContextVar("adapter_request_deadline", default=None)
 
 
+@dataclass
+class CatalogRetryWindow:
+    seconds: float
+    deadline: float | None = None
+
+    def start(self) -> float:
+        if self.deadline is None:
+            self.deadline = time.monotonic() + self.seconds
+        return self.deadline
+
+
+CATALOG_RETRY_WINDOW: ContextVar[CatalogRetryWindow | None] = ContextVar(
+    "adapter_catalog_retry_window", default=None
+)
+
+
+@contextmanager
+def catalog_retry_budget(seconds: float):
+    """One lazy window shared by all backend calls/chunks in this request."""
+    parent = CATALOG_RETRY_WINDOW.get()
+    token = CATALOG_RETRY_WINDOW.set(parent if parent is not None else CatalogRetryWindow(seconds))
+    try:
+        yield
+    finally:
+        CATALOG_RETRY_WINDOW.reset(token)
+
+
 class RequestBudgetExceeded(AdapterError):
     def __init__(self):
         super().__init__(

agent_memory_leaderboard/src/config.py

@@ -16,6 +16,7 @@ class Settings:
     connect_timeout_seconds: float = field(default=30, kw_only=True)
     request_timeout_seconds: float = field(default=1700, kw_only=True)
     visibility_timeout_seconds: float = field(default=60, kw_only=True)
+    catalog_retry_timeout_seconds: float = field(default=5, kw_only=True)
     backend_max_concurrency: int = 0
     backend_requests_per_second: float = 0
     backend_queue_timeout_seconds: float = 120
@@ -58,6 +59,11 @@ def __post_init__(self) -> None:
             raise ValueError("ADAPTER_JOURNAL_LOG_ALL_OPERATIONS must be a boolean")
         if not math.isfinite(self.journal_slow_ms) or self.journal_slow_ms < 0:
             raise ValueError("ADAPTER_JOURNAL_SLOW_MS must be nonnegative and finite")
+        if (
+            not math.isfinite(self.catalog_retry_timeout_seconds)
+            or not 0 <= self.catalog_retry_timeout_seconds <= 120
+        ):
+            raise ValueError("MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS must be in [0, 120]")
         if (
             not math.isfinite(self.visibility_timeout_seconds)
             or not 0 < self.visibility_timeout_seconds <= 120
@@ -89,6 +95,9 @@ def from_env(cls) -> "Settings":
             connect_timeout_seconds=float(os.getenv("MEMORIA_CONNECT_TIMEOUT_SECONDS", "30")),
             request_timeout_seconds=float(os.getenv("ADAPTER_REQUEST_TIMEOUT_SECONDS", "1700")),
             visibility_timeout_seconds=float(os.getenv("MEMORIA_VISIBILITY_TIMEOUT_SECONDS", "60")),
+            catalog_retry_timeout_seconds=float(
+                os.getenv("MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS", "5")
+            ),
             backend_max_concurrency=int(os.getenv("MEMORIA_MAX_CONCURRENCY", "0")),
             backend_requests_per_second=float(os.getenv("MEMORIA_REQUESTS_PER_SECOND", "0")),
             backend_queue_timeout_seconds=float(os.getenv("MEMORIA_QUEUE_TIMEOUT_SECONDS", "120")),

agent_memory_leaderboard/src/errors.py

@@ -17,3 +17,7 @@ def __init__(
 
 class VisibilityPending(AdapterError):
     """A valid empty read/retrieval result; other 503s must not enter this retry loop."""
+
+
+class CatalogVisibilityPending(AdapterError):
+    """Recognized upstream missing-table error; a write may already have committed."""

agent_memory_leaderboard/src/memoria.py

@@ -1,21 +1,32 @@
 """HTTP boundary: use an owner-scoped master credential, never caller headers."""
 
+import logging
 import math
+import re
 import time
 from typing import Annotated, Any
 from urllib.parse import quote
 
 import httpx
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
 
+from .budget import CATALOG_RETRY_WINDOW, CatalogRetryWindow
 from .budget import remaining as budget_remaining
 from .config import Settings
-from .errors import AdapterError, VisibilityPending
+from .errors import AdapterError, CatalogVisibilityPending, VisibilityPending
 from .models import Nonblank
 from .scheduling import BackendScheduler
 
 IDEMPOTENT_PATH = "/v1/memories/idempotent"
 ADAPTER_BRANCH = "main"
+_TABLE_IDENTIFIER = r'''["'`]*(?:[\w-]+["'`]*\.)?["'`]*mem_[a-z0-9_]+["'`]*'''
+_MISSING_TABLE = re.compile(
+    rf"\b1064\s*\(42000\):\s*SQL parser error:\s*table\s+{_TABLE_IDENTIFIER}"
+    rf"\s+does not exist\b|\b1146\s*\(42S02\):\s*no such table\s+"
+    rf"{_TABLE_IDENTIFIER}(?=\s|[,;:)]|$)",
+    re.IGNORECASE,
+)
+logger = logging.getLogger(__name__)
 
 
 class Memory(BaseModel):
@@ -58,14 +69,88 @@ def request(
         body: dict | None = None,
         *,
         deadline: float | None = None,
+    ) -> Any:
+        # Replay only reads and the atomic write route, preserving the exact
+        # payload/key. A failed response does not prove a write was uncommitted.
+        safe_replay = method == "GET" or (
+            method == "POST"
+            and (
+                path in {"/v1/memories/query", "/v1/memories/search"}
+                or (path == IDEMPOTENT_PATH and body and body.get("idempotency_key"))
+            )
+        )
+        window = CATALOG_RETRY_WINDOW.get()
+        if window is None:
+            window = CatalogRetryWindow(self.settings.catalog_retry_timeout_seconds)
+        retry_deadline = None
+        attempt = 0
+        while True:
+            try:
+                result = self._request_once(
+                    method, path, owner, body, deadline=deadline, retry_deadline=retry_deadline
+                )
+                if retry_deadline is not None and time.monotonic() >= retry_deadline:
+                    budget_remaining(float("inf"))
+                    if deadline is not None:
+                        self._remaining_visibility(deadline)
+                    raise CatalogVisibilityPending(502, "Memoria catalog retry window expired")
+                return result
+            except CatalogVisibilityPending:
+                if not safe_replay:
+                    raise
+                retry_deadline = window.start()
+                available = budget_remaining(retry_deadline - time.monotonic())
+                if deadline is not None:
+                    available = min(available, self._remaining_visibility(deadline))
+                if available <= 0:
+                    raise
+                delay = min(0.5 if attempt == 0 else 1.0, available)
+                # No upstream body, URL, owner, credentials or payload in logs.
+                logger.warning(
+                    "Memoria catalog visibility retry wait: attempt=%d delay_ms=%.0f",
+                    attempt + 1,
+                    delay * 1000,
+                )
+                # _request_once released the backend slot before this wait.
+                time.sleep(delay)
+                attempt += 1
+                budget_remaining(float("inf"))
+                if deadline is not None:
+                    self._remaining_visibility(deadline)
+                if time.monotonic() >= retry_deadline:
+                    raise
+            except AdapterError as exc:
+                # Earlier attempts were ambiguous even if this retry never sent.
+                if retry_deadline is not None:
+                    exc.write_not_sent = False
+                raise
+
+    def _request_once(
+        self,
+        method: str,
+        path: str,
+        owner: str,
+        body: dict | None = None,
+        *,
+        deadline: float | None = None,
+        retry_deadline: float | None = None,
     ) -> Any:
         budget_remaining(float("inf"))
         if deadline is not None:
             self._remaining_visibility(deadline)
         visibility_limited = False
         try:
-            with self.scheduler.slot(deadline=deadline):
+            admission_deadline = min(
+                deadline if deadline is not None else math.inf,
+                retry_deadline if retry_deadline is not None else math.inf,
+            )
+            with self.scheduler.slot(deadline=admission_deadline):
                 timeout = budget_remaining(self.settings.timeout_seconds)
+                if retry_deadline is not None:
+                    retry_remaining = retry_deadline - time.monotonic()
+                    if retry_remaining <= 0:
+                        raise CatalogVisibilityPending(502, "Memoria catalog retry window expired")
+                    timeout = min(timeout, retry_remaining)
                 if deadline is not None:
                     remaining = self._remaining_visibility(deadline)
                     visibility_limited = remaining <= timeout
@@ -126,7 +211,12 @@ def request(
             # conflicts and server errors remain ambiguous and require recovery.
             rejected = code in {400, 401, 403, 404, 405, 413, 415, 422, 425, 429}
             retry_after = response.headers.get("Retry-After", "")
-            raise AdapterError(
+            error_class = (
+                CatalogVisibilityPending
+                if code == 500 and _MISSING_TABLE.search(response.text[:16384])
+                else AdapterError
+            )
+            raise error_class(
                 status,
                 reason,
                 write_not_sent=rejected,

agent_memory_leaderboard/src/service.py

@@ -2,8 +2,8 @@
 import json
 from itertools import chain, pairwise
 
+from .budget import catalog_retry_budget, request_budget
 from .budget import remaining as budget_remaining
-from .budget import request_budget
 from .config import Settings
 from .errors import AdapterError
 from .ingestion import message_parts
@@ -30,7 +30,11 @@ def owner(self, user_id: str) -> str:
         return digest([self.settings.namespace, user_id])
 
     def add(self, request: AddRequest) -> AddResponse:
-        with request_budget(self.settings.request_timeout_seconds), self.journal.metrics.add():
+        with (
+            request_budget(self.settings.request_timeout_seconds),
+            catalog_retry_budget(self.settings.catalog_retry_timeout_seconds),
+            self.journal.metrics.add(),
+        ):
             owner = self.owner(request.user_id)
             key = digest([self.settings.memoria_url.rstrip("/"), owner, request.request_id])
             fingerprint = digest(request.model_dump())
@@ -125,7 +129,10 @@ def write(
         )
 
     def search(self, request: SearchRequest) -> SearchResponse:
-        with request_budget(self.settings.request_timeout_seconds):
+        with (
+            request_budget(self.settings.request_timeout_seconds),
+            catalog_retry_budget(self.settings.catalog_retry_timeout_seconds),
+        ):
             result = self._search(request)
             budget_remaining(float("inf"))
             return result