feat: add download link mode hooks to api responses

docs/profile-isolation/README.md

@@ -5,8 +5,9 @@ This folder is the readable map for the **system profile / principal profile / s
 Read order:
 1. `technical-design.md` — architecture and precedence rules
 2. `checklist.md` — implementation and acceptance checklist
-3. `issue-body-templates.md` — GitHub issue body templates for this workstream
-4. `README.md` — quick file map and why these files exist
+3. `implementation-status.md` — what has already landed in code and what still needs review
+4. `issue-body-templates.md` — GitHub issue body templates for this workstream
+5. `README.md` — quick file map and why these files exist
 
 ## What belongs where
 

docs/profile-isolation/checklist.md

@@ -47,6 +47,21 @@ Verify that global system rules, principal-specific profiles, and session-only c
 - [ ] B cannot ask for A's memory or profile notes
 - [ ] Existence leaks are rejected just like full content leaks
 
+### API-server session and response-chain isolation
+- [ ] `api_server` session CRUD routes operate inside the resolved principal profile only
+- [ ] Session message history resolves from the same principal-scoped `state.db`
+- [ ] `previous_response_id` resolves from the active principal-scoped `response_store.db` only
+- [ ] Conversation-name chaining does not cross principal boundaries
+- [ ] `GET /v1/responses/{id}` and `DELETE /v1/responses/{id}` cannot touch another principal's response store
+
+### Managed download isolation
+- [ ] Owner-bound managed downloads reject callers outside the matching request scope
+- [ ] Public managed downloads strip `owner_metadata` instead of silently reusing owner-bound scope
+- [ ] Public and private download modes preserve the same expiry semantics
+- [ ] Callers can be prompted or guided to choose `private` vs `public` link mode when a managed download is generated
+- [ ] Streaming terminal payloads rewrite managed-download URLs consistently with non-streaming payloads
+- [ ] Any remaining token-level streaming leakage risk is either blocked or explicitly tracked for follow-up
+
 ### OpenWebUI / gateway / email
 - [ ] OpenWebUI resolves the live caller, not a cached operator default
 - [ ] Telegram and LINE use the bound principal, not chat labels alone
@@ -61,8 +76,12 @@ Verify that global system rules, principal-specific profiles, and session-only c
 - [ ] `tests/gateway/test_api_server_session_ownership.py`
 - [ ] `tests/hermes_cli/test_web_server_session_access.py`
 - [ ] `tests/gateway/test_verified_email_handoff_guard_helpers.py`
+- [ ] `tests/gateway/test_api_server.py -k 'ResponsesEndpoint or ResponsesStreaming or Truncation or ConversationParameter'`
+- [ ] `tests/gateway/test_multiplex_api_server_routing.py`
+- [ ] `tests/hermes_cli/test_managed_downloads.py`
 
 ## Done means
 - [ ] The docs in this folder describe the current behavior accurately
+- [ ] `implementation-status.md` reflects the latest landed isolation phases
 - [ ] Code comments point back here and to `technical-design.md`
 - [ ] No code path can silently merge private data into the shared system profile

docs/profile-isolation/implementation-status.md

@@ -0,0 +1,108 @@
+# Profile Isolation Implementation Status
+
+## Purpose
+
+This file is the changelog-style bridge between the design docs in this folder and the code that already landed.
+Use it when you need to answer three questions quickly:
+
+1. What isolation work is already implemented?
+2. Which code paths enforce it?
+3. Which follow-up gaps still need explicit review?
+
+## Completed phases
+
+### Phase 1 — Prompt and profile layering
+
+Implemented:
+- layered prompt assembly for system / principal / session ordering
+- shared knowledge prompt block with separate budget and doc pointers
+- principal-aware profile resolution helpers
+
+Primary files:
+- `agent/prompt_builder.py`
+- `agent/system_prompt.py`
+- `agent/shared_knowledge.py`
+- `agent/agent_init.py`
+- `gateway/principal_profiles.py`
+- `hermes_cli/config.py`
+
+### Phase 2 — Session, dashboard, and memory ownership gates
+
+Implemented:
+- principal-aware session visibility in dashboard routes
+- owner-gated session detail / export / descendant access
+- principal-scoped remote durable memory writes
+- principal-aware `session_search` profile access filtering
+
+Primary files:
+- `hermes_cli/web_server.py`
+- `tools/memory_tool.py`
+- `tools/session_search_tool.py`
+- `gateway/platforms/api_server.py`
+
+### Phase 3 — Verified-identity routing and managed-download ownership
+
+Implemented:
+- verified identity carried into API-server session binding
+- principal-aware managed-download owner metadata
+- owner-bound managed downloads rejected outside the matching request scope
+
+Primary files:
+- `gateway/platforms/api_server.py`
+- `hermes_cli/managed_downloads.py`
+- `hermes_cli/web_server.py`
+
+### Phase 4 — API response-chain and session CRUD isolation
+
+Implemented:
+- `api_server` session CRUD / message / fork / chat endpoints now resolve the verified caller first and then operate inside the matching principal profile scope
+- response snapshots used by `previous_response_id`, conversation-name chaining, and `GET/DELETE /v1/responses/{id}` now live in a per-profile `response_store.db`
+- streaming and non-streaming responses share the same principal-scoped response-store behavior
+- `/v1/runs` now resolves `previous_response_id` from the active principal profile instead of a process-global store
+- a lazy compatibility proxy preserves `adapter._response_store` call sites in tests without opening SQLite on every adapter construction
+
+Primary files:
+- `gateway/platforms/api_server.py`
+
+### Phase 5 — Download-link mode control
+
+Implemented:
+- managed downloads now support explicit `link_mode` selection
+- `private` mode keeps `owner_metadata` bound to the link
+- `public` mode strips `owner_metadata` while preserving the existing expiry policy
+- `api_server` request bodies now accept `download_link_mode` and return machine-readable `hermes.download_link_mode*` metadata so callers can prompt for private vs public link behavior
+- `/v1/responses` non-streaming payloads now keep `response.output[*]` aligned with the rewritten managed-download URL instead of exposing a host-local artifact path in the assistant message item
+- `/v1/responses` streaming terminal events now carry the same managed-download rewrite and `hermes.download_link_mode` metadata as the non-streaming path
+
+Primary files:
+- `hermes_cli/managed_downloads.py`
+- `tests/hermes_cli/test_managed_downloads.py`
+
+## Verified tests already exercised for this workstream
+
+- `tests/agent/test_system_prompt_principal_precedence.py`
+- `tests/agent/test_system_prompt_shared_knowledge.py`
+- `tests/gateway/test_principal_profile_routing_phase1.py`
+- `tests/gateway/test_principal_profile_runtime_phase2.py`
+- `tests/gateway/test_api_server_session_ownership.py`
+- `tests/hermes_cli/test_web_server_session_access.py`
+- `tests/gateway/test_api_server.py -k 'ResponsesEndpoint or ResponsesStreaming or Truncation or ConversationParameter'`
+- `tests/gateway/test_multiplex_api_server_routing.py`
+- `tests/hermes_cli/test_managed_downloads.py`
+
+## Remaining follow-up review items
+
+These are not known regressions, but they remain explicit review targets whenever this isolation layer changes again:
+
+- artifact-rewrite call sites outside `api_server` that may still default to public links without a caller-selected mode
+- gateway UX hooks that should ask the user whether a link should stay owner-bound or become public
+- any future response-chain storage outside `response_store.db`
+- non-API file-delivery paths that may need the same principal-aware mode selection contract
+- live token-by-token streaming deltas may still expose a host-local artifact path before the terminal `response.completed` envelope is rewritten; this is a presentation-path follow-up, not a principal-ownership bypass
+
+## Current invariant
+
+If a request is authenticated as principal **A**, then:
+- it must read and write only principal A's session store and response store,
+- it must not resolve principal B's response chain,
+- and any owner-bound download minted from that request must remain inaccessible to principal B.

docs/profile-isolation/issue-body-templates.md

@@ -399,4 +399,33 @@ Search, export, and shared-session handoff are common leak paths.
 - `tests/hermes_cli/test_web_server_session_access.py`
 - `tests/gateway/test_verified_email_handoff_guard_helpers.py`
 ```
+
+### Issue: Add explicit private-vs-public download link selection hooks
+
+```md
+## Goal
+Expose a stable API/gateway hook so clients can let the user choose whether a generated download link should stay owner-bound or become public.
+
+## Problem
+Managed downloads already enforce owner-bound access, but callers also need a safe way to intentionally request a public link without weakening the private default.
+
+## Scope
+- Accept an explicit `download_link_mode` request hint
+- Preserve private-by-default behavior for verified callers
+- Return response metadata that lets clients prompt the user with `private` vs `public`
+- Keep expiry semantics unchanged across both modes
+
+## Acceptance Criteria
+- [ ] Omitted mode defaults to private for owner-bound downloads
+- [ ] `download_link_mode=public` strips owner metadata and produces a public link
+- [ ] Responses expose machine-readable metadata so UI/gateway layers can prompt users before or after generation
+- [ ] `previous_response_id` / response-store isolation remains principal-scoped
+
+## References
+- `gateway/platforms/api_server.py`
+- `hermes_cli/managed_downloads.py`
+- `hermes_cli/web_server.py`
+- `tests/gateway/test_api_server.py`
+- `tests/hermes_cli/test_managed_downloads.py`
+```
 ```

docs/profile-isolation/technical-design.md

@@ -63,8 +63,9 @@ A lower layer may tailor the response, but it may not override a higher layer.
 2. Load the global system profile.
 3. Load the matching principal profile.
 4. Add session-only context.
-5. Build the prompt.
-6. Refuse to build cross-principal context if the caller cannot be resolved.
+5. Route session storage, response-chain storage, and managed-download ownership through the same resolved principal profile.
+6. Build the prompt.
+7. Refuse to build cross-principal context if the caller cannot be resolved.
 
 ## Write rules
 
@@ -87,6 +88,15 @@ A lower layer may tailor the response, but it may not override a higher layer.
 - Sanitize shared knowledge before injection
 - Keep audit trails for every profile write
 
+## API response-chain and file-delivery requirements
+
+- `previous_response_id`, conversation-name chaining, and response retrieval must resolve from the active principal profile only
+- Session CRUD and message history endpoints must read and write `state.db` inside the resolved principal profile only
+- Managed downloads may expose two delivery modes:
+  - `private`: keep `owner_metadata` and enforce caller-bound access
+  - `public`: strip `owner_metadata` but keep the same expiry semantics
+- Any UI or API that can generate a managed download should be able to surface a caller choice between `private` and `public` without changing the underlying expiry contract
+
 ## Files to inspect during implementation
 
 - `agent/prompt_builder.py`
@@ -98,6 +108,9 @@ A lower layer may tailor the response, but it may not override a higher layer.
 - `tests/agent/test_system_prompt_shared_knowledge.py`
 - `tests/gateway/test_principal_profile_routing_phase1.py`
 - `tests/gateway/test_principal_profile_runtime_phase2.py`
+- `tests/gateway/test_api_server.py`
+- `tests/gateway/test_multiplex_api_server_routing.py`
+- `tests/hermes_cli/test_managed_downloads.py`
 
 ## Rollout note
 

gateway/platforms/api_server.py

@@ -774,7 +774,8 @@ def _rewrite_local_artifact_references_for_remote_client(
     text: str,
     *,
     owner_metadata: Optional[Dict[str, Any]] = None,
-) -> str:
+    link_mode: Optional[str] = None,
+) -> tuple[str, Dict[str, Any]]:
     """Turn host-local file references into managed HTTPS download URLs.
 
     Open WebUI / API-server clients cannot open ``file://`` URIs or host-local
@@ -789,16 +790,24 @@ def _rewrite_local_artifact_references_for_remote_client(
       URL; otherwise the original text is left untouched.
     """
     if not text:
-        return text
+        return text, {"rewritten_count": 0, "link_mode": None, "mode_options": []}
     if "file://" not in text and "MEDIA:" not in text and "/" not in text and "\\" not in text:
-        return text
+        return text, {"rewritten_count": 0, "link_mode": None, "mode_options": []}
 
     try:
-        from hermes_cli.managed_downloads import build_public_download_url
+        from hermes_cli.managed_downloads import build_public_download_url, normalize_download_link_mode
     except Exception:
-        return text
+        return text, {"rewritten_count": 0, "link_mode": None, "mode_options": []}
+
+    normalized_owner = owner_metadata or {}
+    mode_options = ["private", "public"] if normalized_owner else ["public"]
+    effective_mode = normalize_download_link_mode(
+        link_mode,
+        default="private" if normalized_owner else "public",
+    )
 
     rewritten: dict[str, str] = {}
+    rewritten_count = 0
 
     def _public_url_for(raw_path: str) -> Optional[str]:
         key = str(raw_path or "")
@@ -809,7 +818,13 @@ def _rewrite_local_artifact_references_for_remote_client(
             rewritten[key] = ""
             return None
         try:
-            url = str(build_public_download_url(safe_path, owner_metadata=owner_metadata) or "").strip()
+            url = str(
+                build_public_download_url(
+                    safe_path,
+                    owner_metadata=normalized_owner,
+                    link_mode=effective_mode,
+                ) or ""
+            ).strip()
         except Exception:
             url = ""
         rewritten[key] = url
@@ -819,9 +834,13 @@ def _rewrite_local_artifact_references_for_remote_client(
 
     # 1) Convert non-image / leftover MEDIA tags to public download URLs.
     def _media_repl(match: "re.Match[str]") -> str:
+        nonlocal rewritten_count
         raw_path = match.group("path")
         normalized = raw_path.strip("`\"'")
-        return _public_url_for(normalized) or match.group(0)
+        replaced = _public_url_for(normalized)
+        if replaced:
+            rewritten_count += 1
+        return replaced or match.group(0)
 
     try:
         out = MEDIA_TAG_CLEANUP_RE.sub(_media_repl, out)
@@ -831,8 +850,11 @@ def _rewrite_local_artifact_references_for_remote_client(
 
     # 2) Replace explicit file:// URIs.
     def _uri_repl(match: "re.Match[str]") -> str:
+        nonlocal rewritten_count
         raw_path = match.group("path")
         url = _public_url_for(raw_path)
+        if url:
+            rewritten_count += 1
         return url or match.group(0)
 
     out = _LOCAL_FILE_URI_RE.sub(_uri_repl, out)
@@ -840,6 +862,7 @@ def _rewrite_local_artifact_references_for_remote_client(
     # 3) Replace bare absolute-path lines / tokens when they resolve to a safe,
     #    existing deliverable file. Keep surrounding punctuation intact.
     def _path_repl(match: "re.Match[str]") -> str:
+        nonlocal rewritten_count
         raw_path = str(match.group("path") or "")
         candidate = raw_path.strip().strip("`\"'")
         # Avoid greedily swallowing natural-language trailing punctuation.
@@ -847,9 +870,35 @@ def _rewrite_local_artifact_references_for_remote_client(
         url = _public_url_for(candidate)
         if not url:
             return match.group(0)
+        rewritten_count += 1
         return f"{match.group('prefix')}{url}{match.group('suffix')}"
 
-    return _LOCAL_PATH_LINE_RE.sub(_path_repl, out)
+    return _LOCAL_PATH_LINE_RE.sub(_path_repl, out), {
+        "rewritten_count": rewritten_count,
+        "link_mode": effective_mode if rewritten_count else None,
+        "mode_options": mode_options if rewritten_count else [],
+    }
+
+
+def _download_link_hook_metadata(
+    rewrite_info: Optional[Dict[str, Any]],
+    *,
+    requested_mode: Optional[str],
+) -> Optional[Dict[str, Any]]:
+    if not rewrite_info:
+        return None
+    rewritten_count = int(rewrite_info.get("rewritten_count", 0) or 0)
+    mode_options = [str(x) for x in (rewrite_info.get("mode_options") or []) if str(x).strip()]
+    selected_mode = str(rewrite_info.get("link_mode") or "").strip() or None
+    if rewritten_count <= 0 or len(mode_options) <= 1:
+        return None
+    raw_requested = str(requested_mode or "").strip().lower()
+    return {
+        "download_link_mode": selected_mode,
+        "download_link_mode_selected_explicitly": raw_requested in {"private", "public"},
+        "download_link_mode_options": mode_options,
+        "download_link_mode_prompt": "偵測到可下載檔案；如需改成公開下載連結，請在下一次請求指定 download_link_mode=public，否則預設維持 private。",
+    }
 
 
 def _redact_api_error_text(value: Any, *, limit: int | None = None) -> str:
@@ -2815,6 +2864,7 @@ class APIServerAdapter(BasePlatformAdapter):
         user_message, err = _session_chat_user_message(body)
         if err is not None:
             return err
+        requested_download_link_mode = body.get("download_link_mode")
         system_prompt = body.get("system_message") or body.get("instructions")
         identity_context, identity_err = self._verified_identity_context_or_error(request, body)
         if identity_err is not None:
@@ -2838,22 +2888,24 @@ class APIServerAdapter(BasePlatformAdapter):
             identity_context=identity_context,
         )
         effective_session_id = result.get("session_id") if isinstance(result, dict) else session_id
-        final_response = _rewrite_local_artifact_references_for_remote_client(
+        final_response, download_link_info = _rewrite_local_artifact_references_for_remote_client(
             _resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else ""),
             owner_metadata=self._managed_download_owner_metadata(identity_context),
+            link_mode=requested_download_link_mode,
         )
+        hermes_meta = _download_link_hook_metadata(download_link_info, requested_mode=requested_download_link_mode)
         headers = {"X-Hermes-Session-Id": effective_session_id or session_id}
         if gateway_session_key:
             headers["X-Hermes-Session-Key"] = gateway_session_key
-        return web.json_response(
-            {
-                "object": "hermes.session.chat.completion",
-                "session_id": effective_session_id or session_id,
-                "message": {"role": "assistant", "content": final_response},
-                "usage": usage,
-            },
-            headers=headers,
-        )
+        payload = {
+            "object": "hermes.session.chat.completion",
+            "session_id": effective_session_id or session_id,
+            "message": {"role": "assistant", "content": final_response},
+            "usage": usage,
+        }
+        if hermes_meta:
+            payload["hermes"] = hermes_meta
+        return web.json_response(payload, headers=headers)
 
     @_admit_api_agent_request
     async def _handle_session_chat_stream(self, request: "web.Request") -> "web.StreamResponse":
@@ -2868,6 +2920,7 @@ class APIServerAdapter(BasePlatformAdapter):
         user_message, err = _session_chat_user_message(body)
         if err is not None:
             return err
+        requested_download_link_mode = body.get("download_link_mode")
         system_prompt = body.get("system_message") or body.get("instructions")
         identity_context, identity_err = self._verified_identity_context_or_error(request, body)
         if identity_err is not None:
@@ -2937,10 +2990,12 @@ class APIServerAdapter(BasePlatformAdapter):
                     gateway_session_key=gateway_session_key,
                     identity_context=identity_context,
                 )
-                final_response = _rewrite_local_artifact_references_for_remote_client(
+                final_response, download_link_info = _rewrite_local_artifact_references_for_remote_client(
                     _resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else ""),
                     owner_metadata=self._managed_download_owner_metadata(identity_context),
+                    link_mode=requested_download_link_mode,
                 )
+                hermes_meta = _download_link_hook_metadata(download_link_info, requested_mode=requested_download_link_mode)
                 effective_session_id = result.get("session_id", session_id) if isinstance(result, dict) else session_id
                 turn_messages = self._turn_transcript_messages(history, user_message, result) if isinstance(result, dict) else []
                 await queue.put(_event_payload("assistant.completed", {
@@ -2950,6 +3005,7 @@ class APIServerAdapter(BasePlatformAdapter):
                     "completed": True,
                     "partial": False,
                     "interrupted": False,
+                    **({"hermes": hermes_meta} if hermes_meta else {}),
                 }))
                 await queue.put(_event_payload("run.completed", {
                     "session_id": effective_session_id,
@@ -3027,6 +3083,7 @@ class APIServerAdapter(BasePlatformAdapter):
             )
 
         stream = _coerce_request_bool(body.get("stream"), default=False)
+        requested_download_link_mode = body.get("download_link_mode")
         identity_context, identity_err = self._verified_identity_context_or_error(request, body)
         if identity_err is not None:
             return identity_err
@@ -3273,10 +3330,12 @@ class APIServerAdapter(BasePlatformAdapter):
                     status=500,
                 )
 
-        final_response = _rewrite_local_artifact_references_for_remote_client(
+        final_response, download_link_info = _rewrite_local_artifact_references_for_remote_client(
             _resolve_media_to_data_urls(result.get("final_response") or ""),
             owner_metadata=self._managed_download_owner_metadata(identity_context),
+            link_mode=requested_download_link_mode,
         )
+        download_link_hook = _download_link_hook_metadata(download_link_info, requested_mode=requested_download_link_mode)
         is_partial = bool(result.get("partial"))
         is_failed = bool(result.get("failed"))
         completed = bool(result.get("completed", True))
@@ -3349,10 +3408,14 @@ class APIServerAdapter(BasePlatformAdapter):
                 "error": err_msg,
                 "error_code": "output_truncated" if finish_reason == "length" else "agent_error",
             }
+            if download_link_hook:
+                response_data["hermes"].update(download_link_hook)
             response_headers["X-Hermes-Completed"] = "false"
             response_headers["X-Hermes-Partial"] = "true" if is_partial else "false"
             if err_msg:
                 response_headers["X-Hermes-Error"] = _redact_api_error_text(err_msg, limit=200)
+        elif download_link_hook:
+            response_data["hermes"] = download_link_hook
 
         return web.json_response(response_data, headers=response_headers)
 
@@ -3569,6 +3632,7 @@ class APIServerAdapter(BasePlatformAdapter):
         session_id: str,
         gateway_session_key: Optional[str] = None,
         identity_context: Optional[Dict[str, Optional[str]]] = None,
+        requested_download_link_mode: Optional[str] = None,
     ) -> "web.StreamResponse":
         """Write an SSE stream for POST /v1/responses (OpenAI Responses API).
 
@@ -3664,6 +3728,7 @@ class APIServerAdapter(BasePlatformAdapter):
         agent_error: Optional[str] = None
         usage: Dict[str, int] = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
         terminal_snapshot_persisted = False
+        download_link_hook: Optional[Dict[str, Any]] = None
 
         def _persist_response_snapshot(
             response_env: Dict[str, Any],
@@ -3969,10 +4034,19 @@ class APIServerAdapter(BasePlatformAdapter):
                 # the full response at the end), emit a single fallback
                 # delta so Responses clients still receive a live text part.
                 agent_final = result.get("final_response", "") if isinstance(result, dict) else ""
-                if agent_final and not final_text_parts:
-                    await _emit_text_delta(agent_final)
-                if agent_final and not final_response_text:
-                    final_response_text = agent_final
+                rewritten_final, download_link_info = _rewrite_local_artifact_references_for_remote_client(
+                    _resolve_media_to_data_urls(agent_final),
+                    owner_metadata=self._managed_download_owner_metadata(identity_context),
+                    link_mode=requested_download_link_mode,
+                )
+                download_link_hook = _download_link_hook_metadata(
+                    download_link_info,
+                    requested_mode=requested_download_link_mode,
+                )
+                if rewritten_final and not final_text_parts:
+                    await _emit_text_delta(rewritten_final)
+                if rewritten_final and not final_response_text:
+                    final_response_text = rewritten_final
                 if isinstance(result, dict) and result.get("error") and not final_response_text:
                     agent_error = _redact_api_error_text(result["error"])
             except Exception as e:  # noqa: BLE001
@@ -4064,10 +4138,13 @@ class APIServerAdapter(BasePlatformAdapter):
                     conversation_history_snapshot=_failed_history,
                 )
                 terminal_snapshot_persisted = True
-                await _write_event("response.failed", {
+                failed_event = {
                     "type": "response.failed",
                     "response": failed_env,
-                })
+                }
+                if download_link_hook:
+                    failed_event["hermes"] = download_link_hook
+                await _write_event("response.failed", failed_event)
             else:
                 completed_env = _envelope("completed")
                 completed_env["output"] = final_items
@@ -4087,10 +4164,14 @@ class APIServerAdapter(BasePlatformAdapter):
                     conversation_history_snapshot=full_history,
                 )
                 terminal_snapshot_persisted = True
-                await _write_event("response.completed", {
+                completed_event = {
                     "type": "response.completed",
                     "response": completed_env,
-                })
+                }
+                if download_link_hook:
+                    completed_env["hermes"] = download_link_hook
+                    completed_event["hermes"] = download_link_hook
+                await _write_event("response.completed", completed_event)
 
         except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError, OSError):
             _persist_incomplete_if_needed()
@@ -4181,6 +4262,7 @@ class APIServerAdapter(BasePlatformAdapter):
         instructions = body.get("instructions")
         previous_response_id = body.get("previous_response_id")
         conversation = body.get("conversation")
+        requested_download_link_mode = body.get("download_link_mode")
         identity_context, identity_err = self._verified_identity_context_or_error(request, body)
         if identity_err is not None:
             return identity_err
@@ -4355,6 +4437,7 @@ class APIServerAdapter(BasePlatformAdapter):
                 session_id=session_id,
                 gateway_session_key=gateway_session_key,
                 identity_context=identity_context,
+                requested_download_link_mode=requested_download_link_mode,
             )
 
         async def _compute_response():
@@ -4392,10 +4475,12 @@ class APIServerAdapter(BasePlatformAdapter):
                     status=500,
                 )
 
-        final_response = _rewrite_local_artifact_references_for_remote_client(
+        final_response, download_link_info = _rewrite_local_artifact_references_for_remote_client(
             _resolve_media_to_data_urls(result.get("final_response", "")),
             owner_metadata=self._managed_download_owner_metadata(identity_context),
+            link_mode=requested_download_link_mode,
         )
+        download_link_hook = _download_link_hook_metadata(download_link_info, requested_mode=requested_download_link_mode)
         if not final_response:
             final_response = _redact_api_error_text(result.get("error", "(No response generated)"))
 
@@ -4420,6 +4505,18 @@ class APIServerAdapter(BasePlatformAdapter):
             result,
         )
         output_items = self._extract_output_items(result, start_index=output_start_index)
+        if output_items:
+            last_item = output_items[-1]
+            if (
+                isinstance(last_item, dict)
+                and last_item.get("type") == "message"
+                and last_item.get("role") == "assistant"
+                and isinstance(last_item.get("content"), list)
+                and last_item["content"]
+                and isinstance(last_item["content"][0], dict)
+                and last_item["content"][0].get("type") == "output_text"
+            ):
+                last_item["content"][0]["text"] = final_response
 
         response_data = {
             "id": response_id,
@@ -4434,6 +4531,8 @@ class APIServerAdapter(BasePlatformAdapter):
                 "total_tokens": usage.get("total_tokens", 0),
             },
         }
+        if download_link_hook:
+            response_data["hermes"] = download_link_hook
 
         # Store the complete response object for future chaining / GET retrieval
         if store and response_store is not None:

tests/gateway/test_api_server.py

@@ -1386,6 +1386,39 @@ class TestChatCompletionsEndpoint:
             fake_task.callbacks[0](fake_task)
             assert stream_q.get_nowait() is None
 
+    @pytest.mark.asyncio
+    async def test_download_link_hook_metadata_is_returned_for_chat_completions(self, adapter):
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            with (
+                patch.object(adapter, "_run_agent", new_callable=AsyncMock) as mock_run,
+                patch(
+                    "gateway.platforms.api_server._rewrite_local_artifact_references_for_remote_client",
+                    return_value=(
+                        "https://hermesadmin.bremen.com.tw/downloads/s/s-abc123",
+                        {"rewritten_count": 1, "link_mode": "private", "mode_options": ["private", "public"]},
+                    ),
+                ),
+            ):
+                mock_run.return_value = (
+                    {"final_response": "/Users/hermes/result.pdf", "messages": [], "api_calls": 1},
+                    {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
+                )
+                resp = await cli.post(
+                    "/v1/chat/completions",
+                    json={
+                        "model": "hermes-agent",
+                        "messages": [{"role": "user", "content": "請給我檔案"}],
+                    },
+                )
+
+            assert resp.status == 200
+            data = await resp.json()
+            assert data["choices"][0]["message"]["content"].startswith("https://hermesadmin")
+            assert data["hermes"]["download_link_mode"] == "private"
+            assert data["hermes"]["download_link_mode_selected_explicitly"] is False
+            assert data["hermes"]["download_link_mode_options"] == ["private", "public"]
+
     @pytest.mark.asyncio
     async def test_stream_sends_keepalive_during_quiet_tool_gap(self, adapter):
         """Idle SSE streams should send keepalive comments while tools run silently."""
@@ -1976,6 +2009,40 @@ class TestResponsesEndpoint:
             assert data["output"][0]["content"][0]["type"] == "output_text"
             assert data["output"][0]["content"][0]["text"] == "Paris is the capital of France."
 
+    @pytest.mark.asyncio
+    async def test_download_link_hook_metadata_is_returned_for_responses(self, adapter):
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            with (
+                patch.object(adapter, "_run_agent", new_callable=AsyncMock) as mock_run,
+                patch(
+                    "gateway.platforms.api_server._rewrite_local_artifact_references_for_remote_client",
+                    return_value=(
+                        "https://hermesadmin.bremen.com.tw/downloads/s/s-public1",
+                        {"rewritten_count": 1, "link_mode": "public", "mode_options": ["private", "public"]},
+                    ),
+                ),
+            ):
+                mock_run.return_value = (
+                    {"final_response": "/Users/hermes/result.pdf", "messages": [], "api_calls": 1},
+                    {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0},
+                )
+                resp = await cli.post(
+                    "/v1/responses",
+                    json={
+                        "model": "hermes-agent",
+                        "input": "請給我檔案",
+                        "download_link_mode": "public",
+                    },
+                )
+
+            assert resp.status == 200
+            data = await resp.json()
+            assert data["hermes"]["download_link_mode"] == "public"
+            assert data["hermes"]["download_link_mode_selected_explicitly"] is True
+            assert data["hermes"]["download_link_mode_options"] == ["private", "public"]
+            assert data["output"][0]["content"][0]["text"] == "https://hermesadmin.bremen.com.tw/downloads/s/s-public1"
+
     @pytest.mark.asyncio
     async def test_successful_response_with_array_input(self, adapter):
         """Array input with role/content objects."""
@@ -2439,6 +2506,36 @@ class TestResponsesStreaming:
                 assert "Hello" in body
                 assert " world" in body
 
+    @pytest.mark.asyncio
+    async def test_streaming_responses_completed_event_includes_download_link_hook(self, adapter):
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            async def _mock_run_agent(**kwargs):
+                return (
+                    {"final_response": "/Users/hermes/result.pdf", "messages": [], "api_calls": 1},
+                    {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
+                )
+
+            with (
+                patch.object(adapter, "_run_agent", side_effect=_mock_run_agent),
+                patch(
+                    "gateway.platforms.api_server._rewrite_local_artifact_references_for_remote_client",
+                    return_value=(
+                        "https://hermesadmin.bremen.com.tw/downloads/s/s-private-stream",
+                        {"rewritten_count": 1, "link_mode": "private", "mode_options": ["private", "public"]},
+                    ),
+                ),
+            ):
+                resp = await cli.post(
+                    "/v1/responses",
+                    json={"model": "hermes-agent", "input": "hi", "stream": True},
+                )
+                assert resp.status == 200
+                body = await resp.text()
+                assert "event: response.completed" in body
+                assert "https://hermesadmin.bremen.com.tw/downloads/s/s-private-stream" in body
+                assert '"download_link_mode": "private"' in body
+
     @pytest.mark.asyncio
     async def test_stream_string_false_returns_json_response(self, adapter):
         """Quoted false must not route Responses API requests into SSE mode."""