diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 592f972cc..73c528496 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -604,6 +604,52 @@ class ResponseStore: return row[0] if row else 0 +class _ResponseStoreProxy: + """Backward-compatible lazy facade for ``adapter._response_store``. + + Tests and older single-profile call paths reach into ``adapter._response_store`` + directly. Keep that surface, but defer opening SQLite until an actual store + method is invoked so high-fixture test runs do not exhaust file descriptors. + """ + + def __init__(self, adapter: "APIServerAdapter") -> None: + self._adapter = adapter + + def _store(self) -> Optional[ResponseStore]: + return self._adapter._ensure_response_store() + + def get(self, response_id: str) -> Optional[Dict[str, Any]]: + store = self._store() + return store.get(response_id) if store is not None else None + + def put(self, response_id: str, data: Dict[str, Any]) -> None: + store = self._store() + if store is not None: + store.put(response_id, data) + + def delete(self, response_id: str) -> bool: + store = self._store() + return store.delete(response_id) if store is not None else False + + def get_conversation(self, name: str) -> Optional[str]: + store = self._store() + return store.get_conversation(name) if store is not None else None + + def set_conversation(self, name: str, response_id: str) -> None: + store = self._store() + if store is not None: + store.set_conversation(name, response_id) + + def close(self) -> None: + store = self._store() + if store is not None: + store.close() + + def __len__(self) -> int: + store = self._store() + return len(store) if store is not None else 0 + + # --------------------------------------------------------------------------- # CORS middleware # --------------------------------------------------------------------------- @@ -1150,7 +1196,15 @@ class APIServerAdapter(BasePlatformAdapter): self._app: Optional["web.Application"] = None self._runner: Optional["web.AppRunner"] = None self._site: Optional["web.TCPSite"] = None - self._response_store = ResponseStore() + # ResponseStore must follow the active profile home just like state.db. + # Verified-principal routing means different users land in different + # principal_* homes, so a single adapter-global response_store would let + # previous_response_id / conversation chains bleed across principals. + # Keep ``_response_store`` as a lazy compatibility facade for tests and + # older single-profile paths; scoped requests still resolve the concrete + # store via ``_ensure_response_store()`` inside the active profile scope. + self._response_store: Optional[Any] = _ResponseStoreProxy(self) + self._response_stores: Dict[str, Any] = {} # Active run streams: run_id -> asyncio.Queue of SSE event dicts self._run_streams: Dict[str, "asyncio.Queue[Optional[Dict]]"] = {} # Creation timestamps for orphaned-run TTL sweep @@ -1672,6 +1726,36 @@ class APIServerAdapter(BasePlatformAdapter): logger.debug("SessionDB unavailable for API server: %s", e) return None + def _ensure_response_store(self): + """Lazily initialise and return the ResponseStore for the active profile home. + + Response snapshots back ``previous_response_id`` and conversation-name + chaining. Under principal-routed profiles this must resolve inside the + current scoped ``get_hermes_home()`` rather than a process-global DB. + """ + try: + from hermes_constants import get_hermes_home + + home = get_hermes_home() + cache = getattr(self, "_response_stores", None) + if cache is None: + cache = {} + self._response_stores = cache + key = str(home) + store = cache.get(key) + if store is None: + default_store = getattr(self, "_response_store", None) + default_path = getattr(default_store, "_db_path", None) if default_store is not None else None + if default_store is not None and default_path == str(home / "response_store.db"): + store = default_store + else: + store = ResponseStore(db_path=str(home / "response_store.db")) + cache[key] = store + return store + except Exception as exc: + logger.debug("ResponseStore unavailable for API server: %s", exc) + return None + # ------------------------------------------------------------------ # Agent creation helper # ------------------------------------------------------------------ @@ -2222,14 +2306,20 @@ class APIServerAdapter(BasePlatformAdapter): return {}, web.json_response(_openai_error("Request body must be a JSON object"), status=400) return body, None - def _get_existing_session_or_404(self, session_id: str) -> tuple[Optional[Dict[str, Any]], Optional["web.Response"]]: - db = self._ensure_session_db() - if db is None: - return None, web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) - session = db.get_session(session_id) - if not session: - return None, web.json_response(_openai_error(f"Session not found: {session_id}", code="session_not_found"), status=404) - return session, None + def _get_existing_session_or_404( + self, + session_id: str, + identity_context: Optional[Dict[str, Optional[str]]] = None, + ) -> tuple[Optional[Dict[str, Any]], Optional["web.Response"]]: + profile = self._effective_identity_profile(identity_context) + with self._profile_scope(profile): + db = self._ensure_session_db() + if db is None: + return None, web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) + session = db.get_session(session_id) + if not session: + return None, web.json_response(_openai_error(f"Session not found: {session_id}", code="session_not_found"), status=404) + return session, None @staticmethod def _normalize_api_identity_text(value: Any) -> str: @@ -2481,30 +2571,35 @@ class APIServerAdapter(BasePlatformAdapter): if auth_err: return auth_err - db = self._ensure_session_db() - if db is None: - return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err - limit = self._parse_nonnegative_int(request.query.get("limit"), default=50, maximum=200) - offset = self._parse_nonnegative_int(request.query.get("offset"), default=0, maximum=1_000_000) - source = request.query.get("source") or None - include_children = _coerce_request_bool(request.query.get("include_children"), default=False) - owner = self._session_owner_from_request_body(request) - sessions = db.list_sessions_rich( - source=source, - user_id=owner, - limit=limit, - offset=offset, - include_children=include_children, - order_by_last_active=True, - ) - return web.json_response({ - "object": "list", - "data": [self._session_response(s) for s in sessions], - "limit": limit, - "offset": offset, - "has_more": len(sessions) == limit, - }) + with self._profile_scope(self._effective_identity_profile(identity_context)): + db = self._ensure_session_db() + if db is None: + return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) + + limit = self._parse_nonnegative_int(request.query.get("limit"), default=50, maximum=200) + offset = self._parse_nonnegative_int(request.query.get("offset"), default=0, maximum=1_000_000) + source = request.query.get("source") or None + include_children = _coerce_request_bool(request.query.get("include_children"), default=False) + owner = self._session_owner_from_request_body(request) + sessions = db.list_sessions_rich( + source=source, + user_id=owner, + limit=limit, + offset=offset, + include_children=include_children, + order_by_last_active=True, + ) + return web.json_response({ + "object": "list", + "data": [self._session_response(s) for s in sessions], + "limit": limit, + "offset": offset, + "has_more": len(sessions) == limit, + }) async def _handle_create_session(self, request: "web.Request") -> "web.Response": """POST /api/sessions — create an empty Hermes session row.""" @@ -2515,48 +2610,56 @@ class APIServerAdapter(BasePlatformAdapter): if err: return err - db = self._ensure_session_db() - if db is None: - return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) + identity_context, identity_err = self._verified_identity_context_or_error(request, body) + if identity_err is not None: + return identity_err - raw_id = body.get("id") or body.get("session_id") - session_id = str(raw_id).strip() if raw_id else f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}" - from gateway.session import _is_path_unsafe - if not session_id or re.search(r'[\r\n\x00]', session_id) or _is_path_unsafe(session_id): - return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400) - if len(session_id) > self._MAX_SESSION_HEADER_LEN: - return web.json_response(_openai_error("Session ID too long", code="invalid_session_id"), status=400) - if db.get_session(session_id): - return web.json_response(_openai_error(f"Session already exists: {session_id}", code="session_exists"), status=409) + with self._profile_scope(self._effective_identity_profile(identity_context)): + db = self._ensure_session_db() + if db is None: + return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) - model = body.get("model") or self._model_name - system_prompt = body.get("system_prompt") - owner = self._session_owner_from_request_body(request, body) - if system_prompt is not None and not isinstance(system_prompt, str): - return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400) - db.create_session( - session_id, - "api_server", - model=str(model) if model else None, - system_prompt=system_prompt, - user_id=owner, - ) - title = body.get("title") - if title is not None: - try: - db.set_session_title(session_id, str(title)) - except ValueError as exc: - db.delete_session(session_id) - return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) - session = db.get_session(session_id) or {"id": session_id, "source": "api_server", "model": model, "title": title} - return web.json_response({"object": "hermes.session", "session": self._session_response(session)}, status=201) + raw_id = body.get("id") or body.get("session_id") + session_id = str(raw_id).strip() if raw_id else f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}" + from gateway.session import _is_path_unsafe + if not session_id or re.search(r'[\r\n\x00]', session_id) or _is_path_unsafe(session_id): + return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400) + if len(session_id) > self._MAX_SESSION_HEADER_LEN: + return web.json_response(_openai_error("Session ID too long", code="invalid_session_id"), status=400) + if db.get_session(session_id): + return web.json_response(_openai_error(f"Session already exists: {session_id}", code="session_exists"), status=409) + + model = body.get("model") or self._model_name + system_prompt = body.get("system_prompt") + owner = self._session_owner_from_request_body(request, body) + if system_prompt is not None and not isinstance(system_prompt, str): + return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400) + db.create_session( + session_id, + "api_server", + model=str(model) if model else None, + system_prompt=system_prompt, + user_id=owner, + ) + title = body.get("title") + if title is not None: + try: + db.set_session_title(session_id, str(title)) + except ValueError as exc: + db.delete_session(session_id) + return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) + session = db.get_session(session_id) or {"id": session_id, "source": "api_server", "model": model, "title": title} + return web.json_response({"object": "hermes.session", "session": self._session_response(session)}, status=201) async def _handle_get_session(self, request: "web.Request") -> "web.Response": """GET /api/sessions/{session_id}.""" auth_err = self._check_auth(request) if auth_err: return auth_err - session, err = self._get_existing_session_or_404(request.match_info["session_id"]) + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err + session, err = self._get_existing_session_or_404(request.match_info["session_id"], identity_context) if err: return err authz_err = self._authorize_session_owner(request, session) @@ -2569,8 +2672,11 @@ class APIServerAdapter(BasePlatformAdapter): auth_err = self._check_auth(request) if auth_err: return auth_err + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err session_id = request.match_info["session_id"] - session, err = self._get_existing_session_or_404(session_id) + session, err = self._get_existing_session_or_404(session_id, identity_context) if err: return err authz_err = self._authorize_session_owner(request, session) @@ -2584,61 +2690,73 @@ class APIServerAdapter(BasePlatformAdapter): if unknown: return web.json_response(_openai_error(f"Unsupported session fields: {', '.join(unknown)}", code="unsupported_session_field"), status=400) - db = self._ensure_session_db() - if "title" in body: - try: - db.set_session_title(session_id, "" if body["title"] is None else str(body["title"])) - except ValueError as exc: - return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) - if body.get("end_reason"): - db.end_session(session_id, str(body["end_reason"])) - session = db.get_session(session_id) or session - return web.json_response({"object": "hermes.session", "session": self._session_response(session)}) + with self._profile_scope(self._effective_identity_profile(identity_context)): + db = self._ensure_session_db() + if "title" in body: + try: + db.set_session_title(session_id, "" if body["title"] is None else str(body["title"])) + except ValueError as exc: + return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) + if body.get("end_reason"): + db.end_session(session_id, str(body["end_reason"])) + session = db.get_session(session_id) or session + return web.json_response({"object": "hermes.session", "session": self._session_response(session)}) async def _handle_delete_session(self, request: "web.Request") -> "web.Response": """DELETE /api/sessions/{session_id}.""" auth_err = self._check_auth(request) if auth_err: return auth_err + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err session_id = request.match_info["session_id"] - session, err = self._get_existing_session_or_404(session_id) + session, err = self._get_existing_session_or_404(session_id, identity_context) if err: return err authz_err = self._authorize_session_owner(request, session) if authz_err: return authz_err - db = self._ensure_session_db() - deleted = db.delete_session(session_id) - return web.json_response({"object": "hermes.session.deleted", "id": session_id, "deleted": bool(deleted)}) + with self._profile_scope(self._effective_identity_profile(identity_context)): + db = self._ensure_session_db() + deleted = db.delete_session(session_id) + return web.json_response({"object": "hermes.session.deleted", "id": session_id, "deleted": bool(deleted)}) async def _handle_session_messages(self, request: "web.Request") -> "web.Response": """GET /api/sessions/{session_id}/messages.""" auth_err = self._check_auth(request) if auth_err: return auth_err + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err session_id = request.match_info["session_id"] - session, err = self._get_existing_session_or_404(session_id) + session, err = self._get_existing_session_or_404(session_id, identity_context) if err: return err authz_err = self._authorize_session_owner(request, session) if authz_err: return authz_err - db = self._ensure_session_db() - resolved_id = db.resolve_resume_session_id(session_id) - messages = db.get_messages(resolved_id) - return web.json_response({ - "object": "list", - "session_id": resolved_id, - "data": [self._message_response(m) for m in messages], - }) + with self._profile_scope(self._effective_identity_profile(identity_context)): + db = self._ensure_session_db() + resolved_id = db.resolve_resume_session_id(session_id) + messages = db.get_messages(resolved_id) + return web.json_response({ + "object": "list", + "session_id": resolved_id, + "data": [self._message_response(m) for m in messages], + }) async def _handle_fork_session(self, request: "web.Request") -> "web.Response": """POST /api/sessions/{session_id}/fork — branch via current SessionDB primitives.""" auth_err = self._check_auth(request) if auth_err: return auth_err + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err source_id = request.match_info["session_id"] - source, err = self._get_existing_session_or_404(source_id) + source, err = self._get_existing_session_or_404(source_id, identity_context) if err: return err authz_err = self._authorize_session_owner(request, source) @@ -2647,41 +2765,42 @@ class APIServerAdapter(BasePlatformAdapter): body, err = await self._read_json_body(request) if err: return err - db = self._ensure_session_db() - fork_id = str(body.get("id") or body.get("session_id") or f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}").strip() - if not fork_id or re.search(r'[\r\n\x00]', fork_id): - return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400) - if db.get_session(fork_id): - return web.json_response(_openai_error(f"Session already exists: {fork_id}", code="session_exists"), status=409) + with self._profile_scope(self._effective_identity_profile(identity_context)): + db = self._ensure_session_db() + fork_id = str(body.get("id") or body.get("session_id") or f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}").strip() + if not fork_id or re.search(r'[\r\n\x00]', fork_id): + return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400) + if db.get_session(fork_id): + return web.json_response(_openai_error(f"Session already exists: {fork_id}", code="session_exists"), status=409) - # Match the CLI /branch semantics: mark the original as branched, then - # create a child session that carries the transcript forward. This uses - # SessionDB's native parent_session_id/end_reason visibility model rather - # than inventing a parallel fork store. - db.end_session(source_id, "branched") - db.create_session( - fork_id, - "api_server", - model=source.get("model"), - system_prompt=source.get("system_prompt"), - parent_session_id=source_id, - user_id=source.get("user_id"), - ) - messages = db.get_messages(source_id) - db.replace_messages(fork_id, messages) - title = body.get("title") - if title is None: - base = source.get("title") or "fork" + # Match the CLI /branch semantics: mark the original as branched, then + # create a child session that carries the transcript forward. This uses + # SessionDB's native parent_session_id/end_reason visibility model rather + # than inventing a parallel fork store. + db.end_session(source_id, "branched") + db.create_session( + fork_id, + "api_server", + model=source.get("model"), + system_prompt=source.get("system_prompt"), + parent_session_id=source_id, + user_id=source.get("user_id"), + ) + messages = db.get_messages(source_id) + db.replace_messages(fork_id, messages) + title = body.get("title") + if title is None: + base = source.get("title") or "fork" + try: + title = db.get_next_title_in_lineage(base) + except Exception: + title = f"{base} fork" try: - title = db.get_next_title_in_lineage(base) - except Exception: - title = f"{base} fork" - try: - db.set_session_title(fork_id, str(title)) - except ValueError as exc: - return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) - fork = db.get_session(fork_id) or {"id": fork_id, "parent_session_id": source_id} - return web.json_response({"object": "hermes.session", "session": self._session_response(fork)}, status=201) + db.set_session_title(fork_id, str(title)) + except ValueError as exc: + return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) + fork = db.get_session(fork_id) or {"id": fork_id, "parent_session_id": source_id} + return web.json_response({"object": "hermes.session", "session": self._session_response(fork)}, status=201) @_admit_api_agent_request async def _handle_session_chat(self, request: "web.Request") -> "web.Response": @@ -2690,12 +2809,6 @@ class APIServerAdapter(BasePlatformAdapter): if key_err is not None: return key_err session_id = request.match_info["session_id"] - session, err = self._get_existing_session_or_404(session_id) - if err: - return err - authz_err = self._authorize_session_owner(request, session) - if authz_err: - return authz_err body, err = await self._read_json_body(request) if err: return err @@ -2706,9 +2819,16 @@ class APIServerAdapter(BasePlatformAdapter): identity_context, identity_err = self._verified_identity_context_or_error(request, body) if identity_err is not None: return identity_err + session, err = self._get_existing_session_or_404(session_id, identity_context) + if err: + return err + authz_err = self._authorize_session_owner(request, session) + if authz_err: + return authz_err if system_prompt is not None and not isinstance(system_prompt, str): return web.json_response(_openai_error("system_message must be a string", code="invalid_system_message"), status=400) - history = self._conversation_history_for_session(session_id) + with self._profile_scope(self._effective_identity_profile(identity_context)): + history = self._conversation_history_for_session(session_id) result, usage = await self._run_agent( user_message=user_message, conversation_history=history, @@ -2742,12 +2862,6 @@ class APIServerAdapter(BasePlatformAdapter): if key_err is not None: return key_err session_id = request.match_info["session_id"] - session, err = self._get_existing_session_or_404(session_id) - if err: - return err - authz_err = self._authorize_session_owner(request, session) - if authz_err: - return authz_err body, err = await self._read_json_body(request) if err: return err @@ -2758,6 +2872,12 @@ class APIServerAdapter(BasePlatformAdapter): identity_context, identity_err = self._verified_identity_context_or_error(request, body) if identity_err is not None: return identity_err + session, err = self._get_existing_session_or_404(session_id, identity_context) + if err: + return err + authz_err = self._authorize_session_owner(request, session) + if authz_err: + return authz_err if system_prompt is not None and not isinstance(system_prompt, str): return web.json_response(_openai_error("system_message must be a string", code="invalid_system_message"), status=400) @@ -2805,7 +2925,8 @@ class APIServerAdapter(BasePlatformAdapter): try: await queue.put(_event_payload("run.started", {"user_message": {"role": "user", "content": user_message}})) await queue.put(_event_payload("message.started", {"message": {"id": message_id, "role": "assistant"}})) - history = self._conversation_history_for_session(session_id) + with self._profile_scope(self._effective_identity_profile(identity_context)): + history = self._conversation_history_for_session(session_id) result, usage = await self._run_agent( user_message=user_message, conversation_history=history, @@ -3447,6 +3568,7 @@ class APIServerAdapter(BasePlatformAdapter): store: bool, session_id: str, gateway_session_key: Optional[str] = None, + identity_context: Optional[Dict[str, Optional[str]]] = None, ) -> "web.StreamResponse": """Write an SSE stream for POST /v1/responses (OpenAI Responses API). @@ -3493,6 +3615,9 @@ class APIServerAdapter(BasePlatformAdapter): sse_headers["X-Hermes-Session-Key"] = gateway_session_key response = web.StreamResponse(status=200, headers=sse_headers) await response.prepare(request) + request_profile = self._effective_identity_profile(identity_context) + with self._profile_scope(request_profile): + response_store = self._ensure_response_store() # State accumulated during the stream final_text_parts: List[str] = [] @@ -3545,19 +3670,19 @@ class APIServerAdapter(BasePlatformAdapter): *, conversation_history_snapshot: Optional[List[Dict[str, Any]]] = None, ) -> None: - if not store: + if not store or response_store is None: return if conversation_history_snapshot is None: conversation_history_snapshot = list(conversation_history) conversation_history_snapshot.append({"role": "user", "content": user_message}) - self._response_store.put(response_id, { + response_store.put(response_id, { "response": response_env, "conversation_history": conversation_history_snapshot, "instructions": instructions, "session_id": session_id, }) if conversation: - self._response_store.set_conversation(conversation, response_id) + response_store.set_conversation(conversation, response_id) def _persist_incomplete_if_needed() -> None: """Persist an ``incomplete`` snapshot if no terminal one was written. @@ -4059,6 +4184,9 @@ class APIServerAdapter(BasePlatformAdapter): identity_context, identity_err = self._verified_identity_context_or_error(request, body) if identity_err is not None: return identity_err + request_profile = self._effective_identity_profile(identity_context) + with self._profile_scope(request_profile): + response_store = self._ensure_response_store() store = _coerce_request_bool(body.get("store"), default=True) # conversation and previous_response_id are mutually exclusive @@ -4067,7 +4195,7 @@ class APIServerAdapter(BasePlatformAdapter): # Resolve conversation name to latest response_id if conversation: - previous_response_id = self._response_store.get_conversation(conversation) + previous_response_id = response_store.get_conversation(conversation) if response_store is not None else None # No error if conversation doesn't exist yet — it's a new conversation # Normalize input to message list @@ -4116,7 +4244,7 @@ class APIServerAdapter(BasePlatformAdapter): stored_session_id = None if not conversation_history and previous_response_id: - stored = self._response_store.get(previous_response_id) + stored = response_store.get(previous_response_id) if response_store is not None else None if stored is None: return web.json_response(_openai_error(f"Previous response not found: {previous_response_id}"), status=404) conversation_history = list(stored.get("conversation_history", [])) @@ -4226,6 +4354,7 @@ class APIServerAdapter(BasePlatformAdapter): store=store, session_id=session_id, gateway_session_key=gateway_session_key, + identity_context=identity_context, ) async def _compute_response(): @@ -4307,8 +4436,8 @@ class APIServerAdapter(BasePlatformAdapter): } # Store the complete response object for future chaining / GET retrieval - if store: - self._response_store.put(response_id, { + if store and response_store is not None: + response_store.put(response_id, { "response": response_data, "conversation_history": full_history, "instructions": instructions, @@ -4317,7 +4446,7 @@ class APIServerAdapter(BasePlatformAdapter): # Update conversation mapping so the next request with the same # conversation name automatically chains to this response if conversation: - self._response_store.set_conversation(conversation, response_id) + response_store.set_conversation(conversation, response_id) response_headers = {"X-Hermes-Session-Id": session_id} if gateway_session_key: @@ -4333,9 +4462,14 @@ class APIServerAdapter(BasePlatformAdapter): auth_err = self._check_auth(request) if auth_err: return auth_err + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err response_id = request.match_info["response_id"] - stored = self._response_store.get(response_id) + with self._profile_scope(self._effective_identity_profile(identity_context)): + response_store = self._ensure_response_store() + stored = response_store.get(response_id) if response_store is not None else None if stored is None: return web.json_response(_openai_error(f"Response not found: {response_id}"), status=404) @@ -4346,9 +4480,14 @@ class APIServerAdapter(BasePlatformAdapter): auth_err = self._check_auth(request) if auth_err: return auth_err + identity_context, identity_err = self._verified_identity_context_or_error(request) + if identity_err is not None: + return identity_err response_id = request.match_info["response_id"] - deleted = self._response_store.delete(response_id) + with self._profile_scope(self._effective_identity_profile(identity_context)): + response_store = self._ensure_response_store() + deleted = response_store.delete(response_id) if response_store is not None else False if not deleted: return web.json_response(_openai_error(f"Response not found: {response_id}"), status=404) @@ -5080,6 +5219,9 @@ class APIServerAdapter(BasePlatformAdapter): identity_context, identity_err = self._verified_identity_context_or_error(request, body) if identity_err is not None: return identity_err + request_profile = self._effective_identity_profile(identity_context) + with self._profile_scope(request_profile): + response_store = self._ensure_response_store() # Accept explicit conversation_history from the request body. # Precedence: explicit conversation_history > previous_response_id. @@ -5103,7 +5245,7 @@ class APIServerAdapter(BasePlatformAdapter): stored_session_id = None if not conversation_history and previous_response_id: - stored = self._response_store.get(previous_response_id) + stored = response_store.get(previous_response_id) if response_store is not None else None if stored: conversation_history = list(stored.get("conversation_history", [])) stored_session_id = stored.get("session_id") @@ -5809,9 +5951,16 @@ class APIServerAdapter(BasePlatformAdapter): (OSError: [Errno 24] Too many open files, #37011). """ self._mark_disconnected() - if self._response_store is not None: + closed_ids = set() + for store in list(getattr(self, "_response_stores", {}).values()): + if store is None: + continue + marker = id(store) + if marker in closed_ids: + continue + closed_ids.add(marker) try: - self._response_store.close() + store.close() except Exception: logger.debug( "Failed to close response store for %s", self.name, exc_info=True, diff --git a/hermes_cli/managed_downloads.py b/hermes_cli/managed_downloads.py index c6b05057d..93b550418 100644 --- a/hermes_cli/managed_downloads.py +++ b/hermes_cli/managed_downloads.py @@ -181,6 +181,15 @@ def _normalize_owner_metadata(owner_metadata: dict | None) -> dict[str, str]: return allowed +def normalize_download_link_mode(mode: str | None, *, default: str = "private") -> str: + raw = str(mode or "").strip().lower() + if raw in {"public", "shared", "open"}: + return "public" + if raw in {"private", "owner", "bound", "owner_bound", "owner-bound", "scoped"}: + return "private" + return "public" if str(default).strip().lower() == "public" else "private" + + def create_signed_download_token( file_path: str, *, @@ -313,11 +322,21 @@ def resolve_short_download_id(short_id: str) -> Path: return Path(resolve_short_download_record(short_id)["path"]) -def build_public_download_url(file_path: str, *, owner_metadata: dict | None = None) -> str: +def build_public_download_url( + file_path: str, + *, + owner_metadata: dict | None = None, + link_mode: str | None = None, +) -> str: base_url = str(resolve_download_public_url() or "").rstrip("/") if not base_url: return "" - short_id = create_short_download_id(file_path, owner_metadata=owner_metadata) + effective_mode = normalize_download_link_mode( + link_mode, + default="private" if _normalize_owner_metadata(owner_metadata) else "public", + ) + effective_owner_metadata = owner_metadata if effective_mode == "private" else None + short_id = create_short_download_id(file_path, owner_metadata=effective_owner_metadata) if not short_id: return "" # Preflight: prove the short link resolves before we hand it out. diff --git a/tests/hermes_cli/test_managed_downloads.py b/tests/hermes_cli/test_managed_downloads.py index a841c76dd..2113538d6 100644 --- a/tests/hermes_cli/test_managed_downloads.py +++ b/tests/hermes_cli/test_managed_downloads.py @@ -162,3 +162,56 @@ def test_owner_metadata_roundtrips_in_short_download_record(_isolate_hermes_home "principal_id": "p-123", "external_user_id": "openwebui-42", } + + +def test_public_link_mode_strips_owner_metadata(_isolate_hermes_home): + from hermes_constants import get_hermes_home + from hermes_cli.managed_downloads import build_public_download_url, resolve_short_download_record + + home = get_hermes_home() + (home / "config.yaml").write_text( + "dashboard:\n" + " public_url: https://hermesadmin.bremen.com.tw\n" + " download_public_url: https://hermesadmin.bremen.com.tw\n", + encoding="utf-8", + ) + source = home / "public-link.txt" + source.write_text("hello public", encoding="utf-8") + + url = build_public_download_url( + str(source), + owner_metadata={"email": "owner@bremen.com.tw", "principal_id": "p-123"}, + link_mode="public", + ) + short_id = Path(urlparse(url).path).name.removeprefix("s-") + record = resolve_short_download_record(short_id) + + assert record["metadata"] == {} + + +def test_private_link_mode_preserves_owner_metadata(_isolate_hermes_home): + from hermes_constants import get_hermes_home + from hermes_cli.managed_downloads import build_public_download_url, resolve_short_download_record + + home = get_hermes_home() + (home / "config.yaml").write_text( + "dashboard:\n" + " public_url: https://hermesadmin.bremen.com.tw\n" + " download_public_url: https://hermesadmin.bremen.com.tw\n", + encoding="utf-8", + ) + source = home / "private-link.txt" + source.write_text("hello private", encoding="utf-8") + + url = build_public_download_url( + str(source), + owner_metadata={"email": "owner@bremen.com.tw", "principal_id": "p-123"}, + link_mode="private", + ) + short_id = Path(urlparse(url).path).name.removeprefix("s-") + record = resolve_short_download_record(short_id) + + assert record["metadata"] == { + "email": "owner@bremen.com.tw", + "principal_id": "p-123", + }