diff --git a/.mcp.json b/.mcp.json index 650ccae8..1933ac40 100644 --- a/.mcp.json +++ b/.mcp.json @@ -1,113 +1,52 @@ { "mcpServers": { "nlt-build": { - "type": "stdio", - "command": "/home/wtthornton/.local/bin/tapps-mcp", - "args": [ - "serve", - "--profile", - "nlt-build" - ], - "env": { - "TAPPS_MCP_PROJECT_ROOT": ".", - "TAPPS_MCP_MEMORY_BRAIN_HTTP_URL": "http://localhost:8080", - "TAPPS_MCP_MEMORY_BRAIN_AUTH_TOKEN": "${TAPPS_BRAIN_AUTH_TOKEN}", - "TAPPS_MCP_MEMORY_BRAIN_PROJECT_ID": "tapps-brain", - "TAPPS_MCP_CONTEXT7_API_KEY": "${TAPPS_MCP_CONTEXT7_API_KEY}", - "TAPPS_BRAIN_PROFILE": "full", - "TAPPS_METRICS_STORAGE": "dual" - }, - "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + "type": "http", + "url": "http://127.0.0.1:8760/mcp", + "headers": { + "X-Tapps-Project-Root": "/home/wtthornton/code/tapps-brain" }, - "nlt-memory": { - "type": "stdio", - "command": "/home/wtthornton/.local/bin/tapps-mcp", - "args": [ - "serve", - "--profile", - "nlt-memory" - ], - "env": { - "TAPPS_MCP_PROJECT_ROOT": ".", - "TAPPS_MCP_MEMORY_BRAIN_HTTP_URL": "http://localhost:8080", - "TAPPS_MCP_MEMORY_BRAIN_AUTH_TOKEN": "${TAPPS_BRAIN_AUTH_TOKEN}", - "TAPPS_BRAIN_PROFILE": "full", - "TAPPS_METRICS_STORAGE": "dual", - "TAPPS_MCP_CONTEXT7_API_KEY": "${TAPPS_MCP_CONTEXT7_API_KEY}", - "TAPPS_MCP_MEMORY_BRAIN_PROJECT_ID": "tapps-brain" - }, - "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + }, + "nlt-memory": { + "type": "http", + "url": "http://127.0.0.1:8761/mcp", + "headers": { + "X-Tapps-Project-Root": "/home/wtthornton/code/tapps-brain" }, - "nlt-setup": { - "type": "stdio", - "command": "/home/wtthornton/.local/bin/tapps-mcp", - "args": [ - "serve", - "--profile", - "nlt-setup" - ], - "env": { - "TAPPS_MCP_PROJECT_ROOT": ".", - "TAPPS_MCP_MEMORY_BRAIN_HTTP_URL": "http://localhost:8080", - "TAPPS_MCP_MEMORY_BRAIN_AUTH_TOKEN": "${TAPPS_BRAIN_AUTH_TOKEN}", - "TAPPS_BRAIN_PROFILE": "full", - "TAPPS_METRICS_STORAGE": "dual", - "TAPPS_MCP_CONTEXT7_API_KEY": "${TAPPS_MCP_CONTEXT7_API_KEY}", - "TAPPS_MCP_MEMORY_BRAIN_PROJECT_ID": "tapps-brain" - }, - "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + }, + "nlt-setup": { + "type": "http", + "url": "http://127.0.0.1:8762/mcp", + "headers": { + "X-Tapps-Project-Root": "/home/wtthornton/code/tapps-brain" }, - "nlt-linear-issues": { - "type": "stdio", - "command": "/home/wtthornton/.local/bin/tapps-platform", - "args": [ - "serve", - "--profile", - "nlt-linear-issues" - ], - "env": { - "TAPPS_MCP_PROJECT_ROOT": ".", - "TAPPS_MCP_MEMORY_BRAIN_HTTP_URL": "http://localhost:8080", - "TAPPS_MCP_MEMORY_BRAIN_AUTH_TOKEN": "${TAPPS_BRAIN_AUTH_TOKEN}", - "TAPPS_BRAIN_PROFILE": "full", - "TAPPS_METRICS_STORAGE": "dual", - "TAPPS_MCP_CONTEXT7_API_KEY": "${TAPPS_MCP_CONTEXT7_API_KEY}", - "TAPPS_MCP_MEMORY_BRAIN_PROJECT_ID": "tapps-brain" - }, - "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + }, + "nlt-linear-issues": { + "type": "http", + "url": "http://127.0.0.1:8763/mcp", + "headers": { + "X-Tapps-Project-Root": "/home/wtthornton/code/tapps-brain" }, - "nlt-project-docs": { - "type": "stdio", - "command": "/home/wtthornton/.local/bin/docsmcp", - "args": [ - "serve", - "--profile", - "nlt-project-docs" - ], - "env": { - "DOCS_MCP_PROJECT_ROOT": ".", - "TAPPS_BRAIN_PROFILE": "agent_brain" - }, - "instructions": "Documentation MCP: epic/story/prompt generators, artifact validation, and planning helpers for Markdown docs in this repo." + "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + }, + "nlt-project-docs": { + "type": "http", + "url": "http://127.0.0.1:8764/mcp", + "headers": { + "X-Tapps-Project-Root": "/home/wtthornton/code/tapps-brain" }, - "nlt-release-ship": { - "type": "stdio", - "command": "/home/wtthornton/.local/bin/tapps-platform", - "args": [ - "serve", - "--profile", - "nlt-release-ship" - ], - "env": { - "TAPPS_MCP_PROJECT_ROOT": ".", - "TAPPS_MCP_MEMORY_BRAIN_HTTP_URL": "http://localhost:8080", - "TAPPS_MCP_MEMORY_BRAIN_AUTH_TOKEN": "${TAPPS_BRAIN_AUTH_TOKEN}", - "TAPPS_BRAIN_PROFILE": "full", - "TAPPS_METRICS_STORAGE": "dual", - "TAPPS_MCP_CONTEXT7_API_KEY": "${TAPPS_MCP_CONTEXT7_API_KEY}", - "TAPPS_MCP_MEMORY_BRAIN_PROJECT_ID": "tapps-brain" - }, - "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." - } + "instructions": "Documentation MCP: epic/story/prompt generators, artifact validation, and planning helpers for Markdown docs in this repo." + }, + "nlt-release-ship": { + "type": "http", + "url": "http://127.0.0.1:8765/mcp", + "headers": { + "X-Tapps-Project-Root": "/home/wtthornton/code/tapps-brain" + }, + "instructions": "Code quality scoring (0-100 across 7 categories), security scanning (Bandit + secret detection), quality gates (pass/fail against configurable presets), documentation lookup, domain expert consultation, and project profiling for Python projects." + } } } diff --git a/src/tapps_brain/diagnostics.py b/src/tapps_brain/diagnostics.py index 636b38ae..3c5a4ecc 100644 --- a/src/tapps_brain/diagnostics.py +++ b/src/tapps_brain/diagnostics.py @@ -251,9 +251,7 @@ def _freshness(store: MemoryStore) -> DimensionScore: tier_age_days_sum[tier_s] = tier_age_days_sum.get(tier_s, 0.0) + age_days avg = sum(scores) / len(scores) n = len(entries) - tier_avg_age = { - k: round(tier_age_days_sum[k] / tier_counts[k], 2) for k in sorted(tier_counts) - } + tier_avg_age = {k: round(tier_age_days_sum[k] / tier_counts[k], 2) for k in sorted(tier_counts)} context_n = tier_counts.get(MemoryTier.context.value, 0) return DimensionScore( name="freshness", diff --git a/src/tapps_brain/http_adapter.py b/src/tapps_brain/http_adapter.py index 4e9694d2..a96a14d3 100644 --- a/src/tapps_brain/http_adapter.py +++ b/src/tapps_brain/http_adapter.py @@ -1036,6 +1036,24 @@ def _build_mcp_server() -> Any: ) +def _resolve_wrapped_default_store(store: Any) -> Any: + """Return the inner default store when ``store`` is a Hive ``_StoreProxy`` wrapper. + + Do not use ``getattr(store, "_default_store", store)`` on :class:`MagicMock` — + mocks auto-create ``_default_store`` children, which breaks tenant matching in + integration tests that inject mock stores. + """ + store_dict = getattr(store, "__dict__", None) + if isinstance(store_dict, dict) and "_default_store" in store_dict: + return store_dict["_default_store"] + slots = getattr(type(store), "__slots__", ()) + if isinstance(slots, str): + slots = (slots,) + if isinstance(slots, (tuple, list)) and "_default_store" in slots: + return store._default_store + return store + + def create_app( *, store: MemoryStore | None = None, @@ -1710,6 +1728,58 @@ def _get_store_or_503() -> Any: ) return cfg.store + def _get_tenant_store_or_503(project_id: str, agent_id: str) -> Any: + """Resolve a tenant-scoped :class:`MemoryStore` for REST data-plane routes (ADR-010). + + ``cfg.store`` is the process-default store (often built at ``serve`` startup + for one project). Per-request ``X-Project-Id`` must select the matching + Postgres scope via the same :func:`_get_store_for_project` helper the MCP + :class:`_StoreProxy` uses — passing ``project_id`` only into + ``memory_service`` while reusing the default store silently breaks tenant + isolation (tapps-mcp #6). + """ + from tapps_brain.mcp_server.context import _get_store_for_project + from tapps_brain.project_registry import ProjectNotRegisteredError + + base = _get_store_or_503() + default_store = _resolve_wrapped_default_store(base) + server_agent = ( + getattr(default_store, "_agent_id", None) + or getattr(default_store, "agent_id", None) + or "http-adapter" + ) + eff_agent = (agent_id or "").strip() or "unknown" + call_agent = eff_agent if eff_agent != server_agent else None + + try: + return _get_store_for_project( + project_id, + default_store=default_store, + enable_hive=True, + agent_id=str(server_agent), + call_agent_id=call_agent, + ) + except ProjectNotRegisteredError as exc: + raise HTTPException( + status_code=404, + detail={ + "error": "project_not_registered", + "detail": f"Project {exc.project_id!r} is not registered.", + "project_id": exc.project_id, + }, + ) from exc + + def _async_store_covers_tenant(project_id: str, agent_id: str) -> bool: + """True when the startup async store matches the request tenant.""" + if _get_async_store_or_none() is None or cfg.store is None: + return False + base = _resolve_wrapped_default_store(cfg.store) + store_pid = getattr(base, "_project_id", None) or "" + store_aid = getattr(base, "_agent_id", None) or "" + eff_agent = (agent_id or "").strip() or "unknown" + agent_ok = eff_agent in {store_aid, "unknown"} or not store_aid + return bool(store_pid) and store_pid == project_id and agent_ok + def _get_async_store_or_none() -> Any: """Return the async-native store when wired at startup, else None.""" return getattr(cfg, "async_store", None) @@ -1810,8 +1880,6 @@ async def _v1_remember(request: Request) -> JSONResponse: "tags"?: list[str], "scope"?: str, "confidence"?: float, "agent_scope"?: str, "group"?: str }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -1819,6 +1887,7 @@ async def _v1_remember(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) # TAP-629: acquire per-key guard BEFORE the cache check so that # concurrent duplicates yield at ``await guard.acquire()`` rather @@ -1890,7 +1959,7 @@ async def _v1_remember(request: Request) -> JSONResponse: from tapps_brain.services import memory_service as _ms _async_store = _get_async_store_or_none() - if _async_store is not None: + if _async_store is not None and _async_store_covers_tenant(project_id, agent_id): # TAP-826: use async-native path — DB write goes through # AsyncPostgresPrivateBackend without blocking a thread. result = await _ms.async_memory_save( @@ -1961,8 +2030,6 @@ async def _v1_reinforce(request: Request) -> JSONResponse: Request body (JSON): ``{ "key": str, "confidence_boost"?: float }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -1970,6 +2037,7 @@ async def _v1_reinforce(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) # TAP-629: acquire per-key guard before cache check (see _v1_remember). ikey, istore = _get_ikey_and_istore(request) @@ -2035,7 +2103,7 @@ async def _v1_reinforce(request: Request) -> JSONResponse: from tapps_brain.services import memory_service as _ms _async_store = _get_async_store_or_none() - if _async_store is not None: + if _async_store is not None and _async_store_covers_tenant(project_id, agent_id): # TAP-1566: async-native reinforce — DB write goes through # AsyncPostgresPrivateBackend without blocking a thread. result = await _ms.async_memory_reinforce( @@ -2087,8 +2155,6 @@ async def _v1_remember_batch(request: Request) -> JSONResponse: Response: ``{ "results": [...], "saved_count": int, "error_count": int }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -2096,6 +2162,7 @@ async def _v1_remember_batch(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) try: raw = await request.body() @@ -2158,8 +2225,6 @@ async def _v1_recall_batch(request: Request) -> JSONResponse: Response: ``{ "results": [...], "query_count": int }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -2167,6 +2232,7 @@ async def _v1_recall_batch(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) try: raw = await request.body() @@ -2229,8 +2295,6 @@ async def _v1_reinforce_batch(request: Request) -> JSONResponse: Response: ``{ "results": [...], "reinforced_count": int, "error_count": int }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -2238,6 +2302,7 @@ async def _v1_reinforce_batch(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) try: raw = await request.body() @@ -2280,7 +2345,7 @@ async def _v1_reinforce_batch(request: Request) -> JSONResponse: from tapps_brain.services import memory_service as _ms _async_store = _get_async_store_or_none() - if _async_store is not None: + if _async_store is not None and _async_store_covers_tenant(project_id, agent_id): # TAP-1566: async-native batch reinforce — each per-item write # goes through AsyncPostgresPrivateBackend. result = await _ms.async_memory_reinforce_many( @@ -2314,8 +2379,6 @@ async def _v1_recall(request: Request) -> JSONResponse: Response: ``{ "results": [...], "query": str }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -2323,6 +2386,7 @@ async def _v1_recall(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) try: raw = await request.body() @@ -2404,8 +2468,6 @@ async def _v1_forget(request: Request) -> JSONResponse: Response: ``{ "forgotten": bool, "key": str, "reason"?: str }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -2413,6 +2475,7 @@ async def _v1_forget(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) ikey, istore = _get_ikey_and_istore(request) guard: asyncio.Lock | None = None @@ -2476,7 +2539,7 @@ async def _v1_forget(request: Request) -> JSONResponse: from tapps_brain.services import memory_service as _ms _async_store = _get_async_store_or_none() - if _async_store is not None: + if _async_store is not None and _async_store_covers_tenant(project_id, agent_id): result = await _ms.async_brain_forget( _async_store, project_id, agent_id, key=mem_key ) @@ -2515,8 +2578,6 @@ async def _v1_learn_success(request: Request) -> JSONResponse: Response: ``{ "learned": true, "key": str }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -2524,6 +2585,7 @@ async def _v1_learn_success(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) ikey, istore = _get_ikey_and_istore(request) guard: asyncio.Lock | None = None @@ -2587,7 +2649,7 @@ async def _v1_learn_success(request: Request) -> JSONResponse: from tapps_brain.services import memory_service as _ms _async_store = _get_async_store_or_none() - if _async_store is not None: + if _async_store is not None and _async_store_covers_tenant(project_id, agent_id): result = await _ms.async_brain_learn_success( _async_store, project_id, @@ -2636,8 +2698,6 @@ async def _v1_learn_failure(request: Request) -> JSONResponse: Response: ``{ "learned": true, "key": str }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -2645,6 +2705,7 @@ async def _v1_learn_failure(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) ikey, istore = _get_ikey_and_istore(request) guard: asyncio.Lock | None = None @@ -2708,7 +2769,7 @@ async def _v1_learn_failure(request: Request) -> JSONResponse: from tapps_brain.services import memory_service as _ms _async_store = _get_async_store_or_none() - if _async_store is not None: + if _async_store is not None and _async_store_covers_tenant(project_id, agent_id): result = await _ms.async_brain_learn_failure( _async_store, project_id, @@ -3324,8 +3385,6 @@ async def _v1_kg_feedback(request: Request) -> JSONResponse: Response: ``{ "recorded": true, "edge_id": str, "feedback_type": str }`` """ - store = _get_store_or_503() - project_id = (request.headers.get("x-project-id") or "").strip() if not project_id: raise HTTPException( @@ -3333,6 +3392,7 @@ async def _v1_kg_feedback(request: Request) -> JSONResponse: detail={"error": "bad_request", "detail": "X-Project-Id header is required."}, ) agent_id = (request.headers.get("x-agent-id") or "").strip() or "unknown" + store = _get_tenant_store_or_503(project_id, agent_id) try: raw = await request.body() diff --git a/src/tapps_brain/mcp_server/context.py b/src/tapps_brain/mcp_server/context.py index b3b53f47..b42b7a5e 100644 --- a/src/tapps_brain/mcp_server/context.py +++ b/src/tapps_brain/mcp_server/context.py @@ -20,6 +20,7 @@ from __future__ import annotations +import contextlib import contextvars import os import threading @@ -29,7 +30,7 @@ from typing import TYPE_CHECKING, Any if TYPE_CHECKING: - from collections.abc import Callable + from collections.abc import Callable, Iterator def _get_logger() -> Any: # noqa: ANN401 @@ -68,6 +69,77 @@ def _get_logger() -> Any: # noqa: ANN401 ) +def _meta_field(meta: Any, field: str) -> str | None: # noqa: ANN401 + """Read *field* from an MCP ``RequestParams.Meta`` (or compatible) object.""" + if meta is None: + return None + val = getattr(meta, field, None) + if val is None: + extra = getattr(meta, "model_extra", None) or {} + val = extra.get(field) + if not val: + return None + return str(val).strip() or None + + +def _header_tenant_from_mcp_request() -> tuple[str | None, str | None]: + """Read the tenant envelope from the inbound HTTP request on MCP ``request_ctx``. + + Streamable HTTP session tasks process ``tools/call`` outside the FastAPI + middleware task, so :data:`REQUEST_PROJECT_ID` is often unset even though + :class:`McpTenantMiddleware` validated ``X-Project-Id`` on the wire. + """ + try: + from mcp.server.lowlevel.server import request_ctx as _request_ctx_var + except Exception: + return None, None + try: + rc = _request_ctx_var.get() + except LookupError: + return None, None + req = getattr(rc, "request", None) + headers = getattr(req, "headers", None) + if headers is None: + return None, None + project_id = (headers.get("x-project-id") or "").strip() or None + agent_id = (headers.get("x-agent-id") or "").strip() or None + tapps_agent = (headers.get("x-tapps-agent") or "").strip() + if tapps_agent: + agent_id = tapps_agent + return project_id, agent_id + + +def _meta_tenant_from_mcp_request() -> tuple[str | None, str | None]: + """Read ``project_id`` / ``agent_id`` from MCP JSON-RPC ``params._meta``.""" + try: + from mcp.server.lowlevel.server import request_ctx as _request_ctx_var + except Exception: + return None, None + try: + rc = _request_ctx_var.get() + except LookupError: + return None, None + meta = getattr(rc, "meta", None) + return _meta_field(meta, "project_id"), _meta_field(meta, "agent_id") + + +@contextlib.contextmanager +def _mcp_tenant_context_for_tool_call() -> Iterator[None]: + """Bridge resolved tenant identity into contextvars for MCP tool execution.""" + pid = _current_request_project_id() + aid = _current_request_agent_id() + tokens: list[tuple[contextvars.ContextVar[str | None], contextvars.Token[str | None]]] = [] + if pid and REQUEST_PROJECT_ID.get() != pid: + tokens.append((REQUEST_PROJECT_ID, REQUEST_PROJECT_ID.set(pid))) + if aid and REQUEST_AGENT_ID.get() != aid: + tokens.append((REQUEST_AGENT_ID, REQUEST_AGENT_ID.set(aid))) + try: + yield + finally: + for var, tok in reversed(tokens): + var.reset(tok) + + # -------------------------------------------------------------------------- # STORY-069.3: Per-call project_id dispatch with a bounded LRU store cache. # -------------------------------------------------------------------------- @@ -214,75 +286,50 @@ def _current_request_project_id() -> str | None: """Resolve the per-request project_id across transports. Precedence: - 1. ``REQUEST_PROJECT_ID`` contextvar (set by the FastAPI tenant - middleware for Streamable HTTP — STORY-070.4). - 2. MCP JSON-RPC ``_meta.project_id`` on the active request context - (legacy per-call override; also works over stdio). - 3. ``TAPPS_BRAIN_PROJECT`` environment variable (stdio transport - set by the client's ``.mcp.json`` ``env``). + 1. ``X-Project-Id`` on the Starlette :class:`~starlette.requests.Request` + attached to MCP ``request_ctx`` (authoritative for Streamable HTTP + ``tools/call`` — session tasks inherit a stale + :data:`REQUEST_PROJECT_ID` from ``initialize``). + 2. MCP JSON-RPC ``_meta.project_id`` on the active request context. + 3. ``REQUEST_PROJECT_ID`` contextvar (FastAPI middleware — same-task + HTTP handlers only). + 4. ``TAPPS_BRAIN_PROJECT`` environment variable (stdio transport). """ + header_pid, _ = _header_tenant_from_mcp_request() + if header_pid: + return header_pid + meta_pid, _ = _meta_tenant_from_mcp_request() + if meta_pid: + return meta_pid pid = REQUEST_PROJECT_ID.get() if pid: return str(pid).strip() or None - try: - from mcp.server.lowlevel.server import request_ctx - except Exception: - request_ctx = None # type: ignore[assignment] - if request_ctx is not None: - try: - rc = request_ctx.get() - except LookupError: - rc = None - if rc is not None: - meta = getattr(rc, "meta", None) - if meta is not None: - mpid = getattr(meta, "project_id", None) - if mpid is None: - extra = getattr(meta, "model_extra", None) or {} - mpid = extra.get("project_id") - if mpid: - return str(mpid).strip() or None env_pid = (os.environ.get("TAPPS_BRAIN_PROJECT") or "").strip() return env_pid or None -def _current_request_agent_id() -> str | None: # noqa: PLR0911 +def _current_request_agent_id() -> str | None: """Return the effective per-request agent_id. Precedence (STORY-070.7): - 1. :data:`REQUEST_AGENT_ID` contextvar (set by HTTP middleware from - ``X-Agent-Id`` / ``X-Tapps-Agent``). - 2. MCP JSON-RPC ``_meta.agent_id`` on the active request context - (mirrors ``_current_request_project_id``). - 3. ``None`` — caller falls back to the server-level default. + 1. ``X-Tapps-Agent`` / ``X-Agent-Id`` on the Starlette request attached + to MCP ``request_ctx`` (see :func:`_current_request_project_id`). + 2. MCP JSON-RPC ``_meta.agent_id`` on the active request context. + 3. :data:`REQUEST_AGENT_ID` contextvar (HTTP middleware — same-task only). + 4. ``None`` — caller falls back to the server-level default. """ + _, header_aid = _header_tenant_from_mcp_request() + if header_aid: + return header_aid + _, meta_aid = _meta_tenant_from_mcp_request() + if meta_aid: + return meta_aid agent = REQUEST_AGENT_ID.get() if agent: val = str(agent).strip() if val: return val - # Also check _meta.agent_id from the MCP JSON-RPC envelope so stdio - # clients can multiplex agents without a header layer. - try: - from mcp.server.lowlevel.server import request_ctx - except Exception: - return None - try: - rc = request_ctx.get() - except LookupError: - return None - if rc is None: - return None - meta = getattr(rc, "meta", None) - if meta is None: - return None - aid = getattr(meta, "agent_id", None) - if aid is None: - extra = getattr(meta, "model_extra", None) or {} - aid = extra.get("agent_id") - if not aid: - return None - return str(aid).strip() or None + return None def _current_request_scope() -> str | None: diff --git a/src/tapps_brain/mcp_server/tool_filter.py b/src/tapps_brain/mcp_server/tool_filter.py index 1c69b8bf..ae27dc84 100644 --- a/src/tapps_brain/mcp_server/tool_filter.py +++ b/src/tapps_brain/mcp_server/tool_filter.py @@ -530,20 +530,23 @@ async def _filtered_call_tool(name: str, arguments: dict[str, Any], **kwargs: An with _METRICS_LOCK: key = (profile, "", "allowed") _MCP_TOOLS_CALL_TOTAL[key] = _MCP_TOOLS_CALL_TOTAL.get(key, 0) + 1 - try: - return await _orig_call_tool(name, arguments, **kwargs) - except Exception as exc: - # TAP-1580: enrich ToolError messages whose underlying cause is a - # pydantic ValidationError reporting missing required fields. - from mcp.server.fastmcp.exceptions import ToolError - - if not isinstance(exc, ToolError): - raise - missing = _missing_required_fields(exc) - if not missing: - raise - enriched = ToolError(_enriched_tool_error_message(name, str(exc), missing)) - raise enriched from exc.__cause__ + from tapps_brain.mcp_server.context import _mcp_tenant_context_for_tool_call + + with _mcp_tenant_context_for_tool_call(): + try: + return await _orig_call_tool(name, arguments, **kwargs) + except Exception as exc: + # TAP-1580: enrich ToolError messages whose underlying cause is a + # pydantic ValidationError reporting missing required fields. + from mcp.server.fastmcp.exceptions import ToolError + + if not isinstance(exc, ToolError): + raise + missing = _missing_required_fields(exc) + if not missing: + raise + enriched = ToolError(_enriched_tool_error_message(name, str(exc), missing)) + raise enriched from exc.__cause__ # Install wrappers on the tool manager instance (not the class) so only # this *mcp* instance is affected. diff --git a/tests/integration/test_http_adapter.py b/tests/integration/test_http_adapter.py index c8771f65..2e6b6b5f 100644 --- a/tests/integration/test_http_adapter.py +++ b/tests/integration/test_http_adapter.py @@ -13,6 +13,7 @@ from __future__ import annotations import asyncio +import contextlib import threading import time from datetime import UTC, datetime @@ -104,6 +105,28 @@ def _make_settings( return s +@contextlib.contextmanager +def _patch_http_settings(settings: _Settings): + """Patch adapter settings and tenant store resolution for in-process tests.""" + + def _fake_get_store_for_project(project_id: str, **kwargs: Any) -> Any: + del project_id, kwargs + if settings.store is None: + msg = "test settings have no store" + raise RuntimeError(msg) + return settings.store + + with ( + patch.object(_http_mod, "_settings", settings), + patch.object(_http_mod, "get_settings", return_value=settings), + patch( + "tapps_brain.mcp_server.context._get_store_for_project", + side_effect=_fake_get_store_for_project, + ), + ): + yield + + # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- @@ -120,10 +143,7 @@ async def test_remember_uses_async_store(self) -> None: _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -147,10 +167,7 @@ async def test_concurrent_remember_no_thread_pool_saturation(self) -> None: _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -180,10 +197,7 @@ async def test_remember_falls_back_to_to_thread_without_async_store(self) -> Non _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -208,10 +222,7 @@ async def test_brain_forget_uses_async_store(self) -> None: _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -237,10 +248,7 @@ async def test_learn_success_uses_async_store(self) -> None: _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -265,10 +273,7 @@ async def test_learn_failure_uses_async_store(self) -> None: _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -298,10 +303,7 @@ async def test_reinforce_uses_async_store(self) -> None: _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -326,10 +328,7 @@ async def test_reinforce_falls_back_to_to_thread_without_async_store(self) -> No _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -352,10 +351,7 @@ async def test_reinforce_batch_uses_async_store(self) -> None: _mcp_dummy = MagicMock() _mcp_dummy.session_manager = None - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=_mcp_dummy) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: @@ -554,10 +550,7 @@ def _make_tool(name: str, description: str) -> Any: settings = _make_settings() - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=mcp_server) with TestClient(app, raise_server_exceptions=True) as client: resp = client.get("/v1/tools/list") @@ -585,10 +578,7 @@ def test_v1_tools_list_no_auth_required(self) -> None: mcp_server.session_manager = None settings = _make_settings() - with ( - patch.object(_http_mod, "_settings", settings), - patch.object(_http_mod, "get_settings", return_value=settings), - ): + with _patch_http_settings(settings): app = create_app(mcp_server=mcp_server) with TestClient(app) as client: resp = client.get("/v1/tools/list") diff --git a/tests/unit/test_diagnostics.py b/tests/unit/test_diagnostics.py index 74791551..5e81b17a 100644 --- a/tests/unit/test_diagnostics.py +++ b/tests/unit/test_diagnostics.py @@ -75,9 +75,7 @@ def test_dimension_recommendations_healthy_store_silent() -> None: "freshness": DimensionScore( name="freshness", score=0.90, raw_details={"context_share": 0.80} ), - "staleness": DimensionScore( - name="staleness", score=1.0, raw_details={"gc_candidates": 0} - ), + "staleness": DimensionScore(name="staleness", score=1.0, raw_details={"gc_candidates": 0}), } assert _dimension_recommendations(scores) == [] @@ -434,7 +432,9 @@ class _HiveCountBackend: def count_by_namespace(self) -> dict[str, int]: return {"universal": 42, "repo-brain": 5} - def search(self, query: str, namespaces: list[str] | None = None, **kwargs: object) -> list[dict[str, str]]: + def search( + self, query: str, namespaces: list[str] | None = None, **kwargs: object + ) -> list[dict[str, str]]: raise AssertionError(f"search should not be called when counts exist: {query!r}") diff --git a/tests/unit/test_http_adapter.py b/tests/unit/test_http_adapter.py index 328146b5..80153b45 100644 --- a/tests/unit/test_http_adapter.py +++ b/tests/unit/test_http_adapter.py @@ -2348,3 +2348,52 @@ def test_recall_no_auth_token_returns_401(self) -> None: headers={"X-Project-Id": "proj-x"}, ) assert resp.status_code == 401 + + +class TestRestTenantStoreResolution: + """REST data-plane routes resolve per-request stores (ADR-010 / tapps-mcp #6).""" + + _AUTH = {"Authorization": "Bearer tok", "X-Project-Id": "tenant-a"} + + def test_remember_resolves_tenant_store(self, monkeypatch: pytest.MonkeyPatch) -> None: + default = MagicMock(name="default_store") + tenant = MagicMock(name="tenant_store") + captured: list[Any] = [] + + def _fake_get_store_for_project(project_id: str | None, **kwargs: Any) -> Any: + return tenant if project_id == "tenant-b" else default + + def _fake_memory_save( + store: Any, project_id: str, agent_id: str, **kwargs: Any + ) -> dict[str, Any]: + captured.append(store) + return { + "status": "saved", + "key": kwargs["key"], + "tier": "pattern", + "confidence": 0.6, + "memory_group": None, + } + + monkeypatch.setattr( + "tapps_brain.mcp_server.context._get_store_for_project", + _fake_get_store_for_project, + ) + monkeypatch.setattr( + "tapps_brain.services.memory_service.memory_save", + _fake_memory_save, + ) + + settings = _make_settings(auth_token="tok", store=default) + with _client(settings) as client: + resp = client.post( + "/v1/remember", + json={"key": "k1", "value": "v1"}, + headers={ + "Authorization": "Bearer tok", + "X-Project-Id": "tenant-b", + "X-Agent-Id": "agent-1", + }, + ) + assert resp.status_code == 200 + assert captured == [tenant] diff --git a/tests/unit/test_mcp_server.py b/tests/unit/test_mcp_server.py index 617f87f9..da5ad0c2 100644 --- a/tests/unit/test_mcp_server.py +++ b/tests/unit/test_mcp_server.py @@ -3843,3 +3843,77 @@ def _boom(*a, **kw): assert err.code == -32002 assert err.message == "project_not_registered" assert err.data["project_id"] == "ghost" + + +class TestMcpHttpHeaderTenantFallback: + """Streamable HTTP session tasks read tenant headers from request_ctx.request.""" + + def test_current_request_project_id_from_http_headers(self) -> None: + from types import SimpleNamespace + + from tapps_brain.mcp_server import context as ctx + + class _Headers: + def get(self, key: str, default: str | None = None) -> str | None: + return {"x-project-id": "tenant-b"}.get(key, default) + + req = SimpleNamespace(headers=_Headers()) + rc = SimpleNamespace(request=req, meta=None) + + from mcp.server.lowlevel.server import request_ctx + + tok = request_ctx.set(rc) + try: + assert ctx._current_request_project_id() == "tenant-b" + finally: + request_ctx.reset(tok) + + def test_current_request_agent_id_prefers_x_tapps_agent(self) -> None: + from types import SimpleNamespace + + from tapps_brain.mcp_server import context as ctx + + class _Headers: + def get(self, key: str, default: str | None = None) -> str | None: + mapping = { + "x-agent-id": "legacy", + "x-tapps-agent": "canonical", + } + return mapping.get(key, default) + + req = SimpleNamespace(headers=_Headers()) + rc = SimpleNamespace(request=req, meta=None) + + from mcp.server.lowlevel.server import request_ctx + + tok = request_ctx.set(rc) + try: + assert ctx._current_request_agent_id() == "canonical" + finally: + request_ctx.reset(tok) + + def test_mcp_tenant_context_binds_contextvars_for_tool_call(self) -> None: + from types import SimpleNamespace + + from tapps_brain.mcp_server import context as ctx + + class _Headers: + def get(self, key: str, default: str | None = None) -> str | None: + return {"x-project-id": "tenant-c", "x-agent-id": "agent-c"}.get(key, default) + + req = SimpleNamespace(headers=_Headers()) + rc = SimpleNamespace(request=req, meta=None) + + from mcp.server.lowlevel.server import request_ctx + + tok = request_ctx.set(rc) + stale = ctx.REQUEST_PROJECT_ID.set("stale-from-init") + try: + assert ctx._current_request_project_id() == "tenant-c" + with ctx._mcp_tenant_context_for_tool_call(): + assert ctx.REQUEST_PROJECT_ID.get() == "tenant-c" + assert ctx.REQUEST_AGENT_ID.get() == "agent-c" + assert ctx.REQUEST_PROJECT_ID.get() == "stale-from-init" + finally: + ctx.REQUEST_PROJECT_ID.reset(stale) + request_ctx.reset(tok) diff --git a/tests/unit/test_story_070_12_labels.py b/tests/unit/test_story_070_12_labels.py index 8392d9a1..46636096 100644 --- a/tests/unit/test_story_070_12_labels.py +++ b/tests/unit/test_story_070_12_labels.py @@ -652,7 +652,9 @@ def test_explicit_ids_take_precedence_over_resolver(self) -> None: "tapps_brain.otel_tracer._resolve_request_tenant", return_value=("ctx-proj", "ctx-agent"), ), - start_mcp_tool_span("brain_recall", project_id="explicit-proj", agent_id="explicit-agent"), + start_mcp_tool_span( + "brain_recall", project_id="explicit-proj", agent_id="explicit-agent" + ), ): pass