Bremen
/
emma-hermes
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: scope api responses by principal profile

fix/verification-admin-route-recovery
Hermes Agent 2026-07-30 01:20:22 +08:00
parent 04b69b3d7c
commit 47e80c769f
3 changed files with 374 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

451
gateway/platforms/api_server.py
View File

@ -604,6 +604,52 @@ class ResponseStore:
return row[0] if row else 0 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 # CORS middleware
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -1150,7 +1196,15 @@ class APIServerAdapter(BasePlatformAdapter):
self._app: Optional["web.Application"] = None self._app: Optional["web.Application"] = None
self._runner: Optional["web.AppRunner"] = None self._runner: Optional["web.AppRunner"] = None
self._site: Optional["web.TCPSite"] = 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 # Active run streams: run_id -> asyncio.Queue of SSE event dicts
self._run_streams: Dict[str, "asyncio.Queue[Optional[Dict]]"] = {} self._run_streams: Dict[str, "asyncio.Queue[Optional[Dict]]"] = {}
# Creation timestamps for orphaned-run TTL sweep # Creation timestamps for orphaned-run TTL sweep
@ -1672,6 +1726,36 @@ class APIServerAdapter(BasePlatformAdapter):
logger.debug("SessionDB unavailable for API server: %s", e) logger.debug("SessionDB unavailable for API server: %s", e)
return None 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 # 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 {}, web.json_response(_openai_error("Request body must be a JSON object"), status=400)
return body, None return body, None
def _get_existing_session_or_404(self, session_id: str) -> tuple[Optional[Dict[str, Any]], Optional["web.Response"]]: def _get_existing_session_or_404(
db = self._ensure_session_db() self,
if db is None: session_id: str,
return None, web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) identity_context: Optional[Dict[str, Optional[str]]] = None,
session = db.get_session(session_id) ) -> tuple[Optional[Dict[str, Any]], Optional["web.Response"]]:
if not session: profile = self._effective_identity_profile(identity_context)
return None, web.json_response(_openai_error(f"Session not found: {session_id}", code="session_not_found"), status=404) with self._profile_scope(profile):
return session, None 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 @staticmethod
def _normalize_api_identity_text(value: Any) -> str: def _normalize_api_identity_text(value: Any) -> str:
@ -2481,30 +2571,35 @@ class APIServerAdapter(BasePlatformAdapter):
if auth_err: if auth_err:
return auth_err return auth_err
db = self._ensure_session_db() identity_context, identity_err = self._verified_identity_context_or_error(request)
if db is None: if identity_err is not None:
return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) return identity_err
limit = self._parse_nonnegative_int(request.query.get("limit"), default=50, maximum=200) with self._profile_scope(self._effective_identity_profile(identity_context)):
offset = self._parse_nonnegative_int(request.query.get("offset"), default=0, maximum=1_000_000) db = self._ensure_session_db()
source = request.query.get("source") or None if db is None:
include_children = _coerce_request_bool(request.query.get("include_children"), default=False) return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503)
owner = self._session_owner_from_request_body(request)
sessions = db.list_sessions_rich( limit = self._parse_nonnegative_int(request.query.get("limit"), default=50, maximum=200)
source=source, offset = self._parse_nonnegative_int(request.query.get("offset"), default=0, maximum=1_000_000)
user_id=owner, source = request.query.get("source") or None
limit=limit, include_children = _coerce_request_bool(request.query.get("include_children"), default=False)
offset=offset, owner = self._session_owner_from_request_body(request)
include_children=include_children, sessions = db.list_sessions_rich(
order_by_last_active=True, source=source,
) user_id=owner,
return web.json_response({ limit=limit,
"object": "list", offset=offset,
"data": [self._session_response(s) for s in sessions], include_children=include_children,
"limit": limit, order_by_last_active=True,
"offset": offset, )
"has_more": len(sessions) == limit, 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": async def _handle_create_session(self, request: "web.Request") -> "web.Response":
"""POST /api/sessions — create an empty Hermes session row.""" """POST /api/sessions — create an empty Hermes session row."""
@ -2515,48 +2610,56 @@ class APIServerAdapter(BasePlatformAdapter):
if err: if err:
return err return err
db = self._ensure_session_db() identity_context, identity_err = self._verified_identity_context_or_error(request, body)
if db is None: if identity_err is not None:
return web.json_response(_openai_error("Session database unavailable", code="session_db_unavailable"), status=503) return identity_err
raw_id = body.get("id") or body.get("session_id") with self._profile_scope(self._effective_identity_profile(identity_context)):
session_id = str(raw_id).strip() if raw_id else f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}" db = self._ensure_session_db()
from gateway.session import _is_path_unsafe if db is None:
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("Session database unavailable", code="session_db_unavailable"), status=503)
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 raw_id = body.get("id") or body.get("session_id")
system_prompt = body.get("system_prompt") session_id = str(raw_id).strip() if raw_id else f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}"
owner = self._session_owner_from_request_body(request, body) from gateway.session import _is_path_unsafe
if system_prompt is not None and not isinstance(system_prompt, str): 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("system_prompt must be a string", code="invalid_system_prompt"), status=400) return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400)
db.create_session( if len(session_id) > self._MAX_SESSION_HEADER_LEN:
session_id, return web.json_response(_openai_error("Session ID too long", code="invalid_session_id"), status=400)
"api_server", if db.get_session(session_id):
model=str(model) if model else None, return web.json_response(_openai_error(f"Session already exists: {session_id}", code="session_exists"), status=409)
system_prompt=system_prompt,
user_id=owner, model = body.get("model") or self._model_name
) system_prompt = body.get("system_prompt")
title = body.get("title") owner = self._session_owner_from_request_body(request, body)
if title is not None: if system_prompt is not None and not isinstance(system_prompt, str):
try: return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400)
db.set_session_title(session_id, str(title)) db.create_session(
except ValueError as exc: session_id,
db.delete_session(session_id) "api_server",
return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) model=str(model) if model else None,
session = db.get_session(session_id) or {"id": session_id, "source": "api_server", "model": model, "title": title} system_prompt=system_prompt,
return web.json_response({"object": "hermes.session", "session": self._session_response(session)}, status=201) 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": async def _handle_get_session(self, request: "web.Request") -> "web.Response":
"""GET /api/sessions/{session_id}.""" """GET /api/sessions/{session_id}."""
auth_err = self._check_auth(request) auth_err = self._check_auth(request)
if auth_err: if auth_err:
return 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: if err:
return err return err
authz_err = self._authorize_session_owner(request, session) authz_err = self._authorize_session_owner(request, session)
@ -2569,8 +2672,11 @@ class APIServerAdapter(BasePlatformAdapter):
auth_err = self._check_auth(request) auth_err = self._check_auth(request)
if auth_err: if auth_err:
return 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_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: if err:
return err return err
authz_err = self._authorize_session_owner(request, session) authz_err = self._authorize_session_owner(request, session)
@ -2584,61 +2690,73 @@ class APIServerAdapter(BasePlatformAdapter):
if unknown: if unknown:
return web.json_response(_openai_error(f"Unsupported session fields: {', '.join(unknown)}", code="unsupported_session_field"), status=400) return web.json_response(_openai_error(f"Unsupported session fields: {', '.join(unknown)}", code="unsupported_session_field"), status=400)
db = self._ensure_session_db() with self._profile_scope(self._effective_identity_profile(identity_context)):
if "title" in body: db = self._ensure_session_db()
try: if "title" in body:
db.set_session_title(session_id, "" if body["title"] is None else str(body["title"])) try:
except ValueError as exc: db.set_session_title(session_id, "" if body["title"] is None else str(body["title"]))
return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400) except ValueError as exc:
if body.get("end_reason"): return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400)
db.end_session(session_id, str(body["end_reason"])) if body.get("end_reason"):
session = db.get_session(session_id) or session db.end_session(session_id, str(body["end_reason"]))
return web.json_response({"object": "hermes.session", "session": self._session_response(session)}) 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": async def _handle_delete_session(self, request: "web.Request") -> "web.Response":
"""DELETE /api/sessions/{session_id}.""" """DELETE /api/sessions/{session_id}."""
auth_err = self._check_auth(request) auth_err = self._check_auth(request)
if auth_err: if auth_err:
return 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_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: if err:
return err return err
authz_err = self._authorize_session_owner(request, session) authz_err = self._authorize_session_owner(request, session)
if authz_err: if authz_err:
return authz_err return authz_err
db = self._ensure_session_db() with self._profile_scope(self._effective_identity_profile(identity_context)):
deleted = db.delete_session(session_id) db = self._ensure_session_db()
return web.json_response({"object": "hermes.session.deleted", "id": session_id, "deleted": bool(deleted)}) 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": async def _handle_session_messages(self, request: "web.Request") -> "web.Response":
"""GET /api/sessions/{session_id}/messages.""" """GET /api/sessions/{session_id}/messages."""
auth_err = self._check_auth(request) auth_err = self._check_auth(request)
if auth_err: if auth_err:
return 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_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: if err:
return err return err
authz_err = self._authorize_session_owner(request, session) authz_err = self._authorize_session_owner(request, session)
if authz_err: if authz_err:
return authz_err return authz_err
db = self._ensure_session_db() with self._profile_scope(self._effective_identity_profile(identity_context)):
resolved_id = db.resolve_resume_session_id(session_id) db = self._ensure_session_db()
messages = db.get_messages(resolved_id) resolved_id = db.resolve_resume_session_id(session_id)
return web.json_response({ messages = db.get_messages(resolved_id)
"object": "list", return web.json_response({
"session_id": resolved_id, "object": "list",
"data": [self._message_response(m) for m in messages], "session_id": resolved_id,
}) "data": [self._message_response(m) for m in messages],
})
async def _handle_fork_session(self, request: "web.Request") -> "web.Response": async def _handle_fork_session(self, request: "web.Request") -> "web.Response":
"""POST /api/sessions/{session_id}/fork — branch via current SessionDB primitives.""" """POST /api/sessions/{session_id}/fork — branch via current SessionDB primitives."""
auth_err = self._check_auth(request) auth_err = self._check_auth(request)
if auth_err: if auth_err:
return 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_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: if err:
return err return err
authz_err = self._authorize_session_owner(request, source) authz_err = self._authorize_session_owner(request, source)
@ -2647,41 +2765,42 @@ class APIServerAdapter(BasePlatformAdapter):
body, err = await self._read_json_body(request) body, err = await self._read_json_body(request)
if err: if err:
return err return err
db = self._ensure_session_db() with self._profile_scope(self._effective_identity_profile(identity_context)):
fork_id = str(body.get("id") or body.get("session_id") or f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}").strip() db = self._ensure_session_db()
if not fork_id or re.search(r'[\r\n\x00]', fork_id): fork_id = str(body.get("id") or body.get("session_id") or f"api_{int(time.time())}_{uuid.uuid4().hex[:8]}").strip()
return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400) if not fork_id or re.search(r'[\r\n\x00]', fork_id):
if db.get_session(fork_id): return web.json_response(_openai_error("Invalid session ID", code="invalid_session_id"), status=400)
return web.json_response(_openai_error(f"Session already exists: {fork_id}", code="session_exists"), status=409) 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 # Match the CLI /branch semantics: mark the original as branched, then
# create a child session that carries the transcript forward. This uses # create a child session that carries the transcript forward. This uses
# SessionDB's native parent_session_id/end_reason visibility model rather # SessionDB's native parent_session_id/end_reason visibility model rather
# than inventing a parallel fork store. # than inventing a parallel fork store.
db.end_session(source_id, "branched") db.end_session(source_id, "branched")
db.create_session( db.create_session(
fork_id, fork_id,
"api_server", "api_server",
model=source.get("model"), model=source.get("model"),
system_prompt=source.get("system_prompt"), system_prompt=source.get("system_prompt"),
parent_session_id=source_id, parent_session_id=source_id,
user_id=source.get("user_id"), user_id=source.get("user_id"),
) )
messages = db.get_messages(source_id) messages = db.get_messages(source_id)
db.replace_messages(fork_id, messages) db.replace_messages(fork_id, messages)
title = body.get("title") title = body.get("title")
if title is None: if title is None:
base = source.get("title") or "fork" base = source.get("title") or "fork"
try:
title = db.get_next_title_in_lineage(base)
except Exception:
title = f"{base} fork"
try: try:
title = db.get_next_title_in_lineage(base) db.set_session_title(fork_id, str(title))
except Exception: except ValueError as exc:
title = f"{base} fork" return web.json_response(_openai_error(str(exc), code="invalid_title"), status=400)
try: fork = db.get_session(fork_id) or {"id": fork_id, "parent_session_id": source_id}
db.set_session_title(fork_id, str(title)) return web.json_response({"object": "hermes.session", "session": self._session_response(fork)}, status=201)
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 @_admit_api_agent_request
async def _handle_session_chat(self, request: "web.Request") -> "web.Response": async def _handle_session_chat(self, request: "web.Request") -> "web.Response":
@ -2690,12 +2809,6 @@ class APIServerAdapter(BasePlatformAdapter):
if key_err is not None: if key_err is not None:
return key_err return key_err
session_id = request.match_info["session_id"] 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) body, err = await self._read_json_body(request)
if err: if err:
return err return err
@ -2706,9 +2819,16 @@ class APIServerAdapter(BasePlatformAdapter):
identity_context, identity_err = self._verified_identity_context_or_error(request, body) identity_context, identity_err = self._verified_identity_context_or_error(request, body)
if identity_err is not None: if identity_err is not None:
return identity_err 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): 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) 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( result, usage = await self._run_agent(
user_message=user_message, user_message=user_message,
conversation_history=history, conversation_history=history,
@ -2742,12 +2862,6 @@ class APIServerAdapter(BasePlatformAdapter):
if key_err is not None: if key_err is not None:
return key_err return key_err
session_id = request.match_info["session_id"] 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) body, err = await self._read_json_body(request)
if err: if err:
return err return err
@ -2758,6 +2872,12 @@ class APIServerAdapter(BasePlatformAdapter):
identity_context, identity_err = self._verified_identity_context_or_error(request, body) identity_context, identity_err = self._verified_identity_context_or_error(request, body)
if identity_err is not None: if identity_err is not None:
return identity_err 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): 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) 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: try:
await queue.put(_event_payload("run.started", {"user_message": {"role": "user", "content": user_message}})) 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"}})) 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( result, usage = await self._run_agent(
user_message=user_message, user_message=user_message,
conversation_history=history, conversation_history=history,
@ -3447,6 +3568,7 @@ class APIServerAdapter(BasePlatformAdapter):
store: bool, store: bool,
session_id: str, session_id: str,
gateway_session_key: Optional[str] = None, gateway_session_key: Optional[str] = None,
identity_context: Optional[Dict[str, Optional[str]]] = None,
) -> "web.StreamResponse": ) -> "web.StreamResponse":
"""Write an SSE stream for POST /v1/responses (OpenAI Responses API). """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 sse_headers["X-Hermes-Session-Key"] = gateway_session_key
response = web.StreamResponse(status=200, headers=sse_headers) response = web.StreamResponse(status=200, headers=sse_headers)
await response.prepare(request) 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 # State accumulated during the stream
final_text_parts: List[str] = [] final_text_parts: List[str] = []
@ -3545,19 +3670,19 @@ class APIServerAdapter(BasePlatformAdapter):
*, *,
conversation_history_snapshot: Optional[List[Dict[str, Any]]] = None, conversation_history_snapshot: Optional[List[Dict[str, Any]]] = None,
) -> None: ) -> None:
if not store: if not store or response_store is None:
return return
if conversation_history_snapshot is None: if conversation_history_snapshot is None:
conversation_history_snapshot = list(conversation_history) conversation_history_snapshot = list(conversation_history)
conversation_history_snapshot.append({"role": "user", "content": user_message}) conversation_history_snapshot.append({"role": "user", "content": user_message})
self._response_store.put(response_id, { response_store.put(response_id, {
"response": response_env, "response": response_env,
"conversation_history": conversation_history_snapshot, "conversation_history": conversation_history_snapshot,
"instructions": instructions, "instructions": instructions,
"session_id": session_id, "session_id": session_id,
}) })
if conversation: if conversation:
self._response_store.set_conversation(conversation, response_id) response_store.set_conversation(conversation, response_id)
def _persist_incomplete_if_needed() -> None: def _persist_incomplete_if_needed() -> None:
"""Persist an ``incomplete`` snapshot if no terminal one was written. """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) identity_context, identity_err = self._verified_identity_context_or_error(request, body)
if identity_err is not None: if identity_err is not None:
return identity_err 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) store = _coerce_request_bool(body.get("store"), default=True)
# conversation and previous_response_id are mutually exclusive # conversation and previous_response_id are mutually exclusive
@ -4067,7 +4195,7 @@ class APIServerAdapter(BasePlatformAdapter):
# Resolve conversation name to latest response_id # Resolve conversation name to latest response_id
if conversation: 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 # No error if conversation doesn't exist yet — it's a new conversation
# Normalize input to message list # Normalize input to message list
@ -4116,7 +4244,7 @@ class APIServerAdapter(BasePlatformAdapter):
stored_session_id = None stored_session_id = None
if not conversation_history and previous_response_id: 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: if stored is None:
return web.json_response(_openai_error(f"Previous response not found: {previous_response_id}"), status=404) return web.json_response(_openai_error(f"Previous response not found: {previous_response_id}"), status=404)
conversation_history = list(stored.get("conversation_history", [])) conversation_history = list(stored.get("conversation_history", []))
@ -4226,6 +4354,7 @@ class APIServerAdapter(BasePlatformAdapter):
store=store, store=store,
session_id=session_id, session_id=session_id,
gateway_session_key=gateway_session_key, gateway_session_key=gateway_session_key,
identity_context=identity_context,
) )
async def _compute_response(): async def _compute_response():
@ -4307,8 +4436,8 @@ class APIServerAdapter(BasePlatformAdapter):
} }
# Store the complete response object for future chaining / GET retrieval # Store the complete response object for future chaining / GET retrieval
if store: if store and response_store is not None:
self._response_store.put(response_id, { response_store.put(response_id, {
"response": response_data, "response": response_data,
"conversation_history": full_history, "conversation_history": full_history,
"instructions": instructions, "instructions": instructions,
@ -4317,7 +4446,7 @@ class APIServerAdapter(BasePlatformAdapter):
# Update conversation mapping so the next request with the same # Update conversation mapping so the next request with the same
# conversation name automatically chains to this response # conversation name automatically chains to this response
if conversation: 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} response_headers = {"X-Hermes-Session-Id": session_id}
if gateway_session_key: if gateway_session_key:
@ -4333,9 +4462,14 @@ class APIServerAdapter(BasePlatformAdapter):
auth_err = self._check_auth(request) auth_err = self._check_auth(request)
if auth_err: if auth_err:
return 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"] 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: if stored is None:
return web.json_response(_openai_error(f"Response not found: {response_id}"), status=404) 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) auth_err = self._check_auth(request)
if auth_err: if auth_err:
return 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"] 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: if not deleted:
return web.json_response(_openai_error(f"Response not found: {response_id}"), status=404) 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) identity_context, identity_err = self._verified_identity_context_or_error(request, body)
if identity_err is not None: if identity_err is not None:
return identity_err 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. # Accept explicit conversation_history from the request body.
# Precedence: explicit conversation_history > previous_response_id. # Precedence: explicit conversation_history > previous_response_id.
@ -5103,7 +5245,7 @@ class APIServerAdapter(BasePlatformAdapter):
stored_session_id = None stored_session_id = None
if not conversation_history and previous_response_id: 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: if stored:
conversation_history = list(stored.get("conversation_history", [])) conversation_history = list(stored.get("conversation_history", []))
stored_session_id = stored.get("session_id") stored_session_id = stored.get("session_id")
@ -5809,9 +5951,16 @@ class APIServerAdapter(BasePlatformAdapter):
(OSError: [Errno 24] Too many open files, #37011). (OSError: [Errno 24] Too many open files, #37011).
""" """
self._mark_disconnected() 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: try:
self._response_store.close() store.close()
except Exception: except Exception:
logger.debug( logger.debug(
"Failed to close response store for %s", self.name, exc_info=True, "Failed to close response store for %s", self.name, exc_info=True,

23
hermes_cli/managed_downloads.py
View File

@ -181,6 +181,15 @@ def _normalize_owner_metadata(owner_metadata: dict | None) -> dict[str, str]:
return allowed 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( def create_signed_download_token(
file_path: str, 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"]) 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("/") base_url = str(resolve_download_public_url() or "").rstrip("/")
if not base_url: if not base_url:
return "" 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: if not short_id:
return "" return ""
# Preflight: prove the short link resolves before we hand it out. # Preflight: prove the short link resolves before we hand it out.

53
tests/hermes_cli/test_managed_downloads.py
View File

@ -162,3 +162,56 @@ def test_owner_metadata_roundtrips_in_short_download_record(_isolate_hermes_home
"principal_id": "p-123", "principal_id": "p-123",
"external_user_id": "openwebui-42", "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",
}