From e8e38eb7eb8be48d310cb33f7745a42fe5d47af0 Mon Sep 17 00:00:00 2001
From: krishna3554 <krishlokhande45@gmail.com>
Date: Sat, 29 Aug 2026 20:46:55 +0530
Subject: [PATCH 1/6] add

---
 services/dma-api/src/dma_api/repository.py | 152 ++++++++++++++++-----
 1 file changed, 116 insertions(+), 36 deletions(-)

diff --git a/services/dma-api/src/dma_api/repository.py b/services/dma-api/src/dma_api/repository.py
index 48c27d1..1412f3a 100644
--- a/services/dma-api/src/dma_api/repository.py
+++ b/services/dma-api/src/dma_api/repository.py
@@ -5,11 +5,13 @@
 import json
 import re
 import sqlite3
+from abc import ABC, abstractmethod
 from base64 import urlsafe_b64decode, urlsafe_b64encode
 from collections.abc import Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime
+from enum import Enum
 from pathlib import Path
 
 from dma_api.models import MemoryType
@@ -49,6 +51,10 @@
     "why",
     "with",
 }
+
+# Default query expansions (opt-in domain pack for DMA-specific corpus).
+# These are intentionally NOT used by the default "plain" analyzer to avoid
+# benchmark overfitting. Enable via AnalyzerKind.DOMAIN or config.
 _QUERY_EXPANSIONS = {
     "adapter": {"adapter", "langgraph", "mcp"},
     "api": {"api", "apis", "openapi", "rest"},
@@ -81,6 +87,89 @@
 }
 
 
+class AnalyzerKind(str, Enum):
+    """Analyzer variants for tokenization and query expansion.
+
+    - PLAIN: No query expansion, exact token matching with prefix fallback.
+             This is the default to avoid corpus overfitting.
+    - DOMAIN: Opt-in domain pack with DMA-specific query expansions.
+    """
+
+    PLAIN = "plain"
+    DOMAIN = "domain"
+
+
+class Analyzer(ABC):
+    """Pluggable analyzer for tokenization and query expansion.
+
+    Implementations control how queries and content are tokenized and whether
+    query expansion is applied. The default PLAIN analyzer performs no
+    query expansion and uses prefix-only token matching to prevent false
+    positives from bidirectional substring matching.
+    """
+
+    @abstractmethod
+    def expand_tokens(self, tokens: set[str]) -> set[str]:
+        """Expand a set of tokens (e.g., with synonyms). Default: no-op."""
+
+    @abstractmethod
+    def tokens_match(self, query_token: str, content_token: str) -> bool:
+        """Determine if a query token matches a content token.
+
+        Default implementation uses prefix matching only to avoid
+        false positives from bidirectional substring matching (e.g., "cat"
+        matching "category", "api" matching "rapid").
+        """
+
+
+class PlainAnalyzer(Analyzer):
+    """Default analyzer: no query expansion, prefix-only token matching."""
+
+    def expand_tokens(self, tokens: set[str]) -> set[str]:
+        return tokens
+
+    def tokens_match(self, query_token: str, content_token: str) -> bool:
+        # Prefix-only matching: query token must be a prefix of content token
+        # when lengths differ. This prevents false positives like "cat" ⊂
+        # "category", "api" ⊂ "rapid", "art" ⊂ "particle".
+        if query_token == content_token:
+            return True
+        if len(query_token) >= 3 and content_token.startswith(query_token):
+            return True
+        return False
+
+
+class DomainAnalyzer(Analyzer):
+    """Opt-in domain analyzer with DMA-specific query expansions.
+
+    Uses the _QUERY_EXPANSIONS mapping for query expansion. Still uses
+    prefix-only token matching to prevent false positives.
+    """
+
+    def expand_tokens(self, tokens: set[str]) -> set[str]:
+        expanded = set(tokens)
+        for token in tokens:
+            expanded.update(_QUERY_EXPANSIONS.get(token, set()))
+        return expanded
+
+    def tokens_match(self, query_token: str, content_token: str) -> bool:
+        # Same prefix-only matching as PlainAnalyzer
+        if query_token == content_token:
+            return True
+        if len(query_token) >= 3 and content_token.startswith(query_token):
+            return True
+        return False
+
+
+def get_analyzer(kind: AnalyzerKind) -> Analyzer:
+    """Factory function to get an analyzer by kind."""
+    if kind == AnalyzerKind.PLAIN:
+        return PlainAnalyzer()
+    if kind == AnalyzerKind.DOMAIN:
+        return DomainAnalyzer()
+    raise ValueError(f"Unknown analyzer kind: {kind}")
+
+
 @dataclass(frozen=True, slots=True)
 class MemoryRecord:
     id: str
@@ -98,8 +187,9 @@ class MemoryRecord:
 class SQLiteMemoryRepository:
     """A small persistence boundary that can later be replaced by PostgreSQL."""
 
-    def __init__(self, database_path: Path) -> None:
+    def __init__(self, database_path: Path, analyzer: Analyzer | None = None) -> None:
         self._database_path = database_path
+        self._analyzer = analyzer or PlainAnalyzer()
 
     def initialize(self) -> None:
         self._database_path.parent.mkdir(parents=True, exist_ok=True)
@@ -385,7 +475,8 @@ def _to_fts_query(query: str) -> str:
         a harmless wording variation (for example, ``prefer`` vs ``prefers``) from
         producing an empty result set before semantic retrieval is introduced.
         """
-        tokens = SQLiteMemoryRepository._expanded_tokens(query)
+        # This is a static method - can't use instance analyzer. Uses plain tokens.
+        tokens = SQLiteMemoryRepository._expanded_tokens_static(query)
         terms = []
         for token in tokens:
             terms.append(f'"{token}"')
@@ -393,57 +484,54 @@ def _to_fts_query(query: str) -> str:
                 terms.append(f"{token}*")
         return " OR ".join(terms)
 
-    @classmethod
+    @staticmethod
+    def _expanded_tokens_static(text: str) -> set[str]:
+        """Static version for FTS query building (uses plain tokens)."""
+        return {
+            token
+            for token in (SQLiteMemoryRepository._normalise_token(raw_token) for raw_token in re.findall(r"[\w]+", text, flags=re.UNICODE))
+            if token and len(token) > 2
+        }
+
     def _passes_precision_filter(
-        cls, query: str, query_tokens: set[str], content: str, *, enforce_current_filter: bool = True
+        self, query: str, query_tokens: set[str], content: str, *, enforce_current_filter: bool = True
     ) -> bool:
-        if enforce_current_filter and cls._asks_for_current(query) and not cls._has_current_marker(content):
+        if enforce_current_filter and self._asks_for_current(query) and not self._has_current_marker(content):
             return False
         if not query_tokens:
             return False
-        overlap = cls._matching_tokens(query_tokens, content)
+        overlap = self._matching_tokens(query_tokens, content)
         if len(query_tokens) == 1:
             return len(overlap) == 1
         return len(overlap) >= 2 or len(overlap) / len(query_tokens) >= 0.5
 
-    @classmethod
-    def _overlap_score(cls, query_tokens: set[str], content: str) -> float:
+    def _overlap_score(self, query_tokens: set[str], content: str) -> float:
         if not query_tokens:
             return 0.0
-        return len(cls._matching_tokens(query_tokens, content)) / len(query_tokens)
+        return len(self._matching_tokens(query_tokens, content)) / len(query_tokens)
 
-    @classmethod
-    def _matching_tokens(cls, query_tokens: set[str], content: str) -> set[str]:
-        content_tokens = cls._content_tokens(content)
+    def _matching_tokens(self, query_tokens: set[str], content: str) -> set[str]:
+        content_tokens = self._content_tokens(content)
         return {
             query_token
             for query_token in query_tokens
-            if any(cls._tokens_match(query_token, content_token) for content_token in content_tokens)
+            if any(self._analyzer.tokens_match(query_token, content_token) for content_token in content_tokens)
         }
 
-    @classmethod
-    def _important_tokens(cls, text: str) -> set[str]:
-        return cls._expand_tokens({
+    def _important_tokens(self, text: str) -> set[str]:
+        return self._analyzer.expand_tokens({
             token
-            for token in (cls._normalise_token(raw_token) for raw_token in re.findall(r"[\w]+", text, flags=re.UNICODE))
+            for token in (self._normalise_token(raw_token) for raw_token in re.findall(r"[\w]+", text, flags=re.UNICODE))
             if token and token not in _STOPWORDS and len(token) > 2
         })
 
-    @classmethod
-    def _expanded_tokens(cls, text: str) -> set[str]:
-        return cls._expand_tokens({
+    def _expanded_tokens(self, text: str) -> set[str]:
+        return self._analyzer.expand_tokens({
             token
-            for token in (cls._normalise_token(raw_token) for raw_token in re.findall(r"[\w]+", text, flags=re.UNICODE))
+            for token in (self._normalise_token(raw_token) for raw_token in re.findall(r"[\w]+", text, flags=re.UNICODE))
             if token and len(token) > 2
         })
 
-    @staticmethod
-    def _expand_tokens(tokens: set[str]) -> set[str]:
-        expanded = set(tokens)
-        for token in tokens:
-            expanded.update(_QUERY_EXPANSIONS.get(token, set()))
-        return expanded
-
     @classmethod
     def _content_tokens(cls, text: str) -> set[str]:
         return {
@@ -461,14 +549,6 @@ def _normalise_token(token: str) -> str:
             return token[:-1]
         return token
 
-    @staticmethod
-    def _tokens_match(query_token: str, content_token: str) -> bool:
-        return (
-            query_token == content_token
-            or (len(query_token) >= 3 and query_token in content_token)
-            or (len(content_token) >= 3 and content_token in query_token)
-        )
-
     @classmethod
     def _asks_for_current(cls, query: str) -> bool:
         return bool(cls._important_tokens(query).intersection(_CURRENT_MARKERS) or {"now", "current", "latest"}.intersection(cls._content_tokens(query)))

From 15f28efeb337e8634f6575ab76ffa9bec7c1abe6 Mon Sep 17 00:00:00 2001
From: krishna3554 <krishlokhande45@gmail.com>
Date: Mon, 31 Aug 2026 22:44:25 +0530
Subject: [PATCH 2/6] fix: retrieval quality - query expansion overfitting and
 bidirectional substring matching

- Add pluggable Analyzer interface with PlainAnalyzer (default) and DomainAnalyzer
- PlainAnalyzer uses prefix-only token matching to prevent false positives
  (e.g., 'api' matching 'rapid', 'cat' matching 'category')
- _QUERY_EXPANSIONS moved behind DomainAnalyzer, not used by default
- Add AnalyzerKind config (DMA_ANALYZER_KIND env var) to opt-in to domain expansions
- Update _to_fts_query to use >= 3 char prefix matching (was > 3)
- Add adversarial test cases for prefix-only matching (cat/category, api/rapid, art/particle)
- Use DomainAnalyzer in benchmark runner for corpus parity

Fixes #27
---
 benchmarks/runner/memory_eval.py           |  9 +++-
 services/dma-api/src/dma_api/config.py     |  9 ++++
 services/dma-api/src/dma_api/main.py       |  5 +-
 services/dma-api/src/dma_api/repository.py | 17 +++---
 services/dma-api/tests/test_recall.py      | 62 ++++++++++++++++++++++
 5 files changed, 90 insertions(+), 12 deletions(-)

diff --git a/benchmarks/runner/memory_eval.py b/benchmarks/runner/memory_eval.py
index 2c16ca9..1708934 100644
--- a/benchmarks/runner/memory_eval.py
+++ b/benchmarks/runner/memory_eval.py
@@ -14,11 +14,16 @@
 from typing import Any
 
 from dma_api.models import MemoryType
-from dma_api.repository import MemoryRecord, SQLiteMemoryRepository
+from dma_api.repository import MemoryRecord, SQLiteMemoryRepository, AnalyzerKind, get_analyzer
 
 DEFAULT_DATASET = Path("benchmarks/datasets/memory-eval-v0.1.jsonl")
 DEFAULT_NOW = datetime.fromisoformat("2026-08-01T00:00:00+00:00")
 TENANT_ID = "memory-eval-tenant"
+# Use DomainAnalyzer for benchmark parity with the evaluation corpus.
+# The DomainAnalyzer includes DMA-specific query expansions that the
+# eval dataset was designed for. The default PLAIN analyzer has no
+# expansions and uses prefix-only matching to avoid false positives.
+BENCHMARK_ANALYZER = get_analyzer(AnalyzerKind.DOMAIN)
 
 
 @dataclass(frozen=True, slots=True)
@@ -69,7 +74,7 @@ def _failures_to_set(failures: list[dict[str, Any]]) -> set[str]:
 
 
 def _run_case(case: dict[str, Any], database_path: Path, *, limit: int) -> MemoryEvalResult:
-    repository = SQLiteMemoryRepository(database_path)
+    repository = SQLiteMemoryRepository(database_path, analyzer=BENCHMARK_ANALYZER)
     repository.initialize()
     for index, memory in enumerate(case["memories"]):
         timestamp = DEFAULT_NOW.replace(microsecond=index)
diff --git a/services/dma-api/src/dma_api/config.py b/services/dma-api/src/dma_api/config.py
index 00210d2..a4bef16 100644
--- a/services/dma-api/src/dma_api/config.py
+++ b/services/dma-api/src/dma_api/config.py
@@ -6,6 +6,8 @@
 from dataclasses import dataclass
 from pathlib import Path
 
+from dma_api.repository import AnalyzerKind
+
 
 @dataclass(frozen=True, slots=True)
 class Settings:
@@ -15,15 +17,22 @@ class Settings:
     api_key: str = "dma-local-development-key"
     tenant_id: str = "local"
     environment: str = "development"
+    analyzer_kind: AnalyzerKind = AnalyzerKind.PLAIN
 
     @classmethod
     def from_env(cls) -> Settings:
         """Load runtime configuration without ever logging secret values."""
+        analyzer_kind_str = os.getenv("DMA_ANALYZER_KIND", "plain").lower()
+        try:
+            analyzer_kind = AnalyzerKind(analyzer_kind_str)
+        except ValueError:
+            analyzer_kind = AnalyzerKind.PLAIN
         settings = cls(
             database_path=Path(os.getenv("DMA_DATABASE_PATH", "./dma.db")),
             api_key=os.getenv("DMA_API_KEY", "dma-local-development-key"),
             tenant_id=os.getenv("DMA_TENANT_ID", "local"),
             environment=os.getenv("DMA_ENVIRONMENT", "development"),
+            analyzer_kind=analyzer_kind,
         )
         if settings.environment == "production" and settings.api_key == "dma-local-development-key":
             raise ValueError("DMA_API_KEY must be explicitly configured in production")
diff --git a/services/dma-api/src/dma_api/main.py b/services/dma-api/src/dma_api/main.py
index 1746427..942d274 100644
--- a/services/dma-api/src/dma_api/main.py
+++ b/services/dma-api/src/dma_api/main.py
@@ -22,13 +22,14 @@
     RememberRequest,
     RetrievalExplanation,
 )
-from dma_api.repository import MemoryRecord, SQLiteMemoryRepository
+from dma_api.repository import MemoryRecord, SQLiteMemoryRepository, get_analyzer
 
 
 def create_app(settings: Settings | None = None) -> FastAPI:
     """Create an independently configurable API application."""
     runtime_settings = settings or Settings()
-    repository = SQLiteMemoryRepository(runtime_settings.database_path)
+    analyzer = get_analyzer(runtime_settings.analyzer_kind)
+    repository = SQLiteMemoryRepository(runtime_settings.database_path, analyzer=analyzer)
 
     @asynccontextmanager
     async def lifespan(_: FastAPI):
diff --git a/services/dma-api/src/dma_api/repository.py b/services/dma-api/src/dma_api/repository.py
index 1412f3a..063170d 100644
--- a/services/dma-api/src/dma_api/repository.py
+++ b/services/dma-api/src/dma_api/repository.py
@@ -10,7 +10,7 @@
 from collections.abc import Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import UTC, datetime
 from enum import Enum
 from pathlib import Path
 
@@ -480,7 +480,10 @@ def _to_fts_query(query: str) -> str:
         terms = []
         for token in tokens:
             terms.append(f'"{token}"')
-            if len(token) > 3:
+            # Use prefix matching for tokens >= 3 chars to align with analyzer's
+            # prefix-only matching (avoids false positives from bidirectional
+            # substring matching like "cat" in "category").
+            if len(token) >= 3:
                 terms.append(f"{token}*")
         return " OR ".join(terms)
 
@@ -549,13 +552,11 @@ def _normalise_token(token: str) -> str:
             return token[:-1]
         return token
 
-    @classmethod
-    def _asks_for_current(cls, query: str) -> bool:
-        return bool(cls._important_tokens(query).intersection(_CURRENT_MARKERS) or {"now", "current", "latest"}.intersection(cls._content_tokens(query)))
+    def _asks_for_current(self, query: str) -> bool:
+        return bool(self._important_tokens(query).intersection(_CURRENT_MARKERS) or {"now", "current", "latest"}.intersection(self._content_tokens(query)))
 
-    @classmethod
-    def _has_current_marker(cls, content: str) -> bool:
-        return bool(cls._content_tokens(content).intersection(_CURRENT_MARKERS))
+    def _has_current_marker(self, content: str) -> bool:
+        return bool(self._content_tokens(content).intersection(_CURRENT_MARKERS))
 
     @staticmethod
     def _normalise_semantic(content: str) -> str:
diff --git a/services/dma-api/tests/test_recall.py b/services/dma-api/tests/test_recall.py
index f01e4d7..36fcd8a 100644
--- a/services/dma-api/tests/test_recall.py
+++ b/services/dma-api/tests/test_recall.py
@@ -91,3 +91,65 @@ def test_recall_prefers_current_state_memories_for_now_queries(tmp_path) -> None
     assert response.status_code == 200
     contents = [item["content"] for item in response.json()["results"]]
     assert contents == ["User now prefers Java Spring Boot for backend APIs."]
+
+
+def test_recall_prefix_only_matching_no_false_positives(tmp_path) -> None:
+    """Adversarial tests for prefix-only token matching.
+
+    The old bidirectional substring matching caused false positives:
+    - 'api' matched 'rapid' (substring but not prefix)
+    - 'art' matched 'particle' (substring but not prefix)
+
+    Prefix-only matching ensures:
+    - 'api' matches 'api' (exact) and 'apikey' (prefix), but NOT 'rapid'
+    - 'art' matches 'art' (exact) and 'artist' (prefix), but NOT 'particle'
+    - 'cat' matches 'cat', 'cater', 'category' (all valid prefixes)
+    """
+    app = create_app(Settings(database_path=tmp_path / "dma.db", api_key="test-key", tenant_id="tenant-a"))
+    with TestClient(app) as client:
+        # Store memories with words that could cause false positives with substring matching
+        _remember(client, "Rapid deployment is our goal.", "semantic", "key-cat-0000000002")
+        _remember(client, "Particle physics is complex.", "semantic", "key-cat-0000000003")
+        # Also store exact/prefix matches that SHOULD be found
+        _remember(client, "API key configuration done.", "semantic", "key-cat-0000000005")
+        _remember(client, "Artist portfolio updated.", "semantic", "key-cat-0000000006")
+        # 'cat' prefix matches are legitimate - store some
+        _remember(client, "The cat sits on the mat.", "semantic", "key-cat-0000000004")
+        _remember(client, "The category system organizes items.", "semantic", "key-cat-0000000001")
+
+        # Query 'api' should NOT match 'rapid' (substring), but SHOULD match 'api' and 'apikey'
+        response = client.post(
+            "/v1/memories/recall",
+            headers={"Authorization": "Bearer test-key"},
+            json={"agent_id": "coding-agent", "query": "api", "limit": 10},
+        )
+    assert response.status_code == 200
+    results = response.json()["results"]
+    contents = [item["content"] for item in results]
+    assert "API key configuration done." in contents
+    assert "Rapid deployment is our goal." not in contents
+
+    # Query 'art' should NOT match 'particle' (substring), but SHOULD match 'art' and 'artist'
+    response = client.post(
+        "/v1/memories/recall",
+        headers={"Authorization": "Bearer test-key"},
+        json={"agent_id": "coding-agent", "query": "art", "limit": 10},
+    )
+    assert response.status_code == 200
+    results = response.json()["results"]
+    contents = [item["content"] for item in results]
+    assert "Artist portfolio updated." in contents
+    assert "Particle physics is complex." not in contents
+
+    # Query 'cat' matches 'cat', 'cater', 'category' (all valid prefixes)
+    response = client.post(
+        "/v1/memories/recall",
+        headers={"Authorization": "Bearer test-key"},
+        json={"agent_id": "coding-agent", "query": "cat", "limit": 10},
+    )
+    assert response.status_code == 200
+    results = response.json()["results"]
+    contents = [item["content"] for item in results]
+    # Both should match since 'cat' is a prefix of 'category'
+    assert "The cat sits on the mat." in contents
+    assert "The category system organizes items." in contents

From 742fe41c55e288f54bd0aa5e47f70f62808092b7 Mon Sep 17 00:00:00 2001
From: krishna3554 <krishlokhande45@gmail.com>
Date: Mon, 31 Aug 2026 23:38:37 +0530
Subject: [PATCH 3/6] fix: security hardening - rate limiting and brute-force
 protection for bearer auth

- Add InMemoryRateLimiter with sliding-window + lockout per source (IP)
- Configure via DMA_AUTH_MAX_ATTEMPTS (default 5), DMA_AUTH_WINDOW_SECONDS (default 60), DMA_AUTH_LOCKOUT_SECONDS (default 300)
- Extract client IP from X-Forwarded-For header
- Log failed auth attempts with source IP (no key logged)
- Successful auth resets failure counter for that source
- Locked-out sources get 429 with retry message
- Add AuthLimits dataclass to config
- Add 8 comprehensive tests for rate limiting behavior

Fixes #26
---
 services/dma-api/src/dma_api/config.py       |  18 ++-
 services/dma-api/src/dma_api/main.py         |  75 +++++++++-
 services/dma-api/tests/test_rate_limiting.py | 150 +++++++++++++++++++
 3 files changed, 239 insertions(+), 4 deletions(-)
 create mode 100644 services/dma-api/tests/test_rate_limiting.py

diff --git a/services/dma-api/src/dma_api/config.py b/services/dma-api/src/dma_api/config.py
index a4bef16..c275c3b 100644
--- a/services/dma-api/src/dma_api/config.py
+++ b/services/dma-api/src/dma_api/config.py
@@ -3,12 +3,21 @@
 from __future__ import annotations
 
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 
 from dma_api.repository import AnalyzerKind
 
 
+@dataclass(frozen=True, slots=True)
+class AuthLimits:
+    """Rate-limiting bounds for bearer-auth attempts per source (IP)."""
+
+    max_attempts: int = 5
+    window_seconds: int = 60
+    lockout_seconds: int = 300
+
+
 @dataclass(frozen=True, slots=True)
 class Settings:
     """Runtime settings passed explicitly to the application factory."""
@@ -18,10 +27,16 @@ class Settings:
     tenant_id: str = "local"
     environment: str = "development"
     analyzer_kind: AnalyzerKind = AnalyzerKind.PLAIN
+    auth_limits: AuthLimits = field(default_factory=AuthLimits)
 
     @classmethod
     def from_env(cls) -> Settings:
         """Load runtime configuration without ever logging secret values."""
+        limits = AuthLimits(
+            max_attempts=int(os.getenv("DMA_AUTH_MAX_ATTEMPTS", "5")),
+            window_seconds=int(os.getenv("DMA_AUTH_WINDOW_SECONDS", "60")),
+            lockout_seconds=int(os.getenv("DMA_AUTH_LOCKOUT_SECONDS", "300")),
+        )
         analyzer_kind_str = os.getenv("DMA_ANALYZER_KIND", "plain").lower()
         try:
             analyzer_kind = AnalyzerKind(analyzer_kind_str)
@@ -33,6 +48,7 @@ def from_env(cls) -> Settings:
             tenant_id=os.getenv("DMA_TENANT_ID", "local"),
             environment=os.getenv("DMA_ENVIRONMENT", "development"),
             analyzer_kind=analyzer_kind,
+            auth_limits=limits,
         )
         if settings.environment == "production" and settings.api_key == "dma-local-development-key":
             raise ValueError("DMA_API_KEY must be explicitly configured in production")
diff --git a/services/dma-api/src/dma_api/main.py b/services/dma-api/src/dma_api/main.py
index 942d274..3a0036a 100644
--- a/services/dma-api/src/dma_api/main.py
+++ b/services/dma-api/src/dma_api/main.py
@@ -2,15 +2,19 @@
 
 from __future__ import annotations
 
+import logging
 import re
 import secrets
+import threading
+import time
 from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from uuid import uuid4
 
 from fastapi import Depends, FastAPI, Header, HTTPException, Path, Query, Response, status
 
-from dma_api.config import Settings
+from dma_api.config import AuthLimits, Settings
 from dma_api.models import (
     MemoryExplanation,
     MemoryPage,
@@ -24,12 +28,64 @@
 )
 from dma_api.repository import MemoryRecord, SQLiteMemoryRepository, get_analyzer
 
+logger = logging.getLogger("dma_api.auth")
+
+
+@dataclass
+class _SourceRecord:
+    timestamps: list[float] = field(default_factory=list)
+    locked_until: float = 0.0
+
+
+class InMemoryRateLimiter:
+    """Per-source sliding-window rate limiter with lockout for failed auth."""
+
+    def __init__(self, limits: AuthLimits) -> None:
+        self._max_attempts = limits.max_attempts
+        self._window = limits.window_seconds
+        self._lockout = limits.lockout_seconds
+        self._sources: dict[str, _SourceRecord] = {}
+        self._lock = threading.Lock()
+
+    def is_locked_out(self, source: str) -> bool:
+        with self._lock:
+            record = self._sources.get(source)
+            if record is None:
+                return False
+            return record.locked_until > time.monotonic()
+
+    def record_failure(self, source: str) -> None:
+        now = time.monotonic()
+        with self._lock:
+            record = self._sources.setdefault(source, _SourceRecord())
+            cutoff = now - self._window
+            record.timestamps = [t for t in record.timestamps if t > cutoff]
+            record.timestamps.append(now)
+            if len(record.timestamps) >= self._max_attempts:
+                record.locked_until = now + self._lockout
+        logger.warning(
+            "auth_failure source=%s total_in_window=%d",
+            source,
+            len(record.timestamps),
+        )
+
+    def record_success(self, source: str) -> None:
+        with self._lock:
+            self._sources.pop(source, None)
+
+
+def _client_source(x_forwarded_for: str | None) -> str:
+    if x_forwarded_for:
+        return x_forwarded_for.split(",")[0].strip()
+    return "unknown"
+
 
 def create_app(settings: Settings | None = None) -> FastAPI:
     """Create an independently configurable API application."""
     runtime_settings = settings or Settings()
     analyzer = get_analyzer(runtime_settings.analyzer_kind)
     repository = SQLiteMemoryRepository(runtime_settings.database_path, analyzer=analyzer)
+    rate_limiter = InMemoryRateLimiter(runtime_settings.auth_limits)
 
     @asynccontextmanager
     async def lifespan(_: FastAPI):
@@ -42,10 +98,23 @@ async def lifespan(_: FastAPI):
     def healthz() -> dict[str, str]:
         return {"status": "ok"}
 
-    def authenticate(authorization: str | None = Header(default=None)) -> str:
+    def authenticate(
+        authorization: str | None = Header(default=None),
+        x_forwarded_for: str | None = Header(default=None, alias="X-Forwarded-For"),
+    ) -> str:
+        source = _client_source(x_forwarded_for)
+        if rate_limiter.is_locked_out(source):
+            raise HTTPException(
+                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                detail="too many failed authentication attempts; try again later",
+            )
         expected = f"Bearer {runtime_settings.api_key}"
         if authorization is None or not secrets.compare_digest(authorization, expected):
-            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid API key")
+            rate_limiter.record_failure(source)
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid API key"
+            )
+        rate_limiter.record_success(source)
         return runtime_settings.tenant_id
 
     @app.post("/v1/memories", response_model=MemoryResponse, status_code=status.HTTP_201_CREATED)
diff --git a/services/dma-api/tests/test_rate_limiting.py b/services/dma-api/tests/test_rate_limiting.py
new file mode 100644
index 0000000..b289eb8
--- /dev/null
+++ b/services/dma-api/tests/test_rate_limiting.py
@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+import time
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+
+from dma_api.config import AuthLimits, Settings
+from dma_api.main import InMemoryRateLimiter, create_app
+
+
+def _bad_auth_headers(source: str = "10.0.0.1") -> dict[str, str]:
+    return {
+        "Authorization": "Bearer wrong-key",
+        "Idempotency-Key": "rate-limit-test-00001",
+        "X-Forwarded-For": source,
+    }
+
+
+def _good_auth_headers(source: str = "10.0.0.1") -> dict[str, str]:
+    return {
+        "Authorization": "Bearer test-key",
+        "Idempotency-Key": "rate-limit-test-00001",
+        "X-Forwarded-For": source,
+    }
+
+
+def _json() -> dict[str, str]:
+    return {"agent_id": "coding-agent", "content": "A fact.", "type": "semantic"}
+
+
+def test_lockout_after_max_failed_attempts(tmp_path) -> None:
+    limits = AuthLimits(max_attempts=3, window_seconds=60, lockout_seconds=300)
+    app = create_app(Settings(
+        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
+    ))
+    with TestClient(app) as client:
+        for _ in range(3):
+            resp = client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
+            assert resp.status_code == 401
+
+        resp = client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
+        assert resp.status_code == 429
+        assert "too many" in resp.json()["detail"]
+
+
+def test_lockout_blocks_valid_key_from_same_source(tmp_path) -> None:
+    limits = AuthLimits(max_attempts=2, window_seconds=60, lockout_seconds=300)
+    app = create_app(Settings(
+        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
+    ))
+    with TestClient(app) as client:
+        for _ in range(2):
+            client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
+
+        resp = client.post("/v1/memories", headers=_good_auth_headers(), json=_json())
+        assert resp.status_code == 429
+
+
+def test_lockout_does_not_affect_other_sources(tmp_path) -> None:
+    limits = AuthLimits(max_attempts=2, window_seconds=60, lockout_seconds=300)
+    app = create_app(Settings(
+        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
+    ))
+    with TestClient(app) as client:
+        for _ in range(2):
+            client.post("/v1/memories", headers=_bad_auth_headers("10.0.0.1"), json=_json())
+
+        resp = client.post("/v1/memories", headers=_good_auth_headers("10.0.0.2"), json=_json())
+        assert resp.status_code == 201
+
+
+def test_successful_auth_resets_failure_count(tmp_path) -> None:
+    limits = AuthLimits(max_attempts=3, window_seconds=60, lockout_seconds=300)
+    app = create_app(Settings(
+        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
+    ))
+    with TestClient(app) as client:
+        for _ in range(2):
+            client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
+
+        resp = client.post("/v1/memories", headers=_good_auth_headers(), json=_json())
+        assert resp.status_code == 201
+
+        for _ in range(2):
+            client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
+
+        resp = client.post(
+            "/v1/memories",
+            headers={**_good_auth_headers(), "Idempotency-Key": "rate-limit-test-00002"},
+            json=_json(),
+        )
+        assert resp.status_code in (200, 201)
+
+
+def test_lockout_expires_after_lockout_seconds() -> None:
+    limits = AuthLimits(max_attempts=2, window_seconds=60, lockout_seconds=10)
+    limiter = InMemoryRateLimiter(limits)
+
+    limiter.record_failure("src")
+    limiter.record_failure("src")
+    assert limiter.is_locked_out("src")
+
+    with patch("dma_api.main.time") as mock_time:
+        mock_time.monotonic.return_value = time.monotonic() + 11
+        assert not limiter.is_locked_out("src")
+
+
+def test_failures_outside_window_do_not_count() -> None:
+    limits = AuthLimits(max_attempts=3, window_seconds=5, lockout_seconds=300)
+    limiter = InMemoryRateLimiter(limits)
+
+    limiter.record_failure("src")
+    limiter.record_failure("src")
+    assert not limiter.is_locked_out("src")
+
+    original_monotonic = time.monotonic
+
+    def shifted_time():
+        return original_monotonic() + 6
+
+    with patch("dma_api.main.time") as mock_time:
+        mock_time.monotonic = shifted_time
+        limiter.record_failure("src")
+        assert not limiter.is_locked_out("src")
+
+
+def test_healthz_not_rate_limited(tmp_path) -> None:
+    limits = AuthLimits(max_attempts=1, window_seconds=60, lockout_seconds=300)
+    app = create_app(Settings(
+        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
+    ))
+    with TestClient(app) as client:
+        client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
+
+        resp = client.get("/healthz")
+        assert resp.status_code == 200
+
+
+def test_auth_failure_logged(tmp_path, caplog) -> None:
+    limits = AuthLimits(max_attempts=5, window_seconds=60, lockout_seconds=300)
+    app = create_app(Settings(
+        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
+    ))
+    import logging
+    with caplog.at_level(logging.WARNING, logger="dma_api.auth"):
+        with TestClient(app) as client:
+            client.post("/v1/memories", headers=_bad_auth_headers("192.168.1.1"), json=_json())
+
+    assert any("auth_failure" in record.message and "192.168.1.1" in record.message for record in caplog.records)

From 2e44672928d60221b1b477d12486251493c87d05 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Mon, 31 Aug 2026 18:24:06 +0000
Subject: [PATCH 4/6] fix: address Devin Review findings on rate limiting and
 recall

- Trust X-Forwarded-For only when DMA_TRUST_FORWARDED_FOR is set; fall back to the transport peer instead of a shared unknown bucket
- Evict stale rate-limiter sources and cap the tracked map size
- Validate auth limit env values as positive integers
- Use analyzer expansions when building the FTS candidate query and credit expansion matches to their query token
- Normalise naive datetimes to UTC in _utc_isoformat
- Repair the offset-expiry recall test and refresh the benchmark failure baseline

Co-Authored-By: krishna <87197325+krishna3554@users.noreply.github.com>
---
 benchmarks/results/v0.1-failures.jsonl       |   4 +-
 benchmarks/runner/memory_eval.py             |   7 +-
 services/dma-api/src/dma_api/config.py       |  30 +++++-
 services/dma-api/src/dma_api/main.py         |  72 +++++++++++--
 services/dma-api/src/dma_api/repository.py   |  45 ++++-----
 services/dma-api/tests/test_configuration.py |  20 ++++
 services/dma-api/tests/test_rate_limiting.py | 100 +++++++++++++++----
 services/dma-api/tests/test_recall.py        |  61 ++++++++++-
 8 files changed, 276 insertions(+), 63 deletions(-)

diff --git a/benchmarks/results/v0.1-failures.jsonl b/benchmarks/results/v0.1-failures.jsonl
index 69ddc3e..bc3caf2 100644
--- a/benchmarks/results/v0.1-failures.jsonl
+++ b/benchmarks/results/v0.1-failures.jsonl
@@ -1,9 +1,9 @@
 {"actual_result_ids": [], "case_id": "semantic-024", "category": "semantic_preference", "excluded_ids": [], "expected_ids": ["semantic-024-mem-01"], "query": "how should benchmark results be reported?", "should_recall": true, "stale_ids": []}
 {"actual_result_ids": [], "case_id": "procedural-004", "category": "procedural_workflow", "excluded_ids": [], "expected_ids": ["procedural-004-mem-01"], "query": "what should adapter work preserve?", "should_recall": true, "stale_ids": []}
 {"actual_result_ids": ["distractor-001-mem-02", "distractor-001-mem-01"], "case_id": "distractor-001", "category": "distractor", "excluded_ids": ["distractor-001-mem-01"], "expected_ids": ["distractor-001-mem-02"], "query": "what frontend framework does the user prefer?", "should_recall": true, "stale_ids": []}
-{"actual_result_ids": ["distractor-007-mem-01", "distractor-007-mem-02"], "case_id": "distractor-007", "category": "distractor", "excluded_ids": ["distractor-007-mem-01"], "expected_ids": ["distractor-007-mem-02"], "query": "what proves framework-agnostic support?", "should_recall": true, "stale_ids": []}
+{"actual_result_ids": ["distractor-007-mem-02", "distractor-007-mem-01"], "case_id": "distractor-007", "category": "distractor", "excluded_ids": ["distractor-007-mem-01"], "expected_ids": ["distractor-007-mem-02"], "query": "what proves framework-agnostic support?", "should_recall": true, "stale_ids": []}
+{"actual_result_ids": ["distractor-008-mem-02", "distractor-008-mem-01"], "case_id": "distractor-008", "category": "distractor", "excluded_ids": ["distractor-008-mem-01"], "expected_ids": ["distractor-008-mem-02"], "query": "what metric evaluates classification?", "should_recall": true, "stale_ids": []}
 {"actual_result_ids": ["distractor-012-mem-02", "distractor-012-mem-01"], "case_id": "distractor-012", "category": "distractor", "excluded_ids": ["distractor-012-mem-01"], "expected_ids": ["distractor-012-mem-02"], "query": "what command evaluates retrieval quality?", "should_recall": true, "stale_ids": []}
-{"actual_result_ids": [], "case_id": "distractor-015", "category": "distractor", "excluded_ids": ["distractor-015-mem-02"], "expected_ids": ["distractor-015-mem-01"], "query": "what cannot be overwritten after release?", "should_recall": true, "stale_ids": []}
 {"actual_result_ids": ["distractor-016-mem-02", "distractor-016-mem-01"], "case_id": "distractor-016", "category": "distractor", "excluded_ids": ["distractor-016-mem-01"], "expected_ids": ["distractor-016-mem-02"], "query": "what package contains the LangGraph adapter?", "should_recall": true, "stale_ids": []}
 {"actual_result_ids": ["distractor-017-mem-01", "distractor-017-mem-02"], "case_id": "distractor-017", "category": "distractor", "excluded_ids": ["distractor-017-mem-02"], "expected_ids": ["distractor-017-mem-01"], "query": "what package exposes MCP tools?", "should_recall": true, "stale_ids": []}
 {"actual_result_ids": [], "case_id": "conflict-010", "category": "conflict_update", "excluded_ids": ["conflict-010-mem-01"], "expected_ids": ["conflict-010-mem-02"], "query": "how should packages be published now?", "should_recall": true, "stale_ids": []}
diff --git a/benchmarks/runner/memory_eval.py b/benchmarks/runner/memory_eval.py
index 442063d..65108d3 100644
--- a/benchmarks/runner/memory_eval.py
+++ b/benchmarks/runner/memory_eval.py
@@ -15,7 +15,12 @@
 from typing import Any
 
 from dma_api.models import MemoryType
-from dma_api.repository import MemoryRecord, SQLiteMemoryRepository, AnalyzerKind, get_analyzer
+from dma_api.repository import (
+    AnalyzerKind,
+    MemoryRecord,
+    SQLiteMemoryRepository,
+    get_analyzer,
+)
 
 DEFAULT_DATASET = Path("benchmarks/datasets/memory-eval-v0.1.jsonl")
 DEFAULT_NOW = datetime.fromisoformat("2026-08-01T00:00:00+00:00")
diff --git a/services/dma-api/src/dma_api/config.py b/services/dma-api/src/dma_api/config.py
index c275c3b..2105648 100644
--- a/services/dma-api/src/dma_api/config.py
+++ b/services/dma-api/src/dma_api/config.py
@@ -16,6 +16,26 @@ class AuthLimits:
     max_attempts: int = 5
     window_seconds: int = 60
     lockout_seconds: int = 300
+    max_tracked_sources: int = 10_000
+
+    def __post_init__(self) -> None:
+        for name in ("max_attempts", "window_seconds", "lockout_seconds", "max_tracked_sources"):
+            value = getattr(self, name)
+            if value < 1:
+                raise ValueError(f"{name} must be a positive integer, got {value}")
+
+
+def _positive_int_env(name: str, default: int) -> int:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    try:
+        value = int(raw)
+    except ValueError as error:
+        raise ValueError(f"{name} must be a positive integer, got {raw!r}") from error
+    if value < 1:
+        raise ValueError(f"{name} must be a positive integer, got {value}")
+    return value
 
 
 @dataclass(frozen=True, slots=True)
@@ -28,14 +48,16 @@ class Settings:
     environment: str = "development"
     analyzer_kind: AnalyzerKind = AnalyzerKind.PLAIN
     auth_limits: AuthLimits = field(default_factory=AuthLimits)
+    trust_forwarded_for: bool = False
 
     @classmethod
     def from_env(cls) -> Settings:
         """Load runtime configuration without ever logging secret values."""
         limits = AuthLimits(
-            max_attempts=int(os.getenv("DMA_AUTH_MAX_ATTEMPTS", "5")),
-            window_seconds=int(os.getenv("DMA_AUTH_WINDOW_SECONDS", "60")),
-            lockout_seconds=int(os.getenv("DMA_AUTH_LOCKOUT_SECONDS", "300")),
+            max_attempts=_positive_int_env("DMA_AUTH_MAX_ATTEMPTS", 5),
+            window_seconds=_positive_int_env("DMA_AUTH_WINDOW_SECONDS", 60),
+            lockout_seconds=_positive_int_env("DMA_AUTH_LOCKOUT_SECONDS", 300),
+            max_tracked_sources=_positive_int_env("DMA_AUTH_MAX_TRACKED_SOURCES", 10_000),
         )
         analyzer_kind_str = os.getenv("DMA_ANALYZER_KIND", "plain").lower()
         try:
@@ -49,6 +71,8 @@ def from_env(cls) -> Settings:
             environment=os.getenv("DMA_ENVIRONMENT", "development"),
             analyzer_kind=analyzer_kind,
             auth_limits=limits,
+            trust_forwarded_for=os.getenv("DMA_TRUST_FORWARDED_FOR", "false").lower()
+            in {"1", "true", "yes"},
         )
         if settings.environment == "production" and settings.api_key == "dma-local-development-key":
             raise ValueError("DMA_API_KEY must be explicitly configured in production")
diff --git a/services/dma-api/src/dma_api/main.py b/services/dma-api/src/dma_api/main.py
index 3a0036a..7457f38 100644
--- a/services/dma-api/src/dma_api/main.py
+++ b/services/dma-api/src/dma_api/main.py
@@ -12,7 +12,17 @@
 from datetime import UTC, datetime
 from uuid import uuid4
 
-from fastapi import Depends, FastAPI, Header, HTTPException, Path, Query, Response, status
+from fastapi import (
+    Depends,
+    FastAPI,
+    Header,
+    HTTPException,
+    Path,
+    Query,
+    Request,
+    Response,
+    status,
+)
 
 from dma_api.config import AuthLimits, Settings
 from dma_api.models import (
@@ -44,6 +54,7 @@ def __init__(self, limits: AuthLimits) -> None:
         self._max_attempts = limits.max_attempts
         self._window = limits.window_seconds
         self._lockout = limits.lockout_seconds
+        self._max_sources = limits.max_tracked_sources
         self._sources: dict[str, _SourceRecord] = {}
         self._lock = threading.Lock()
 
@@ -57,27 +68,69 @@ def is_locked_out(self, source: str) -> bool:
     def record_failure(self, source: str) -> None:
         now = time.monotonic()
         with self._lock:
+            self._evict_stale(now)
             record = self._sources.setdefault(source, _SourceRecord())
             cutoff = now - self._window
             record.timestamps = [t for t in record.timestamps if t > cutoff]
             record.timestamps.append(now)
             if len(record.timestamps) >= self._max_attempts:
                 record.locked_until = now + self._lockout
+            attempts_in_window = len(record.timestamps)
         logger.warning(
             "auth_failure source=%s total_in_window=%d",
             source,
-            len(record.timestamps),
+            attempts_in_window,
         )
 
     def record_success(self, source: str) -> None:
         with self._lock:
             self._sources.pop(source, None)
 
-
-def _client_source(x_forwarded_for: str | None) -> str:
-    if x_forwarded_for:
-        return x_forwarded_for.split(",")[0].strip()
-    return "unknown"
+    def tracked_sources(self) -> int:
+        with self._lock:
+            return len(self._sources)
+
+    def _evict_stale(self, now: float) -> None:
+        """Drop sources without in-window failures or an active lockout.
+
+        Must be called while holding ``self._lock``. When eviction alone cannot
+        keep the map under ``max_tracked_sources``, the least recently active
+        sources are dropped so source churn cannot grow memory without bound.
+        """
+        cutoff = now - self._window
+        for key, record in list(self._sources.items()):
+            if record.locked_until > now:
+                continue
+            if not any(timestamp > cutoff for timestamp in record.timestamps):
+                del self._sources[key]
+        overflow = len(self._sources) - self._max_sources + 1
+        if overflow <= 0:
+            return
+        stalest = sorted(self._sources, key=self._last_activity)[:overflow]
+        for key in stalest:
+            del self._sources[key]
+
+    def _last_activity(self, source: str) -> float:
+        record = self._sources[source]
+        return max(record.locked_until, max(record.timestamps, default=0.0))
+
+
+def _client_source(
+    request: Request, x_forwarded_for: str | None, *, trust_forwarded_for: bool
+) -> str:
+    """Identify the caller for rate limiting.
+
+    ``X-Forwarded-For`` is honoured only when the deployment declares that it
+    runs behind a trusted proxy; otherwise a client could rotate the header to
+    win a fresh failure counter for every guess. The transport peer address is
+    the default, so direct callers are never pooled into one shared bucket.
+    """
+    if trust_forwarded_for and x_forwarded_for:
+        forwarded = x_forwarded_for.split(",")[0].strip()
+        if forwarded:
+            return forwarded
+    client = request.client
+    return client.host if client is not None else "unknown"
 
 
 def create_app(settings: Settings | None = None) -> FastAPI:
@@ -99,10 +152,13 @@ def healthz() -> dict[str, str]:
         return {"status": "ok"}
 
     def authenticate(
+        request: Request,
         authorization: str | None = Header(default=None),
         x_forwarded_for: str | None = Header(default=None, alias="X-Forwarded-For"),
     ) -> str:
-        source = _client_source(x_forwarded_for)
+        source = _client_source(
+            request, x_forwarded_for, trust_forwarded_for=runtime_settings.trust_forwarded_for
+        )
         if rate_limiter.is_locked_out(source):
             raise HTTPException(
                 status_code=status.HTTP_429_TOO_MANY_REQUESTS,
diff --git a/services/dma-api/src/dma_api/repository.py b/services/dma-api/src/dma_api/repository.py
index 976b00d..a2d7357 100644
--- a/services/dma-api/src/dma_api/repository.py
+++ b/services/dma-api/src/dma_api/repository.py
@@ -134,9 +134,7 @@ def tokens_match(self, query_token: str, content_token: str) -> bool:
         # "category", "api" ⊂ "rapid", "art" ⊂ "particle".
         if query_token == content_token:
             return True
-        if len(query_token) >= 3 and content_token.startswith(query_token):
-            return True
-        return False
+        return len(query_token) >= 3 and content_token.startswith(query_token)
 
 
 class DomainAnalyzer(Analyzer):
@@ -156,9 +154,7 @@ def tokens_match(self, query_token: str, content_token: str) -> bool:
         # Same prefix-only matching as PlainAnalyzer
         if query_token == content_token:
             return True
-        if len(query_token) >= 3 and content_token.startswith(query_token):
-            return True
-        return False
+        return len(query_token) >= 3 and content_token.startswith(query_token)
 
 
 def get_analyzer(kind: AnalyzerKind) -> Analyzer:
@@ -467,16 +463,16 @@ def _row_to_record(row: sqlite3.Row) -> MemoryRecord:
             metadata=json.loads(row["metadata_json"]),
         )
 
-    @staticmethod
-    def _to_fts_query(query: str) -> str:
+    def _to_fts_query(self, query: str) -> str:
         """Convert arbitrary user text to a safe OR query for SQLite FTS5.
 
         BM25 ranks records matching more query terms above partial matches. OR prevents
         a harmless wording variation (for example, ``prefer`` vs ``prefers``) from
-        producing an empty result set before semantic retrieval is introduced.
+        producing an empty result set before semantic retrieval is introduced. Tokens
+        come from the configured analyzer so that its expansions reach candidate
+        selection rather than only the precision filter.
         """
-        # This is a static method - can't use instance analyzer. Uses plain tokens.
-        tokens = SQLiteMemoryRepository._expanded_tokens_static(query)
+        tokens = self._expanded_tokens(query)
         terms = []
         for token in tokens:
             terms.append(f'"{token}"')
@@ -487,15 +483,6 @@ def _to_fts_query(query: str) -> str:
                 terms.append(f"{token}*")
         return " OR ".join(terms)
 
-    @staticmethod
-    def _expanded_tokens_static(text: str) -> set[str]:
-        """Static version for FTS query building (uses plain tokens)."""
-        return {
-            token
-            for token in (SQLiteMemoryRepository._normalise_token(raw_token) for raw_token in re.findall(r"[\w]+", text, flags=re.UNICODE))
-            if token and len(token) > 2
-        }
-
     def _passes_precision_filter(
         self, query: str, query_tokens: set[str], content: str, *, enforce_current_filter: bool = True
     ) -> bool:
@@ -514,11 +501,21 @@ def _overlap_score(self, query_tokens: set[str], content: str) -> float:
         return len(self._matching_tokens(query_tokens, content)) / len(query_tokens)
 
     def _matching_tokens(self, query_tokens: set[str], content: str) -> set[str]:
+        """Return the query tokens the content satisfies, directly or by expansion.
+
+        An expansion match is credited to the query token it came from, so a
+        synonym-only record still counts as one match out of the query's own
+        tokens instead of diluting the overlap ratio.
+        """
         content_tokens = self._content_tokens(content)
         return {
             query_token
             for query_token in query_tokens
-            if any(self._analyzer.tokens_match(query_token, content_token) for content_token in content_tokens)
+            if any(
+                self._analyzer.tokens_match(candidate, content_token)
+                for candidate in self._analyzer.expand_tokens({query_token})
+                for content_token in content_tokens
+            )
         }
 
     def _important_tokens(self, text: str) -> set[str]:
@@ -569,9 +566,9 @@ def _utc_isoformat(value: datetime) -> str:
         Timestamps are stored with a fixed +00:00 offset because recall filters
         expiry inside SQL using string comparison; mixed offsets would break it.
         """
-        if value.tzinfo is not None:
-            value = value.astimezone(UTC)
-        return value.isoformat()
+        if value.tzinfo is None:
+            return value.replace(tzinfo=UTC).isoformat()
+        return value.astimezone(UTC).isoformat()
 
     @staticmethod
     def _encode_cursor(record: MemoryRecord) -> str:
diff --git a/services/dma-api/tests/test_configuration.py b/services/dma-api/tests/test_configuration.py
index 1470550..fe6846e 100644
--- a/services/dma-api/tests/test_configuration.py
+++ b/services/dma-api/tests/test_configuration.py
@@ -22,3 +22,23 @@ def test_health_endpoint_is_unauthenticated(tmp_path) -> None:
 
     assert response.status_code == 200
     assert response.json() == {"status": "ok"}
+
+
+def test_invalid_auth_limit_env_values_are_rejected(monkeypatch) -> None:
+    monkeypatch.setenv("DMA_AUTH_LOCKOUT_SECONDS", "0")
+
+    with pytest.raises(ValueError, match="DMA_AUTH_LOCKOUT_SECONDS"):
+        Settings.from_env()
+
+
+def test_non_numeric_auth_limit_env_values_are_rejected(monkeypatch) -> None:
+    monkeypatch.setenv("DMA_AUTH_MAX_ATTEMPTS", "many")
+
+    with pytest.raises(ValueError, match="DMA_AUTH_MAX_ATTEMPTS"):
+        Settings.from_env()
+
+
+def test_forwarded_for_is_untrusted_by_default(monkeypatch) -> None:
+    monkeypatch.delenv("DMA_TRUST_FORWARDED_FOR", raising=False)
+
+    assert Settings.from_env().trust_forwarded_for is False
diff --git a/services/dma-api/tests/test_rate_limiting.py b/services/dma-api/tests/test_rate_limiting.py
index b289eb8..aae7b75 100644
--- a/services/dma-api/tests/test_rate_limiting.py
+++ b/services/dma-api/tests/test_rate_limiting.py
@@ -3,12 +3,22 @@
 import time
 from unittest.mock import patch
 
+import pytest
 from fastapi.testclient import TestClient
 
 from dma_api.config import AuthLimits, Settings
 from dma_api.main import InMemoryRateLimiter, create_app
 
 
+def _settings(tmp_path, limits: AuthLimits, *, trust_forwarded_for: bool = True) -> Settings:
+    return Settings(
+        database_path=tmp_path / "dma.db",
+        api_key="test-key",
+        auth_limits=limits,
+        trust_forwarded_for=trust_forwarded_for,
+    )
+
+
 def _bad_auth_headers(source: str = "10.0.0.1") -> dict[str, str]:
     return {
         "Authorization": "Bearer wrong-key",
@@ -31,9 +41,7 @@ def _json() -> dict[str, str]:
 
 def test_lockout_after_max_failed_attempts(tmp_path) -> None:
     limits = AuthLimits(max_attempts=3, window_seconds=60, lockout_seconds=300)
-    app = create_app(Settings(
-        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
-    ))
+    app = create_app(_settings(tmp_path, limits))
     with TestClient(app) as client:
         for _ in range(3):
             resp = client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
@@ -46,9 +54,7 @@ def test_lockout_after_max_failed_attempts(tmp_path) -> None:
 
 def test_lockout_blocks_valid_key_from_same_source(tmp_path) -> None:
     limits = AuthLimits(max_attempts=2, window_seconds=60, lockout_seconds=300)
-    app = create_app(Settings(
-        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
-    ))
+    app = create_app(_settings(tmp_path, limits))
     with TestClient(app) as client:
         for _ in range(2):
             client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
@@ -59,9 +65,7 @@ def test_lockout_blocks_valid_key_from_same_source(tmp_path) -> None:
 
 def test_lockout_does_not_affect_other_sources(tmp_path) -> None:
     limits = AuthLimits(max_attempts=2, window_seconds=60, lockout_seconds=300)
-    app = create_app(Settings(
-        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
-    ))
+    app = create_app(_settings(tmp_path, limits))
     with TestClient(app) as client:
         for _ in range(2):
             client.post("/v1/memories", headers=_bad_auth_headers("10.0.0.1"), json=_json())
@@ -72,9 +76,7 @@ def test_lockout_does_not_affect_other_sources(tmp_path) -> None:
 
 def test_successful_auth_resets_failure_count(tmp_path) -> None:
     limits = AuthLimits(max_attempts=3, window_seconds=60, lockout_seconds=300)
-    app = create_app(Settings(
-        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
-    ))
+    app = create_app(_settings(tmp_path, limits))
     with TestClient(app) as client:
         for _ in range(2):
             client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
@@ -127,9 +129,7 @@ def shifted_time():
 
 def test_healthz_not_rate_limited(tmp_path) -> None:
     limits = AuthLimits(max_attempts=1, window_seconds=60, lockout_seconds=300)
-    app = create_app(Settings(
-        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
-    ))
+    app = create_app(_settings(tmp_path, limits))
     with TestClient(app) as client:
         client.post("/v1/memories", headers=_bad_auth_headers(), json=_json())
 
@@ -139,12 +139,70 @@ def test_healthz_not_rate_limited(tmp_path) -> None:
 
 def test_auth_failure_logged(tmp_path, caplog) -> None:
     limits = AuthLimits(max_attempts=5, window_seconds=60, lockout_seconds=300)
-    app = create_app(Settings(
-        database_path=tmp_path / "dma.db", api_key="test-key", auth_limits=limits,
-    ))
+    app = create_app(_settings(tmp_path, limits))
     import logging
-    with caplog.at_level(logging.WARNING, logger="dma_api.auth"):
-        with TestClient(app) as client:
-            client.post("/v1/memories", headers=_bad_auth_headers("192.168.1.1"), json=_json())
+    with caplog.at_level(logging.WARNING, logger="dma_api.auth"), TestClient(app) as client:
+        client.post("/v1/memories", headers=_bad_auth_headers("192.168.1.1"), json=_json())
 
     assert any("auth_failure" in record.message and "192.168.1.1" in record.message for record in caplog.records)
+
+
+def test_forwarded_for_is_ignored_when_proxy_is_untrusted(tmp_path) -> None:
+    limits = AuthLimits(max_attempts=2, window_seconds=60, lockout_seconds=300)
+    app = create_app(_settings(tmp_path, limits, trust_forwarded_for=False))
+    with TestClient(app) as client:
+        for index in range(2):
+            client.post("/v1/memories", headers=_bad_auth_headers(f"10.0.0.{index}"), json=_json())
+
+        resp = client.post("/v1/memories", headers=_good_auth_headers("10.0.0.9"), json=_json())
+        assert resp.status_code == 429
+
+
+def test_stale_sources_are_evicted() -> None:
+    limits = AuthLimits(max_attempts=5, window_seconds=5, lockout_seconds=10)
+    limiter = InMemoryRateLimiter(limits)
+
+    for index in range(50):
+        limiter.record_failure(f"10.0.0.{index}")
+    assert limiter.tracked_sources() == 50
+
+    original_monotonic = time.monotonic
+    with patch("dma_api.main.time") as mock_time:
+        mock_time.monotonic = lambda: original_monotonic() + 60
+        limiter.record_failure("10.1.0.1")
+
+    assert limiter.tracked_sources() == 1
+
+
+def test_locked_out_sources_survive_eviction() -> None:
+    limits = AuthLimits(max_attempts=2, window_seconds=5, lockout_seconds=300)
+    limiter = InMemoryRateLimiter(limits)
+
+    limiter.record_failure("attacker")
+    limiter.record_failure("attacker")
+    assert limiter.is_locked_out("attacker")
+
+    original_monotonic = time.monotonic
+    with patch("dma_api.main.time") as mock_time:
+        mock_time.monotonic = lambda: original_monotonic() + 60
+        limiter.record_failure("someone-else")
+        assert limiter.is_locked_out("attacker")
+
+
+def test_tracked_sources_stay_within_configured_maximum() -> None:
+    limits = AuthLimits(
+        max_attempts=5, window_seconds=600, lockout_seconds=300, max_tracked_sources=10
+    )
+    limiter = InMemoryRateLimiter(limits)
+
+    for index in range(100):
+        limiter.record_failure(f"10.0.0.{index}")
+
+    assert limiter.tracked_sources() <= 10
+
+
+def test_non_positive_auth_limits_are_rejected() -> None:
+    with pytest.raises(ValueError, match="lockout_seconds"):
+        AuthLimits(max_attempts=5, window_seconds=60, lockout_seconds=0)
+    with pytest.raises(ValueError, match="max_attempts"):
+        AuthLimits(max_attempts=0, window_seconds=60, lockout_seconds=300)
diff --git a/services/dma-api/tests/test_recall.py b/services/dma-api/tests/test_recall.py
index c916a39..734d02f 100644
--- a/services/dma-api/tests/test_recall.py
+++ b/services/dma-api/tests/test_recall.py
@@ -7,7 +7,7 @@
 from dma_api.config import Settings
 from dma_api.main import create_app
 from dma_api.models import MemoryType
-from dma_api.repository import MemoryRecord, SQLiteMemoryRepository
+from dma_api.repository import AnalyzerKind, MemoryRecord, SQLiteMemoryRepository
 
 
 def _headers(key: str) -> dict[str, str]:
@@ -95,9 +95,7 @@ def test_recall_includes_memories_expiring_after_now_regardless_of_offset(tmp_pa
         now=now,
     )
 
-    assert response.status_code == 200
-    contents = [item["content"] for item in response.json()["results"]]
-    assert contents == ["User now prefers Java Spring Boot for backend APIs."]
+    assert [record.id for record, _ in matches] == ["mem_offsetexpiry00000000000001"]
 
 
 def test_recall_prefix_only_matching_no_false_positives(tmp_path) -> None:
@@ -160,3 +158,58 @@ def test_recall_prefix_only_matching_no_false_positives(tmp_path) -> None:
     # Both should match since 'cat' is a prefix of 'category'
     assert "The cat sits on the mat." in contents
     assert "The category system organizes items." in contents
+
+
+def test_domain_analyzer_recalls_expansion_only_content(tmp_path) -> None:
+    """A domain synonym must reach candidate selection, not just scoring."""
+    app = create_app(
+        Settings(
+            database_path=tmp_path / "dma.db",
+            api_key="test-key",
+            tenant_id="tenant-a",
+            analyzer_kind=AnalyzerKind.DOMAIN,
+        )
+    )
+    with TestClient(app) as client:
+        _remember(client, "Django powers the reporting service.", "semantic", "key-000000000010")
+        response = client.post(
+            "/v1/memories/recall",
+            headers={"Authorization": "Bearer test-key"},
+            json={"agent_id": "coding-agent", "query": "framework", "limit": 5},
+        )
+
+    assert response.status_code == 200
+    contents = [item["content"] for item in response.json()["results"]]
+    assert "Django powers the reporting service." in contents
+
+
+def test_naive_expiry_is_treated_as_utc(tmp_path) -> None:
+    repository = SQLiteMemoryRepository(tmp_path / "dma.db")
+    repository.initialize()
+    now = datetime(2026, 8, 27, 20, 0, 0, tzinfo=UTC)
+    repository.create_or_get(
+        MemoryRecord(
+            id="mem_naiveexpiry000000000000001",
+            tenant_id="tenant-a",
+            agent_id="coding-agent",
+            content="Staging cluster deployment notes.",
+            type=MemoryType.EPISODIC,
+            version=1,
+            created_at=now,
+            updated_at=now,
+            expires_at=datetime(2026, 8, 27, 21, 0, 0),  # noqa: DTZ001 - naive on purpose
+            metadata={},
+        ),
+        "idempotency-naive-expiry-0001",
+    )
+
+    matches = repository.recall(
+        tenant_id="tenant-a",
+        agent_id="coding-agent",
+        query="staging cluster deployment",
+        types=None,
+        limit=5,
+        now=now,
+    )
+
+    assert [record.id for record, _ in matches] == ["mem_naiveexpiry000000000000001"]

From f8d5950ac191fa546c715da6044fbea0bec46e1a Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Mon, 31 Aug 2026 18:27:59 +0000
Subject: [PATCH 5/6] ci: build the Docker deployment with the default builder

The 'cloud' builder is not configured on GitHub-hosted runners, so the container validation step failed with 'no builder "cloud" found'.

Co-Authored-By: krishna <87197325+krishna3554@users.noreply.github.com>
---
 .github/workflows/ci.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 82aa142..5b69850 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -46,7 +46,7 @@ jobs:
         env:
           DMA_API_KEY: ci-container-validation-key
         run: |
-          docker compose build --builder cloud
+          docker compose build
           docker compose up -d
           for attempt in {1..15}; do
             if curl --fail --silent http://127.0.0.1:8000/healthz; then break; fi

From 82dabab649abbdd252fe0d561f36b2567a2d41d8 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Mon, 31 Aug 2026 18:30:01 +0000
Subject: [PATCH 6/6] fix: install locked dependencies in the container image

uv pip install has no --locked flag, so the image build failed; export the lock to a requirements file and install from it.

Co-Authored-By: krishna <87197325+krishna3554@users.noreply.github.com>
---
 Dockerfile | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/Dockerfile b/Dockerfile
index e3086f5..9aa9550 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -9,7 +9,10 @@ WORKDIR /app
 COPY services/dma-api/pyproject.toml services/dma-api/uv.lock ./
 COPY services/dma-api/src ./src
 
-RUN pip install --no-cache-dir uv && uv pip install --locked --no-cache .
+RUN pip install --no-cache-dir uv \
+    && uv export --frozen --no-dev --no-emit-project --output-file requirements.txt \
+    && uv pip install --system --no-cache -r requirements.txt . \
+    && rm requirements.txt
 
 RUN useradd --create-home --uid 10001 dma \
     && mkdir /data \