fix(security): isolate bearer authorization paths

core/database.py

@@ -5,7 +5,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
 from urllib.parse import unquote, urlparse
-from sqlalchemy import DDL, event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, inspect, text
+from sqlalchemy import DDL, event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, UniqueConstraint, func, inspect, text
 from sqlalchemy.engine import Engine, make_url
 from sqlalchemy.types import TypeDecorator
 from sqlalchemy.ext.declarative import declarative_base, declared_attr
@@ -280,6 +280,47 @@ class ChatMessage(Base):
         Index('ix_messages_session_time', 'session_id', 'timestamp'),  # Composite for efficient message retrieval
     )
 
+
+class ChatSessionApprovalGrant(Base):
+    """Server-owned, durable approval provenance for one chat session.
+
+    A resolved tool-approval card is display/history data, not authority. This
+    separate row is inserted only by the interactive approval continuation and
+    is keyed by the real session owner plus session id. It deliberately has no
+    update path; deleting the owning session cascades the grant so an old id
+    cannot carry approval authority into a newly-created conversation.
+    """
+
+    __tablename__ = "chat_session_approval_grants"
+
+    id = Column(String, primary_key=True, index=True)
+    session_id = Column(
+        String,
+        ForeignKey("sessions.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    owner = Column(String, nullable=False, index=True)
+    approval_id = Column(String, nullable=False, index=True)
+    provenance_version = Column(Integer, nullable=False, default=1)
+    created_at = Column(DateTime, default=utcnow_naive, nullable=False)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "session_id",
+            "owner",
+            "approval_id",
+            name="uq_chat_session_approval_grant",
+        ),
+        Index(
+            "ix_chat_session_approval_grant_lookup",
+            "session_id",
+            "owner",
+            "provenance_version",
+        ),
+    )
+
+
 class Document(TimestampMixin, Base):
     """Living document that the AI can create and edit in-place."""
     __tablename__ = "documents"

core/middleware.py

@@ -11,6 +11,7 @@ from starlette.responses import Response
 from starlette.routing import get_route_path
 
 from src.owner_identity import INTERNAL_TOOL_USER, auth_disabled
+from src.auth_helpers import is_bearer_principal
 
 
 # Per-process token that lets the in-app tool layer hit admin-gated
@@ -59,6 +60,13 @@ def require_admin(request: Request):
     Allows access when auth is explicitly disabled, or when the request carries
     the in-process internal-tool token used by loopback agent tools.
     """
+    # A bearer principal never inherits admin authority, even when the token
+    # carries a legacy cookbook scope or auth is disabled in a direct-entry
+    # test. Host-control routes use this centralized gate, so rejecting here
+    # covers shell, model-serving, MCP, runtime, and other admin surfaces.
+    if is_bearer_principal(request):
+        raise HTTPException(403, "API tokens cannot use admin host-control surfaces")
+
     # In-process bypass for tool-layer loopback calls. Two paths:
     # (a) header-direct (caller set X-Odysseus-Internal-Token), or
     # (b) the auth middleware already validated the token and stamped

core/models.py

@@ -10,8 +10,9 @@ from typing import Dict, List, Any, Optional, TYPE_CHECKING
 
 from src.tool_approval_scopes import (
     CHAT_SESSION_APPROVAL_CONTEXT_MARKER,
-    CHAT_SESSION_APPROVAL_DECISION,
 )
+from src.message_metadata import sanitize_projected_message_metadata
+from src.tool_approval_provenance import has_chat_session_approval_grant
 
 if TYPE_CHECKING:
     from .session_manager import SessionManager
@@ -40,29 +41,11 @@ def _history_grants_chat_session_approval(
     history: List["ChatMessage"],
     session_id: str,
 ) -> bool:
-    """Return whether this exact chat has a resolved session-scope grant."""
+    """Compatibility shim: durable history is never an authority source.
 
-    expected_session = str(session_id or "")
-    if not expected_session:
-        return False
-    for message in reversed(history or []):
-        metadata = getattr(message, "metadata", None)
-        if not isinstance(metadata, dict):
-            continue
-        tool_events = metadata.get("tool_events")
-        if not isinstance(tool_events, list):
-            continue
-        for event in reversed(tool_events):
-            ask_user = event.get("ask_user") if isinstance(event, dict) else None
-            if not isinstance(ask_user, dict):
-                continue
-            if (
-                ask_user.get("kind") == "tool_approval"
-                and ask_user.get("resolved") == CHAT_SESSION_APPROVAL_DECISION
-                and ask_user.get("approved_by_interactive_session") is True
-                and str(ask_user.get("session_id") or "") == expected_session
-            ):
-                return True
+    Keep the old private symbol for downstream imports, but deliberately return
+    false.  The live projection checks the separate server-owned grant table.
+    """
     return False
 
 
@@ -165,21 +148,13 @@ class Session:
                 projected.pop("metadata", None)
                 messages.append(projected)
                 continue
-            if isinstance(metadata, dict) and CHAT_SESSION_APPROVAL_CONTEXT_MARKER in metadata:
-                # The marker is derived below from a verified persisted
-                # approval event. Never pass a raw durable/client marker
-                # through to the model context.
-                metadata = {
-                    key: value
-                    for key, value in metadata.items()
-                    if key != CHAT_SESSION_APPROVAL_CONTEXT_MARKER
-                }
-                if metadata:
-                    projected["metadata"] = metadata
-                else:
-                    projected.pop("metadata", None)
+            metadata = sanitize_projected_message_metadata(metadata)
+            if metadata:
+                projected["metadata"] = metadata
+            else:
+                projected.pop("metadata", None)
             messages.append(projected)
-        if not _history_grants_chat_session_approval(self.history, self.id):
+        if not has_chat_session_approval_grant(self.id, self.owner):
             return messages
 
         # Keep the grant close to the latest user request so route-neutral

core/session_manager.py

@@ -62,6 +62,22 @@ def _parse_msg_content(raw):
     return raw
 
 
+def _parse_message_metadata(raw) -> dict:
+    """Decode only JSON objects from durable message metadata.
+
+    Legacy rows may contain a JSON list (including list-of-pairs) or another
+    scalar.  Such values have no trusted message fields and must not reach the
+    ``_db_id``/timestamp merge below or any approval projection.
+    """
+    if not raw:
+        return {}
+    try:
+        parsed = json.loads(raw) if isinstance(raw, str) else raw
+    except (json.JSONDecodeError, TypeError, ValueError):
+        return {}
+    return dict(parsed) if isinstance(parsed, dict) else {}
+
+
 class SessionManager:
     """
     Manages chat sessions with database persistence.
@@ -161,8 +177,7 @@ class SessionManager:
         # Try relationship first, then direct query
         if db_session.messages:
             for db_msg in db_session.messages:
-                meta = json.loads(db_msg.meta_data) if db_msg.meta_data else {}
-                if meta is None: meta = {}
+                meta = _parse_message_metadata(db_msg.meta_data)
                 meta['_db_id'] = db_msg.id
                 meta.setdefault('timestamp', _message_timestamp_iso(db_msg.timestamp))
                 history.append(ChatMessage(
@@ -176,8 +191,7 @@ class SessionManager:
             ).order_by(DbChatMessage.timestamp).all()
 
             for db_msg in db_messages:
-                meta = json.loads(db_msg.meta_data) if db_msg.meta_data else {}
-                if meta is None: meta = {}
+                meta = _parse_message_metadata(db_msg.meta_data)
                 meta['_db_id'] = db_msg.id
                 meta.setdefault('timestamp', _message_timestamp_iso(db_msg.timestamp))
                 history.append(ChatMessage(
@@ -254,6 +268,8 @@ class SessionManager:
                 logger.warning("Dropping message for deleted session %s", session_id)
                 return
 
+            if not isinstance(message.metadata, dict):
+                message.metadata = None
             missing_upload_id = reserve_message_upload_references(
                 getattr(self, "upload_handler", None),
                 getattr(db_session, "owner", None),
@@ -366,6 +382,8 @@ class SessionManager:
             # ownership check/access touch and the replacement transaction.
             # A failed reservation must leave the existing transcript intact.
             for message in messages:
+                if not isinstance(message.metadata, dict):
+                    message.metadata = None
                 missing_upload_id = reserve_message_upload_references(
                     getattr(self, "upload_handler", None),
                     getattr(db_session, "owner", None),

routes/chat_helpers.py

@@ -16,7 +16,12 @@ from src.llm_core import normalize_model_id
 from src.endpoint_resolver import normalize_base
 from src.context_compactor import maybe_compact, trim_for_context
 from src.model_context import estimate_tokens, get_context_length
-from src.auth_helpers import effective_user
+from src.auth_helpers import (
+    RequestCapability,
+    effective_user,
+    request_capability as build_request_capability,
+)
+from src.tool_approval_scopes import CHAT_SESSION_APPROVAL_CONTEXT_MARKER
 from src.prompt_security import untrusted_context_message
 from src.attachment_refs import attachment_ref
 from routes.prefs_routes import _load_for_user as load_prefs_for_user
@@ -104,6 +109,33 @@ def _append_incognito_message(session_id: str, role: str, content: Any, metadata
     bundle["updated_at"] = time.time()
 
 
+def _history_for_request_capability(sess, capability: RequestCapability) -> list[dict[str, Any]]:
+    """Project persisted history without interactive approval authority for bearers."""
+    history = sess.get_context_messages()
+    if not capability.is_bearer:
+        return history
+
+    # Session.get_context_messages() derives the marker only from the separate
+    # server-owned grant table. A pure bearer chat may still read its owner's
+    # ordinary transcript, but it must not receive even that interactive
+    # approval signal as model context or future tool authority.
+    projected = []
+    for item in history or []:
+        if not isinstance(item, dict):
+            continue
+        message = dict(item)
+        metadata = message.get("metadata")
+        if isinstance(metadata, dict) and CHAT_SESSION_APPROVAL_CONTEXT_MARKER in metadata:
+            metadata = dict(metadata)
+            metadata.pop(CHAT_SESSION_APPROVAL_CONTEXT_MARKER, None)
+            if metadata:
+                message["metadata"] = metadata
+            else:
+                message.pop("metadata", None)
+        projected.append(message)
+    return projected
+
+
 # ── Data containers ────────────────────────────────────────────────────── #
 
 @dataclass
@@ -406,7 +438,13 @@ def build_uploaded_file_manifest(att_ids: list, upload_handler, owner: Optional[
     return manifest
 
 
-def add_user_message(sess, chat_handler, preprocessed: PreprocessedMessage, incognito: bool = False):
+def add_user_message(
+    sess,
+    chat_handler,
+    preprocessed: PreprocessedMessage,
+    incognito: bool = False,
+    capability: RequestCapability | None = None,
+):
     """Add user message to session history and update session name.
     Incognito messages must not mutate persistent session history, even in
     memory, because a later normal turn can persist the same session object."""
@@ -414,11 +452,23 @@ def add_user_message(sess, chat_handler, preprocessed: PreprocessedMessage, inco
         return
     user_meta = {"attachments": preprocessed.attachment_meta} if preprocessed.attachment_meta else None
     sess.add_message(ChatMessage("user", preprocessed.user_content, metadata=user_meta))
-    chat_handler.update_session_name_if_needed(sess, preprocessed.text_for_context)
+    if capability is None or capability.allow_auto_naming:
+        chat_handler.update_session_name_if_needed(sess, preprocessed.text_for_context)
 
 
-def fire_message_event(request, webhook_manager, session_id: str, sess, message: str, compare_mode: bool = False):
+def fire_message_event(
+    request,
+    webhook_manager,
+    session_id: str,
+    sess,
+    message: str,
+    compare_mode: bool = False,
+    capability: RequestCapability | None = None,
+):
     """Fire webhook and event_bus events for a new user message."""
+    capability = capability or build_request_capability(request)
+    if not capability.allow_message_events:
+        return
     if webhook_manager and not compare_mode:
         webhook_manager.fire_and_forget("chat.message", {
             "session_id": session_id, "model": sess.model, "message": message[:2000],
@@ -626,12 +676,15 @@ async def build_chat_context(
     defer_context_shaping: bool = False,
     continuation_context_message: str | None = None,
     persist_user_message: bool = True,
+    capability: RequestCapability | None = None,
 ) -> ChatContext:
     """Build the full context (preface + messages) for an LLM call.
 
     This is the shared logic between /chat and /chat_stream — preset extraction,
     message preprocessing, memory/RAG/web injection, compaction, normalization.
     """
+    capability = capability or build_request_capability(request)
+
     # Preset
     preset = extract_preset(chat_handler, preset_id)
 
@@ -653,11 +706,25 @@ async def build_chat_context(
         user_meta = {"attachments": preprocessed.attachment_meta} if preprocessed.attachment_meta else None
         _append_incognito_message(session_id, "user", preprocessed.user_content, user_meta)
     elif persist_user_message:
-        add_user_message(sess, chat_handler, preprocessed, incognito=False)
+        add_user_message(
+            sess,
+            chat_handler,
+            preprocessed,
+            incognito=False,
+            capability=capability,
+        )
 
     # Fire events
     if persist_user_message and not incognito:
-        fire_message_event(request, webhook_manager, session_id, sess, message, compare_mode)
+        fire_message_event(
+            request,
+            webhook_manager,
+            session_id,
+            sess,
+            message,
+            compare_mode,
+            capability=capability,
+        )
 
     # Resolve owner-scoped prefs/context. Browser requests keep the cookie user;
     # bearer-token chat requests use the token owner instead of the "api" sentinel.
@@ -761,7 +828,11 @@ async def build_chat_context(
     # Build messages. In Nobody/incognito mode, never read saved session
     # history: the session id may be a temporary wrapper or, in buggy clients, a
     # stale normal session id. Only the ephemeral incognito transcript is safe.
-    messages = preface + (_incognito_messages(session_id) if incognito else sess.get_context_messages())
+    messages = preface + (
+        _incognito_messages(session_id)
+        if incognito
+        else _history_for_request_capability(sess, capability)
+    )
 
     # Current date/time — injected as a standalone *user*-role context message
     # placed immediately before the latest user turn, NOT folded into the
@@ -1173,6 +1244,7 @@ def run_post_response_tasks(
     owner: str = None,
     extract_skills: bool = True,
     allow_background_extraction: bool = True,
+    capability: RequestCapability | None = None,
 ):
     """Fire background tasks after a completed response: memory extraction, webhooks, auto-name, skill extraction.
 
@@ -1188,6 +1260,12 @@ def run_post_response_tasks(
     ``_queue_background_extraction`` keeps them from overlapping the *next*
     turn's request too.
     """
+    if capability is not None and not capability.allow_deferred_work:
+        # Pure bearer chat is intentionally synchronous and request-bound.
+        # Do not schedule extraction, teacher/model work, callbacks, or
+        # auto-naming after the authorized request has returned/disconnected.
+        return
+
     _extraction_jobs: list = []
 
     # Memory extraction — only every 4th message pair to avoid excess LLM calls

routes/chat_routes.py

@@ -44,6 +44,7 @@ from src.auth_helpers import (
     effective_user,
     enforce_api_token_chat_controls,
     get_current_user,
+    request_capability as build_request_capability,
     require_chat_scope,
     require_interactive_request,
 )
@@ -74,6 +75,7 @@ from src.tool_policy import (
     web_search_enabled_for_turn,
 )
 from src.tool_approvals import tool_approval_store
+from src.tool_approval_provenance import create_chat_session_approval_grant
 
 logger = logging.getLogger(__name__)
 
@@ -96,7 +98,11 @@ def _stream_failure_status(chunk: str) -> Optional[int]:
 
 
 def _mark_tool_approval_resolved(sess, approval_id: Any, decision: Any) -> bool:
-    """Persist a consumed approval decision on its existing tool event."""
+    """Persist display-only resolution state on the existing tool event.
+
+    This metadata is intentionally never consulted for authorization; the
+    separate provenance row is written by the interactive approval path.
+    """
 
     approval_key = str(approval_id or "")
     normalized_decision = str(decision or "").strip().lower()
@@ -786,7 +792,8 @@ def setup_chat_routes(
         # non-streaming path can't be used to bypass).
         _enforce_chat_privileges(request, sess)
 
-        api_token_request = getattr(request.state, "api_token", False) is True
+        request_capability = build_request_capability(request)
+        api_token_request = request_capability.is_bearer
         tool_policy = build_effective_tool_policy(last_user_message=message)
         allow_tool_preprocessing = (
             not api_token_request
@@ -817,6 +824,7 @@ def setup_chat_routes(
             webhook_manager=webhook_manager,
             allow_tool_preprocessing=allow_tool_preprocessing,
             defer_context_shaping=foreground_policy.enabled,
+            capability=request_capability,
         )
 
         # Research injection
@@ -925,6 +933,7 @@ def setup_chat_routes(
             character_name=ctx.preset.character_name,
             owner=ctx.user,
             allow_background_extraction=allow_tool_preprocessing,
+            capability=request_capability,
         )
 
         return {
@@ -1003,6 +1012,11 @@ def setup_chat_routes(
             approval_id=tool_approval_id,
             allow_bash=allow_bash,
         )
+        request_capability = build_request_capability(request)
+        # Keep the route decision and the downstream capability derived from
+        # the same verified principal. The explicit control gate above remains
+        # the source of the chat-mode rejection message.
+        api_token_request = request_capability.is_bearer
 
         # A bearer token is a chat-only integration credential. Reject every
         # remaining control-plane input before approval lookup, intent
@@ -1216,6 +1230,21 @@ def setup_chat_routes(
                         409,
                         "This tool approval could not be consumed.",
                     )
+                if decision == "approve":
+                    # The transcript card is display data. Only the exact
+                    # interactive, one-use store result may create durable
+                    # session-scope provenance for later tool gates.
+                    if not create_chat_session_approval_grant(
+                        request,
+                        approval=exact_tool_approval,
+                        approval_id=tool_approval_id,
+                        session_id=session,
+                        owner=owner,
+                    ):
+                        logger.warning(
+                            "Tool approval %s ran without a durable chat-session grant",
+                            tool_approval_id,
+                        )
                 if not _mark_tool_approval_resolved(
                     sess,
                     tool_approval_id,
@@ -1418,6 +1447,7 @@ def setup_chat_routes(
                 else None
             ),
             persist_user_message=not tool_approval_continuation,
+            capability=request_capability,
         )
 
         _research_flags = {"do": do_research}  # Mutable container for generator scope
@@ -1652,7 +1682,7 @@ def setup_chat_routes(
         # Persist session mode after policy/privilege gates so blocked research
         # turns remain ordinary chat/agent streams and saved messages.
         _effective_mode = 'research' if effective_do_research else (chat_mode or 'chat')
-        if _effective_mode in ('agent', 'research', 'chat'):
+        if _effective_mode in ('agent', 'research', 'chat') and not request_capability.is_bearer:
             set_session_mode(session, _effective_mode)
 
         async def stream_with_save() -> AsyncGenerator[str, None]:
@@ -2299,6 +2329,7 @@ def setup_chat_routes(
                                         allow_tool_preprocessing
                                         and not tool_approval_continuation
                                     ),
+                                    capability=request_capability,
                                 )
                             _stream_set(session, status="done")
                             yield chunk
@@ -2573,6 +2604,7 @@ def setup_chat_routes(
                                         allow_tool_preprocessing
                                         and not tool_approval_continuation
                                     ),
+                                    capability=request_capability,
                                 )
                             _stream_set(session, status="done")
                             yield chunk
@@ -2647,7 +2679,7 @@ def setup_chat_routes(
         # buffered output + live); dropping the SSE only removes a subscriber —
         # the run keeps going and saves the assistant message on completion
         # regardless. Reconnect via /api/chat/resume.
-        if compare_mode:
+        if compare_mode or not request_capability.allow_detached_execution:
             return StreamingResponse(_safe_stream(), media_type="text/event-stream")
 
         _detached_run = agent_runs.start(session, _safe_stream())

routes/codex_routes.py

@@ -1,8 +1,9 @@
 """Codex integration routes.
 
 These are small HTTP surfaces intended for the Codex plugin/MCP bridge. They
-reuse existing Odysseus helpers and enforce API-token scopes before touching
-user data.
+reuse existing Odysseus helpers. The bridge is an interactive host-control
+plane and is unavailable to bearer principals; cookie/admin callers retain the
+documented operation path.
 """
 
 import asyncio
@@ -12,13 +13,14 @@ from io import BytesIO
 from pathlib import Path
 from typing import Any
 
-from fastapi import APIRouter, BackgroundTasks, Body, HTTPException, Request
+from fastapi import APIRouter, BackgroundTasks, Body, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
 
 from core.middleware import require_admin
 from src.auth_helpers import (
     require_api_token_owner,
     require_authenticated_request,
+    require_non_bearer_request,
     require_user,
 )
 from src.tool_implementations import do_manage_notes
@@ -88,6 +90,7 @@ async def _as_owner(request: Request, owner: str, fn, *args, **kwargs):
 
 def _scope_owner(request: Request, allowed: set[str]) -> str:
     """Return the data owner if the caller is allowed for this Codex action."""
+    require_non_bearer_request(request)
     if getattr(request.state, "api_token", False) is True:
         scopes = set(getattr(request.state, "api_token_scopes", []) or [])
         if not scopes.intersection(allowed):
@@ -99,6 +102,7 @@ def _scope_owner(request: Request, allowed: set[str]) -> str:
 
 def _scope_owner_all(request: Request, required: set[str]) -> str:
     """Return owner only when an API token has every required scope."""
+    require_non_bearer_request(request)
     if getattr(request.state, "api_token", False) is True:
         scopes = set(getattr(request.state, "api_token_scopes", []) or [])
         missing = required - scopes
@@ -111,9 +115,9 @@ def _scope_owner_all(request: Request, required: set[str]) -> str:
 def _require_cookbook_scope(request: Request, allowed: set[str]) -> str:
     """Authorize a Codex cookbook route.
 
-    For API-token callers, enforce the given scope set.
-    For cookie-session callers, additionally require admin privileges
-    because cookbook surfaces expose host topology, task logs, tmux
+    Bearer callers are rejected by the host-control boundary regardless of
+    legacy scope labels. Cookie-session callers additionally require admin
+    privileges because cookbook surfaces expose host topology, task logs, tmux
     commands, and model-serving controls.
     """
     owner = _scope_owner(request, allowed)
@@ -149,7 +153,11 @@ def setup_codex_routes(
     calendar_router: APIRouter | None = None,
     document_router: APIRouter | None = None,
 ) -> APIRouter:
-    router = APIRouter(prefix="/api/codex", tags=["codex"])
+    router = APIRouter(
+        prefix="/api/codex",
+        tags=["codex"],
+        dependencies=[Depends(require_non_bearer_request)],
+    )
     email_list_endpoint = _find_endpoint(email_router, "GET", "/api/email/list")
     email_read_endpoint = _find_endpoint(email_router, "GET", "/api/email/read/{uid}")
     email_send_endpoint = _find_endpoint(email_router, "POST", "/api/email/send")
@@ -164,6 +172,7 @@ def setup_codex_routes(
 
     @router.get("/capabilities")
     def capabilities(request: Request):
+        require_non_bearer_request(request)
         token_scopes = set(getattr(request.state, "api_token_scopes", []) or [])
         has_token = getattr(request.state, "api_token", False) is True
         def scoped(allowed):
@@ -215,6 +224,7 @@ def setup_codex_routes(
 
     @router.get("/plugin.zip")
     def plugin_zip(request: Request):
+        require_non_bearer_request(request)
         require_authenticated_request(request)
         root = Path(__file__).resolve().parent.parent / "integrations" / "codex"
         if not root.exists():
@@ -511,15 +521,10 @@ def setup_codex_routes(
         return await _as_owner(request, owner, documents_create_endpoint, request, req)
 
     # ── Cookbook surface ──
-    # Lets the agent run the same launch / monitor / kill loop the user
-    # would do by hand in the Cookbook UI: read the current task list +
-    # tmux output, launch a serve task, stop one.  Two scopes:
-    #   cookbook:read   — list tasks + tail output + list servers
-    #   cookbook:launch — also start/stop serves (host shell exec)
-    # `cookbook:launch` is genuinely powerful: /api/model/serve runs SSH'd
-    # commands on the user's hosts. The existing _validate_serve_cmd
-    # allowlist (vllm/python3/sglang/llama-server/etc., no shell metachars)
-    # keeps the agent inside the same sandbox the UI uses.
+    # These handlers retain their legacy scope constants for compatibility
+    # with callers and tests, but the bridge is now an interactive-only
+    # host-control plane. Bearer principals are rejected before any task-list,
+    # tmux-output, launch, stop, or model-serving operation.
 
     async def _run_shell(cmd: str, timeout: float = 15.0) -> dict:
         """Run a shell command, return {exit_code, stdout, stderr}."""
@@ -884,10 +889,15 @@ def setup_claude_routes() -> APIRouter:
     this router only exists to deliver the skill zip via `/api/claude/plugin.zip`
     so the user-facing setup commands stay in the Claude namespace.
     """
-    router = APIRouter(prefix="/api/claude", tags=["claude"])
+    router = APIRouter(
+        prefix="/api/claude",
+        tags=["claude"],
+        dependencies=[Depends(require_non_bearer_request)],
+    )
 
     @router.get("/plugin.zip")
     def plugin_zip(request: Request):
+        require_non_bearer_request(request)
         require_authenticated_request(request)
         # Only ship the skills/ subtree so extracting at ~/.claude/ doesn't dump
         # README.md or other bundle metadata into the user's claude config dir.

routes/history/history_routes.py

@@ -27,6 +27,24 @@ _HISTORY_INLINE_MEDIA_THRESHOLD = 200_000
 _DATA_IMAGE_RE = re.compile(r"data:image/[^;,\"]+;base64,[A-Za-z0-9+/=\s]+")
 
 
+def _metadata_dict(value: Any) -> dict:
+    """Return only mapping-shaped message metadata.
+
+    Legacy rows and client payloads can contain JSON lists/scalars. They are
+    display noise, not trusted fields, and must not reach ``dict.update`` or
+    approval projection code.
+    """
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+        except (json.JSONDecodeError, TypeError, ValueError):
+            return {}
+        return parsed if isinstance(parsed, dict) else {}
+    return {}
+
+
 def _history_display_content(content: Any) -> Any:
     """Return a lightweight browser-display copy of stored message content.
 
@@ -126,12 +144,9 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
 
     def _db_history_entry(m: DbChatMessage) -> Dict[str, Any]:
         entry = {"role": m.role, "content": _history_display_content(m.content)}
-        meta = {}
-        if m.meta_data:
-            try:
-                meta = json.loads(m.meta_data) or {}
-            except (json.JSONDecodeError, ValueError):
-                meta = {}
+        meta = _metadata_dict(m.meta_data)
+        if meta:
+            meta = dict(meta)
         if m.timestamp and "timestamp" not in meta:
             meta["timestamp"] = m.timestamp.isoformat() + "Z"
         if meta:
@@ -199,21 +214,23 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
         for msg in session.history:
             if isinstance(msg, ChatMessage):
                 # Skip hidden messages (e.g. compaction summaries for AI context)
-                if msg.metadata and msg.metadata.get("hidden"):
+                msg_meta = _metadata_dict(msg.metadata)
+                if msg_meta.get("hidden"):
                     continue
                 entry = {"role": msg.role, "content": _history_display_content(msg.content)}
-                if msg.metadata:
-                    entry["metadata"] = msg.metadata
+                if msg_meta:
+                    entry["metadata"] = msg_meta
                 history_dict.append(entry)
             elif isinstance(msg, dict):
-                if msg.get("metadata", {}).get("hidden"):
+                msg_meta = _metadata_dict(msg.get("metadata"))
+                if msg_meta.get("hidden"):
                     continue
                 entry = {
                     "role": msg.get("role", ""),
                     "content": _history_display_content(msg.get("content", "")),
                 }
-                if msg.get("metadata"):
-                    entry["metadata"] = msg["metadata"]
+                if msg_meta:
+                    entry["metadata"] = msg_meta
                 history_dict.append(entry)
 
         # Fallback: load from DB if in-memory renders empty. Display only —
@@ -370,9 +387,8 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
 
                 db_msg.content = content
                 meta = {}
-                if db_msg.meta_data:
-                    try: meta = json.loads(db_msg.meta_data)
-                    except (json.JSONDecodeError, ValueError): pass
+                meta = _metadata_dict(db_msg.meta_data)
+                meta = dict(meta)
                 meta['edited'] = True
                 db_msg.meta_data = json.dumps(meta)
 
@@ -412,13 +428,13 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
                 if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
                    (isinstance(msg, dict) and msg.get('role') == 'assistant'):
                     if isinstance(msg, ChatMessage):
-                        if not msg.metadata:
+                        if not isinstance(msg.metadata, dict):
                             msg.metadata = {}
                         msg.metadata['stopped'] = True
                         if not msg.metadata.get('model'):
                             msg.metadata['model'] = session.model
                     else:
-                        if 'metadata' not in msg:
+                        if not isinstance(msg.get('metadata'), dict):
                             msg['metadata'] = {}
                         msg['metadata']['stopped'] = True
                         if not msg['metadata'].get('model'):
@@ -436,11 +452,8 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
                 )
                 if db_messages:
                     meta = {}
-                    if db_messages.meta_data:
-                        try:
-                            meta = _json.loads(db_messages.meta_data)
-                        except (json.JSONDecodeError, ValueError):
-                            pass
+                    meta = _metadata_dict(db_messages.meta_data)
+                    meta = dict(meta)
                     meta['stopped'] = True
                     if not meta.get('model'):
                         meta['model'] = session.model
@@ -471,11 +484,11 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
                 if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
                    (isinstance(msg, dict) and msg.get('role') == 'assistant'):
                     if isinstance(msg, ChatMessage):
-                        if not msg.metadata:
+                        if not isinstance(msg.metadata, dict):
                             msg.metadata = {}
                         msg.metadata.update(meta_update)
                     else:
-                        if 'metadata' not in msg:
+                        if not isinstance(msg.get('metadata'), dict):
                             msg['metadata'] = {}
                         msg['metadata'].update(meta_update)
                     break
@@ -491,10 +504,7 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
                     .first()
                 )
                 if db_msg:
-                    meta = {}
-                    if db_msg.meta_data:
-                        try: meta = _json.loads(db_msg.meta_data)
-                        except (json.JSONDecodeError, ValueError): pass
+                    meta = dict(_metadata_dict(db_msg.meta_data))
                     meta.update(meta_update)
                     db_msg.meta_data = _json.dumps(meta)
                     db.commit()
@@ -536,8 +546,12 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
             merged_content = content1 + separator + content2
 
             # Merge metadata
-            meta1 = (msg1.metadata if isinstance(msg1, ChatMessage) else msg1.get('metadata')) or {}
-            meta2 = (msg2.metadata if isinstance(msg2, ChatMessage) else msg2.get('metadata')) or {}
+            meta1 = dict(_metadata_dict(
+                msg1.metadata if isinstance(msg1, ChatMessage) else msg1.get('metadata')
+            ))
+            meta2 = dict(_metadata_dict(
+                msg2.metadata if isinstance(msg2, ChatMessage) else msg2.get('metadata')
+            ))
             merged_meta = {**meta1, **meta2}
             merged_meta.pop('stopped', None)  # no longer stopped after continue
 
@@ -693,11 +707,11 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
             pct = max(0.0, min(100.0, pct))
             visible_messages = sum(
                 1 for m in session.history
-                if not (getattr(m, "metadata", None) or {}).get("hidden")
+                if not _metadata_dict(getattr(m, "metadata", None)).get("hidden")
             )
             compacted_messages = sum(
                 1 for m in session.history
-                if (getattr(m, "metadata", None) or {}).get("compacted")
+                if _metadata_dict(getattr(m, "metadata", None)).get("compacted")
             )
             can_compact = used > 0
             return {

routes/hwfit_routes.py

@@ -5,10 +5,11 @@ import shlex
 import subprocess
 from copy import deepcopy
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 
 from core.platform_compat import run_ssh_command
 from routes._validators import validate_remote_host, validate_ssh_port
+from src.auth_helpers import require_non_bearer_request
 
 
 # Backends the manual hardware simulator accepts. Must stay a subset of what
@@ -180,24 +181,32 @@ def _inspect_model_path(model_path: str, host: str = "", ssh_port: str = "") ->
 
 
 def setup_hwfit_routes():
-    router = APIRouter(prefix="/api/hwfit", tags=["hwfit"])
+    router = APIRouter(
+        prefix="/api/hwfit",
+        tags=["hwfit"],
+        dependencies=[Depends(require_non_bearer_request)],
+    )
 
     @router.get("/system")
-    def get_system(host: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False):
+    def get_system(host: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False, request: Request = None):
         """Detect and return current system hardware info. Pass host=user@server for remote.
         fresh=true bypasses the per-host cache (the Rescan button)."""
+        if request is not None:
+            require_non_bearer_request(request)
         from services.hwfit.hardware import detect_system
         host, ssh_port = _validate_detection_target(host, ssh_port)
         return detect_system(host=host, ssh_port=ssh_port, platform=platform, fresh=fresh)
 
     @router.get("/models")
-    def get_models(use_case: str = "", sort: str = "newest", limit: int = 50, search: str = "", host: str = "", quant: str = "", ctx: str = "", gpu_count: str = "", gpu_group: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False, refresh_catalog: bool = False, manual_mode: str = "", manual_gpu_count: str = "", manual_vram_gb: str = "", manual_ram_gb: str = "", manual_backend: str = "", ignore_detected_gpu: bool = False, ignore_detected_ram: bool = False, fit_only: bool = False):
+    def get_models(use_case: str = "", sort: str = "newest", limit: int = 50, search: str = "", host: str = "", quant: str = "", ctx: str = "", gpu_count: str = "", gpu_group: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False, refresh_catalog: bool = False, manual_mode: str = "", manual_gpu_count: str = "", manual_vram_gb: str = "", manual_ram_gb: str = "", manual_backend: str = "", ignore_detected_gpu: bool = False, ignore_detected_ram: bool = False, fit_only: bool = False, request: Request = None):
         """Rank LLM models against detected hardware and return scored results.
         gpu_count: override GPU count (0 = CPU only, 1-N = simulate N GPUs of the
             active group). gpu_group: index into system.gpu_groups (the homogeneous
             pools) to target — empty/auto = the largest pool. vLLM can only
             tensor-parallel across identical GPUs, so we never mix pools.
         fresh=true bypasses the hardware-detection cache."""
+        if request is not None:
+            require_non_bearer_request(request)
         from services.hwfit.hardware import detect_system
         from services.hwfit.fit import rank_models
         from services.hwfit.models import get_models, model_catalog_path, refresh_dynamic_catalogs
@@ -316,7 +325,7 @@ def setup_hwfit_routes():
         return payload
 
     @router.get("/profiles")
-    def get_serve_profiles(model: str = "", model_path: str = "", host: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False, serve_weights_gb: float = 0.0, serve_quant: str = ""):
+    def get_serve_profiles(model: str = "", model_path: str = "", host: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False, serve_weights_gb: float = 0.0, serve_quant: str = "", request: Request = None):
         """Compute llama.cpp serve profiles (Quality/Balanced/Speed) for `model`
         against the detected hardware on `host` (or local). Returns concrete
         flags (n_gpu_layers, n_cpu_moe, cache_type, ctx) the serve UI can apply.
@@ -325,6 +334,8 @@ def setup_hwfit_routes():
         catalog (e.g. an ad-hoc HF repo), pass enough hints via a minimal synthetic
         entry isn't possible here, so we return [] and the UI keeps manual flags.
         """
+        if request is not None:
+            require_non_bearer_request(request)
         from services.hwfit.hardware import detect_system
         from services.hwfit.models import get_models
         from services.hwfit.profiles import compute_serve_profiles
@@ -410,8 +421,10 @@ def setup_hwfit_routes():
         }
 
     @router.get("/image-models")
-    def get_image_models(sort: str = "fit", search: str = "", host: str = "", gpu_count: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False, manual_mode: str = "", manual_gpu_count: str = "", manual_vram_gb: str = "", manual_ram_gb: str = "", manual_backend: str = "", ignore_detected_gpu: bool = False, ignore_detected_ram: bool = False):
+    def get_image_models(sort: str = "fit", search: str = "", host: str = "", gpu_count: str = "", ssh_port: str = "", platform: str = "", fresh: bool = False, manual_mode: str = "", manual_gpu_count: str = "", manual_vram_gb: str = "", manual_ram_gb: str = "", manual_backend: str = "", ignore_detected_gpu: bool = False, ignore_detected_ram: bool = False, request: Request = None):
         """Rank image generation models against detected hardware."""
+        if request is not None:
+            require_non_bearer_request(request)
         from services.hwfit.hardware import detect_system
         from services.hwfit.image_models import rank_image_models
         host, ssh_port = _validate_detection_target(host, ssh_port)

routes/memory/memory_routes.py

@@ -1,5 +1,5 @@
 # routes/memory_routes.py
-from fastapi import APIRouter, Form, HTTPException, Request, UploadFile, File
+from fastapi import APIRouter, Depends, Form, HTTPException, Request, UploadFile, File
 from typing import Dict, Any, Optional, List
 import json
 import os
@@ -53,9 +53,19 @@ def _load_for_update(memory_manager) -> List[Dict[str, Any]]:
 
 def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionManager, memory_vector=None):
     """Set up memory-related routes."""
-    router = APIRouter(prefix="/api/memory", tags=["memory"])
+    router = APIRouter(
+        prefix="/api/memory",
+        tags=["memory"],
+        dependencies=[Depends(require_user)],
+    )
 
     def _owner(request: Request) -> Optional[str]:
+        # Router dependencies do not run when a handler is called directly
+        # (including through an integration router), so keep the same bearer
+        # rejection at the owner-resolution seam. ``None`` is retained only
+        # for legacy unit callers; real ASGI requests always carry Request.
+        if request is not None:
+            require_user(request)
         return get_current_user(request)
 
     def _assert_session_owner(session_obj, user):

routes/model_routes.py

@@ -29,7 +29,12 @@ from src.endpoint_resolver import (
     build_models_url,
     build_headers,
 )
-from src.auth_helpers import _auth_disabled, owner_filter, require_chat_scope
+from src.auth_helpers import (
+    _auth_disabled,
+    is_bearer_principal,
+    owner_filter,
+    require_chat_scope,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -1536,7 +1541,12 @@ def setup_model_routes(model_discovery):
                 _refresh_inflight["v"] = False
         threading.Thread(target=_do, daemon=True).start()
 
-    def _fetch_models(owner: str = "", is_admin: bool = False):
+    def _fetch_models(
+        owner: str = "",
+        is_admin: bool = False,
+        *,
+        read_only: bool = False,
+    ):
         """Return model list from cached data (instant). Background refresh keeps caches fresh.
 
         SECURITY: filters endpoints by `owner` — without this the picker
@@ -1551,7 +1561,7 @@ def setup_model_routes(model_discovery):
 
         db = SessionLocal()
         try:
-            if _disable_stale_cookbook_local_endpoints(db):
+            if not read_only and _disable_stale_cookbook_local_endpoints(db):
                 _invalidate_models_cache()
             q = db.query(ModelEndpoint).filter(ModelEndpoint.is_enabled == True)
             if owner and not is_admin:
@@ -1634,6 +1644,17 @@ def setup_model_routes(model_discovery):
         except Exception as e:
             logger.error("Auth gate error in GET /api/models, failing closed: %s", e)
             raise HTTPException(status_code=500, detail="Internal error")
+        bearer = is_bearer_principal(request)
+        if bearer and (refresh or background):
+            raise HTTPException(
+                403,
+                "API tokens may only read the owner-scoped cached model list",
+            )
+        if bearer:
+            # The bearer-compatible path is deliberately read-only: no global
+            # admin view, stale-row cleanup, process cache writes, background
+            # probes, stored endpoint credentials, or refresh state changes.
+            return _fetch_models(owner=owner, is_admin=False, read_only=True)
         # Admins see every endpoint (they manage the global pool); regular
         # users get the owner-scoped view.
         _is_admin = False
@@ -2413,11 +2434,11 @@ def setup_model_routes(model_discovery):
         # no per-user default yet, we resolve via the owner-scoped endpoint
         # lookup below (last-resort: first enabled endpoint THIS user owns).
         # Unauthenticated single-user mode keeps the old behavior.
-        from src.auth_helpers import get_current_user as _gcu
-        try:
-            _user = _gcu(request) or ""
-        except Exception:
-            _user = ""
+        # Resolve through the same owner/scope gate as the model picker. In an
+        # auth-disabled process there is no middleware to stamp token state, so
+        # raw bearer detection must still prevent a token from resolving
+        # global/admin defaults.
+        _user = require_chat_scope(request) or ""
         # Admins resolve via the global defaults (they own them, and the
         # scoped resolution was making the picker disappear for them).
         # Regular users get per-user prefs with NO global fallback for the
@@ -2427,7 +2448,12 @@ def setup_model_routes(model_discovery):
         _is_admin = False
         try:
             auth_mgr = getattr(request.app.state, "auth_manager", None)
-            if _user and auth_mgr is not None and getattr(auth_mgr, "is_admin", None):
+            if (
+                _user
+                and not is_bearer_principal(request)
+                and auth_mgr is not None
+                and getattr(auth_mgr, "is_admin", None)
+            ):
                 _is_admin = bool(auth_mgr.is_admin(_user))
         except Exception:
             _is_admin = False
@@ -2686,8 +2712,13 @@ def setup_model_routes(model_discovery):
     # ── Tool management ──
 
     @router.get("/tools")
-    def list_tools():
+    def list_tools(request: Request):
         """List all available tools with their enabled/disabled status."""
+        # Tool inventory is an interactive/agent capability description, not
+        # part of the narrow bearer chat contract. Cookie/local callers retain
+        # the historical response.
+        from src.auth_helpers import require_non_bearer_request
+        require_non_bearer_request(request)
         from src.agent_tools import TOOL_TAGS
         settings = _load_settings()
         disabled = set(settings.get("disabled_tools", []))

routes/shell_routes.py

@@ -16,6 +16,7 @@ from pathlib import Path
 from typing import Dict, Any
 from core.platform_compat import IS_APPLE_SILICON, which_tool
 from core.middleware import INTERNAL_TOOL_USER
+from src.auth_helpers import is_bearer_principal
 from src.host_docker_access import (
     HOST_DOCKER_ACCESS_HINT,
     host_docker_access_enabled as _host_docker_access_enabled,
@@ -53,6 +54,11 @@ from core.platform_compat import (
 def _require_admin(request: Request):
     """Reject non-admin callers. Shell exec is admin-only — never expose to
     regular users; that's RCE-after-signup."""
+    # This route predates the shared middleware helper and is also called
+    # directly by a few integration paths.  Reject the credential class before
+    # trusting a caller-supplied current_user that might look administrative.
+    if is_bearer_principal(request):
+        raise HTTPException(403, "API tokens cannot use admin host-control surfaces")
     auth_manager = getattr(request.app.state, "auth_manager", None)
     if not auth_manager:
         # No auth at all — only safe in fully-trusted localhost dev mode

routes/workspace_routes.py

@@ -2,7 +2,7 @@
 import os
 from fastapi import APIRouter, Request, HTTPException, Query
 
-from src.auth_helpers import get_current_user
+from src.auth_helpers import get_current_user, require_non_bearer_request
 from src.tool_security import owner_is_admin_or_single_user
 
 # Cap entries returned per directory (mirrors filesystem_tools._CODENAV_MAX_HITS).
@@ -24,6 +24,7 @@ def setup_workspace_routes():
         NON_ADMIN_BLOCKED_TOOLS). A non-admin who can't use those tools must not
         be able to map the host's directory tree either.
         """
+        require_non_bearer_request(request)
         owner = get_current_user(request)
         if not owner_is_admin_or_single_user(owner):
             raise HTTPException(status_code=403, detail="Workspace browsing is admin-only")
@@ -75,6 +76,7 @@ def setup_workspace_routes():
         instead of being stored client-side and silently dropped at chat time.
         Admin-gated like /browse: it confirms path existence on the host.
         """
+        require_non_bearer_request(request)
         owner = get_current_user(request)
         if not owner_is_admin_or_single_user(owner):
             raise HTTPException(status_code=403, detail="Workspace selection is admin-only")

src/auth_helpers.py

@@ -1,6 +1,7 @@
 """Shared auth helpers used by all route files."""
 
 import os
+from dataclasses import dataclass
 from typing import Optional
 from fastapi import Request, HTTPException
 
@@ -11,6 +12,48 @@ from src.owner_identity import (
 )
 
 
+@dataclass(frozen=True)
+class RequestCapability:
+    """Immutable request authority passed through chat execution helpers.
+
+    A bearer token that has the narrow ``chat`` scope is still a pure chat
+    capability.  It may complete the synchronous model call, but it cannot
+    create detached execution, emit interactive events, or schedule follow-up
+    work that would run after the request's authorization context is gone.
+    Cookie and AUTH_ENABLED=false requests retain the existing interactive
+    behavior.
+    """
+
+    principal: str
+    owner: Optional[str]
+    is_bearer: bool
+    allow_deferred_work: bool
+    allow_detached_execution: bool
+    allow_message_events: bool
+    allow_auto_naming: bool
+
+
+def is_bearer_principal(request: Request) -> bool:
+    """Return whether the request is attributable to an API-token principal.
+
+    The auth middleware stamps ``state.api_token`` for a verified token.  The
+    header/sentinel checks keep direct endpoint calls and auth-disabled
+    alternate entry points fail-closed instead of treating the ``api``
+    sentinel as a normal cookie user.
+    """
+    state = getattr(request, "state", None)
+    if getattr(state, "api_token", False) is True:
+        return True
+    current_user = getattr(state, "current_user", None)
+    if isinstance(current_user, str) and current_user.strip().casefold() == "api":
+        return True
+    try:
+        auth_header = request.headers.get("authorization", "")
+    except Exception:
+        auth_header = ""
+    return isinstance(auth_header, str) and auth_header.strip().casefold().startswith("bearer ody_")
+
+
 def get_current_user(request: Request) -> Optional[str]:
     """Get current username from request state (set by auth middleware)."""
     state = getattr(request, "state", None)
@@ -43,9 +86,22 @@ def effective_user(request: Request) -> Optional[str]:
 
 
 def _is_api_token_request(request: Request) -> bool:
-    """Return True when middleware authenticated a bearer API token."""
-    state = getattr(request, "state", None)
-    return getattr(state, "api_token", False) is True
+    """Return True when the request has a bearer API-token principal."""
+    return is_bearer_principal(request)
+
+
+def request_capability(request: Request) -> RequestCapability:
+    """Build the one request capability shared by chat downstream helpers."""
+    bearer = is_bearer_principal(request)
+    return RequestCapability(
+        principal="bearer" if bearer else "interactive",
+        owner=effective_user(request),
+        is_bearer=bearer,
+        allow_deferred_work=not bearer,
+        allow_detached_execution=not bearer,
+        allow_message_events=not bearer,
+        allow_auto_naming=not bearer,
+    )
 
 
 def require_api_token_owner(request: Request) -> str:
@@ -65,7 +121,22 @@ def require_api_token_owner(request: Request) -> str:
         or is_request_sentinel_owner(owner)
     ):
         raise HTTPException(403, "API token has no owner")
-    return owner.strip()
+    normalized_owner = owner.strip()
+    # The normal auth middleware has already resolved this identity from the
+    # token row. Keep the same invariant for direct endpoint calls and
+    # alternate ASGI entry points when a configured auth manager is available.
+    auth_state = getattr(getattr(request, "app", None), "state", None)
+    auth_manager = getattr(auth_state, "auth_manager", None)
+    users = getattr(auth_manager, "users", None)
+    if (
+        getattr(auth_manager, "is_configured", False)
+        and isinstance(users, dict)
+        and normalized_owner.casefold() not in {
+            str(username).strip().casefold() for username in users
+        }
+    ):
+        raise HTTPException(403, "API token owner is not a configured user")
+    return normalized_owner
 
 
 def require_api_token_scope(request: Request, required_scope: str) -> Optional[str]:
@@ -102,13 +173,18 @@ def require_interactive_request(request: Request) -> Optional[str]:
     approve, or otherwise control interactive agent work.
     """
     current_user = get_current_user(request)
-    if _is_api_token_request(request) or (
-        isinstance(current_user, str) and current_user.strip().casefold() == "api"
-    ):
+    if is_bearer_principal(request):
         raise HTTPException(403, "API tokens cannot use this interactive surface")
     return current_user
 
 
+def require_non_bearer_request(request: Request) -> Optional[str]:
+    """Reject bearer principals while preserving cookie/local route behavior."""
+    if is_bearer_principal(request):
+        raise HTTPException(403, "API tokens cannot use this host-control surface")
+    return get_current_user(request)
+
+
 def enforce_api_token_chat_controls(
     request: Request,
     *,
@@ -137,7 +213,7 @@ def require_authenticated_request(request: Request) -> str:
     user data. Owner-scoped routes should use ``require_user`` for browser
     sessions or their own API-token scope/owner gate.
     """
-    if _is_api_token_request(request):
+    if is_bearer_principal(request):
         return require_api_token_owner(request)
     return require_user(request)
 
@@ -178,7 +254,7 @@ def require_user(request: Request) -> str:
     Use this on routes that touch user data so middleware misconfig can't
     open them up.
     """
-    if _is_api_token_request(request):
+    if is_bearer_principal(request):
         raise HTTPException(403, "API tokens must use a scope-aware API route")
 
     u = get_current_user(request)

src/message_metadata.py

@@ -10,6 +10,65 @@ _SERVER_OWNED_MESSAGE_METADATA = frozenset({
     CHAT_SESSION_APPROVAL_CONTEXT_MARKER,
 })
 
+_APPROVAL_PROVENANCE_FIELDS = frozenset({
+    "approval_id",
+    "approved_by_interactive_session",
+    "resolved",
+    "session_id",
+})
+
+
+def _scrub_approval_metadata(value: Any, *, projection: bool, in_approval: bool = False):
+    """Copy metadata while removing fields that can imply approval authority.
+
+    Client ingress drops every server-owned tool-event container.  Context
+    projection keeps harmless server-generated tool-event display data, but
+    strips the approval selectors and resolution/provenance fields from every
+    nested shape.  This makes old rows useful for display without allowing a
+    legacy dict, nested dict, or list-shaped payload to become authority.
+    """
+    if isinstance(value, dict):
+        kind = value.get("kind")
+        approval_scope = in_approval or kind == "tool_approval"
+        cleaned = {}
+        for key, item in value.items():
+            if key in _SERVER_OWNED_MESSAGE_METADATA and not (
+                projection and key == "tool_events"
+            ):
+                continue
+            if key == "tool_approval":
+                continue
+            # These fields have no safe client/display meaning in a message
+            # projection. Strip them even when a legacy writer placed them at
+            # the metadata root instead of under a recognizable approval node.
+            if key in _APPROVAL_PROVENANCE_FIELDS:
+                continue
+            if key == "ask_user":
+                scrubbed = _scrub_approval_metadata(
+                    item, projection=projection, in_approval=True
+                )
+                if scrubbed:
+                    cleaned[key] = scrubbed
+                continue
+            if key == "tool_events":
+                # Some legacy writers placed approval fields directly on an
+                # event rather than under ask_user. Treat the complete event
+                # container as non-authoritative approval-shaped metadata.
+                cleaned[key] = _scrub_approval_metadata(
+                    item, projection=projection, in_approval=True
+                )
+                continue
+            cleaned[key] = _scrub_approval_metadata(
+                item, projection=projection, in_approval=approval_scope
+            )
+        return cleaned
+    if isinstance(value, list):
+        return [
+            _scrub_approval_metadata(item, projection=projection, in_approval=in_approval)
+            for item in value
+        ]
+    return value
+
 
 def sanitize_client_message_metadata(metadata: Any) -> Optional[dict]:
     """Normalize client metadata and drop server-owned fields.
@@ -24,9 +83,19 @@ def sanitize_client_message_metadata(metadata: Any) -> Optional[dict]:
         return None
     if not isinstance(metadata, dict):
         return None
-    sanitized = {
-        key: value
-        for key, value in metadata.items()
-        if key not in _SERVER_OWNED_MESSAGE_METADATA
-    }
+    sanitized = _scrub_approval_metadata(metadata, projection=False)
     return sanitized or None
+
+
+def sanitize_projected_message_metadata(metadata: Any) -> Optional[dict]:
+    """Return a model-context copy with approval provenance stripped.
+
+    The projection path may retain non-authoritative tool-event details for
+    continuity, but it never projects the raw chat-session marker or the
+    legacy fields that used to be interpreted as a durable approval grant.
+    A separate server-owned grant store is the only source for that marker.
+    """
+    if not isinstance(metadata, dict):
+        return None
+    cleaned = _scrub_approval_metadata(metadata, projection=True)
+    return cleaned or None

src/tool_approval_provenance.py

@@ -0,0 +1,178 @@
+"""Durable server-owned provenance for chat-session tool approvals.
+
+Approval cards and their resolution fields live in the chat transcript for
+display compatibility.  They are intentionally not authority.  This module
+owns the separate database row that can be created only after an interactive
+server-side ``ExactToolApproval`` was consumed for the matching session and
+owner.  Legacy transcript rows are never migrated into this table.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any, Optional
+
+from fastapi import HTTPException
+
+from src.owner_identity import auth_disabled, is_request_sentinel_owner
+
+logger = logging.getLogger(__name__)
+
+_PROVENANCE_VERSION = 1
+
+
+def _owner_key(owner: Any) -> str:
+    value = str(owner or "").strip().casefold()
+    if not value or is_request_sentinel_owner(value):
+        return ""
+    return value
+
+
+def _approval_binding_is_valid(
+    approval: Any,
+    *,
+    approval_id: str,
+    session_id: str,
+    owner_key: str,
+) -> bool:
+    """Require the exact consumed chat-scope grant before inserting a row."""
+    if approval is None or not getattr(approval, "grants_chat_session", False):
+        return False
+    pending = getattr(approval, "pending", None)
+    if pending is None:
+        return False
+    return (
+        str(getattr(pending, "approval_id", "") or "") == approval_id
+        and str(getattr(pending, "session_id", "") or "") == session_id
+        and _owner_key(getattr(pending, "owner", None)) == owner_key
+    )
+
+
+def create_chat_session_approval_grant(
+    request,
+    *,
+    approval: Any,
+    approval_id: Any,
+    session_id: Any,
+    owner: Any,
+) -> bool:
+    """Persist one interactive chat-session approval grant.
+
+    The caller must supply the exact in-memory approval object returned by the
+    one-use store.  Bearer principals are rejected even if they present a
+    client-shaped approval payload.  A database failure fails closed by
+    returning ``False``: it never manufactures an in-memory durable grant.
+    """
+    from src.auth_helpers import effective_user, require_interactive_request
+
+    require_interactive_request(request)
+    from src.tool_approvals import ExactToolApproval
+
+    # This proof is set only by ToolApprovalStore.consume().  In particular,
+    # a client-shaped dict or a hand-constructed ExactToolApproval is not an
+    # interactive approval event and cannot mint durable authority.
+    if not isinstance(approval, ExactToolApproval) or not getattr(
+        approval, "_consumed_from_store", False
+    ):
+        return False
+    approval_key = str(approval_id or "")
+    session_key = str(session_id or "")
+    requested_owner_key = _owner_key(owner)
+    if not approval_key or not session_key or (
+        not requested_owner_key and not auth_disabled()
+    ):
+        return False
+    if not _approval_binding_is_valid(
+        approval,
+        approval_id=approval_key,
+        session_id=session_key,
+        owner_key=requested_owner_key,
+    ):
+        return False
+
+    request_owner_key = _owner_key(effective_user(request))
+    if not auth_disabled() and request_owner_key != requested_owner_key:
+        raise HTTPException(403, "Approval owner does not match the interactive principal")
+
+    # Import lazily so the pure request/auth helpers do not create a database
+    # import cycle during application startup.
+    from core.database import (
+        ChatSessionApprovalGrant,
+        Session as DbSession,
+        SessionLocal,
+    )
+
+    db = SessionLocal()
+    try:
+        session_row = db.query(DbSession).filter(DbSession.id == session_key).first()
+        if session_row is None:
+            return False
+        stored_owner_key = _owner_key(getattr(session_row, "owner", None))
+        if stored_owner_key != requested_owner_key:
+            # AUTH_ENABLED=false is a deliberate single-user compatibility
+            # mode. It may reopen an owner-stamped legacy session, but the
+            # grant remains bound to that stored owner for projection.
+            if not auth_disabled() or requested_owner_key:
+                return False
+            grant_owner_key = stored_owner_key
+        else:
+            grant_owner_key = requested_owner_key
+
+        existing = db.query(ChatSessionApprovalGrant).filter(
+            ChatSessionApprovalGrant.session_id == session_key,
+            ChatSessionApprovalGrant.owner == grant_owner_key,
+            ChatSessionApprovalGrant.approval_id == approval_key,
+            ChatSessionApprovalGrant.provenance_version == _PROVENANCE_VERSION,
+        ).first()
+        if existing is not None:
+            return True
+
+        db.add(ChatSessionApprovalGrant(
+            id=uuid.uuid4().hex,
+            session_id=session_key,
+            owner=grant_owner_key,
+            approval_id=approval_key,
+            provenance_version=_PROVENANCE_VERSION,
+        ))
+        db.commit()
+        return True
+    except Exception:
+        db.rollback()
+        logger.warning("Could not persist chat-session approval provenance", exc_info=True)
+        return False
+    finally:
+        db.close()
+
+
+def has_chat_session_approval_grant(
+    session_id: Any,
+    owner: Optional[Any],
+) -> bool:
+    """Return whether the exact owner/session has a server-owned grant."""
+    session_key = str(session_id or "")
+    owner_key = _owner_key(owner)
+    if (
+        not session_key
+        or (isinstance(owner, str) and is_request_sentinel_owner(owner))
+        or (not owner_key and not auth_disabled())
+    ):
+        return False
+
+    from core.database import ChatSessionApprovalGrant, SessionLocal
+
+    db = SessionLocal()
+    try:
+        return db.query(ChatSessionApprovalGrant).filter(
+            ChatSessionApprovalGrant.session_id == session_key,
+            ChatSessionApprovalGrant.owner == owner_key,
+            ChatSessionApprovalGrant.provenance_version == _PROVENANCE_VERSION,
+        ).first() is not None
+    except Exception:
+        # Existing installations are upgraded lazily by Base.metadata.create_all
+        # at startup. Until that has happened, ignoring the absent table is the
+        # safe migration behavior: legacy history can never grant authority.
+        logger.debug("Chat-session approval provenance lookup unavailable", exc_info=True)
+        return False
+    finally:
+        db.close()

src/tool_approval_scopes.py

@@ -12,8 +12,9 @@ TASK_APPROVAL_DECISION = "approve_task"
 CHAT_SESSION_APPROVAL_DECISION = "approve"
 DENY_APPROVAL_DECISION = "deny"
 
-# Session.get_context_messages() adds this server-owned marker only when the
-# session history contains a matching, resolved chat-session approval.
+# Session.get_context_messages() adds this server-owned marker only when a
+# separate, immutable, owner/session-bound approval grant exists. Transcript
+# metadata is display-only and never establishes the marker.
 CHAT_SESSION_APPROVAL_CONTEXT_MARKER = "_tool_approval_chat_session_granted"
 
 

src/tool_approvals.py

@@ -228,6 +228,10 @@ class ExactToolApproval:
     # sealed action.
     allow_remaining_actions: bool = True
     _claimed: bool = field(default=False, init=False, repr=False)
+    # Only ToolApprovalStore.consume() may set this proof.  A caller cannot
+    # manufacture chat-session provenance by constructing an ExactToolApproval
+    # around a browser-shaped PendingToolApproval.
+    _consumed_from_store: bool = field(default=False, init=False, repr=False)
     _lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False)
 
     @property
@@ -462,16 +466,19 @@ class ToolApprovalStore:
         if scope is None:
             return None
         if not allow_continuation:
-            return ExactToolApproval(
+            grant = ExactToolApproval(
                 pending,
                 scope=ToolApprovalScope.SINGLE_ACTION,
                 allow_remaining_actions=False,
             )
-        return ExactToolApproval(
-            pending,
-            scope=scope,
-            allow_remaining_actions=True,
-        )
+        else:
+            grant = ExactToolApproval(
+                pending,
+                scope=scope,
+                allow_remaining_actions=True,
+            )
+        grant._consumed_from_store = True
+        return grant
 
     def peek(self, approval_id: Any) -> PendingToolApproval | None:
         now = time.time()

tests/test_api_token_forward_fixes.py

@@ -552,3 +552,298 @@ async def test_stream_bearer_chat_cannot_dispatch_image_generation(monkeypatch):
     assert exc.value.status_code == 403
     assert "image" in str(exc.value.detail).lower()
     assert "chat" not in captured
+
+
+def test_nested_legacy_approval_shapes_are_stripped_on_ingress_and_projection():
+    from src.message_metadata import sanitize_projected_message_metadata
+
+    metadata = {
+        "safe": {"label": "keep"},
+        CHAT_SESSION_APPROVAL_CONTEXT_MARKER: True,
+        "approval_id": "legacy-root",
+        "resolved": "approve",
+        "session_id": "session-1",
+        "tool_events": [
+            {
+                "ask_user": {
+                    "kind": "tool_approval",
+                    "approval_id": "legacy-nested",
+                    "resolved": "approve",
+                    "approved_by_interactive_session": True,
+                    "session_id": "session-1",
+                    "label": "Allow",
+                }
+            },
+            {
+                "kind": "tool_approval",
+                "approval_id": "legacy-direct",
+                "resolved": "approve",
+                "session_id": "session-1",
+                "label": "Allow",
+            },
+            ["not-a-metadata-mapping"],
+        ],
+    }
+
+    client = sanitize_client_message_metadata(metadata)
+    assert client == {"safe": {"label": "keep"}}
+
+    projected = sanitize_projected_message_metadata(metadata)
+    assert projected["safe"] == {"label": "keep"}
+    assert CHAT_SESSION_APPROVAL_CONTEXT_MARKER not in projected
+    assert "approval_id" not in projected
+    assert "resolved" not in projected
+    assert "session_id" not in projected
+    assert "tool_events" in projected
+    assert projected["tool_events"][0]["ask_user"] == {
+        "kind": "tool_approval",
+        "label": "Allow",
+    }
+    assert projected["tool_events"][1] == {
+        "kind": "tool_approval",
+        "label": "Allow",
+    }
+
+
+def test_session_metadata_parser_rejects_list_of_pairs_and_non_dict_values():
+    from core.session_manager import _parse_message_metadata
+
+    assert _parse_message_metadata(
+        '[["tool_events", [{"ask_user": {"resolved": "approve"}}]]]'
+    ) == {}
+    assert _parse_message_metadata('["approval_id", "forged"]') == {}
+    assert _parse_message_metadata('"forged"') == {}
+    assert _parse_message_metadata("not-json") == {}
+    assert _parse_message_metadata('{"safe": true}') == {"safe": True}
+
+
+def test_hand_constructed_approval_cannot_mint_durable_chat_provenance():
+    from src.tool_approval_provenance import create_chat_session_approval_grant
+    from src.tool_approvals import ExactToolApproval, ToolApprovalStore
+    from src.tool_capabilities import capabilities_for_action
+
+    store = ToolApprovalStore()
+    pending = store.create(
+        owner="alice",
+        session_id="session-1",
+        origin_run_id="run-1",
+        tool_name="bash",
+        content="printf safe",
+        workspace=None,
+        external_untrusted_context_seen=False,
+        capabilities=capabilities_for_action("bash", "printf safe"),
+    )
+    forged = ExactToolApproval(pending)
+    with pytest.raises(HTTPException) as bearer_exc:
+        create_chat_session_approval_grant(
+            _request(),
+            approval=forged,
+            approval_id=pending.approval_id,
+            session_id="session-1",
+            owner="alice",
+        )
+    assert bearer_exc.value.status_code == 403
+
+    request = _request(api_token=False, owner="alice", scopes=(), current_user="alice")
+    assert create_chat_session_approval_grant(
+        request,
+        approval=forged,
+        approval_id=pending.approval_id,
+        session_id="session-1",
+        owner="alice",
+    ) is False
+
+
+@pytest.mark.asyncio
+async def test_bearer_memory_routes_reject_router_and_direct_entry_points(monkeypatch):
+    import routes.memory_routes as memory_routes
+    import inspect
+
+    memory_manager = MagicMock()
+    session_manager = MagicMock()
+    router = memory_routes.setup_memory_routes(memory_manager, session_manager)
+    request = _request()
+    direct_cases = [
+        ("/api/memory/debug", "POST", {"query": "secret"}),
+        ("/api/memory/add", "POST", {}),
+        ("/api/memory", "GET", {}),
+        ("/api/memory/search", "POST", {"query": "secret", "session_id": None, "category": None}),
+        ("/api/memory/timeline", "GET", {}),
+        ("/api/memory/by-session/{session_id}", "GET", {"session_id": "session-1"}),
+        ("/api/memory/extract", "POST", {"session": "session-1"}),
+        ("/api/memory/audit", "POST", {"session": None}),
+        ("/api/memory/import", "POST", {"session": None, "file": None}),
+        ("/api/memory/{memory_id}/pin", "POST", {"memory_id": "memory-1"}),
+        ("/api/memory/{memory_id}", "GET", {"memory_id": "memory-1"}),
+        ("/api/memory/{memory_id}", "PUT", {"memory_id": "memory-1", "text": "replacement", "category": None}),
+        ("/api/memory/{memory_id}", "DELETE", {"memory_id": "memory-1"}),
+    ]
+    for path, method, kwargs in direct_cases:
+        endpoint = next(
+            route.endpoint
+            for route in router.routes
+            if route.path == path and method in route.methods
+        )
+        with pytest.raises(HTTPException) as exc:
+            result = endpoint(request, **kwargs)
+            if inspect.isawaitable(result):
+                await result
+        assert exc.value.status_code == 403, path
+    memory_manager.load.assert_not_called()
+
+    app = FastAPI()
+    app.include_router(router)
+    headers = {
+        "x-api-token": "1",
+        "x-api-owner": "alice",
+        "x-api-scopes": "chat",
+    }
+    async with _client(_PrincipalState(app)) as client:
+        response = await client.get("/api/memory", headers=headers)
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_bearer_capability_suppresses_deferred_callbacks_and_message_events(monkeypatch):
+    from routes import chat_helpers
+    from src.auth_helpers import request_capability
+
+    request = _request()
+    capability = request_capability(request)
+    assert capability.is_bearer is True
+    assert capability.allow_deferred_work is False
+    assert capability.allow_detached_execution is False
+    assert capability.allow_message_events is False
+    assert capability.allow_auto_naming is False
+
+    sess = SimpleNamespace(
+        history=[object()] * 8,
+        endpoint_url="https://selected.example/v1",
+        model="selected-model",
+        headers={"Authorization": "Bearer selected"},
+        name="New chat",
+        add_message=MagicMock(),
+    )
+    webhook_manager = MagicMock()
+    monkeypatch.setattr(
+        chat_helpers,
+        "_spawn_bg",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("bearer scheduled work")),
+    )
+    monkeypatch.setattr(
+        chat_helpers,
+        "accumulate_token_usage",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("bearer usage callback")),
+    )
+    chat_helpers.run_post_response_tasks(
+        sess,
+        SimpleNamespace(),
+        "session-1",
+        "hello",
+        "answer",
+        {"prompt_tokens": 1},
+        {"auto_memory": True, "auto_skills": True},
+        MagicMock(),
+        MagicMock(),
+        webhook_manager,
+        agent_rounds=3,
+        agent_tool_calls=3,
+        skills_manager=MagicMock(),
+        owner="alice",
+        capability=capability,
+    )
+    webhook_manager.fire_and_forget.assert_not_called()
+
+    with_marker = SimpleNamespace(
+        get_context_messages=lambda: [{
+            "role": "user",
+            "content": "prior",
+            "metadata": {CHAT_SESSION_APPROVAL_CONTEXT_MARKER: True, "safe": "yes"},
+        }]
+    )
+    projected = chat_helpers._history_for_request_capability(with_marker, capability)
+    assert projected == [{"role": "user", "content": "prior", "metadata": {"safe": "yes"}}]
+
+    chat_helpers.fire_message_event(
+        request,
+        webhook_manager,
+        "session-1",
+        sess,
+        "hello",
+        capability=capability,
+    )
+    webhook_manager.fire_and_forget.assert_not_called()
+
+    chat_handler = MagicMock()
+    chat_helpers.add_user_message(
+        sess,
+        chat_handler,
+        SimpleNamespace(
+            attachment_meta=[],
+            user_content="hello",
+            text_for_context="hello",
+        ),
+        capability=capability,
+    )
+    chat_handler.update_session_name_if_needed.assert_not_called()
+
+
+def test_bearer_cannot_reach_workspace_or_hwfit_direct_handlers(monkeypatch):
+    from routes import hwfit_routes, workspace_routes
+
+    bearer = _request()
+    workspace_router = workspace_routes.setup_workspace_routes()
+    browse = next(route.endpoint for route in workspace_router.routes if route.path == "/api/workspace/browse")
+    vet = next(route.endpoint for route in workspace_router.routes if route.path == "/api/workspace/vet")
+    with pytest.raises(HTTPException):
+        browse(bearer, path="/")
+    with pytest.raises(HTTPException):
+        vet(bearer, path="/")
+
+    hwfit_router = hwfit_routes.setup_hwfit_routes()
+    for path in ("/api/hwfit/system", "/api/hwfit/models", "/api/hwfit/profiles", "/api/hwfit/image-models"):
+        endpoint = next(route.endpoint for route in hwfit_router.routes if route.path == path)
+        with pytest.raises(HTTPException):
+            endpoint(request=bearer)
+
+
+@pytest.mark.asyncio
+async def test_codex_bearer_rejected_before_direct_and_router_host_control(monkeypatch):
+    import routes.codex_routes as codex_routes
+
+    router = codex_routes.setup_codex_routes()
+    bearer = _request(scopes=("chat", "cookbook:read", "cookbook:launch"))
+    direct_cases = [
+        ("/api/codex/capabilities", "GET", (bearer,)),
+        ("/api/codex/plugin.zip", "GET", (bearer,)),
+        ("/api/codex/cookbook/tasks", "GET", (bearer,)),
+        ("/api/codex/cookbook/serve", "POST", (bearer, {})),
+        ("/api/codex/cookbook/output/{session_id}", "GET", (bearer, "serve-1")),
+    ]
+    for path, method, args in direct_cases:
+        endpoint = next(
+            route.endpoint
+            for route in router.routes
+            if route.path == path and method in route.methods
+        )
+        with pytest.raises(HTTPException) as exc:
+            result = endpoint(*args)
+            if hasattr(result, "__await__"):
+                await result
+        assert exc.value.status_code == 403
+
+    app = FastAPI()
+    app.include_router(router)
+    headers = {
+        "x-api-token": "1",
+        "x-api-owner": "alice",
+        "x-api-scopes": "cookbook:read,cookbook:launch",
+    }
+    async with _client(_PrincipalState(app)) as client:
+        for method, path, kwargs in (
+            ("GET", "/api/codex/capabilities", {}),
+            ("GET", "/api/codex/cookbook/tasks", {}),
+            ("POST", "/api/codex/cookbook/serve", {"json": {}}),
+        ):
+            response = await client.request(method, path, headers=headers, **kwargs)
+            assert response.status_code == 403, (path, response.text)

tests/test_codex_cookbook_admin_gate.py

@@ -5,8 +5,8 @@ routes (tasks, servers, output, stop, adopt, presets, etc.) through
 normal cookie sessions because _scope_owner only checked login status,
 not admin privileges.
 
-After the fix, cookie-session callers must be admin; API-token callers
-are still governed by scope checks only.
+After the fix, cookie-session callers must be admin and bearer callers are
+rejected from the Codex host-control plane regardless of legacy scopes.
 """
 import pytest
 from types import SimpleNamespace
@@ -80,13 +80,14 @@ class TestCookieSessionAdminGate:
 
 
 class TestApiTokenScopeGate:
-    """API-token callers are governed by scope, not admin status."""
+    """Bearer callers cannot enter Codex host-control routes."""
 
-    def test_token_with_scope_allowed(self, monkeypatch):
+    def test_token_with_legacy_scope_rejected(self, monkeypatch):
         monkeypatch.setenv("AUTH_ENABLED", "true")
         req = _api_token_request(scopes=["cookbook:read"])
-        owner = _require_cookbook_scope(req, COOKBOOK_READ_SCOPES)
-        assert owner == "alice"
+        with pytest.raises(HTTPException) as exc:
+            _require_cookbook_scope(req, COOKBOOK_READ_SCOPES)
+        assert exc.value.status_code == 403
 
     def test_token_missing_scope_rejected(self, monkeypatch):
         monkeypatch.setenv("AUTH_ENABLED", "true")

tests/test_codex_ssh_host_validation.py

@@ -8,6 +8,7 @@ These pin validation on the host/port before they reach the ssh string, matching
 the validators the rest of the cookbook routes already apply.
 """
 import asyncio
+from types import SimpleNamespace
 
 import pytest
 from fastapi import APIRouter, HTTPException
@@ -56,6 +57,30 @@ def _codex_request(scopes) -> Request:
     return request
 
 
+def _interactive_request(path="/api/codex/documents") -> Request:
+    app = SimpleNamespace(
+        state=SimpleNamespace(
+            auth_manager=SimpleNamespace(
+                is_configured=True,
+                is_admin=lambda username: username == "alice",
+            )
+        )
+    )
+    request = Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "path": path,
+            "headers": [],
+            "state": {},
+            "app": app,
+        }
+    )
+    request.state.current_user = "alice"
+    request.state.api_token = False
+    return request
+
+
 def test_rejects_remote_host_with_shell_metacharacters():
     task = {"remoteHost": "box; rm -rf ~", "sshPort": ""}
     with pytest.raises(HTTPException) as exc:
@@ -135,7 +160,7 @@ def _documents_endpoint(total: int):
 async def test_documents_pagination_clamps_offset_and_limit():
     endpoint, calls = _documents_endpoint(total=99)
 
-    result = await endpoint(_codex_request(["documents:read"]), offset=-10, limit=500)
+    result = await endpoint(_interactive_request(), offset=-10, limit=500)
 
     assert calls[-1]["owner"] == "alice"
     assert calls[-1]["offset"] == 0
@@ -148,7 +173,7 @@ async def test_documents_pagination_clamps_offset_and_limit():
 async def test_documents_pagination_clamps_zero_limit_to_one():
     endpoint, calls = _documents_endpoint(total=3)
 
-    result = await endpoint(_codex_request(["documents:read"]), offset=0, limit=0)
+    result = await endpoint(_interactive_request(), offset=0, limit=0)
 
     assert calls[-1]["limit"] == 1
     assert len(result["documents"]) == 1
@@ -159,7 +184,7 @@ async def test_documents_pagination_clamps_zero_limit_to_one():
 async def test_documents_pagination_returns_next_offset_when_truncated():
     endpoint, _calls = _documents_endpoint(total=7)
 
-    result = await endpoint(_codex_request(["documents:read"]), offset=2, limit=3)
+    result = await endpoint(_interactive_request(), offset=2, limit=3)
 
     assert [doc["id"] for doc in result["documents"]] == ["doc-2", "doc-3", "doc-4"]
     assert result["next_offset"] == 5
@@ -170,7 +195,7 @@ async def test_documents_pagination_rejects_invalid_offset():
     endpoint, _calls = _documents_endpoint(total=7)
 
     with pytest.raises(HTTPException) as exc:
-        await endpoint(_codex_request(["documents:read"]), offset="soon", limit=3)
+        await endpoint(_interactive_request(), offset="soon", limit=3)
 
     assert exc.value.status_code == 400
     assert exc.value.detail == "Invalid offset"
@@ -181,7 +206,7 @@ async def test_documents_pagination_rejects_invalid_limit():
     endpoint, _calls = _documents_endpoint(total=7)
 
     with pytest.raises(HTTPException) as exc:
-        await endpoint(_codex_request(["documents:read"]), offset=0, limit="many")
+        await endpoint(_interactive_request(), offset=0, limit="many")
 
     assert exc.value.status_code == 400
     assert exc.value.detail == "Invalid limit"
@@ -191,7 +216,7 @@ async def test_documents_pagination_rejects_invalid_limit():
 async def test_documents_pagination_out_of_range_offset_returns_empty_page():
     endpoint, calls = _documents_endpoint(total=3)
 
-    result = await endpoint(_codex_request(["documents:read"]), offset=10, limit=2)
+    result = await endpoint(_interactive_request(), offset=10, limit=2)
 
     assert calls[-1]["offset"] == 10
     assert calls[-1]["limit"] == 2
@@ -217,7 +242,7 @@ def test_adopt_rejects_ssh_option_host_before_shell(monkeypatch, host_field):
     }
 
     with pytest.raises(HTTPException) as exc:
-        asyncio.run(endpoint(_launch_request(), body))
+        asyncio.run(endpoint(_interactive_request("/api/codex/cookbook/adopt"), body))
 
     assert exc.value.status_code == 400
     assert calls == []
@@ -237,7 +262,7 @@ async def test_email_draft_document_accepts_send_scope_with_document_write():
     endpoint = _route_endpoint("/api/codex/emails/draft-document", "POST", router=router)
 
     result = await endpoint(
-        _codex_request(["email:send", "documents:write"]),
+        _interactive_request("/api/codex/emails/draft-document"),
         {"to": "recipient@example.com", "subject": "Subject", "body": "Body"},
     )
 

tests/test_kv_cache_invalidation_2927.py

@@ -69,7 +69,7 @@ def _build_context_harness(monkeypatch, chat_helpers, history):
             temperature=0.7, max_tokens=1024, system_prompt="You are Odysseus.", character_name=None,
         )
 
-    def fake_add_user_message(sess, chat_handler, preprocessed, incognito=False):
+    def fake_add_user_message(sess, chat_handler, preprocessed, incognito=False, capability=None):
         sess.messages.append({"role": "user", "content": preprocessed.user_content})
 
     async def fake_maybe_compact(sess, endpoint_url, model, messages, headers, owner=None):

tests/test_model_routes.py

@@ -1744,6 +1744,11 @@ def test_api_models_scopes_api_token_to_token_owner(monkeypatch):
     monkeypatch.setattr(model_routes, "ModelEndpoint", _RouteModelEndpoint)
     monkeypatch.setattr(model_routes, "SessionLocal", lambda: db)
     monkeypatch.setattr(threading, "Thread", _NoopThread)
+    monkeypatch.setattr(
+        model_routes,
+        "_disable_stale_cookbook_local_endpoints",
+        lambda _db: (_ for _ in ()).throw(AssertionError("bearer read touched stale-row state")),
+    )
 
     request = SimpleNamespace(
         state=SimpleNamespace(
@@ -1765,7 +1770,58 @@ def test_api_models_scopes_api_token_to_token_owner(monkeypatch):
     result = _route_endpoint(router, "/api/models")(request)
 
     assert [item["endpoint_name"] for item in result["items"]] == ["alice", "shared"]
-    assert admin_checks == ["alice"]
+    assert admin_checks == []
+
+
+def test_bearer_model_refresh_and_background_flags_fail_before_management_work(monkeypatch):
+    router = model_routes.setup_model_routes(model_discovery=None)
+
+    def fail_session():
+        raise AssertionError("bearer refresh reached the model database")
+
+    monkeypatch.setattr(model_routes, "SessionLocal", fail_session)
+    request = SimpleNamespace(
+        state=SimpleNamespace(
+            current_user="api",
+            api_token=True,
+            api_token_owner="alice",
+            api_token_scopes=["chat"],
+        ),
+        app=SimpleNamespace(
+            state=SimpleNamespace(
+                auth_manager=SimpleNamespace(is_configured=True),
+            ),
+        ),
+    )
+    endpoint = _route_endpoint(router, "/api/models")
+    for flags in ({"refresh": True}, {"background": True}, {"refresh": True, "background": True}):
+        with pytest.raises(HTTPException) as exc:
+            endpoint(request, **flags)
+        assert exc.value.status_code == 403
+
+
+def test_bearer_is_rejected_from_model_management_and_tool_inventory_routes():
+    from fastapi import Response
+
+    router = model_routes.setup_model_routes(model_discovery=None)
+    request = SimpleNamespace(
+        state=SimpleNamespace(
+            current_user="api",
+            api_token=True,
+            api_token_owner="alice",
+            api_token_scopes=["chat"],
+        ),
+        app=SimpleNamespace(state=SimpleNamespace(auth_manager=None)),
+    )
+    with pytest.raises(HTTPException) as tools_exc:
+        _route_endpoint(router, "/api/tools")(request)
+    assert tools_exc.value.status_code == 403
+
+    with pytest.raises(HTTPException) as endpoint_exc:
+        _route_endpoint(router, "/api/model-endpoints/{ep_id}/models")(
+            "endpoint-1", request, Response()
+        )
+    assert endpoint_exc.value.status_code == 403
 
 
 def test_api_models_returns_only_pinned_proxy_models_without_refresh_probe(monkeypatch):

tests/test_tool_approval_task_scope.py

@@ -7,6 +7,8 @@ from pathlib import Path
 from types import SimpleNamespace
 
 from core.models import ChatMessage, Session
+from core.database import Session as DbSession
+from tests.helpers.sqlite_db import make_temp_sqlite
 from src.tool_approval_scopes import (
     CHAT_SESSION_APPROVAL_CONTEXT_MARKER,
     ToolApprovalScope,
@@ -92,7 +94,7 @@ def test_allow_for_task_bypasses_only_the_resumed_run_gate():
     assert fresh.decision_for("bash").allowed is False
 
 
-def test_allow_for_chat_session_applies_to_later_turns_in_only_that_chat():
+def test_allow_for_chat_session_applies_to_later_turns_in_only_that_chat(monkeypatch):
     store = ToolApprovalStore()
     pending = _pending(store, selected_tools=["bash", "manage_skills"])
     grant = store.consume(
@@ -109,6 +111,45 @@ def test_allow_for_chat_session_applies_to_later_turns_in_only_that_chat():
     assert grant.pending.selected_tools == ("bash", "manage_skills")
     assert grant.pending.continuation_query.startswith("inspect the project")
 
+    # The browser-shaped resolution card is not enough to create authority.
+    # Persist the separate grant through the same helper used by the route,
+    # backed by a fresh database so reload behavior is real rather than a
+    # monkeypatched history predicate.
+    monkeypatch.delenv("AUTH_ENABLED", raising=False)
+    import core.database as database
+    from src.tool_approval_provenance import create_chat_session_approval_grant
+
+    db_factory, _engine, _tmpfile = make_temp_sqlite(database.Base.metadata)
+    monkeypatch.setattr(database, "SessionLocal", db_factory)
+    db = db_factory()
+    try:
+        db.add(
+            DbSession(
+                id="session-1",
+                name="Chat",
+                endpoint_url="http://example.invalid",
+                model="test",
+                owner="Alice",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+    interactive_request = SimpleNamespace(
+        state=SimpleNamespace(
+            api_token=False,
+            current_user="alice",
+        ),
+        headers={},
+    )
+    assert create_chat_session_approval_grant(
+        interactive_request,
+        approval=grant,
+        approval_id=pending.approval_id,
+        session_id="session-1",
+        owner="alice",
+    ) is True
+
     resolved_card = pending.public_payload()
     resolved_card["resolved"] = "approve"
     resolved_card["approved_by_interactive_session"] = True
@@ -125,6 +166,7 @@ def test_allow_for_chat_session_applies_to_later_turns_in_only_that_chat():
         name="Chat",
         endpoint_url="http://example.invalid",
         model="test",
+        owner="Alice",
         history=history,
     )
 
@@ -137,6 +179,31 @@ def test_allow_for_chat_session_applies_to_later_turns_in_only_that_chat():
     assert future_turn.approval_gate_bypassed is True
     assert future_turn.decision_for("bash").allowed is True
 
+    # A fresh in-memory Session object with the same durable id/owner sees the
+    # grant after a simulated reload; the transcript card itself is still only
+    # display metadata.
+    reloaded = Session(
+        id="session-1",
+        name="Reloaded",
+        endpoint_url="http://example.invalid",
+        model="test",
+        owner="alice",
+        history=[ChatMessage("user", "after reload")],
+    )
+    assert reloaded.get_context_messages()[-1]["metadata"][CHAT_SESSION_APPROVAL_CONTEXT_MARKER] is True
+
+    wrong_owner = Session(
+        id="session-1",
+        name="Wrong owner",
+        endpoint_url="http://example.invalid",
+        model="test",
+        owner="bob",
+        history=[ChatMessage("user", "cross-owner")],
+    )
+    assert CHAT_SESSION_APPROVAL_CONTEXT_MARKER not in (
+        wrong_owner.get_context_messages()[-1].get("metadata") or {}
+    )
+
     # The persisted card is bound to its original chat id, so a fork/copy does
     # not inherit the grant merely by copying transcript metadata.
     other_session = Session(