From b1984cf4abdb362163ab2bdfdb493c9810d74c45 Mon Sep 17 00:00:00 2001
From: ShitK <80999801+ShitK@users.noreply.github.com>
Date: Tue, 28 Jul 2026 21:31:23 +0800
Subject: [PATCH] fix(security): reject legacy MCP credentials in run metadata
 (#4448)

* docs: design run metadata secret admission

* docs: refine run metadata secret boundaries

* docs: plan run metadata secret fix

* fix(security): centralize legacy run metadata policy

* fix(security): reject secrets at run admission

* fix(security): hide legacy secrets from history APIs

* docs(security): migrate MCP credentials to secret context

* fix(security): redact legacy runnable config metadata

* fix(security): reject legacy config metadata credentials

* fix(security): hide legacy secrets from run kwargs

* docs(security): clarify config redaction boundary

* docs: keep issue 4416 planning local
---
 README.md                                     |   6 +
 backend/AGENTS.md                             |   6 +-
 backend/app/gateway/routers/thread_runs.py    |  27 +-
 backend/app/gateway/routers/threads.py        |  14 +-
 backend/app/gateway/services.py               |  14 +-
 backend/docs/MCP_SERVER.md                    |  55 +-
 .../deerflow/runtime/secret_context.py        |  43 +-
 backend/tests/test_gateway_services.py        | 174 ++++
 backend/tests/test_mcp_session_pool.py        |  53 ++
 backend/tests/test_run_events_endpoint.py     |  46 +
 .../tests/test_run_metadata_secret_safety.py  | 144 ++++
 .../test_skill_request_scoped_secrets.py      |  10 +-
 ...6-07-24-issue-4416-run-metadata-secrets.md | 805 ++++++++++++++++++
 13 files changed, 1370 insertions(+), 27 deletions(-)
 create mode 100644 backend/tests/test_run_metadata_secret_safety.py
 create mode 100644 docs/superpowers/plans/2026-07-24-issue-4416-run-metadata-secrets.md

diff --git a/README.md b/README.md
index 4d5d7d9d7..c2ee2919c 100644
--- a/README.md
+++ b/README.md
@@ -385,6 +385,12 @@ For stdio MCP servers, per-tool call timeouts can be configured with `tool_call_
 MCP routing hints can also prefer a specific MCP tool for matching requests without forbidding other tools. When `tool_search` defers MCP schemas, matching routing metadata can auto-promote up to `tool_search.auto_promote_top_k` deferred schemas before the model call.
 See the [MCP Server Guide](backend/docs/MCP_SERVER.md) for detailed instructions.
 
+Security: pass per-request MCP credentials only through `config.context.secrets`;
+credentials must never be placed in either run metadata surface
+(`metadata.auth_token` or `config.metadata.auth_token`). See [MCP credential migration and cleanup](backend/docs/MCP_SERVER.md#migrating-legacy-mcp-credentials)
+for the supported interceptor flow and the required rotation and retained-copy
+cleanup when migrating from legacy metadata credentials.
+
 #### IM Channels
 
 DeerFlow supports receiving tasks from messaging apps. Channels auto-start when configured — no public IP required for any of them.
diff --git a/backend/AGENTS.md b/backend/AGENTS.md
index 77662356a..e55ed8b87 100644
--- a/backend/AGENTS.md
+++ b/backend/AGENTS.md
@@ -705,13 +705,15 @@ E2B output sync records remote file versions and actual host file metadata in a
 Lets a caller pass per-request, short-lived end-user credentials (e.g. an ERP token) to a skill's sandbox scripts without the value entering the prompt, tool arguments, the executed command string, or traces (issue #3861).
 
 - **Declare**: a skill lists the secrets it needs in `SKILL.md` frontmatter — `required-secrets:` as a string list or `{name, optional}` mappings. `name` is both the lookup key and the env var name exposed to scripts. Parsed by `skills/parser.py::parse_required_secrets` into `Skill.required_secrets` (`SecretRequirement`); malformed entries are dropped with a warning.
-- **Carry**: the caller sends values out-of-band in the run request's `context.secrets` mapping (never a message). `runtime/secret_context.py` owns the contract (`SECRETS_CONTEXT_KEY`, `extract_request_secrets`). The existing `context` passthrough carries it to `runtime.context` without mirroring into `configurable`. `build_run_config` still sets `configurable.thread_id` on the context path — the checkpointer requires it.
+- **Carry**: the caller sends values out-of-band in the run request's `context.secrets` mapping (never a message). `runtime/secret_context.py` owns the contract (`SECRETS_CONTEXT_KEY`, `extract_request_secrets`). The existing `context` passthrough carries it to `runtime.context` without mirroring into `configurable`. `build_run_config` still sets `configurable.thread_id` on the context path — the checkpointer requires it. MCP tool interceptors can read the same live carrier from `langgraph.config.get_config()["context"]["secrets"]`; see `docs/MCP_SERVER.md`.
+- **Admission and redaction ownership**: `services.py::start_run()` validates both legacy request mappings, `metadata.auth_token` and `config.metadata.auth_token`, before any run or thread persistence. `runtime/secret_context.py::redact_config_secrets()` also removes nested config metadata secrets from observable and persisted config copies; historical `RunResponse.kwargs` applies the same redaction non-mutatively, leaving stored `RunRecord` data unchanged. Keep callers on `config.context.secrets` rather than adding another credential carrier. Scheduled task definitions have no durable credential carrier: `ScheduledTaskService` supplies only `scheduled_task_id`, `scheduled_task_run_id`, and `scheduled_trigger` as run metadata.
 - **Bind (point A+)**: `SkillActivationMiddleware._resolve_secret_bindings` recomputes the injection set (`runtime.context[__active_skill_secrets]`) on every model call from two unioned sources, then REPLACES the key. (1) *Slash*: the run's most recent `/skill` activation, persisted as a source on the run context (only the activated skill's **canonical container path**, never its declared secrets) so the whole tool loop after the activation call keeps the binding; a new activation replaces it. Slash reads the genuine user text via `get_original_user_content_text`; `InputSanitizationMiddleware` preserves it (`ORIGINAL_USER_CONTENT_KEY`), so activation fires even after sanitization. (2) *In-context* (autonomous invocation): skills the model actually loaded in this thread — `ThreadState.skill_context` entries. **Both sources resolve the live registry skill by normalized container path on every call** (`_resolve_registry_skill`) and bind only that skill's own declared secrets — enabled + allowlist checked for both; the `secrets-autonomous: false` opt-out (malformed values fail closed to `false`) additionally gates the in-context path but exempts explicit slash. Resolving by registry — not by trusting the source's stored data — is what makes a caller-forged `__slash_skill_secret_source` harmless (`runtime.context` is caller-mergeable; the gateway also strips caller `__`-keys in `build_run_config`), #3938. Authorization is three-gated regardless of activation style: skill **enabled** by the operator × values **supplied per-request** by the caller (`context.secrets`) × names **declared** in frontmatter (∩ semantics). Because the set is recomputed per call, a skill evicted from `skill_context` (capacity) or a caller that stops supplying a value loses injection on the next call. The injected value always comes from the caller's request, never the host environment (scrubbed first — see below), so a declared name that also exists in the host env is safe: the caller's value wins and the host value is dropped (the #3861 per-user-key-overrides-shared-key case). Missing required secrets are logged once per binding change, not injected; binding changes are recorded as a `middleware:skill_secrets` journal event (skill and secret names only, never values).
 - **Inject**: `bash_tool` reads the injection set and passes it as `execute_command(env=...)`. Scope is the activation turn/run only — a run without `/skill` activation injects nothing.
 - **AIO image requirement**: on `AioSandbox` the env path uses the `bash.exec` API (`POST /v1/bash/exec`), which upstream all-in-one-sandbox only ships since `1.9.3` — older images (including a `latest` tag frozen on the `1.0.0.x` line) 404 the whole `/v1/bash/*` namespace. `AioSandbox` detects the 404, remembers the capability gap on the instance, and fails fast with an actionable upgrade error instead of letting the model retry raw 404s; there is deliberately **no** fallback through the legacy shell path because none keeps the secret values out of the command string (#3921). Regression tests: `tests/test_aio_sandbox.py::TestBashExecUnsupportedFailFast`.
 - **Inherited-env scrub**: `execute_command` no longer leaks the Gateway's `os.environ` to skill subprocesses — `env_policy.build_sandbox_env` drops secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`/`*DSN*` + a connection-string denylist like `DATABASE_URL`/`REDIS_URL`/`GH_PAT`, plus no-flag credential sources like `MYSQL_PWD`/`REDISCLI_AUTH`/`PGPASSFILE`/`PGSERVICEFILE`) so platform credentials never reach a skill; a skill that needs one must declare it.
 - **Leak surfaces sealed** (verified by a real-gateway e2e run — secret reaches the sandbox but none of these): prompt (value never in a message), trace (`tracing/metadata.py` never copies `context`), checkpoint (secrets live on `runtime.context`, not graph state), audit (journal records names only), stdout (`tools.py::mask_secret_values` redacts injected values from bash output), and **run-record persistence + run API** (`services.py::start_run` stores `redact_config_secrets(body.config)` so `runs.kwargs_json` and `RunResponse.kwargs` never carry the secret).
-- **Scope / non-goals**: no persistence/vaulting — values are request-scoped and never stored server-side, so long-lived use means the caller re-supplies `context.secrets` on each request while the skill stays in `skill_context`; subagents do not inherit the injection set; the MCP per-user-credential gap (#3322) is a sibling, not covered here. Tests: `tests/test_skill_request_scoped_secrets.py`.
+- **Historical retention**: API response hiding prevents legacy `metadata.auth_token` and `config.metadata.auth_token` from being returned now; it does not delete values already retained in databases, run events, logs, snapshots, exports, or backups. Deployments that ever used either legacy carrier must rotate the credential and clean every retained copy under their retention policy. Restarting or upgrading DeerFlow performs neither action.
+- **Scope / non-goals**: no persistence/vaulting — values are request-scoped and never stored server-side, so long-lived use means the caller re-supplies `context.secrets` on each request while the skill stays in `skill_context`; subagents do not inherit the skill injection set. MCP interceptors may independently consume the same supported request-scoped carrier. Tests: `tests/test_skill_request_scoped_secrets.py`, `tests/test_mcp_session_pool.py`.
 
 ### Model Factory (`packages/harness/deerflow/models/factory.py`)
 
diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py
index 76f0fa1de..fa9ffbcb4 100644
--- a/backend/app/gateway/routers/thread_runs.py
+++ b/backend/app/gateway/routers/thread_runs.py
@@ -39,6 +39,7 @@ from app.gateway.run_models import RunCreateRequest
 from app.gateway.services import build_checkpoint_state_accessor, build_thread_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion
 from app.gateway.utils import sanitize_log_param
 from deerflow.runtime import CancelOutcome, RunRecord, RunStatus, serialize_channel_values_for_api
+from deerflow.runtime.secret_context import redact_config_secrets, redact_metadata_secrets
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text, message_to_text
 from deerflow.workspace_changes import get_workspace_changes_response
 
@@ -213,13 +214,17 @@ async def _raise_lease_valid_elsewhere(
 
 
 def _record_to_response(record: RunRecord) -> RunResponse:
+    kwargs = dict(record.kwargs or {})
+    if "config" in kwargs:
+        kwargs["config"] = redact_config_secrets(kwargs["config"])
+
     return RunResponse(
         run_id=record.run_id,
         thread_id=record.thread_id,
         assistant_id=record.assistant_id,
         status=record.status.value,
-        metadata=record.metadata,
-        kwargs=record.kwargs,
+        metadata=redact_metadata_secrets(record.metadata),
+        kwargs=kwargs,
         multitask_strategy=record.multitask_strategy,
         created_at=record.created_at,
         updated_at=record.updated_at,
@@ -1300,7 +1305,23 @@ async def list_run_events(
     """
     event_store = get_run_event_store(request)
     types = event_types.split(",") if event_types else None
-    return await event_store.list_events(thread_id, run_id, event_types=types, task_id=task_id, limit=limit, after_seq=after_seq)
+    events = await event_store.list_events(
+        thread_id,
+        run_id,
+        event_types=types,
+        task_id=task_id,
+        limit=limit,
+        after_seq=after_seq,
+    )
+    return [
+        {
+            **event,
+            "metadata": redact_metadata_secrets(event.get("metadata")),
+        }
+        if isinstance(event, dict) and "metadata" in event
+        else event
+        for event in events
+    ]
 
 
 @router.get("/{thread_id}/runs/{run_id}/workspace-changes")
diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py
index 31ac9ac8b..acc67c214 100644
--- a/backend/app/gateway/routers/threads.py
+++ b/backend/app/gateway/routers/threads.py
@@ -65,6 +65,7 @@ from deerflow.runtime.goal import (
 from deerflow.runtime.journal import build_branch_history_seed_events
 from deerflow.runtime.runs.manager import ConflictError
 from deerflow.runtime.runs.worker import valid_duration_entry
+from deerflow.runtime.secret_context import redact_metadata_secrets
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.utils.file_io import run_file_io
 from deerflow.utils.time import coerce_iso, now_iso
@@ -349,7 +350,14 @@ class ThreadDeleteResponse(BaseModel):
     message: str
 
 
-class ThreadResponse(BaseModel):
+class _MetadataRedactingResponse(BaseModel):
+    @field_validator("metadata", mode="before", check_fields=False)
+    @classmethod
+    def _redact_legacy_metadata_secret(cls, value: Any) -> Any:
+        return redact_metadata_secrets(value)
+
+
+class ThreadResponse(_MetadataRedactingResponse):
     """Response model for a single thread."""
 
     thread_id: str = Field(description="Unique thread identifier")
@@ -402,7 +410,7 @@ class ThreadSearchRequest(BaseModel):
         return v
 
 
-class ThreadStateResponse(BaseModel):
+class ThreadStateResponse(_MetadataRedactingResponse):
     """Response model for thread state."""
 
     values: dict[str, Any] = Field(default_factory=dict, description="Current channel values")
@@ -472,7 +480,7 @@ class ThreadCompactResponse(BaseModel):
     total_tokens: int = 0
 
 
-class HistoryEntry(BaseModel):
+class HistoryEntry(_MetadataRedactingResponse):
     """Single checkpoint history entry."""
 
     checkpoint_id: str
diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py
index c46d5cb29..bdc0461a8 100644
--- a/backend/app/gateway/services.py
+++ b/backend/app/gateway/services.py
@@ -61,7 +61,11 @@ from deerflow.runtime.checkpoint_mode import (
 from deerflow.runtime.checkpoint_state import graph_state_schema
 from deerflow.runtime.goal import goal_thread_lock
 from deerflow.runtime.runs.naming import resolve_root_run_name
-from deerflow.runtime.secret_context import redact_config_secrets
+from deerflow.runtime.secret_context import (
+    LegacyRunMetadataSecretError,
+    redact_config_secrets,
+    validate_run_metadata_secrets,
+)
 from deerflow.runtime.stream_modes import normalize_stream_modes
 from deerflow.runtime.user_context import reset_current_user, set_current_user
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
@@ -978,6 +982,14 @@ async def start_run(
     request : Request
         FastAPI request — used to retrieve singletons from ``app.state``.
     """
+    body_config = getattr(body, "config", None)
+    config_metadata = body_config.get("metadata") if isinstance(body_config, dict) else None
+    try:
+        validate_run_metadata_secrets(getattr(body, "metadata", None))
+        validate_run_metadata_secrets(config_metadata)
+    except LegacyRunMetadataSecretError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
     stream_modes = normalize_stream_modes(body.stream_mode)
     bridge = get_stream_bridge(request)
     run_mgr = get_run_manager(request)
diff --git a/backend/docs/MCP_SERVER.md b/backend/docs/MCP_SERVER.md
index db2b18e38..cc04597e8 100644
--- a/backend/docs/MCP_SERVER.md
+++ b/backend/docs/MCP_SERVER.md
@@ -154,24 +154,65 @@ Declare interceptors in `extensions_config.json` using the `mcpInterceptors` fie
 
 Each entry is a Python import path in `module:variable` format (resolved via `resolve_variable`). The variable must be a **no-arg builder function** that returns an async interceptor compatible with `MultiServerMCPClient`’s `tool_interceptors` interface, or `None` to skip.
 
-Example interceptor that injects auth headers from LangGraph metadata:
+Example interceptor that injects an authorization header from the request-scoped
+LangGraph secret context:
 
 ```python
+from langgraph.config import get_config
+
+
 def build_auth_interceptor():
     async def interceptor(request, handler):
-        from langgraph.config import get_config
-        metadata = get_config().get("metadata", {})
-        headers = dict(request.headers or {})
-        if token := metadata.get("auth_token"):
-            headers["X-Auth-Token"] = token
-        return await handler(request.override(headers=headers))
+        config = get_config()
+        secrets = (config.get("context") or {}).get("secrets") or {}
+        token = secrets.get("MCP_AUTH_TOKEN")
+        if token:
+            request = request.override(
+                headers={**(request.headers or {}), "Authorization": f"Bearer {token}"}
+            )
+        return await handler(request)
+
     return interceptor
 ```
 
+Supply the credential on each run request through `config.context.secrets`:
+
+```json
+{
+  "metadata": {"source": "my-client"},
+  "config": {
+    "context": {
+      "secrets": {"MCP_AUTH_TOKEN": "<request-scoped credential>"}
+    }
+  }
+}
+```
+
+Both `metadata.auth_token` and `config.metadata.auth_token` are rejected with HTTP 422 at run admission and are never supported
+interceptor paths. Do not put credentials in either metadata surface; use
+`config.context.secrets`, whose values remain available to the live interceptor
+but are removed from persisted and API-visible run configuration copies.
+
 - A single string value is accepted and normalized to a one-element list.
 - Invalid paths or builder failures are logged as warnings without blocking other interceptors.
 - The builder return value must be `callable`; non-callable values are skipped with a warning.
 
+### Migrating legacy MCP credentials
+
+Deployments that previously sent `metadata.auth_token` or `config.metadata.auth_token` must:
+
+1. Update the caller and interceptor to use `config.context.secrets` as shown
+   above.
+2. Rotate the exposed credential before resuming authenticated MCP traffic.
+3. Locate and remove every retained legacy copy according to the deployment's
+   retention policy, including database rows, run events, application or proxy
+   logs, snapshots, exports, and backups.
+
+Current history APIs hide legacy `metadata.auth_token` and `config.metadata.auth_token` values, but hiding a response does not erase
+material already retained by those systems. Restarting or upgrading DeerFlow does
+not rotate credentials or perform historical cleanup; operators must complete
+both actions explicitly.
+
 ## How It Works
 
 MCP servers expose tools that are automatically discovered and integrated into DeerFlow’s agent system at runtime. Once enabled, these tools become available to agents without additional code changes.
diff --git a/backend/packages/harness/deerflow/runtime/secret_context.py b/backend/packages/harness/deerflow/runtime/secret_context.py
index f564a7caf..c94fc1276 100644
--- a/backend/packages/harness/deerflow/runtime/secret_context.py
+++ b/backend/packages/harness/deerflow/runtime/secret_context.py
@@ -31,6 +31,25 @@ ACTIVE_SECRETS_CONTEXT_KEY = "__active_skill_secrets"
 # entire value must be stripped from every observable serialization surface.
 SKILL_TOOL_POLICY_DECISION_CONTEXT_KEY = "__skill_tool_policy_decision"
 
+LEGACY_AUTH_TOKEN_METADATA_KEY = "auth_token"
+
+
+class LegacyRunMetadataSecretError(ValueError):
+    """Raised when a run puts a request credential in persisted metadata."""
+
+
+def validate_run_metadata_secrets(metadata: Any) -> None:
+    """Reject the legacy credential field at run admission."""
+    if isinstance(metadata, dict) and LEGACY_AUTH_TOKEN_METADATA_KEY in metadata:
+        raise LegacyRunMetadataSecretError("Run metadata key 'auth_token' is not allowed; pass request-scoped credentials via config.context.secrets instead.")
+
+
+def redact_metadata_secrets(metadata: Any) -> Any:
+    """Return API-safe metadata without mutating historical storage objects."""
+    if not isinstance(metadata, dict):
+        return metadata
+    return {key: value for key, value in metadata.items() if key != LEGACY_AUTH_TOKEN_METADATA_KEY}
+
 
 def _string_pairs(raw: Any) -> dict[str, str]:
     if not isinstance(raw, dict):
@@ -130,17 +149,23 @@ def redact_secret_context_keys(context: Any) -> Any:
 def redact_config_secrets(config: Any) -> Any:
     """Return a copy of a run config safe to persist or echo back to clients.
 
-    The request config (``body.config``) is stored verbatim on the run record
-    (``runs.kwargs_json``) and echoed by the run API. Strip the secret-bearing
-    keys from its ``context`` so a request-scoped secret is never persisted or
-    returned, while the live config that drives the run (built separately) keeps
-    them. Non-dict / context-less configs pass through unchanged.
+    The request config (``body.config``) would otherwise be stored verbatim on
+    the run record (``runs.kwargs_json``) and echoed by the run API. Strip secret-bearing keys
+    from its ``context`` and legacy credentials from its ``metadata`` so neither
+    protected config surface is persisted or returned, while the live config
+    that drives the run (built separately) keeps them. Ordinary metadata is
+    preserved. Non-dict configs pass through unchanged.
     """
     if not isinstance(config, dict):
         return config
-    context = config.get("context")
-    if not isinstance(context, dict):
-        return config
+
     redacted = dict(config)
-    redacted["context"] = redact_secret_context_keys(context)
+    context = config.get("context")
+    if isinstance(context, dict):
+        redacted["context"] = redact_secret_context_keys(context)
+
+    metadata = config.get("metadata")
+    if isinstance(metadata, dict):
+        redacted["metadata"] = redact_metadata_secrets(metadata)
+
     return redacted
diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py
index a9bd9b44f..510734299 100644
--- a/backend/tests/test_gateway_services.py
+++ b/backend/tests/test_gateway_services.py
@@ -1330,6 +1330,155 @@ async def _capture_start_run_graph_input(body, *, auth_source=None):
     return captured["graph_input"]
 
 
+def _make_start_run_persistence_context():
+    from types import SimpleNamespace
+
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.store.memory import InMemoryStore
+
+    from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
+    from deerflow.runtime import RunManager
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    run_store = MemoryRunStore()
+    thread_store = MemoryThreadMetaStore(InMemoryStore())
+    state = SimpleNamespace(
+        stream_bridge=SimpleNamespace(),
+        run_manager=RunManager(store=run_store),
+        checkpointer=InMemorySaver(),
+        store=InMemoryStore(),
+        run_event_store=SimpleNamespace(),
+        run_events_config=None,
+        thread_store=thread_store,
+        checkpoint_channel_mode="full",
+        scheduled_task_service=None,
+    )
+    request = SimpleNamespace(
+        headers={},
+        state=SimpleNamespace(),
+        app=SimpleNamespace(state=state),
+    )
+    return request, run_store, thread_store
+
+
+def test_start_run_rejects_legacy_auth_token_before_persistence():
+    import asyncio
+    from unittest.mock import AsyncMock, patch
+
+    from fastapi import HTTPException
+
+    from app.gateway.routers.thread_runs import RunCreateRequest
+    from app.gateway.services import start_run
+
+    async def _scenario():
+        request, run_store, thread_store = _make_start_run_persistence_context()
+        body = RunCreateRequest(
+            assistant_id="lead_agent",
+            input={"messages": [{"role": "user", "content": "hi"}]},
+            metadata={"auth_token": "legacy-secret", "token_usage": 7},
+        )
+
+        with patch("app.gateway.services.run_agent", new_callable=AsyncMock) as run_agent:
+            with pytest.raises(HTTPException) as exc_info:
+                await start_run(body, "thread-secret-admission", request)
+
+        assert exc_info.value.status_code == 422
+        assert "config.context.secrets" in str(exc_info.value.detail)
+        assert await run_store.list_by_thread("thread-secret-admission") == []
+        assert await thread_store.get("thread-secret-admission") is None
+        run_agent.assert_not_called()
+
+    asyncio.run(_scenario())
+
+
+def test_start_run_rejects_legacy_auth_token_in_config_metadata_before_persistence(
+    _stub_app_config,
+):
+    import asyncio
+    from unittest.mock import AsyncMock, patch
+
+    from fastapi import HTTPException
+
+    from app.gateway.routers.thread_runs import RunCreateRequest
+    from app.gateway.services import start_run
+
+    async def _scenario():
+        request, run_store, thread_store = _make_start_run_persistence_context()
+        body = RunCreateRequest(
+            assistant_id="lead_agent",
+            input={"messages": [{"role": "user", "content": "hi"}]},
+            metadata={"token_usage": 7},
+            config={
+                "metadata": {
+                    "auth_token": "legacy-secret",
+                    "nested": {"auth_token": "ordinary-nested-metadata"},
+                }
+            },
+        )
+        create_or_reject = AsyncMock(side_effect=AssertionError("run persistence was reached"))
+
+        with patch.object(
+            request.app.state.run_manager,
+            "create_or_reject",
+            new=create_or_reject,
+        ):
+            with pytest.raises(HTTPException) as exc_info:
+                await start_run(body, "thread-config-secret-admission", request)
+
+        assert exc_info.value.status_code == 422
+        assert "config.context.secrets" in str(exc_info.value.detail)
+        create_or_reject.assert_not_awaited()
+        assert await run_store.list_by_thread("thread-config-secret-admission") == []
+        assert await thread_store.get("thread-config-secret-admission") is None
+
+    asyncio.run(_scenario())
+
+
+def test_start_run_preserves_ordinary_metadata(_stub_app_config):
+    import asyncio
+    from typing import Any
+    from unittest.mock import patch
+
+    from app.gateway.routers.thread_runs import RunCreateRequest
+    from app.gateway.services import start_run
+
+    async def _scenario():
+        thread_id = "thread-ordinary-metadata"
+        metadata = {"token_usage": 7, "source": "regression"}
+        request, _run_store, thread_store = _make_start_run_persistence_context()
+        captured: dict[str, Any] = {}
+
+        async def fake_run_agent(*args, **kwargs):
+            captured["config"] = kwargs["config"]
+
+        with (
+            patch(
+                "app.gateway.services.resolve_agent_factory",
+                return_value=object(),
+            ),
+            patch(
+                "app.gateway.services.run_agent",
+                side_effect=fake_run_agent,
+            ),
+        ):
+            record = await start_run(
+                RunCreateRequest(
+                    assistant_id="lead_agent",
+                    input={"messages": [{"role": "user", "content": "hi"}]},
+                    metadata=metadata,
+                ),
+                thread_id,
+                request,
+            )
+            await record.task
+
+        assert record.metadata == metadata
+        assert (await thread_store.get(thread_id))["metadata"] == metadata
+        assert captured["config"]["metadata"] == metadata
+
+    asyncio.run(_scenario())
+
+
 def test_start_run_translates_resume_command_to_langgraph_command(_stub_app_config):
     import asyncio
 
@@ -1723,6 +1872,31 @@ def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_con
     assert result == {"run_id": "run-1", "thread_id": "thread-scheduled"}
 
 
+def test_launch_scheduled_thread_run_rejects_legacy_auth_token():
+    """The internal launcher shares run admission; task API/model state has no metadata field."""
+    import asyncio
+    from types import SimpleNamespace
+
+    from fastapi import HTTPException
+
+    from app.gateway.services import launch_scheduled_thread_run
+
+    async def _scenario():
+        with pytest.raises(HTTPException) as exc_info:
+            await launch_scheduled_thread_run(
+                thread_id="thread-scheduled",
+                assistant_id="lead_agent",
+                prompt="Run in background",
+                app=SimpleNamespace(state=SimpleNamespace()),
+                metadata={"auth_token": "legacy-secret"},
+            )
+
+        assert exc_info.value.status_code == 422
+        assert "config.context.secrets" in str(exc_info.value.detail)
+
+    asyncio.run(_scenario())
+
+
 # ---------------------------------------------------------------------------
 # build_run_config — context / configurable precedence (LangGraph >= 0.6.0)
 # ---------------------------------------------------------------------------
diff --git a/backend/tests/test_mcp_session_pool.py b/backend/tests/test_mcp_session_pool.py
index 00395db54..b45846346 100644
--- a/backend/tests/test_mcp_session_pool.py
+++ b/backend/tests/test_mcp_session_pool.py
@@ -590,6 +590,59 @@ async def test_session_pool_tool_forwards_interceptor_headers():
     mock_session.call_tool.assert_awaited_once_with("act", {"x": 1}, meta={"headers": {"X-User-Id": "u-42"}})
 
 
+@pytest.mark.asyncio
+async def test_session_pool_interceptor_reads_request_scoped_secret():
+    """Interceptors can read request-scoped secrets from LangGraph context."""
+    from langchain_core.tools import StructuredTool
+    from pydantic import BaseModel, Field
+
+    from deerflow.mcp.tools import _make_session_pool_tool
+
+    class Args(BaseModel):
+        x: int = Field(..., description="x")
+
+    original_tool = StructuredTool(
+        name="srv_act",
+        description="test",
+        args_schema=Args,
+        coroutine=AsyncMock(),
+        response_format="content_and_artifact",
+    )
+
+    mock_session = AsyncMock()
+    mock_session.call_tool = AsyncMock(return_value=MagicMock(content=[], isError=False, structuredContent=None))
+    mock_cm = MagicMock()
+    mock_cm.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_cm.__aexit__ = AsyncMock(return_value=False)
+
+    async def secret_header_interceptor(request, handler):
+        from langgraph.config import get_config
+
+        secrets = (get_config().get("context") or {}).get("secrets") or {}
+        return await handler(request.override(headers={"Authorization": f"Bearer {secrets['MCP_AUTH_TOKEN']}"}))
+
+    with (
+        patch("langchain_mcp_adapters.sessions.create_session", return_value=mock_cm),
+        patch(
+            "langgraph.config.get_config",
+            return_value={"context": {"secrets": {"MCP_AUTH_TOKEN": "nested-secret"}}},
+        ),
+    ):
+        wrapped = _make_session_pool_tool(
+            original_tool,
+            "srv",
+            {"transport": "stdio", "command": "x", "args": []},
+            tool_interceptors=[secret_header_interceptor],
+        )
+        await wrapped.coroutine(runtime=None, x=1)
+
+    mock_session.call_tool.assert_awaited_once_with(
+        "act",
+        {"x": 1},
+        meta={"headers": {"Authorization": "Bearer nested-secret"}},
+    )
+
+
 @pytest.mark.asyncio
 async def test_session_pool_tool_no_headers_omits_meta():
     """When no interceptor sets headers, the pooled call must not pass a ``meta``
diff --git a/backend/tests/test_run_events_endpoint.py b/backend/tests/test_run_events_endpoint.py
index c4301280f..ee85e43a0 100644
--- a/backend/tests/test_run_events_endpoint.py
+++ b/backend/tests/test_run_events_endpoint.py
@@ -54,6 +54,52 @@ async def test_list_run_events_forwards_task_id_and_after_seq():
     assert calls["event_types"] == ["subagent.start", "subagent.step", "subagent.end"]
 
 
+@pytest.mark.anyio
+async def test_list_run_events_redacts_historical_run_start_metadata():
+    from app.gateway.routers.thread_runs import list_run_events
+
+    stored_row = {
+        "seq": 1,
+        "event_type": "run.start",
+        "metadata": {
+            "caller": "lead_agent",
+            "auth_token": "legacy-secret",
+            "token_usage": 7,
+        },
+    }
+
+    class FakeStore:
+        async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None):
+            return [stored_row]
+
+    class FakeState:
+        run_event_store = FakeStore()
+
+    class FakeApp:
+        state = FakeState()
+
+    class FakeRequest:
+        app = FakeApp()
+        _deerflow_test_bypass_auth = True
+
+    events = await list_run_events(
+        thread_id="legacy-thread",
+        run_id="legacy-run",
+        request=FakeRequest(),
+        event_types=None,
+        task_id=None,
+        limit=500,
+        after_seq=None,
+    )
+
+    assert events[0]["metadata"] == {
+        "caller": "lead_agent",
+        "token_usage": 7,
+    }
+    assert stored_row["metadata"]["auth_token"] == "legacy-secret"
+    assert events[0] is not stored_row
+
+
 @pytest.mark.anyio
 async def test_effective_memory_flows_from_injection_to_the_existing_debug_api():
     """The production run-events route is the field-level consumer for M1."""
diff --git a/backend/tests/test_run_metadata_secret_safety.py b/backend/tests/test_run_metadata_secret_safety.py
new file mode 100644
index 000000000..8d8f5e10d
--- /dev/null
+++ b/backend/tests/test_run_metadata_secret_safety.py
@@ -0,0 +1,144 @@
+import pytest
+
+from app.gateway.routers.thread_runs import _record_to_response
+from app.gateway.routers.threads import HistoryEntry, ThreadResponse, ThreadStateResponse
+from deerflow.runtime.runs.manager import RunRecord
+from deerflow.runtime.runs.schemas import DisconnectMode, RunStatus
+from deerflow.runtime.secret_context import (
+    LegacyRunMetadataSecretError,
+    redact_config_secrets,
+    redact_metadata_secrets,
+    validate_run_metadata_secrets,
+)
+
+
+@pytest.mark.parametrize("value", ["secret", "", None, {"nested": True}])
+def test_validate_run_metadata_rejects_auth_token_key_by_presence(value):
+    with pytest.raises(LegacyRunMetadataSecretError, match=r"config\.context\.secrets"):
+        validate_run_metadata_secrets({"auth_token": value, "token_usage": 7})
+
+
+@pytest.mark.parametrize(
+    "metadata",
+    [None, "not-a-mapping", {"token": "keep", "nested": {"auth_token": "keep"}}],
+)
+def test_validate_run_metadata_accepts_non_legacy_shapes(metadata):
+    validate_run_metadata_secrets(metadata)
+
+
+def test_redact_metadata_secrets_removes_exact_key_without_mutating_source():
+    source = {
+        "auth_token": "legacy-secret",
+        "token_usage": 7,
+        "nested": {"auth_token": "ordinary-nested-metadata"},
+    }
+
+    redacted = redact_metadata_secrets(source)
+
+    assert redacted == {
+        "token_usage": 7,
+        "nested": {"auth_token": "ordinary-nested-metadata"},
+    }
+    assert source["auth_token"] == "legacy-secret"
+    assert redacted is not source
+
+
+def test_redact_config_secrets_hides_legacy_config_metadata_without_mutating_source():
+    source = {
+        "metadata": {
+            "auth_token": "legacy-secret",
+            "token_usage": 7,
+            "nested": {"auth_token": "ordinary-nested-metadata"},
+        },
+        "context": {
+            "secrets": {"MCP_AUTH_TOKEN": "request-secret"},
+            "model_name": "default",
+        },
+    }
+
+    redacted = redact_config_secrets(source)
+
+    assert redacted == {
+        "metadata": {
+            "token_usage": 7,
+            "nested": {"auth_token": "ordinary-nested-metadata"},
+        },
+        "context": {"model_name": "default"},
+    }
+    assert source["metadata"]["auth_token"] == "legacy-secret"
+    assert source["context"]["secrets"] == {"MCP_AUTH_TOKEN": "request-secret"}
+    assert redacted is not source
+    assert redacted["metadata"] is not source["metadata"]
+    assert redacted["context"] is not source["context"]
+
+
+def test_run_response_hides_historical_auth_token_without_mutating_record():
+    legacy_metadata = {"auth_token": "legacy-secret", "token_usage": 7}
+    record = RunRecord(
+        run_id="legacy-run",
+        thread_id="legacy-thread",
+        assistant_id="lead_agent",
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.cancel,
+        metadata=legacy_metadata,
+    )
+
+    response = _record_to_response(record)
+
+    assert response.metadata == {"token_usage": 7}
+    assert record.metadata["auth_token"] == "legacy-secret"
+
+
+def test_run_response_hides_historical_config_metadata_without_mutating_record():
+    legacy_config = {
+        "metadata": {
+            "auth_token": "legacy-secret",
+            "token_usage": 7,
+            "nested": {"auth_token": "ordinary-nested-metadata"},
+        },
+        "context": {
+            "secrets": {"MCP_AUTH_TOKEN": "request-secret"},
+            "model_name": "default",
+        },
+    }
+    record = RunRecord(
+        run_id="legacy-config-run",
+        thread_id="legacy-thread",
+        assistant_id="lead_agent",
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.cancel,
+        metadata={"token_usage": 7},
+        kwargs={"input": {}, "config": legacy_config},
+    )
+
+    response = _record_to_response(record)
+
+    assert response.kwargs["config"] == {
+        "metadata": {
+            "token_usage": 7,
+            "nested": {"auth_token": "ordinary-nested-metadata"},
+        },
+        "context": {"model_name": "default"},
+    }
+    assert record.kwargs["config"]["metadata"]["auth_token"] == "legacy-secret"
+    assert record.kwargs["config"]["context"]["secrets"] == {"MCP_AUTH_TOKEN": "request-secret"}
+
+
+@pytest.mark.parametrize(
+    ("response_class", "required_fields"),
+    [
+        (ThreadResponse, {"thread_id": "legacy-thread"}),
+        (ThreadStateResponse, {}),
+        (HistoryEntry, {"checkpoint_id": "legacy-checkpoint"}),
+    ],
+)
+def test_thread_metadata_response_models_hide_historical_auth_token(
+    response_class,
+    required_fields,
+):
+    source = {"auth_token": "legacy-secret", "token_usage": 7}
+
+    response = response_class(**required_fields, metadata=source)
+
+    assert response.metadata == {"token_usage": 7}
+    assert source["auth_token"] == "legacy-secret"
diff --git a/backend/tests/test_skill_request_scoped_secrets.py b/backend/tests/test_skill_request_scoped_secrets.py
index d43492dcc..88894639d 100644
--- a/backend/tests/test_skill_request_scoped_secrets.py
+++ b/backend/tests/test_skill_request_scoped_secrets.py
@@ -1147,7 +1147,10 @@ class TestLeakSurfaces:
 
         config = {
             "context": {
-                "secrets": {"ERP_TOKEN": _SECRET},
+                "secrets": {
+                    "ERP_TOKEN": _SECRET,
+                    "nested": {"secondary": _SECRET},
+                },
                 "thread_id": "t",
                 "model_name": "m",
                 "__slash_skill_secret_source": {
@@ -1170,7 +1173,10 @@ class TestLeakSurfaces:
         assert "secrets" not in redacted["context"]
         assert SKILL_TOOL_POLICY_DECISION_CONTEXT_KEY not in redacted["context"]
         # Original is untouched (live config still has secrets).
-        assert config["context"]["secrets"] == {"ERP_TOKEN": _SECRET}
+        assert config["context"]["secrets"] == {
+            "ERP_TOKEN": _SECRET,
+            "nested": {"secondary": _SECRET},
+        }
 
     def test_redact_config_secrets_handles_none_and_no_context(self):
         from deerflow.runtime.secret_context import redact_config_secrets
diff --git a/docs/superpowers/plans/2026-07-24-issue-4416-run-metadata-secrets.md b/docs/superpowers/plans/2026-07-24-issue-4416-run-metadata-secrets.md
new file mode 100644
index 000000000..63b6edfc8
--- /dev/null
+++ b/docs/superpowers/plans/2026-07-24-issue-4416-run-metadata-secrets.md
@@ -0,0 +1,805 @@
+# Issue 4416 Run Metadata Secrets Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Reject the legacy top-level `metadata.auth_token` field before a run can persist anything, preserve request-scoped MCP credentials in `config.context.secrets`, and hide the legacy key from historical API responses.
+
+**Architecture:** `deerflow.runtime.secret_context` remains the single owner of secret-carrier and legacy-key policy. `app.gateway.services.start_run()` is the sole new-run admission boundary; API serializers call the same non-mutating metadata redactor only to hide historical records, without changing stores or callback implementations.
+
+**Tech Stack:** Python 3.12, FastAPI, Pydantic v2, LangGraph/LangChain runnable context, pytest/anyio, ruff.
+
+## Global Constraints
+
+- Reject only the exact top-level metadata key `auth_token`; do not add heuristic matching for names such as `token`, `api_key`, or nested arbitrary metadata.
+- Return HTTP 422 with migration guidance to `config.context.secrets`.
+- Validate before `RunManager.create_or_reject()`, thread upsert/status changes, `build_run_config()`, callback creation, `RunJournal` event emission, and background task creation.
+- Reuse `backend/packages/harness/deerflow/runtime/secret_context.py`; do not maintain duplicate secret-key lists in stores, callbacks, routers, or response models.
+- Historical API hiding must not mutate the stored run, thread, checkpoint, or event object.
+- Independent `POST /api/threads` and `PATCH /api/threads/{thread_id}` metadata contracts are outside #4416; they do not admit a run and must not silently adopt a second policy behavior.
+- Existing nested values in `config.context.secrets` must remain available to the live MCP interceptor while remaining absent from persisted run config.
+- Features and bug fixes ship with tests; run backend formatting and lint checks before completion.
+- Update `README.md`, `backend/docs/MCP_SERVER.md`, and `backend/AGENTS.md` in the same change set.
+
+---
+
+### Task 1: Central legacy metadata policy
+
+**Files:**
+- Create: `backend/tests/test_run_metadata_secret_safety.py`
+- Modify: `backend/packages/harness/deerflow/runtime/secret_context.py`
+
+**Interfaces:**
+- Consumes: existing `redact_config_secrets(config: Any) -> Any`.
+- Produces: `LEGACY_AUTH_TOKEN_METADATA_KEY: str`, `LegacyRunMetadataSecretError(ValueError)`, `validate_run_metadata_secrets(metadata: Any) -> None`, and `redact_metadata_secrets(metadata: Any) -> Any`.
+
+- [ ] **Step 1: Write failing unit tests for exact-key admission and non-mutating redaction**
+
+```python
+import pytest
+
+from deerflow.runtime.secret_context import (
+    LegacyRunMetadataSecretError,
+    redact_metadata_secrets,
+    validate_run_metadata_secrets,
+)
+
+
+@pytest.mark.parametrize("value", ["secret", "", None, {"nested": True}])
+def test_validate_run_metadata_rejects_auth_token_key_by_presence(value):
+    with pytest.raises(LegacyRunMetadataSecretError, match=r"config\.context\.secrets"):
+        validate_run_metadata_secrets({"auth_token": value, "token_usage": 7})
+
+
+@pytest.mark.parametrize(
+    "metadata",
+    [None, "not-a-mapping", {"token": "keep", "nested": {"auth_token": "keep"}}],
+)
+def test_validate_run_metadata_accepts_non_legacy_shapes(metadata):
+    validate_run_metadata_secrets(metadata)
+
+
+def test_redact_metadata_secrets_removes_exact_key_without_mutating_source():
+    source = {
+        "auth_token": "legacy-secret",
+        "token_usage": 7,
+        "nested": {"auth_token": "ordinary-nested-metadata"},
+    }
+
+    redacted = redact_metadata_secrets(source)
+
+    assert redacted == {
+        "token_usage": 7,
+        "nested": {"auth_token": "ordinary-nested-metadata"},
+    }
+    assert source["auth_token"] == "legacy-secret"
+    assert redacted is not source
+```
+
+- [ ] **Step 2: Run the new tests and observe RED**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest tests/test_run_metadata_secret_safety.py -q
+```
+
+Expected: collection fails because the four new symbols do not yet exist.
+
+- [ ] **Step 3: Add the minimal centralized policy**
+
+Add to `secret_context.py`:
+
+```python
+LEGACY_AUTH_TOKEN_METADATA_KEY = "auth_token"
+
+
+class LegacyRunMetadataSecretError(ValueError):
+    """Raised when a run puts a request credential in persisted metadata."""
+
+
+def validate_run_metadata_secrets(metadata: Any) -> None:
+    """Reject the legacy credential field at run admission."""
+    if isinstance(metadata, dict) and LEGACY_AUTH_TOKEN_METADATA_KEY in metadata:
+        raise LegacyRunMetadataSecretError(
+            "Run metadata key 'auth_token' is not allowed; "
+            "pass request-scoped credentials via config.context.secrets instead."
+        )
+
+
+def redact_metadata_secrets(metadata: Any) -> Any:
+    """Return API-safe metadata without mutating historical storage objects."""
+    if not isinstance(metadata, dict):
+        return metadata
+    return {
+        key: value
+        for key, value in metadata.items()
+        if key != LEGACY_AUTH_TOKEN_METADATA_KEY
+    }
+```
+
+- [ ] **Step 4: Run the policy tests and observe GREEN**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest tests/test_run_metadata_secret_safety.py -q
+```
+
+Expected: all Task 1 tests pass.
+
+- [ ] **Step 5: Commit the policy unit**
+
+```bash
+git add backend/packages/harness/deerflow/runtime/secret_context.py backend/tests/test_run_metadata_secret_safety.py
+git commit -m "fix(security): centralize legacy run metadata policy"
+```
+
+### Task 2: Unified run admission before persistence
+
+**Files:**
+- Modify: `backend/tests/test_gateway_services.py`
+- Modify: `backend/app/gateway/services.py`
+
+**Interfaces:**
+- Consumes: `validate_run_metadata_secrets(metadata: Any) -> None` and `LegacyRunMetadataSecretError`.
+- Produces: `start_run()` rejection with `HTTPException(status_code=422)` before any run/thread persistence; scheduled launches inherit the same boundary.
+
+- [ ] **Step 1: Add a failing real-store regression test**
+
+Add this reusable test setup beside `_capture_start_run_graph_input`:
+
+```python
+def _make_start_run_persistence_context():
+    from types import SimpleNamespace
+
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.store.memory import InMemoryStore
+
+    from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
+    from deerflow.runtime import RunManager
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    run_store = MemoryRunStore()
+    thread_store = MemoryThreadMetaStore(InMemoryStore())
+    state = SimpleNamespace(
+        stream_bridge=SimpleNamespace(),
+        run_manager=RunManager(store=run_store),
+        checkpointer=InMemorySaver(),
+        store=InMemoryStore(),
+        run_event_store=SimpleNamespace(),
+        run_events_config=None,
+        thread_store=thread_store,
+        checkpoint_channel_mode="full",
+        scheduled_task_service=None,
+    )
+    request = SimpleNamespace(
+        headers={},
+        state=SimpleNamespace(),
+        app=SimpleNamespace(state=state),
+    )
+    return request, run_store, thread_store
+```
+
+Use it in a synchronous pytest test with an `asyncio.run()` scenario, matching
+the existing `test_gateway_services.py` style:
+
+```python
+def test_start_run_rejects_legacy_auth_token_before_persistence():
+    async def _scenario():
+        request, run_store, thread_store = _make_start_run_persistence_context()
+        body = RunCreateRequest(
+            assistant_id="lead_agent",
+            input={"messages": [{"role": "user", "content": "hi"}]},
+            metadata={"auth_token": "legacy-secret", "token_usage": 7},
+        )
+
+        with patch(
+            "app.gateway.services.run_agent", new_callable=AsyncMock
+        ) as run_agent:
+            with pytest.raises(HTTPException) as exc_info:
+                await start_run(body, "thread-secret-admission", request)
+
+        assert exc_info.value.status_code == 422
+        assert "config.context.secrets" in str(exc_info.value.detail)
+        assert await run_store.list_by_thread("thread-secret-admission") == []
+        assert await thread_store.get("thread-secret-admission") is None
+        run_agent.assert_not_called()
+
+    asyncio.run(_scenario())
+```
+
+- [ ] **Step 2: Add a failing ordinary-metadata preservation test**
+
+Start a run with:
+
+```python
+metadata = {"token_usage": 7, "source": "regression"}
+```
+
+The test function must accept the existing `_stub_app_config` fixture because
+an admitted run calls `get_run_context()` and resolves the live `AppConfig`.
+Use `_make_start_run_persistence_context()`, patch `resolve_agent_factory`, and
+capture the live config in an async `fake_run_agent`:
+
+```python
+def test_start_run_preserves_ordinary_metadata(_stub_app_config):
+    async def _scenario():
+        thread_id = "thread-ordinary-metadata"
+        metadata = {"token_usage": 7, "source": "regression"}
+        request, _run_store, thread_store = (
+            _make_start_run_persistence_context()
+        )
+        captured: dict[str, Any] = {}
+
+        async def fake_run_agent(*args, **kwargs):
+            captured["config"] = kwargs["config"]
+
+        with (
+            patch(
+                "app.gateway.services.resolve_agent_factory",
+                return_value=object(),
+            ),
+            patch(
+                "app.gateway.services.run_agent",
+                side_effect=fake_run_agent,
+            ),
+        ):
+            record = await start_run(
+                RunCreateRequest(
+                    assistant_id="lead_agent",
+                    input={
+                        "messages": [
+                            {"role": "user", "content": "hi"}
+                        ]
+                    },
+                    metadata=metadata,
+                ),
+                thread_id,
+                request,
+            )
+            await record.task
+
+        assert record.metadata == metadata
+        assert (await thread_store.get(thread_id))["metadata"] == metadata
+        assert captured["config"]["metadata"] == metadata
+
+    asyncio.run(_scenario())
+```
+
+This prevents the exact-key rule from becoming broad metadata stripping.
+
+- [ ] **Step 3: Run both admission tests and observe RED**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest \
+  tests/test_gateway_services.py::test_start_run_rejects_legacy_auth_token_before_persistence \
+  tests/test_gateway_services.py::test_start_run_preserves_ordinary_metadata \
+  -q
+```
+
+Expected: the legacy request is currently accepted and persisted.
+
+- [ ] **Step 4: Install validation at the first line of `start_run()`**
+
+Import the shared symbols and add before `get_stream_bridge(request)`:
+
+```python
+try:
+    validate_run_metadata_secrets(getattr(body, "metadata", None))
+except LegacyRunMetadataSecretError as exc:
+    raise HTTPException(status_code=422, detail=str(exc)) from exc
+```
+
+Do not add checks to `RunStore`, `ThreadMetaStore`, `RunJournal`, `build_run_config`, or the router handlers.
+
+- [ ] **Step 5: Run both admission tests and observe GREEN**
+
+Run the exact Task 2 Step 3 command.
+
+Expected: both tests pass; the rejected path creates no row or task.
+
+- [ ] **Step 6: Add and run the scheduled-launch regression**
+
+Call the real scheduled launcher with a deliberately minimal app. Because validation must be the first operation in `start_run()`, the request must receive the policy error before the empty app state is accessed:
+
+```python
+def test_launch_scheduled_thread_run_rejects_legacy_auth_token():
+    async def _scenario():
+        with pytest.raises(HTTPException) as exc_info:
+            await launch_scheduled_thread_run(
+                thread_id="thread-scheduled",
+                assistant_id="lead_agent",
+                prompt="Run in background",
+                app=SimpleNamespace(state=SimpleNamespace()),
+                metadata={"auth_token": "legacy-secret"},
+            )
+
+        assert exc_info.value.status_code == 422
+        assert "config.context.secrets" in str(exc_info.value.detail)
+
+    asyncio.run(_scenario())
+```
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest \
+  tests/test_gateway_services.py::test_launch_scheduled_thread_run_rejects_legacy_auth_token \
+  -q
+```
+
+Expected: PASS, proving scheduled metadata reaches the same admission policy rather than bypassing it.
+
+- [ ] **Step 7: Commit the admission boundary**
+
+```bash
+git add backend/app/gateway/services.py backend/tests/test_gateway_services.py
+git commit -m "fix(security): reject secrets at run admission"
+```
+
+### Task 3: Non-mutating historical API hiding
+
+**Files:**
+- Modify: `backend/tests/test_run_metadata_secret_safety.py`
+- Modify: `backend/tests/test_run_events_endpoint.py`
+- Modify: `backend/app/gateway/routers/thread_runs.py`
+- Modify: `backend/app/gateway/routers/threads.py`
+
+**Interfaces:**
+- Consumes: `redact_metadata_secrets(metadata: Any) -> Any`.
+- Produces: redacted `RunResponse`, `ThreadResponse`, `ThreadStateResponse`, `HistoryEntry`, and run-event rows while leaving their input records unchanged.
+
+- [ ] **Step 1: Add failing response-model tests**
+
+Construct a historical `RunRecord` containing:
+
+```python
+legacy_metadata = {"auth_token": "legacy-secret", "token_usage": 7}
+record = RunRecord(
+    run_id="legacy-run",
+    thread_id="legacy-thread",
+    assistant_id="lead_agent",
+    status=RunStatus.success,
+    on_disconnect=DisconnectMode.cancel,
+    metadata=legacy_metadata,
+)
+
+response = _record_to_response(record)
+
+assert response.metadata == {"token_usage": 7}
+assert record.metadata["auth_token"] == "legacy-secret"
+```
+
+Import `RunRecord` from `deerflow.runtime.runs.manager` and `DisconnectMode` / `RunStatus` from `deerflow.runtime.runs.schemas`.
+
+For each thread response model:
+
+```python
+@pytest.mark.parametrize(
+    ("response_class", "required_fields"),
+    [
+        (ThreadResponse, {"thread_id": "legacy-thread"}),
+        (ThreadStateResponse, {}),
+        (HistoryEntry, {"checkpoint_id": "legacy-checkpoint"}),
+    ],
+)
+def test_thread_metadata_response_models_hide_historical_auth_token(
+    response_class, required_fields
+):
+    source = {"auth_token": "legacy-secret", "token_usage": 7}
+    response = response_class(**required_fields, metadata=source)
+
+    assert response.metadata == {"token_usage": 7}
+    assert source["auth_token"] == "legacy-secret"
+```
+
+Cover all three classes explicitly: `ThreadResponse`, `ThreadStateResponse`, and `HistoryEntry`.
+
+- [ ] **Step 2: Run the response tests and observe RED**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest tests/test_run_metadata_secret_safety.py -q
+```
+
+Expected: historical response objects still contain `auth_token`.
+
+- [ ] **Step 3: Apply the shared redactor to response construction**
+
+In `_record_to_response()` use:
+
+```python
+metadata=redact_metadata_secrets(record.metadata),
+```
+
+In `threads.py`, create one local response base class:
+
+```python
+class _MetadataRedactingResponse(BaseModel):
+    @field_validator("metadata", mode="before", check_fields=False)
+    @classmethod
+    def _redact_legacy_metadata_secret(cls, value: Any) -> Any:
+        return redact_metadata_secrets(value)
+```
+
+Make `ThreadResponse`, `ThreadStateResponse`, and `HistoryEntry` inherit from this base. Do not alter `ThreadCreateRequest`, `ThreadPatchRequest`, `_SERVER_RESERVED_METADATA_KEYS`, or stored rows.
+
+Keep `check_fields=False`: the base class intentionally declares a validator
+for a field that exists only on its subclasses. In the worktree's installed
+Pydantic 2.13.3, `field_validator` still accepts `check_fields`, and removing
+it raises `PydanticUserError` while defining the base class.
+
+Scope note: `record.kwargs["config"]["metadata"]` is the caller's raw
+LangChain `RunnableConfig.metadata`, not `RunCreateRequest.metadata`, and is
+outside the exact documented legacy request field addressed by #4416. Do not
+claim the admission validator cleans this separate mapping. Existing
+historical values there remain subject to the operator rotation and retained
+data cleanup guidance in Task 4; changing the raw RunnableConfig contract
+requires separate policy review.
+
+- [ ] **Step 4: Run the response tests and observe GREEN**
+
+Run the exact Task 3 Step 2 command.
+
+Expected: all helper and response tests pass.
+
+- [ ] **Step 5: Add a failing historical `run.start` event test**
+
+Extend `test_run_events_endpoint.py` with a fake store returning one row:
+
+```python
+stored_row = {
+    "seq": 1,
+    "event_type": "run.start",
+    "metadata": {
+        "caller": "lead_agent",
+        "auth_token": "legacy-secret",
+        "token_usage": 7,
+    },
+}
+```
+
+Call `list_run_events()` and assert:
+
+```python
+assert events[0]["metadata"] == {
+    "caller": "lead_agent",
+    "token_usage": 7,
+}
+assert stored_row["metadata"]["auth_token"] == "legacy-secret"
+assert events[0] is not stored_row
+```
+
+- [ ] **Step 6: Run the event test and observe RED**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest \
+  tests/test_run_events_endpoint.py::test_list_run_events_redacts_historical_run_start_metadata \
+  -q
+```
+
+Expected: returned event metadata still exposes `auth_token`.
+
+- [ ] **Step 7: Redact event rows without mutating the store result**
+
+Replace the direct return in `list_run_events()` with:
+
+```python
+events = await event_store.list_events(
+    thread_id,
+    run_id,
+    event_types=types,
+    task_id=task_id,
+    limit=limit,
+    after_seq=after_seq,
+)
+return [
+    {
+        **event,
+        "metadata": redact_metadata_secrets(event.get("metadata")),
+    }
+    if isinstance(event, dict) and "metadata" in event
+    else event
+    for event in events
+]
+```
+
+- [ ] **Step 8: Run historical output tests and observe GREEN**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest \
+  tests/test_run_metadata_secret_safety.py \
+  tests/test_run_events_endpoint.py \
+  -q
+```
+
+Expected: all tests pass, including existing event forwarding behavior.
+
+- [ ] **Step 9: Commit historical API hiding**
+
+```bash
+git add \
+  backend/app/gateway/routers/thread_runs.py \
+  backend/app/gateway/routers/threads.py \
+  backend/tests/test_run_metadata_secret_safety.py \
+  backend/tests/test_run_events_endpoint.py
+git commit -m "fix(security): hide legacy secrets from history APIs"
+```
+
+### Task 4: Supported MCP carrier and operator documentation
+
+**Files:**
+- Modify: `backend/tests/test_mcp_session_pool.py`
+- Modify: `backend/tests/test_skill_request_scoped_secrets.py`
+- Modify: `backend/docs/MCP_SERVER.md`
+- Modify: `README.md`
+- Modify: `backend/AGENTS.md`
+
+**Interfaces:**
+- Consumes: LangGraph `get_config()["context"]["secrets"]` and existing `redact_config_secrets`.
+- Produces: tested MCP header injection from the supported carrier and migration/rotation/cleanup guidance.
+
+- [ ] **Step 1: Add a failing MCP interceptor carrier regression**
+
+Extend the existing pooled-tool header test pattern with an interceptor that reads the live LangGraph config:
+
+```python
+async def secret_header_interceptor(request, handler):
+    from langgraph.config import get_config
+
+    secrets = (get_config().get("context") or {}).get("secrets") or {}
+    return await handler(
+        request.override(headers={"Authorization": f"Bearer {secrets['MCP_AUTH_TOKEN']}"})
+    )
+```
+
+Patch `langgraph.config.get_config` to return:
+
+```python
+{"context": {"secrets": {"MCP_AUTH_TOKEN": "nested-secret"}}}
+```
+
+Invoke the wrapped tool and assert:
+
+```python
+mock_session.call_tool.assert_awaited_once_with(
+    "act",
+    {"x": 1},
+    meta={"headers": {"Authorization": "Bearer nested-secret"}},
+)
+```
+
+- [ ] **Step 2: Run the MCP regression**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest \
+  tests/test_mcp_session_pool.py::test_session_pool_interceptor_reads_request_scoped_secret \
+  -q
+```
+
+Expected before any production change: PASS if the documented supported carrier is already wired correctly. A failure is a real compatibility gap and must be fixed in the narrowest runtime layer before proceeding.
+
+- [ ] **Step 3: Strengthen persisted-config coverage for nested values**
+
+Change the existing config-redaction fixture to:
+
+```python
+"secrets": {
+    "ERP_TOKEN": _SECRET,
+    "nested": {"secondary": _SECRET},
+},
+```
+
+Keep assertions that the live source remains intact and the redacted copy contains no secret values.
+
+- [ ] **Step 4: Run the supported-carrier regression group**
+
+Run:
+
+```bash
+cd backend
+.venv/bin/python -m pytest \
+  tests/test_mcp_session_pool.py::test_session_pool_interceptor_reads_request_scoped_secret \
+  tests/test_skill_request_scoped_secrets.py::TestLeakSurfaces::test_redact_config_secrets_strips_from_persisted_config \
+  -q
+```
+
+Expected: both tests pass.
+
+- [ ] **Step 5: Replace the unsafe MCP documentation example**
+
+In `backend/docs/MCP_SERVER.md`, replace `get_config()["metadata"]["auth_token"]` with:
+
+```python
+from langgraph.config import get_config
+
+
+def build_auth_interceptor():
+    async def interceptor(request, handler):
+        config = get_config()
+        secrets = (config.get("context") or {}).get("secrets") or {}
+        token = secrets.get("MCP_AUTH_TOKEN")
+        if token:
+            request = request.override(
+                headers={**(request.headers or {}), "Authorization": f"Bearer {token}"}
+            )
+        return await handler(request)
+
+    return interceptor
+```
+
+Document the request shape:
+
+```json
+{
+  "metadata": {"source": "my-client"},
+  "config": {
+    "context": {
+      "secrets": {"MCP_AUTH_TOKEN": "<request-scoped credential>"}
+    }
+  }
+}
+```
+
+State that `metadata.auth_token` is rejected with 422 and is never the supported interceptor path.
+
+- [ ] **Step 6: Add repository and operator guidance**
+
+Add a concise security note to `README.md` linking to `backend/docs/MCP_SERVER.md`.
+
+Update `backend/AGENTS.md` to state:
+
+- `start_run()` validates the exact legacy key before run/thread persistence;
+- `secret_context.py` owns admission and output-redaction policy;
+- historical API hiding does not delete old database/event/log/snapshot/backup material;
+- deployments that previously used `metadata.auth_token` must rotate the credential and clean all retained copies;
+- restarting or upgrading DeerFlow does not perform that cleanup.
+
+- [ ] **Step 7: Commit supported flow and docs**
+
+```bash
+git add \
+  backend/tests/test_mcp_session_pool.py \
+  backend/tests/test_skill_request_scoped_secrets.py \
+  backend/docs/MCP_SERVER.md \
+  README.md \
+  backend/AGENTS.md
+git commit -m "docs(security): migrate MCP credentials to secret context"
+```
+
+### Task 5: Full verification and handoff
+
+**Files:**
+- Verify all modified files.
+
+**Interfaces:**
+- Consumes: all preceding commits.
+- Produces: formatting-clean, lint-clean, regression-tested branch ready for review.
+
+- [ ] **Step 1: Run the focused security regression suite**
+
+```bash
+cd backend
+.venv/bin/python -m pytest \
+  tests/test_run_metadata_secret_safety.py \
+  tests/test_gateway_services.py \
+  tests/test_run_events_endpoint.py \
+  tests/test_mcp_session_pool.py \
+  tests/test_skill_request_scoped_secrets.py \
+  -q
+```
+
+Expected: all selected tests pass.
+
+- [ ] **Step 2: Format and verify formatting**
+
+```bash
+cd backend
+.venv/bin/ruff format \
+  packages/harness/deerflow/runtime/secret_context.py \
+  app/gateway/services.py \
+  app/gateway/routers/thread_runs.py \
+  app/gateway/routers/threads.py \
+  tests/test_run_metadata_secret_safety.py \
+  tests/test_gateway_services.py \
+  tests/test_run_events_endpoint.py \
+  tests/test_mcp_session_pool.py \
+  tests/test_skill_request_scoped_secrets.py
+.venv/bin/ruff format --check \
+  packages/harness/deerflow/runtime/secret_context.py \
+  app/gateway/services.py \
+  app/gateway/routers/thread_runs.py \
+  app/gateway/routers/threads.py \
+  tests/test_run_metadata_secret_safety.py \
+  tests/test_gateway_services.py \
+  tests/test_run_events_endpoint.py \
+  tests/test_mcp_session_pool.py \
+  tests/test_skill_request_scoped_secrets.py
+```
+
+Expected: no files require further formatting.
+
+- [ ] **Step 3: Run lint**
+
+```bash
+cd backend
+.venv/bin/ruff check \
+  packages/harness/deerflow/runtime/secret_context.py \
+  app/gateway/services.py \
+  app/gateway/routers/thread_runs.py \
+  app/gateway/routers/threads.py \
+  tests/test_run_metadata_secret_safety.py \
+  tests/test_gateway_services.py \
+  tests/test_run_events_endpoint.py \
+  tests/test_mcp_session_pool.py \
+  tests/test_skill_request_scoped_secrets.py
+```
+
+Expected: `All checks passed!`
+
+- [ ] **Step 4: Run the full backend suite**
+
+```bash
+cd backend
+.venv/bin/python -m pytest -q
+```
+
+Expected: the full backend suite passes. If an unrelated environment-dependent test fails, record the exact command, output, and why it is unrelated; do not claim a clean full suite.
+
+- [ ] **Step 5: Inspect repository hygiene**
+
+```bash
+git diff --check upstream/main...HEAD
+git status --short
+git log --oneline --decorate upstream/main..HEAD
+```
+
+Expected: no whitespace errors; only intentional files are changed; the branch contains the design, plan, implementation, tests, and documentation commits.
+
+- [ ] **Step 6: Commit any formatter-only changes**
+
+If Step 2 changed files:
+
+```bash
+git add \
+  backend/packages/harness/deerflow/runtime/secret_context.py \
+  backend/app/gateway/services.py \
+  backend/app/gateway/routers/thread_runs.py \
+  backend/app/gateway/routers/threads.py \
+  backend/tests/test_run_metadata_secret_safety.py \
+  backend/tests/test_gateway_services.py \
+  backend/tests/test_run_events_endpoint.py \
+  backend/tests/test_mcp_session_pool.py \
+  backend/tests/test_skill_request_scoped_secrets.py
+git commit -m "style: format issue 4416 security fix"
+```
+
+If Step 2 changed nothing, skip this commit.
+
+- [ ] **Step 7: Summarize evidence for review**
+
+Report:
+
+- the admission boundary and exact 422 migration message;
+- the five historical response surfaces;
+- proof that rejected requests create neither run nor thread records;
+- proof that nested `config.context.secrets` reaches an MCP interceptor but not persisted run config;
+- focused and full-suite test counts;
+- any remaining operational requirement to rotate and clean historical credentials.