diff --git a/headroom/memory/storage_router.py b/headroom/memory/storage_router.py index 45f980584..0572a3bd3 100644 --- a/headroom/memory/storage_router.py +++ b/headroom/memory/storage_router.py @@ -90,7 +90,7 @@ class ResolvedScope: """ mode: MemoryStorageMode - db_path: Path + db_path: Path | None display_name: str # human-readable label, e.g. project basename project_key: str | None # stable hash, None for USER/GLOBAL @@ -282,13 +282,19 @@ class BackendRouter: self._backends: OrderedDict[Path, LocalBackend] = OrderedDict() self._lock = threading.Lock() - def backend_for(self, ctx: RequestContext) -> tuple[LocalBackend, ResolvedScope]: + def backend_for(self, ctx: RequestContext) -> tuple[LocalBackend | None, ResolvedScope]: """Return the backend + scope metadata to use for this request.""" scope = self._resolve_scope(ctx) + if scope.db_path is None: + return None, scope backend = self._get_or_create_backend(scope.db_path) return backend, scope + def scope_for(self, ctx: RequestContext) -> ResolvedScope: + """Resolve request scope without acquiring or creating a backend.""" + return self._resolve_scope(ctx) + def _resolve_scope(self, ctx: RequestContext) -> ResolvedScope: mode = self._config.mode @@ -340,7 +346,7 @@ class BackendRouter: ) return ResolvedScope( mode=MemoryStorageMode.PROJECT, - db_path=self._config.global_db_path, # Unused — caller checks project_key. + db_path=None, display_name="unresolved (no memory)", project_key=None, ) diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index 5f5e4a510..2782e6847 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/headroom/proxy/handlers/anthropic.py @@ -2509,29 +2509,34 @@ class AnthropicHandlerMixin: # Traffic Learner: Extract patterns from inbound tool results if self.traffic_learner: try: - # Wire backend on first use (lazy init after memory handler is ready) - if ( - self.traffic_learner._backend is None - and self.memory_handler - and self.memory_handler.initialized - and self.memory_handler.backend + if self.memory_handler and self.memory_handler.is_project_unresolved( + memory_request_ctx ): - self.traffic_learner.set_backend(self.memory_handler.backend) + logger.info(f"[{request_id}] Traffic learner skipped: project_unresolved") + else: + # Wire backend on first use (lazy init after memory handler is ready) + if ( + self.traffic_learner._backend is None + and self.memory_handler + and self.memory_handler.initialized + and self.memory_handler.backend + ): + self.traffic_learner.set_backend(self.memory_handler.backend) - # Extract tool results from messages and learn from them - tool_results = self.traffic_learner.extract_tool_results_from_messages( - optimized_messages - ) - for tr in tool_results[-5:]: # Only recent results - await self.traffic_learner.on_tool_result( - tool_name=tr["tool_name"], - tool_input=tr["input"], - tool_output=tr["output"], - is_error=tr["is_error"], + # Extract tool results from messages and learn from them + tool_results = self.traffic_learner.extract_tool_results_from_messages( + optimized_messages ) + for tr in tool_results[-5:]: # Only recent results + await self.traffic_learner.on_tool_result( + tool_name=tr["tool_name"], + tool_input=tr["input"], + tool_output=tr["output"], + is_error=tr["is_error"], + ) - # Also extract preference signals from user messages - await self.traffic_learner.on_messages(optimized_messages) + # Also extract preference signals from user messages + await self.traffic_learner.on_messages(optimized_messages) except Exception as e: logger.debug(f"[{request_id}] Traffic learner: {e}") diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 42ac1a8c0..e7ae1e893 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -1753,6 +1753,7 @@ class OpenAIHandlerMixin: body: dict[str, Any], *, request_id: str, + request_context: Any = None, ) -> None: """Feed one Responses HTTP request into the live traffic learner.""" traffic_learner = getattr(self, "traffic_learner", None) @@ -1760,6 +1761,9 @@ class OpenAIHandlerMixin: return try: memory_handler = getattr(self, "memory_handler", None) + if memory_handler and memory_handler.is_project_unresolved(request_context): + logger.info("[%s] Traffic learner skipped: project_unresolved", request_id) + return if ( traffic_learner._backend is None and memory_handler @@ -1789,6 +1793,7 @@ class OpenAIHandlerMixin: messages: list[dict[str, Any]], *, request_id: str, + request_context: Any = None, ) -> None: """Feed one chat/completions request into the live traffic learner. @@ -1805,6 +1810,9 @@ class OpenAIHandlerMixin: return try: memory_handler = getattr(self, "memory_handler", None) + if memory_handler and memory_handler.is_project_unresolved(request_context): + logger.info("[%s] Traffic learner skipped: project_unresolved", request_id) + return if ( traffic_learner._backend is None and memory_handler @@ -1832,6 +1840,7 @@ class OpenAIHandlerMixin: seen_call_ids: set[str], baseline: bool, request_id: str, + request_context: Any = None, ) -> None: """Feed one Codex WS ``response.create`` turn into the traffic learner. @@ -1855,6 +1864,9 @@ class OpenAIHandlerMixin: return try: memory_handler = getattr(self, "memory_handler", None) + if memory_handler and memory_handler.is_project_unresolved(request_context): + logger.info("[%s] Traffic learner skipped: project_unresolved", request_id) + return if ( traffic_learner._backend is None and memory_handler @@ -3341,12 +3353,6 @@ class OpenAIHandlerMixin: stream = body.get("stream", False) - # Learn from the original client payload before memory context or - # compression mutates it, mirroring the Responses and Anthropic - # ingestion paths. Without this, chat/completions traffic (Copilot CLI, - # opencode, OpenAI SDKs) fed nothing to the learner (part of #2060). - await self._observe_openai_chat_traffic(original_client_messages, request_id=request_id) - # Bypass: skip ALL compression for explicit opt-out _bypass = self._headroom_bypass_enabled(request.headers) if _bypass: @@ -3480,6 +3486,12 @@ class OpenAIHandlerMixin: ), ) + await self._observe_openai_chat_traffic( + original_client_messages, + request_id=request_id, + request_context=memory_request_ctx, + ) + # Canonical memory-injection gate (parallels Anthropic). Pre- # PR-this the inline conjunction at the memory site silently # ignored `x-headroom-bypass: true`, mutating request bytes @@ -5459,10 +5471,6 @@ class OpenAIHandlerMixin: bind_scope(tags, request.scope) client = classify_client(headers) - # Learn from the original client payload before memory context or - # compression mutates it. This mirrors the Anthropic ingestion path. - await self._observe_openai_responses_traffic(body, request_id=request_id) - # PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound # headers AFTER `_extract_tags` reads them. Memory user-id reads # `request.headers` below. @@ -5564,6 +5572,12 @@ class OpenAIHandlerMixin: ), ) + await self._observe_openai_responses_traffic( + body, + request_id=request_id, + request_context=memory_request_ctx, + ) + # Rate limiting if self.rate_limiter: rate_key = headers.get("authorization", "default")[:20] @@ -6373,13 +6387,13 @@ class OpenAIHandlerMixin: except (json.JSONDecodeError, TypeError): args = {} - await self.memory_handler._ensure_initialized() - if self.memory_handler._backend: - result = await self.memory_handler._execute_memory_tool( - name, args, memory_user_id, "openai" - ) - else: - result = json.dumps({"error": "Memory backend not initialized"}) + result = await self.memory_handler._execute_memory_tool( + name, + args, + memory_user_id, + "openai", + request_context=memory_request_ctx, + ) tool_outputs.append( { @@ -7184,6 +7198,39 @@ class OpenAIHandlerMixin: ws_recorded_tokens_saved_total = 0 ws_recorded_attempted_input_tokens_total = 0 ws_response_create_frames = 1 + memory_user_id: str | None = None + memory_request_ctx = None + learner_memory_user_id: str | None = ( + ws_headers.get( + "x-headroom-user-id", + os.environ.get("USER", os.environ.get("USERNAME", "default")), + ) + if self.memory_handler + else None + ) + from headroom.memory.storage_router import RequestContext as _MemRequestContext + + def _ws_request_context( + frame_body: dict[str, Any], + base_user_id: str | None, + ) -> Any: + if not self.memory_handler or base_user_id is None: + return None + response_body = frame_body.get("response", frame_body) + if not isinstance(response_body, dict): + return None + return _MemRequestContext( + headers=dict(ws_headers), + system_prompt=str(response_body.get("instructions") or ""), + base_user_id=base_user_id, + project_root_override=( + getattr(self.memory_handler.config, "project_root_override", "") or None + ), + ) + + def _ws_learner_request_context(frame_body: dict[str, Any]) -> Any: + return _ws_request_context(frame_body, learner_memory_user_id) + # Per-connection traffic-learner dedup: tool-call ids already # observed on this WS, so a replayed transcript (each turn resends # the full history; reconnect replays it wholesale) is not counted @@ -7204,6 +7251,7 @@ class OpenAIHandlerMixin: seen_call_ids=ws_learner_seen_call_ids, baseline=True, request_id=request_id, + request_context=_ws_learner_request_context(_ws_first_inner), ) ws_client_frames_total = 1 ws_upstream_frames_total = 0 @@ -7316,8 +7364,6 @@ class OpenAIHandlerMixin: ), ) - memory_user_id: str | None = None - memory_request_ctx = None 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 @@ -7339,27 +7385,12 @@ class OpenAIHandlerMixin: return frame_raw memory_user_id = memory_user_id_candidate + memory_request_ctx = _ws_request_context(frame_body, memory_user_id) try: # Unwrap response.create envelope to access the response body ws_response_body = frame_body.get("response", frame_body) if not isinstance(ws_response_body, dict): return frame_raw - # Per-project memory routing (GH #462). For WS, - # ``ws_response_body`` carries ``instructions`` — - # that's the system-prompt-equivalent we feed to the - # resolver. - from headroom.memory.storage_router import ( - RequestContext as _MemRequestContext, - ) - - memory_request_ctx = _MemRequestContext( - headers=dict(ws_headers), - system_prompt=str(ws_response_body.get("instructions") or ""), - base_user_id=memory_user_id, - project_root_override=( - getattr(self.memory_handler.config, "project_root_override", "") or None - ), - ) # Debug: log what Codex sends so we can see the full tool list existing_tool_names = [ @@ -7914,6 +7945,7 @@ class OpenAIHandlerMixin: seen_call_ids=ws_learner_seen_call_ids, baseline=False, request_id=request_id, + request_context=_ws_learner_request_context(inner_payload), ) store_forced = _ensure_chatgpt_responses_store_false( inner_payload, @@ -8703,16 +8735,13 @@ class OpenAIHandlerMixin: except (json.JSONDecodeError, TypeError): fc_args = {} - await self.memory_handler._ensure_initialized() - if self.memory_handler._backend: - result = await self.memory_handler._execute_memory_tool( - fc_name, - fc_args, - memory_user_id, - "openai", - ) - else: - result = json.dumps({"error": "backend not ready"}) + result = await self.memory_handler._execute_memory_tool( + fc_name, + fc_args, + memory_user_id, + "openai", + request_context=memory_request_ctx, + ) tool_outputs.append( { diff --git a/headroom/proxy/memory_handler.py b/headroom/proxy/memory_handler.py index 10917396d..3238dbe28 100644 --- a/headroom/proxy/memory_handler.py +++ b/headroom/proxy/memory_handler.py @@ -184,12 +184,18 @@ class MemoryHandler: self.config = config self.agent_type = agent_type self._backend: LocalBackend | Any = None - # Per-project routing for the local backend. Built in - # ``_init_backend_locked`` so a single, shared resolver / LRU is - # kept on the handler. Qdrant deployments use composite user-id - # partitioning instead (see ``_compose_effective_user_id``) — the - # router stays None in that case. - self._router: BackendRouter | None = None + storage_root = ( + Path(config.storage_root) + if config.storage_root + else (Path(config.db_path).resolve().parent / "memories") + ) + self._router = BackendRouter( + BackendRouterConfig( + mode=config.storage_mode, + root_dir=storage_root, + global_db_path=Path(config.db_path).resolve(), + ) + ) self._initialized = False # Async singleflight guard for backend init. Ensures concurrent first # callers land on one init (double-checked pattern inside @@ -385,19 +391,8 @@ class MemoryHandler: # remains the GLOBAL-mode fallback / legacy compatibility # backend; callers that pass a ``RequestContext`` route # through ``self._router`` instead. - storage_root = ( - Path(self.config.storage_root) - if self.config.storage_root - else (Path(self.config.db_path).resolve().parent / "memories") - ) global_db_path = Path(self.config.db_path).resolve() - router_cfg = BackendRouterConfig( - mode=self.config.storage_mode, - root_dir=storage_root, - global_db_path=global_db_path, - backend_config_template=backend_config, - ) - self._router = BackendRouter(router_cfg) + self._router._config.backend_config_template = backend_config # Seed the router's LRU with the already-initialized # legacy backend so GLOBAL-mode requests reuse it instead # of opening a second handle to the same file. @@ -406,7 +401,7 @@ class MemoryHandler: logger.info( "event=memory_router_initialized mode=%s root=%s global_db=%s", self.config.storage_mode.value, - storage_root, + self._router._config.root_dir, global_db_path, ) @@ -645,7 +640,9 @@ class MemoryHandler: # Non-local backends: derive scope but keep one shared backend # and compose the user_id so the partition lives in the user_id # column instead of in a separate file. - scope = self._router._resolve_scope(request_context) + scope = self._router.scope_for(request_context) + if scope.db_path is None: + return None, scope, base_user_id composed = ( base_user_id if scope.project_key is None or scope.mode is MemoryStorageMode.GLOBAL @@ -653,6 +650,17 @@ class MemoryHandler: ) return self._backend, scope, composed + def is_project_unresolved(self, request_context: RequestContext | None) -> bool: + """Return whether fail-closed project routing has no usable target.""" + if request_context is None: + return False + scope = self._router.scope_for(request_context) + return ( + scope.mode is MemoryStorageMode.PROJECT + and scope.project_key is None + and scope.db_path is None + ) + @staticmethod def _format_memory_block_header(scope: ResolvedScope | None) -> str: """Workspace / scope provenance header for the injected memory block. @@ -738,32 +746,19 @@ class MemoryHandler: ) return None + if self.is_project_unresolved(request_context): + logger.info( + "event=memory_inject_skipped reason=project_unresolved user_id=%s", + user_id, + ) + return None + await self._ensure_initialized() if not self._backend: return None backend, scope, effective_user_id = self._resolve_for_request(user_id, request_context) - - # Fail-closed when the router was unable to resolve a project in - # PROJECT mode and `unresolved_project_fallback="empty"` (the - # default after the 2026-05-26 incident). The sentinel signal is - # `mode=PROJECT` + `project_key=None`: project mode was requested - # but no x-headroom-project-id / x-headroom-cwd / system-prompt - # cwd: was available, so we have no idea which project this - # request belongs to. Returning None here skips injection - # entirely — better than pooling into GLOBAL and surfacing - # memories from unrelated past sessions (the TAM-550 imperative- - # misread bug). - if ( - scope is not None - and scope.mode is MemoryStorageMode.PROJECT - and scope.project_key is None - ): - logger.info( - "event=memory_inject_skipped reason=project_unresolved user_id=%s scope_display=%s", - effective_user_id, - scope.display_name, - ) + if backend is None: return None # Build the embedding query. When the handler provides a @@ -1106,6 +1101,8 @@ your responses, not to drive new actions.""" """ tool_calls = self._extract_tool_calls(response, provider) results: list[dict[str, Any]] = [] + project_unresolved = self.is_project_unresolved(request_context) + unresolved_result = json.dumps({"status": "skipped", "reason": "project_unresolved"}) for tc in tool_calls: # `tc.get("function", {})` returns None for an explicit @@ -1130,12 +1127,19 @@ your responses, not to drive new actions.""" input_data = {} # Handle native memory tool - if tool_name == NATIVE_MEMORY_TOOL_NAME: - result_content = await self._execute_native_memory_tool(input_data, user_id) + if tool_name == NATIVE_MEMORY_TOOL_NAME and project_unresolved: + result_content = unresolved_result + elif tool_name in MEMORY_TOOL_NAMES and project_unresolved: + result_content = unresolved_result + elif tool_name == NATIVE_MEMORY_TOOL_NAME: + result_content = await self._execute_native_memory_tool( + input_data, + user_id, + request_context=request_context, + ) elif tool_name in MEMORY_TOOL_NAMES: - # Custom memory tools need backend await self._ensure_initialized() - if not self._backend: + if self._backend is None: continue result_content = await self._execute_memory_tool( tool_name, @@ -1179,7 +1183,12 @@ your responses, not to drive new actions.""" request_context: RequestContext | None = None, ) -> str: """Execute a memory tool and return result string.""" + if self.is_project_unresolved(request_context): + return json.dumps({"status": "skipped", "reason": "project_unresolved"}) try: + await self._ensure_initialized() + if self._backend is None: + return json.dumps({"status": "error", "error": "Memory backend not initialized"}) if tool_name == "memory_save": return await self._execute_save(input_data, user_id, provider, request_context) elif tool_name == "memory_search": @@ -1528,7 +1537,13 @@ your responses, not to drive new actions.""" # str_replace → Update memory content # ========================================================================= - async def _execute_native_memory_tool(self, input_data: dict[str, Any], user_id: str) -> str: + async def _execute_native_memory_tool( + self, + input_data: dict[str, Any], + user_id: str, + *, + request_context: RequestContext | None = None, + ) -> str: """Execute Anthropic's native memory tool with semantic backend. This is a TRANSLATION LAYER: Claude thinks it's doing file operations, @@ -1542,24 +1557,28 @@ your responses, not to drive new actions.""" - delete: Remove from vector store - rename: Update memory tags/path """ - # Ensure our semantic backend is initialized - await self._ensure_initialized() + if self.is_project_unresolved(request_context): + return json.dumps({"status": "skipped", "reason": "project_unresolved"}) command = input_data.get("command", "") try: + await self._ensure_initialized() + backend, _scope, effective_user_id = self._resolve_for_request(user_id, request_context) + if backend is None: + return json.dumps({"status": "skipped", "reason": "project_unresolved"}) if command == "view": - return await self._native_view_semantic(input_data, user_id) + return await self._native_view_semantic(input_data, effective_user_id, backend) elif command == "create": - return await self._native_create_semantic(input_data, user_id) + return await self._native_create_semantic(input_data, effective_user_id, backend) elif command == "str_replace": - return await self._native_update_semantic(input_data, user_id) + return await self._native_update_semantic(input_data, effective_user_id, backend) elif command == "insert": - return await self._native_append_semantic(input_data, user_id) + return await self._native_append_semantic(input_data, effective_user_id, backend) elif command == "delete": - return await self._native_delete_semantic(input_data, user_id) + return await self._native_delete_semantic(input_data, effective_user_id, backend) elif command == "rename": - return await self._native_rename_semantic(input_data, user_id) + return await self._native_rename_semantic(input_data, effective_user_id, backend) else: return f"Error: Unknown command '{command}'" except Exception as e: @@ -1833,7 +1852,9 @@ your responses, not to drive new actions.""" # Semantic Translation Methods (Native Tool → Vector Store) # ========================================================================= - async def _native_view_semantic(self, input_data: dict[str, Any], user_id: str) -> str: + async def _native_view_semantic( + self, input_data: dict[str, Any], user_id: str, backend: Any = None + ) -> str: """Handle VIEW command with semantic search capabilities. Path patterns: @@ -1843,6 +1864,7 @@ your responses, not to drive new actions.""" - /memories/all → List all memories (paginated) - /memories/ → Search by topic/path """ + backend = backend or self._backend path = input_data.get("path", "/memories") # Normalize path @@ -1856,31 +1878,36 @@ your responses, not to drive new actions.""" query = subpath[len("search/") :] if not query: return "Error: Please provide a search query. Example: view /memories/search/food preferences" - return await self._semantic_search(query, user_id) + return await self._semantic_search(query, user_id, backend=backend) # CASE 2: /memories/recent → Recent memories if subpath == "recent": - return await self._get_recent_memories(user_id, limit=10) + return await self._get_recent_memories(user_id, backend=backend, limit=10) # CASE 3: /memories/all → List all (paginated) if subpath == "all": - return await self._list_all_memories(user_id, limit=20) + return await self._list_all_memories(user_id, backend=backend, limit=20) # CASE 4: /memories (root) → Overview with instructions if not subpath or subpath == "": - return await self._get_memory_overview(user_id) + return await self._get_memory_overview(user_id, backend=backend) # CASE 5: /memories/ → Search by topic # Treat the path as a search query - return await self._semantic_search(subpath.replace("/", " ").replace("_", " "), user_id) + return await self._semantic_search( + subpath.replace("/", " ").replace("_", " "), user_id, backend=backend + ) - async def _semantic_search(self, query: str, user_id: str, top_k: int = 5) -> str: + async def _semantic_search( + self, query: str, user_id: str, *, backend: Any = None, top_k: int = 5 + ) -> str: """Perform semantic search and format results.""" - if not self._backend: + backend = backend or self._backend + if not backend: return "Error: Memory backend not initialized" try: - results = await self._backend.search_memories( + results = await backend.search_memories( query=query, user_id=user_id, top_k=top_k, @@ -1911,15 +1938,18 @@ your responses, not to drive new actions.""" logger.error(f"Memory: Semantic search failed: {e}") return f"Error searching memories: {e}" - async def _get_recent_memories(self, user_id: str, limit: int = 10) -> str: + async def _get_recent_memories( + self, user_id: str, *, backend: Any = None, limit: int = 10 + ) -> str: """Get most recent memories.""" - if not self._backend: + backend = backend or self._backend + if not backend: return "Error: Memory backend not initialized" try: # Use a generic query to get recent items # Most backends will return by recency when query is broad - results = await self._backend.search_memories( + results = await backend.search_memories( query="recent memories", user_id=user_id, top_k=limit, @@ -1946,14 +1976,17 @@ your responses, not to drive new actions.""" logger.error(f"Memory: Get recent failed: {e}") return f"Error getting recent memories: {e}" - async def _list_all_memories(self, user_id: str, limit: int = 20) -> str: + async def _list_all_memories( + self, user_id: str, *, backend: Any = None, limit: int = 20 + ) -> str: """List all memories (paginated).""" - if not self._backend: + backend = backend or self._backend + if not backend: return "Error: Memory backend not initialized" try: # Get all memories with a broad search - results = await self._backend.search_memories( + results = await backend.search_memories( query="*", # Broad query user_id=user_id, top_k=limit, @@ -1978,14 +2011,15 @@ your responses, not to drive new actions.""" logger.error(f"Memory: List all failed: {e}") return f"Error listing memories: {e}" - async def _get_memory_overview(self, user_id: str) -> str: + async def _get_memory_overview(self, user_id: str, *, backend: Any = None) -> str: """Get memory directory overview with search instructions.""" - if not self._backend: + backend = backend or self._backend + if not backend: return "Error: Memory backend not initialized" try: # Get count of memories - results = await self._backend.search_memories( + results = await backend.search_memories( query="*", user_id=user_id, top_k=100, # Just to get a count @@ -2037,8 +2071,11 @@ To see RECENT: view /memories/recent To SAVE: create /memories/.txt "content" """ - async def _native_create_semantic(self, input_data: dict[str, Any], user_id: str) -> str: + async def _native_create_semantic( + self, input_data: dict[str, Any], user_id: str, backend: Any = None + ) -> str: """Handle CREATE command - save to semantic vector store.""" + backend = backend or self._backend path = input_data.get("path", "") file_text = input_data.get("file_text", "") @@ -2047,7 +2084,7 @@ To SAVE: create /memories/.txt "content" if not file_text: return "Error: file_text is required (the memory content)" - if not self._backend: + if not backend: return "Error: Memory backend not initialized" try: @@ -2060,7 +2097,7 @@ To SAVE: create /memories/.txt "content" ) # Save to our semantic backend - memory = await self._backend.save_memory( + memory = await backend.save_memory( content=file_text, user_id=user_id, importance=0.5, @@ -2074,8 +2111,11 @@ To SAVE: create /memories/.txt "content" logger.error(f"Memory: Semantic create failed: {e}") return f"Error: {e}" - async def _native_update_semantic(self, input_data: dict[str, Any], user_id: str) -> str: + async def _native_update_semantic( + self, input_data: dict[str, Any], user_id: str, backend: Any = None + ) -> str: """Handle STR_REPLACE command - update memory content.""" + backend = backend or self._backend path = input_data.get("path", "") old_str = input_data.get("old_str", "") new_str = input_data.get("new_str", "") @@ -2085,12 +2125,12 @@ To SAVE: create /memories/.txt "content" if not old_str: return "Error: old_str is required" - if not self._backend: + if not backend: return "Error: Memory backend not initialized" try: # Search for memory containing old_str - results = await self._backend.search_memories( + results = await backend.search_memories( query=old_str, user_id=user_id, top_k=5, @@ -2114,15 +2154,15 @@ To SAVE: create /memories/.txt "content" new_content = matching_memory.content.replace(old_str, new_str, 1) # Update via delete + create (or update if backend supports it) - if hasattr(self._backend, "update_memory"): - await self._backend.update_memory( + if hasattr(backend, "update_memory"): + await backend.update_memory( memory_id=matching_memory.id, new_content=new_content, user_id=user_id, ) else: - await self._backend.delete_memory(matching_memory.id) - await self._backend.save_memory( + await backend.delete_memory(matching_memory.id) + await backend.save_memory( content=new_content, user_id=user_id, importance=0.5, @@ -2139,8 +2179,11 @@ To SAVE: create /memories/.txt "content" logger.error(f"Memory: Semantic update failed: {e}") return f"Error: {e}" - async def _native_append_semantic(self, input_data: dict[str, Any], user_id: str) -> str: + async def _native_append_semantic( + self, input_data: dict[str, Any], user_id: str, backend: Any = None + ) -> str: """Handle INSERT command - append to memory or create new.""" + backend = backend or self._backend path = input_data.get("path", "") insert_text = input_data.get("insert_text", "") _insert_line = input_data.get("insert_line", 0) # Unused in semantic mode @@ -2150,7 +2193,7 @@ To SAVE: create /memories/.txt "content" if not insert_text: return "Error: insert_text is required" - if not self._backend: + if not backend: return "Error: Memory backend not initialized" try: @@ -2158,7 +2201,7 @@ To SAVE: create /memories/.txt "content" # with the additional context topic = path.replace("/memories/", "").replace("/", "_").replace(".txt", "") - await self._backend.save_memory( + await backend.save_memory( content=insert_text, user_id=user_id, importance=0.5, @@ -2172,14 +2215,17 @@ To SAVE: create /memories/.txt "content" logger.error(f"Memory: Semantic append failed: {e}") return f"Error: {e}" - async def _native_delete_semantic(self, input_data: dict[str, Any], user_id: str) -> str: + async def _native_delete_semantic( + self, input_data: dict[str, Any], user_id: str, backend: Any = None + ) -> str: """Handle DELETE command - remove from vector store.""" + backend = backend or self._backend path = input_data.get("path", "") if not path: return "Error: path is required" - if not self._backend: + if not backend: return "Error: Memory backend not initialized" try: @@ -2191,7 +2237,7 @@ To SAVE: create /memories/.txt "content" .replace(".txt", "") ) - results = await self._backend.search_memories( + results = await backend.search_memories( query=topic, user_id=user_id, top_k=10, @@ -2206,7 +2252,7 @@ To SAVE: create /memories/.txt "content" # Check if metadata matches path metadata = getattr(r.memory, "metadata", {}) or {} if metadata.get("virtual_path") == path or r.score > 0.8: - await self._backend.delete_memory(r.memory.id) + await backend.delete_memory(r.memory.id) deleted_count += 1 if deleted_count == 0: @@ -2221,8 +2267,11 @@ To SAVE: create /memories/.txt "content" logger.error(f"Memory: Semantic delete failed: {e}") return f"Error: {e}" - async def _native_rename_semantic(self, input_data: dict[str, Any], user_id: str) -> str: + async def _native_rename_semantic( + self, input_data: dict[str, Any], user_id: str, backend: Any = None + ) -> str: """Handle RENAME command - update memory path/topic.""" + backend = backend or self._backend old_path = input_data.get("old_path", "") new_path = input_data.get("new_path", "") @@ -2231,7 +2280,7 @@ To SAVE: create /memories/.txt "content" if not new_path: return "Error: new_path is required" - if not self._backend: + if not backend: return "Error: Memory backend not initialized" try: @@ -2243,7 +2292,7 @@ To SAVE: create /memories/.txt "content" .replace(".txt", "") ) - results = await self._backend.search_memories( + results = await backend.search_memories( query=old_topic, user_id=user_id, top_k=10, @@ -2260,8 +2309,8 @@ To SAVE: create /memories/.txt "content" metadata = getattr(r.memory, "metadata", {}) or {} if metadata.get("virtual_path") == old_path or r.score > 0.8: # Delete old and create with new path - await self._backend.delete_memory(r.memory.id) - await self._backend.save_memory( + await backend.delete_memory(r.memory.id) + await backend.save_memory( content=r.memory.content, user_id=user_id, importance=getattr(r.memory, "importance", 0.5), diff --git a/tests/test_memory_handler_native_ops.py b/tests/test_memory_handler_native_ops.py index 278552c4d..415ff8f2b 100644 --- a/tests/test_memory_handler_native_ops.py +++ b/tests/test_memory_handler_native_ops.py @@ -245,11 +245,11 @@ async def test_execute_native_memory_tool_dispatches_and_wraps_errors( async def fake_ensure_initialized() -> None: return None - async def fake_view(input_data, user_id): # noqa: ANN001 + async def fake_view(input_data, user_id, backend=None): # noqa: ANN001 called.append(("view", input_data, user_id)) return "viewed" - async def fake_create(input_data, user_id): # noqa: ANN001 + async def fake_create(input_data, user_id, backend=None): # noqa: ANN001 called.append(("create", input_data, user_id)) return "created" @@ -264,7 +264,7 @@ async def test_execute_native_memory_tool_dispatches_and_wraps_errors( == "Error: Unknown command 'bad'" ) - async def boom(input_data, user_id): # noqa: ANN001 + async def boom(input_data, user_id, backend=None): # noqa: ANN001 raise RuntimeError("oops") monkeypatch.setattr(handler, "_native_view_semantic", boom) @@ -335,19 +335,19 @@ async def test_native_view_semantic_routes_paths( ) -> None: seen: list[tuple[str, object]] = [] - async def fake_search(query, user_id, top_k=5): # noqa: ANN001 + async def fake_search(query, user_id, *, backend=None, top_k=5): # noqa: ANN001 seen.append(("search", query)) return "search-result" - async def fake_recent(user_id, limit=10): # noqa: ANN001 + async def fake_recent(user_id, *, backend=None, limit=10): # noqa: ANN001 seen.append(("recent", limit)) return "recent-result" - async def fake_all(user_id, limit=20): # noqa: ANN001 + async def fake_all(user_id, *, backend=None, limit=20): # noqa: ANN001 seen.append(("all", limit)) return "all-result" - async def fake_overview(user_id): # noqa: ANN001 + async def fake_overview(user_id, *, backend=None): # noqa: ANN001 seen.append(("overview", user_id)) return "overview-result" @@ -1004,7 +1004,7 @@ async def test_search_and_format_context_and_handle_memory_tool_calls( ): # noqa: ANN001 return f"ran:{tool_name}:{user_id}:{provider}:{input_data}" - async def fake_execute_native(input_data, user_id): # noqa: ANN001 + async def fake_execute_native(input_data, user_id, *, request_context=None): # noqa: ANN001 return f"native:{user_id}:{input_data}" monkeypatch.setattr(handler, "_ensure_initialized", fake_ensure_initialized) diff --git a/tests/test_memory_handler_project_isolation.py b/tests/test_memory_handler_project_isolation.py index 3414bdc84..e760d224e 100644 --- a/tests/test_memory_handler_project_isolation.py +++ b/tests/test_memory_handler_project_isolation.py @@ -40,6 +40,7 @@ class _FakeBackend: def __init__(self, cfg: Any) -> None: self.cfg = cfg + self.initialize_calls = 0 self.saved_contents: list[str] = [] self.search_results: list[Any] = [] # Tag the backend with its db_path so tests can assert on it. @@ -47,7 +48,7 @@ class _FakeBackend: _FakeBackend.instances.append(self) async def _ensure_initialized(self) -> None: - return None + self.initialize_calls += 1 async def search_memories(self, **kwargs: Any) -> list[Any]: return list(self.search_results) @@ -282,8 +283,6 @@ def test_unresolved_project_returns_no_context(tmp_path: Path) -> None: handler = MemoryHandler(cfg, agent_type="test") async def run() -> None: - await handler._ensure_initialized() - # Request with NO project-resolution signal: no header, no cwd, # no parseable system-prompt cwd: line. ctx_unresolved = sr_mod.RequestContext( @@ -314,5 +313,193 @@ def test_unresolved_project_returns_no_context(tmp_path: Path) -> None: "incident on 2026-05-26 (TAM-550) was caused by the GLOBAL " "fallback pooling prior-session content into a fresh thread." ) + assert _FakeBackend.instances == [] + + asyncio.run(run()) + + +def test_all_unresolved_tools_skip_without_backend_or_file_side_effects( + tmp_path: Path, +) -> None: + """Every custom/native operation returns the same fail-closed result.""" + handler = MemoryHandler( + MemoryConfig( + enabled=True, + backend="local", + db_path=str(tmp_path / "global-memory.db"), + storage_mode=sr_mod.MemoryStorageMode.PROJECT, + use_native_tool=True, + native_memory_dir=str(tmp_path / "native"), + ), + agent_type="test", + ) + unresolved = sr_mod.RequestContext( + headers={}, + system_prompt="You are helpful.", + base_user_id="alice", + ) + custom_inputs = { + "memory_save": {"content": "canary"}, + "memory_search": {"query": "canary"}, + "memory_update": {"memory_id": "canary", "new_content": "changed"}, + "memory_delete": {"memory_id": "canary"}, + "memory_list": {}, + } + native_inputs = { + "view": {"path": "/memories/canary.txt"}, + "create": {"path": "/memories/canary.txt", "file_text": "canary"}, + "str_replace": { + "path": "/memories/canary.txt", + "old_str": "canary", + "new_str": "changed", + }, + "insert": { + "path": "/memories/canary.txt", + "insert_line": 0, + "insert_text": "changed", + }, + "delete": {"path": "/memories/canary.txt"}, + "rename": { + "old_path": "/memories/canary.txt", + "new_path": "/memories/renamed.txt", + }, + } + content = [ + { + "type": "tool_use", + "id": name, + "name": name, + "input": input_data, + } + for name, input_data in custom_inputs.items() + ] + content += [ + { + "type": "tool_use", + "id": f"native-{command}", + "name": "memory", + "input": {"command": command, **input_data}, + } + for command, input_data in native_inputs.items() + ] + native_before = sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*")) + + results = asyncio.run( + handler.handle_memory_tool_calls( + {"content": content}, + "alice", + "anthropic", + request_context=unresolved, + ) + ) + + assert len(results) == len(content) + assert all( + result["content"] == '{"status": "skipped", "reason": "project_unresolved"}' + for result in results + ) + assert _FakeBackend.instances == [] + assert sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*")) == native_before + + +def test_resolved_native_create_uses_project_backend(tmp_path: Path) -> None: + """Native translation writes only to the resolved project backend.""" + handler = MemoryHandler( + MemoryConfig( + enabled=True, + backend="local", + db_path=str(tmp_path / "global-memory.db"), + storage_mode=sr_mod.MemoryStorageMode.PROJECT, + use_native_tool=True, + native_memory_dir=str(tmp_path / "native"), + ), + agent_type="test", + ) + + results = asyncio.run( + handler.handle_memory_tool_calls( + { + "content": [ + { + "type": "tool_use", + "id": "native-create", + "name": "memory", + "input": { + "command": "create", + "path": "/memories/project.txt", + "file_text": "PROJECT_NATIVE_CANARY", + }, + } + ] + }, + "alice", + "anthropic", + request_context=_ctx_for_cwd("/tmp/project-native"), + ) + ) + + assert results[0]["content"] == "File created successfully at: /memories/project.txt" + writers = [backend for backend in _FakeBackend.instances if backend.saved_contents] + assert len(writers) == 1 + assert writers[0].saved_contents == ["PROJECT_NATIVE_CANARY"] + assert "projects" in Path(writers[0].db_path).parts + + +def test_unresolved_native_create_has_no_backend_side_effect(tmp_path: Path) -> None: + """Native memory create must fail closed before touching the global backend.""" + handler = MemoryHandler( + MemoryConfig( + enabled=True, + backend="local", + db_path=str(tmp_path / "global-memory.db"), + storage_mode=sr_mod.MemoryStorageMode.PROJECT, + use_native_tool=True, + native_memory_dir=str(tmp_path / "native"), + ), + agent_type="test", + ) + unresolved = sr_mod.RequestContext( + headers={}, + system_prompt="You are helpful.", + base_user_id="alice", + ) + + async def run() -> None: + results = await handler.handle_memory_tool_calls( + { + "content": [ + { + "type": "tool_use", + "id": "native-create", + "name": "memory", + "input": { + "command": "create", + "path": "/memories/canary.txt", + "file_text": "UNRESOLVED_NATIVE_CANARY", + }, + } + ] + }, + "alice", + "anthropic", + request_context=unresolved, + ) + direct_result = await handler._execute_native_memory_tool( + { + "command": "create", + "path": "/memories/direct-canary.txt", + "file_text": "UNRESOLVED_DIRECT_NATIVE_CANARY", + }, + "alice", + request_context=unresolved, + ) + + assert '"reason": "project_unresolved"' in results[0]["content"] + assert direct_result == '{"status": "skipped", "reason": "project_unresolved"}' + assert not any(backend.initialize_calls for backend in _FakeBackend.instances) + assert all( + "UNRESOLVED_NATIVE_CANARY" not in backend.saved_contents + for backend in _FakeBackend.instances + ) asyncio.run(run()) diff --git a/tests/test_memory_storage_router.py b/tests/test_memory_storage_router.py index 90b36c946..40b150a57 100644 --- a/tests/test_memory_storage_router.py +++ b/tests/test_memory_storage_router.py @@ -300,11 +300,14 @@ def test_router_project_mode_unresolved_fails_closed_by_default( """ router = _make_router(tmp_path, MemoryStorageMode.PROJECT, monkeypatch) - _, scope = router.backend_for(_ctx(system_prompt="no env block")) + backend, scope = router.backend_for(_ctx(system_prompt="no env block")) # Fail-closed signal: PROJECT mode preserved, project_key is None. + assert backend is None assert scope.mode is MemoryStorageMode.PROJECT assert scope.project_key is None + assert scope.db_path is None assert scope.display_name == "unresolved (no memory)" + assert router.open_backends() == [] def test_router_project_mode_unresolved_global_fallback_when_opted_in( diff --git a/tests/test_openai_codex_routing.py b/tests/test_openai_codex_routing.py index 471d0d8ca..a3a47c943 100644 --- a/tests/test_openai_codex_routing.py +++ b/tests/test_openai_codex_routing.py @@ -311,6 +311,7 @@ class _MemoryContinuationHandler(_MemoryToolsOnlyHandler): args: dict, user_id: str, provider: str, + request_context=None, ) -> str: assert (name, args, user_id, provider) == ( "memory_search", diff --git a/tests/test_openai_codex_ws_lifecycle.py b/tests/test_openai_codex_ws_lifecycle.py index e24e79200..b7bd0c683 100644 --- a/tests/test_openai_codex_ws_lifecycle.py +++ b/tests/test_openai_codex_ws_lifecycle.py @@ -101,7 +101,11 @@ class _MemoryWsHandler: args: dict, user_id: str, provider: str, + request_context=None, ) -> str: + await self._ensure_initialized() + if not self._backend: + return '{"error": "backend not ready"}' assert (name, args, user_id, provider) == ( "memory_search", {}, @@ -1930,6 +1934,68 @@ async def test_ws_memory_frame_shape_guards_fail_open(initial_frame): assert upstream.sent[2] == later_frames[1] +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("memory_mode", "extra_headers"), + [ + pytest.param("live_zone_tail", {"x-headroom-bypass": "true"}, id="bypass"), + pytest.param("disabled", {}, id="disabled-memory"), + ], +) +async def test_ws_memory_tool_response_passes_through_when_injection_rejected( + monkeypatch, + memory_mode, + extra_headers, +): + monkeypatch.setenv("HEADROOM_MEMORY_INJECTION_MODE", memory_mode) + function_call = { + "type": "function_call", + "id": "fc-1", + "call_id": "call-1", + "name": "memory_search", + "arguments": "{}", + } + upstream_events = [ + json.dumps({"type": "response.created", "response": {"id": "r-1"}}), + json.dumps({"type": "response.output_item.added", "item": function_call}), + json.dumps({"type": "response.output_item.done", "item": function_call}), + json.dumps( + { + "type": "response.completed", + "response": {"id": "r-1", "output": [function_call]}, + } + ), + ] + upstream = _FakeUpstream(upstream_events) + fake_ws_mod = _make_fake_websockets_module(upstream) + initial_frame = _first_frame() + client_ws = _FakeWebSocket( + frames=[initial_frame], + headers={ + "authorization": "Bearer test", + "x-headroom-user-id": "user-1", + **extra_headers, + }, + hold_after_initial=True, + ) + handler = _DummyOpenAIHandler() + handler.memory_handler = _MemoryWsHandler() + executed: list[tuple[str, dict, str, str]] = [] + + async def _execute_memory_tool(name, args, user_id, provider, request_context=None): + executed.append((name, args, user_id, provider)) + return '{"memories": []}' + + handler.memory_handler._execute_memory_tool = _execute_memory_tool + + with patch.dict(sys.modules, {"websockets": fake_ws_mod}): + await handler.handle_openai_responses_ws(client_ws) + + assert client_ws.sent_text == upstream_events + assert upstream.sent == [initial_frame] + assert executed == [] + + @pytest.mark.asyncio async def test_ws_memory_enabled_non_memory_response_streams_completion(): message_item = { @@ -1994,7 +2060,7 @@ async def test_ws_late_memory_call_after_streamed_message_passes_through(): handler.memory_handler = _MemoryWsHandler() executed: list[tuple[str, dict, str, str]] = [] - async def _execute_memory_tool(name, args, user_id, provider): + async def _execute_memory_tool(name, args, user_id, provider, request_context=None): executed.append((name, args, user_id, provider)) return '{"memories": []}' @@ -2081,7 +2147,7 @@ async def test_ws_memory_continuation_normalizes_malformed_arguments(): handler.memory_handler = _MemoryWsHandler() executed: list[tuple[str, dict, str, str]] = [] - async def _execute_memory_tool(name, args, user_id, provider): + async def _execute_memory_tool(name, args, user_id, provider, request_context=None): executed.append((name, args, user_id, provider)) return '{"memories": []}' @@ -2204,7 +2270,7 @@ async def test_ws_memory_continuation_continues_pre_stream_and_passes_late_call( handler.memory_handler = _MemoryWsHandler() executed: list[tuple[str, dict, str, str]] = [] - async def _execute_memory_tool(name, args, user_id, provider): + async def _execute_memory_tool(name, args, user_id, provider, request_context=None): executed.append((name, args, user_id, provider)) return '{"memories": []}' diff --git a/tests/test_openai_responses_traffic_learner.py b/tests/test_openai_responses_traffic_learner.py index 9f1f7397c..c90684ec9 100644 --- a/tests/test_openai_responses_traffic_learner.py +++ b/tests/test_openai_responses_traffic_learner.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from types import SimpleNamespace from typing import Any import httpx @@ -111,6 +112,55 @@ def test_responses_input_does_not_promote_unknown_role_to_user() -> None: ] +def test_unresolved_project_skips_openai_learner_entry_points() -> None: + learner = _RecordingLearner() + backend_bindings: list[object] = [] + setattr(learner, "set_backend", backend_bindings.append) + memory_handler = SimpleNamespace( + is_project_unresolved=lambda _ctx: True, + initialized=True, + backend=object(), + ) + proxy = SimpleNamespace(traffic_learner=learner, memory_handler=memory_handler) + + async def run() -> None: + await OpenAIHandlerMixin._observe_openai_responses_traffic( + proxy, + {"input": _responses_input()}, + request_id="responses-unresolved", + request_context=object(), + ) + await OpenAIHandlerMixin._observe_openai_chat_traffic( + proxy, + [{"role": "user", "content": "remember this"}], + request_id="chat-unresolved", + request_context=object(), + ) + seen_call_ids: set[str] = set() + await OpenAIHandlerMixin._observe_openai_ws_response_create( + proxy, + _ws_frame(["baseline"]), + seen_call_ids=seen_call_ids, + baseline=True, + request_id="ws-baseline-unresolved", + request_context=object(), + ) + await OpenAIHandlerMixin._observe_openai_ws_response_create( + proxy, + _ws_frame(["later"]), + seen_call_ids=seen_call_ids, + baseline=False, + request_id="ws-later-unresolved", + request_context=object(), + ) + assert seen_call_ids == set() + + asyncio.run(run()) + assert backend_bindings == [] + assert learner.message_batches == [] + assert learner.tool_results == [] + + def test_responses_http_request_reaches_traffic_learner() -> None: config = ProxyConfig( optimize=False,