fix(memory): audit passive context injection (#2212)

## Description

Close the passive-memory observability loop by recording access for
context rows that survive the final injection budget and tagging
requests where context is actually appended.

Closes #2211

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Track only memory IDs retained after ranking, similarity filtering,
entry limits, and final text truncation.
- Call optional backend `record_access` with stable de-duplication and
fail-open error handling.
- Extend structured injection logging to stamp `memory_injected=true`
when injected bytes are positive.
- Thread request tags through successful Anthropic, OpenAI Chat, OpenAI
Responses, Gemini, and Codex WebSocket injection sites.
- Add a static contract test that all current successful handler
injection logs pass tags.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
$ uv run --with pytest --with pytest-asyncio --with numpy --with fastapi pytest \
    tests/test_memory_handler_native_ops.py \
    tests/test_memory_auto_tail.py \
    tests/test_memory_handler_project_isolation.py \
    tests/test_memory_injection_logging.py -q
51 passed

$ uv run --with ruff ruff check <touched Python files>
All checks passed!

$ git diff --check
(no output)
```

## Real Behavior Proof

- Environment: Python 3.13 with synthetic backend and handler fixtures
- Exact command / steps: run the focused test set above
- Observed result: only IDs present after the final text budget are
access-recorded; access-write failures remain fail-open; positive
injection logs stamp `memory_injected=true`; all six current successful
injection call sites pass tags
- Not tested: live provider requests, full repository suite, third-party
backends without `record_access`

## Review Readiness

- [x] I have performed a self-review
- [ ] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A

## Additional Notes

Access accounting is intentionally best-effort: unsupported backends and
write failures do not delay or fail the upstream model request.
Documentation and changelog changes are not needed for this internal
observability fix.

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>

headroom/proxy/handlers/anthropic.py

@@ -2092,6 +2092,7 @@ class AnthropicHandlerMixin:
                                         decision="injected_live_zone_tail",
                                         bytes_injected=len(memory_context),
                                         query=user_query,
+                                        tags=tags,
                                     )
                                 else:
                                     log_memory_injection(

headroom/proxy/handlers/gemini.py

@@ -336,7 +336,7 @@ class GeminiHandlerMixin:
         # Pre-PR-this Gemini's memory site silently ignored
         # `x-headroom-bypass: true`, mutating request bytes under the
         # user's "don't touch my bytes" signal.
-        from headroom.proxy.helpers import get_memory_injection_mode
+        from headroom.proxy.helpers import get_memory_injection_mode, log_memory_injection
         from headroom.proxy.memory_decision import MemoryDecision
         from headroom.proxy.memory_query import MemoryQuery
 
@@ -574,6 +574,14 @@ class GeminiHandlerMixin:
                                 f"[{request_id}] Memory: Injected {bytes_appended} chars "
                                 f"into latest user message tail for user {memory_user_id} (gemini)"
                             )
+                            log_memory_injection(
+                                request_id=request_id,
+                                session_id=None,
+                                decision="injected_live_zone_tail_gemini",
+                                bytes_injected=bytes_appended,
+                                query=None,
+                                tags=tags,
+                            )
                         else:
                             logger.debug(
                                 f"[{request_id}] Memory: no eligible user message; "

headroom/proxy/handlers/openai.py

@@ -3062,6 +3062,7 @@ class OpenAIHandlerMixin:
                                     decision="injected_live_zone_tail_chat",
                                     bytes_injected=bytes_appended,
                                     query=None,
+                                    tags=tags,
                                 )
                                 logger.info(
                                     f"[{request_id}] Memory: Injected {bytes_appended} chars "
@@ -4206,6 +4207,7 @@ class OpenAIHandlerMixin:
                                     decision="injected_live_zone_tail_string",
                                     bytes_injected=len(memory_context),
                                     query=user_query,
+                                    tags=tags,
                                 )
                             elif isinstance(current_input, list):
                                 new_input, bytes_appended = append_text_to_latest_user_input_item(
@@ -4220,6 +4222,7 @@ class OpenAIHandlerMixin:
                                         decision="injected_live_zone_tail",
                                         bytes_injected=bytes_appended,
                                         query=user_query,
+                                        tags=tags,
                                     )
                                 else:
                                     log_memory_injection(
@@ -5544,7 +5547,7 @@ class OpenAIHandlerMixin:
 
             memory_user_id: str | None = None
             memory_request_ctx = None
-            from headroom.proxy.helpers import get_memory_injection_mode
+            from headroom.proxy.helpers import get_memory_injection_mode, log_memory_injection
             from headroom.proxy.memory_decision import MemoryDecision
             from headroom.proxy.memory_query import MemoryQuery
 
@@ -5659,6 +5662,14 @@ class OpenAIHandlerMixin:
                                     f"[{request_id}] WS Memory: Injected {len(memory_context)} chars "
                                     f"into input tail (string-shaped input)"
                                 )
+                                log_memory_injection(
+                                    request_id=request_id,
+                                    session_id=session_id,
+                                    decision="injected_live_zone_tail_ws",
+                                    bytes_injected=len(memory_context),
+                                    query=None,
+                                    tags=ws_tags,
+                                )
                             else:
                                 # List-shaped WS input is owned by the
                                 # Rust handler (per PR-C5 comment). The

headroom/proxy/helpers.py

@@ -353,6 +353,7 @@ def log_memory_injection(
     decision: str,
     bytes_injected: int,
     query: str | None = None,
+    tags: dict[str, str] | None = None,
 ) -> None:
     """Emit a structured log line for every memory-context routing decision.
 
@@ -360,6 +361,8 @@ def log_memory_injection(
     Never log raw query content or Authorization header — only a stable
     hash of the query.
     """
+    if tags is not None and bytes_injected > 0:
+        tags["memory_injected"] = "true"
     query_hash = hash_query_for_log(query) if query else ""
     logger.info(
         "event=memory_injection request_id=%s session_id=%s decision=%s "

headroom/proxy/memory_handler.py

@@ -821,6 +821,7 @@ class MemoryHandler:
             # Both branches below render the same `i. [id] content` shape
             # so the format is stable regardless of whether a ranker is
             # in play.
+            selected_memory_ids: list[str] = []
             if ranker is not None:
                 from headroom.proxy.memory_ranker import MemoryCandidate
 
@@ -839,6 +840,8 @@ class MemoryHandler:
                 memory_lines = []
                 for i, candidate in enumerate(ranked, 1):
                     memory_id = candidate.id or "?"
+                    if candidate.id:
+                        selected_memory_ids.append(candidate.id)
                     memory_lines.append(f"{i}. [{memory_id}] {candidate.content}")
                     if candidate.related_entities:
                         entities_str = ", ".join(candidate.related_entities[:3])
@@ -864,6 +867,8 @@ class MemoryHandler:
                 memory_lines = []
                 for i, result in enumerate(filtered_results, 1):
                     memory_id = getattr(result.memory, "id", None) or "?"
+                    if memory_id != "?":
+                        selected_memory_ids.append(memory_id)
                     memory_lines.append(f"{i}. [{memory_id}] {result.memory.content}")
                     if hasattr(result, "related_entities") and result.related_entities:
                         entities_str = ", ".join(result.related_entities[:3])
@@ -909,6 +914,25 @@ your responses, not to drive new actions."""
         # the input query (which stays full-fidelity per MemoryQuery).
         context = effective_budget.apply_to_text(context)
 
+        # Track only memories that survived ranking, entry limits, and the
+        # final text budget. Backends without access tracking keep working,
+        # and an audit write failure must never block the upstream request.
+        accessed_memory_ids = list(
+            dict.fromkeys(
+                memory_id for memory_id in selected_memory_ids if f"[{memory_id}]" in context
+            )
+        )
+        record_access = getattr(backend, "record_access", None)
+        if accessed_memory_ids and callable(record_access):
+            try:
+                await record_access(accessed_memory_ids)
+            except Exception as e:
+                logger.debug(
+                    "Memory: Failed to record passive retrieval access for %d memories: %s",
+                    len(accessed_memory_ids),
+                    e,
+                )
+
         logger.info(
             "event=memory_inject user=%s scope=%s count=%d chars=%d budget_tokens=%d",
             effective_user_id,

tests/test_memory_handler_native_ops.py

@@ -30,6 +30,7 @@ class FakeBackend:
         self.saved: list[dict[str, object]] = []
         self.updated: list[dict[str, object]] = []
         self.deleted: list[str] = []
+        self.accessed: list[list[str]] = []
         self.raise_on: str | None = None
 
     async def search_memories(self, **kwargs):  # noqa: ANN003
@@ -55,6 +56,12 @@ class FakeBackend:
         self.deleted.append(memory_id)
         return True
 
+    async def record_access(self, memory_ids: list[str]) -> int:
+        if self.raise_on == "record_access":
+            raise RuntimeError("access tracking failed")
+        self.accessed.append(memory_ids)
+        return len(memory_ids)
+
 
 def make_result(
     memory_id: str,
@@ -967,6 +974,7 @@ async def test_search_and_format_context_and_handle_memory_tool_calls(
     # tripping through memory_search.
     assert "1. [m1] Alice likes pizza" in context
     assert "(Related: Alice, pizza)" in context
+    assert backend.accessed == [["m1"]]
 
     backend.raise_on = "search"
     assert (
@@ -975,6 +983,14 @@ async def test_search_and_format_context_and_handle_memory_tool_calls(
     )
     backend.raise_on = None
 
+    backend.raise_on = "record_access"
+    context = await handler.search_and_format_context(
+        "u1",
+        [{"role": "user", "content": "What food does Alice like?"}],
+    )
+    assert "1. [m1] Alice likes pizza" in context
+    backend.raise_on = None
+
     async def fake_ensure_initialized() -> None:
         return None
 
@@ -1062,6 +1078,38 @@ async def test_search_and_format_context_and_handle_memory_tool_calls(
     assert skipped == []
 
 
+@pytest.mark.asyncio
+async def test_search_records_only_memories_left_by_final_text_budget(
+    handler: MemoryHandler,
+) -> None:
+    backend = FakeBackend()
+    handler._backend = backend
+    handler._initialized = True
+    backend.search_results = [
+        make_result("m1", "First preference"),
+        make_result("m2", "Second preference"),
+    ]
+
+    class FirstEntryBudget:
+        max_entries = 2
+        min_similarity = 0.3
+        max_tokens = 1024
+
+        @staticmethod
+        def apply_to_text(text: str) -> str:
+            return text.split("2. [m2]", maxsplit=1)[0]
+
+    context = await handler.search_and_format_context(
+        "u1",
+        [{"role": "user", "content": "What are my preferences?"}],
+        budget=FirstEntryBudget(),
+    )
+
+    assert "[m1]" in context
+    assert "[m2]" not in context
+    assert backend.accessed == [["m1"]]
+
+
 @pytest.mark.asyncio
 async def test_ensure_initialized_timeout_and_cancellation(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path

tests/test_memory_injection_logging.py

@@ -0,0 +1,59 @@
+"""Tests for memory-injection request tags."""
+
+import ast
+from pathlib import Path
+
+from headroom.proxy.helpers import log_memory_injection
+
+HANDLER_FILES = [
+    Path("headroom/proxy/handlers/anthropic.py"),
+    Path("headroom/proxy/handlers/openai.py"),
+    Path("headroom/proxy/handlers/gemini.py"),
+]
+
+
+def test_log_memory_injection_marks_only_successful_injection() -> None:
+    tags: dict[str, str] = {}
+
+    log_memory_injection(
+        request_id="hr_test_memory",
+        session_id=None,
+        decision="no_eligible_user_turn",
+        bytes_injected=0,
+        tags=tags,
+    )
+    assert "memory_injected" not in tags
+
+    log_memory_injection(
+        request_id="hr_test_memory",
+        session_id=None,
+        decision="injected_live_zone_tail",
+        bytes_injected=42,
+        tags=tags,
+    )
+    assert tags["memory_injected"] == "true"
+
+
+def test_successful_handler_injection_logs_pass_tags() -> None:
+    missing: list[tuple[Path, int]] = []
+    successful_sites = 0
+
+    for file_path in HANDLER_FILES:
+        tree = ast.parse(file_path.read_text(encoding="utf-8"), filename=str(file_path))
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.Call):
+                continue
+            if not isinstance(node.func, ast.Name) or node.func.id != "log_memory_injection":
+                continue
+            kwargs = {kw.arg: kw.value for kw in node.keywords if kw.arg is not None}
+            bytes_injected = kwargs.get("bytes_injected")
+            if isinstance(bytes_injected, ast.Constant) and bytes_injected.value == 0:
+                continue
+            successful_sites += 1
+            if "tags" not in kwargs:
+                missing.append((file_path, node.lineno))
+
+    assert successful_sites >= 6, "Expected all current provider injection sites"
+    assert not missing, "Successful memory injections missing tags: " + ", ".join(
+        f"{path}:{line}" for path, line in missing
+    )