feat(skills): bind request-scoped secrets for autonomously-invoked skills (A+) (#3938)

* feat(skills): bind request-scoped secrets for in-context (autonomously invoked) skills

Extends the #3861 binding point A (slash-activation only) to A+: the
injection set is recomputed on every model call from two unioned
sources — the run's most recent slash activation (persisted on the run
context so the tool loop keeps the binding) and skills the model
actually loaded in this thread (ThreadState.skill_context), re-validated
against the live registry each call.

Authorization stays three-gated regardless of activation style: skill
enabled by the operator, values supplied per-request by the caller in
context.secrets (never persisted server-side, never from the host env),
names declared in the skill's required-secrets frontmatter. Because the
set is replaced per call, eviction from skill_context or a caller that
stops supplying a value revokes injection on the next call.

New frontmatter field secrets-autonomous (default true) lets a skill
restrict binding to explicit slash activation; malformed values fail
closed to false. Binding changes are recorded as a
middleware:skill_secrets journal event carrying names only.

Design informed by a survey of peer systems (Claude Code, Codex CLI,
opencode, pi, deepagents, hermes-agent, QwenPaw) and specs
(agentskills.io, MCP 2025-11-25): the industry trust boundary is
enable-time consent plus caller-scoped credentials, not per-invocation
ceremony; no surveyed system scopes secrets to an activation turn.

Part of #3914

* refactor(skills): centralize secret context keys, document intentional per-call reload

Review follow-ups (no behavior change): move the two private binding keys
(__slash_skill_secret_source, __skill_secrets_binding_audit) into
secret_context.py and add them to REDACTED_CONTEXT_KEYS so the redaction
allowlist stays a complete guard even though both keys hold names only.
Document why _in_context_secret_sources reloads skills every call rather
than caching: load_skills re-reads enabled state so an operator disabling
a skill revokes its binding on the next model call — an mtime cache would
miss enable/disable toggles and keep injecting after a disable.

* fix(skills): match in-context secret bindings by path only, never by name

Review finding (confused deputy): _in_context_secret_sources fell back to
name matching when a skill_context path did not resolve. DeerFlow lets a
custom skill shadow a same-named public/legacy one (load_skills de-dupes
by name, custom wins), so a thread that read public/foo could bind the
custom foo's declared secrets although the custom skill was never loaded
in the thread. The recent user-isolation path changes make by-path misses
(and thus the dangerous fallback) more likely. Drop the by-name fallback:
match strictly by the exact container file path the model read; an
unresolved path simply does not bind (the safe direction). Regression
tests cover the shadowing case and a stale path.

Part of #3914

* fix(skills): resolve secret-binding sources via registry; strip caller __-keys

Security review (willem-bd, #3938):

1. Forged `__slash_skill_secret_source` bypassed the enabled/allowlist/
   secrets-autonomous gates. runtime.context is caller-mergeable, and the
   slash source was trusted as authoritative (its stored requirements were
   injected directly). Now the slash source records only the activated
   skill's canonical container path, and BOTH the slash and in-context
   sources resolve the live registry skill by normalized path each call
   (_resolve_registry_skill) — binding only that real, enabled, allowlisted
   skill's own declared secrets. A forged path resolves to nothing. As
   defense in depth, build_run_config strips caller-supplied __-prefixed
   context keys at the gateway boundary.
2. Malformed caller requirements crashed the run (unguarded tuple unpack /
   DoS). The middleware no longer unpacks caller-provided requirement data
   at all — declarations come from the registry — so a malformed source
   fails closed instead of raising.
3. Path-normalization asymmetry silently disabled in-context binding on a
   trailing-slash container_path config. Both the registry keys and the
   lookup path are now posixpath.normpath'd.

Regression tests: forged source rejected, forged-but-real path ignores
caller requirements + allowlist, malformed source fails closed, trailing-
slash config binds, gateway strips __-keys.

Part of #3914

* docs(skills): correct _SLASH_SECRET_SOURCE_KEY comment and note fail-closed trade-off

Post-review cleanup: the key now stores only the canonical container path
(the comment still described the pre-fix skill-name+requirements shape),
and document that a transient registry-load failure fails closed (drops
the binding for that call) rather than trusting stale data.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/AGENTS.md

@@ -418,12 +418,12 @@ Lets a caller pass per-request, short-lived end-user credentials (e.g. an ERP to
 
 - **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.
-- **Bind (point A)**: on slash-activation, `SkillActivationMiddleware._apply_skill_secrets` resolves the activated skill's declared secrets against `context.secrets` and writes the per-run injection set to `runtime.context[__active_skill_secrets]`. Slash activation reads the genuine user text via `get_original_user_content_text`; `InputSanitizationMiddleware` preserves it (`ORIGINAL_USER_CONTENT_KEY`) when it wraps input in BEGIN/END markers, so activation fires even after sanitization. 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, not injected.
+- **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*`/`*PASSWORD*`/`*CREDENTIAL*`/`*DSN*` + a connection-string denylist like `DATABASE_URL`/`REDIS_URL`/`GH_PAT`) 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**: only `/slash`-activated skills receive secrets (autonomously invoked enabled skills do not); no persistence/vaulting; the MCP per-user-credential gap (#3322) is a sibling, not covered here. Tests: `tests/test_skill_request_scoped_secrets.py`.
+- **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`.
 
 ### Model Factory (`packages/harness/deerflow/models/factory.py`)
 

backend/app/gateway/services.py

@@ -378,7 +378,15 @@ def build_run_config(
             if context_value is None:
                 context = {}
             elif isinstance(context_value, Mapping):
-                context = dict(context_value)
+                # Strip caller-supplied ``__``-prefixed keys: those are the
+                # harness's private run-context channels (skill secret-binding
+                # sources, the active-secret set, the run journal). A caller must
+                # not be able to seed them and forge internal state — e.g. a
+                # forged ``__slash_skill_secret_source`` would otherwise bypass the
+                # skill enabled/allowlist/declaration gates (#3938). Legitimate
+                # caller keys (``secrets``, ``user_id``, model overrides) never use
+                # the ``__`` prefix.
+                context = {key: value for key, value in context_value.items() if not (isinstance(key, str) and key.startswith("__"))}
             else:
                 raise ValueError("request config 'context' must be a mapping or null.")
             context["thread_id"] = thread_id

backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py

@@ -1,4 +1,4 @@
-"""Middleware for explicit slash skill activation."""
+"""Middleware for skill activation: explicit slash + in-context secret binding."""
 
 from __future__ import annotations
 
@@ -6,6 +6,7 @@ import asyncio
 import hashlib
 import html
 import logging
+import posixpath
 import uuid
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
@@ -16,11 +17,16 @@ from langchain.agents.middleware import AgentMiddleware
 from langchain.agents.middleware.types import ModelRequest, ModelResponse
 from langchain_core.messages import AIMessage, HumanMessage
 
-from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY, extract_request_secrets
+from deerflow.runtime.secret_context import (
+    _SECRETS_BINDING_AUDIT_KEY,
+    _SLASH_SECRET_SOURCE_KEY,
+    ACTIVE_SECRETS_CONTEXT_KEY,
+    extract_request_secrets,
+)
 from deerflow.skills.slash import parse_slash_skill_reference, resolve_slash_skill
 from deerflow.skills.storage import get_or_new_skill_storage, get_or_new_user_skill_storage
 from deerflow.skills.storage.skill_storage import SkillStorage
-from deerflow.skills.types import SKILL_MD_FILE, SecretRequirement, SkillCategory
+from deerflow.skills.types import SKILL_MD_FILE, SecretRequirement, Skill, SkillCategory
 from deerflow.utils.messages import get_original_user_content_text
 
 if TYPE_CHECKING:
@@ -32,6 +38,16 @@ _SLASH_SKILL_ACTIVATION_KEY = "slash_skill_activation"
 _SLASH_SKILL_ACTIVATION_TARGET_ID_KEY = "slash_skill_activation_target_id"
 _SUMMARY_MESSAGE_NAME = "summary"
 
+# _SECRETS_BINDING_AUDIT_KEY: last audited binding (skill and secret names only,
+# never values) so unchanged bindings are not re-recorded each call.
+# _SLASH_SECRET_SOURCE_KEY: latest slash activation as a secret source, holding
+# ONLY the activated skill's canonical container path (never its declared
+# secrets — those are read from the live registry on each call, #3938). The
+# injection set is recomputed every model call, but a slash-activated skill must
+# stay bound for the rest of the run — the model's tool loop issues many model
+# calls after the single activation call (#3861 semantics). Both live in
+# secret_context so they are covered by REDACTED_CONTEXT_KEYS in one place.
+
 
 @dataclass(frozen=True, slots=True)
 class _Activation:
@@ -241,18 +257,18 @@ Follow this skill before choosing a general workflow. Load supporting resources
         except Exception:
             logger.debug("Failed to record slash skill activation audit event", exc_info=True)
 
-    def _prepare_model_request(self, request: ModelRequest, *, hook: str) -> ModelRequest | AIMessage | None:
+    def _prepare_model_request(self, request: ModelRequest, *, hook: str) -> tuple[ModelRequest | AIMessage | None, _Activation | None]:
         target_and_resolution = self._find_activation_target(list(request.messages))
         if target_and_resolution is None:
-            return None
+            return None, None
 
         target_index, target, resolution = target_and_resolution
         if resolution.failure_message:
-            return AIMessage(content=resolution.failure_message)
+            return AIMessage(content=resolution.failure_message), None
 
         activation = resolution.activation
         if activation is None:
-            return None
+            return None, None
 
         logger.info(
             "SkillActivationMiddleware: activating slash skill %s category=%s path=%s hash=%s",
@@ -262,59 +278,204 @@ Follow this skill before choosing a general workflow. Load supporting resources
             activation.content_hash,
         )
         self._record_activation(request, activation, hook=hook)
-        self._apply_skill_secrets(request, activation)
         activation_msg = self._make_activation_message(target, self._build_activation_reminder(activation))
         messages = list(request.messages)
         messages.insert(target_index, activation_msg)
-        return request.override(messages=messages)
+        return request.override(messages=messages), activation
 
-    @staticmethod
-    def _apply_skill_secrets(request: ModelRequest, activation: _Activation) -> None:
-        """Resolve the activated skill's declared secrets into the per-run injection
-        set (binding point A, issue #3861).
+    def _handle_model_request(self, request: ModelRequest, *, hook: str) -> ModelRequest | AIMessage:
+        prepared, activation = self._prepare_model_request(request, hook=hook)
+        if isinstance(prepared, AIMessage):
+            return prepared
+        effective = prepared if prepared is not None else request
+        self._resolve_secret_bindings(effective, activation, hook=hook)
+        return effective
 
-        For each declared secret present in the request's ``context.secrets``,
-        record its value in the injection set stored under
-        ``ACTIVE_SECRETS_CONTEXT_KEY`` on the shared run context, so the bash tool
-        can build the subprocess env for this turn. The injected value always comes
-        from the caller's request — never from the host environment, which is
-        scrubbed of secret-looking names by ``env_policy.build_sandbox_env`` before
-        injection. A skill can therefore never harvest a host platform credential
-        (it only ever receives what the caller explicitly supplied), so a declared
-        name that also exists in the host env is fine: the caller's value wins and
-        the host value is dropped. Secret *values* are never logged.
+    def _resolve_secret_bindings(self, request: ModelRequest, activation: _Activation | None, *, hook: str) -> None:
+        """Recompute the per-run secret injection set (binding point A+, #3861/#3914).
+
+        Sources, unioned on every model call:
+
+        - the most recent slash activation of this run (persisted as a source on
+          the run context so the whole tool loop after the activation call keeps
+          the binding — a new slash activation replaces it). The slash source is
+          validated once, at activation (enabled + allowlist checks in
+          ``_resolve_activation``), and deliberately NOT re-validated per call:
+          slash is a run-scoped commitment made by the user, and it dies with
+          the run anyway;
+        - skills the model loaded earlier in the thread (``ThreadState.skill_context``),
+          re-validated against the live registry on each call: enabled,
+          runtime-allowed for this agent, and not opted out via
+          ``secrets-autonomous: false``. Slash activation is exempt from the
+          opt-out — it is the explicit-ceremony path.
+
+        The set is recomputed and REPLACED each call, so a skill evicted from
+        skill_context, or a caller that stops supplying a value, loses its
+        injection on the next call automatically. Injected values always come
+        from the caller's request (``context.secrets``) — never the host
+        environment, which ``env_policy.build_sandbox_env`` scrubs before
+        injection — so a skill can never harvest a host platform credential.
+        Secret *values* are never logged; the audit journal records names only.
         """
         runtime = getattr(request, "runtime", None)
         context = getattr(runtime, "context", None)
         if not isinstance(context, dict):
             return
-        # Unconditionally clear any active-secret set a previous activation in
-        # the same run may have written, before this turn's resolution decides
-        # what (if anything) to install. Otherwise a later skill that declares
-        # no secrets, or whose required secrets the caller did not supply, would
-        # inherit the previous skill's injection set and the bash tool would
-        # inject those values into a subprocess that never declared them (#3861).
-        context.pop(ACTIVE_SECRETS_CONTEXT_KEY, None)
-        if not activation.required_secrets:
-            return
+
+        # The slash source records only the canonical container path of the
+        # activated skill — never its declared secrets. Both sources resolve the
+        # live registry skill by path on read, so a caller-forged source (the
+        # context is caller-mergeable) can never inject secrets a real, enabled,
+        # allowlisted skill did not declare (#3938).
+        if activation is not None:
+            context[_SLASH_SECRET_SOURCE_KEY] = {"path": activation.container_file_path}
 
         request_secrets = extract_request_secrets(context)
+        sources: list[tuple[str, tuple[SecretRequirement, ...]]] = []
+        if request_secrets:
+            registry = self._load_skill_registry_by_path()
+            if registry is not None:
+                # Slash source: exempt from the ``secrets-autonomous`` opt-out
+                # (explicit ceremony), but still enabled + allowlist checked.
+                slash_source = context.get(_SLASH_SECRET_SOURCE_KEY)
+                slash_path = slash_source.get("path") if isinstance(slash_source, dict) else None
+                slash_skill = self._resolve_registry_skill(registry, slash_path, require_autonomous=False)
+                if slash_skill is not None:
+                    sources.append((slash_skill.name, tuple(slash_skill.required_secrets)))
+                sources.extend(self._in_context_secret_sources(request, registry))
+
         injected: dict[str, str] = {}
-        missing: list[str] = []
-        for req in activation.required_secrets:
-            if req.name in request_secrets:
-                injected[req.name] = request_secrets[req.name]
-            elif not req.optional:
-                missing.append(req.name)
+        bound_skills: set[str] = set()
+        missing: dict[str, list[str]] = {}
+        for skill_name, requirements in sources:
+            for req in requirements:
+                if req.name in request_secrets:
+                    injected[req.name] = request_secrets[req.name]
+                    bound_skills.add(skill_name)
+                elif not req.optional:
+                    missing.setdefault(skill_name, []).append(req.name)
 
         if injected:
             context[ACTIVE_SECRETS_CONTEXT_KEY] = injected
-        if missing:
+        else:
+            context.pop(ACTIVE_SECRETS_CONTEXT_KEY, None)
+
+        audit_state = {
+            "skills": sorted(bound_skills),
+            "secrets": sorted(injected),
+            "missing": {name: sorted(values) for name, values in sorted(missing.items())},
+        }
+        previous = context.get(_SECRETS_BINDING_AUDIT_KEY)
+        if previous == audit_state:
+            return
+        if previous is None and not injected and not missing:
+            return
+        context[_SECRETS_BINDING_AUDIT_KEY] = audit_state
+        for skill_name, names in sorted(missing.items()):
             logger.warning(
-                "Skill %s activated but required secrets are missing from the request context: %s",
-                activation.skill_name,
-                ", ".join(sorted(missing)),
+                "Skill %s is active but required secrets are missing from the request context: %s",
+                skill_name,
+                ", ".join(names),
             )
+        self._record_secret_binding(context, audit_state, hook=hook)
+
+    def _load_skill_registry_by_path(self) -> dict[str, Skill] | None:
+        """Load the live skill registry keyed by normalized container file path.
+
+        Reloaded every call on purpose (not cached): load_skills re-reads the
+        enabled state from extensions_config so an operator disabling a skill
+        revokes its secret binding on the very next model call. A cache keyed on
+        file mtimes would miss enable/disable toggles (which do not touch
+        SKILL.md) and keep injecting after a disable — trading the
+        immediate-revocation security property for speed. The cost is gated: the
+        only caller runs this only when the caller supplied secrets.
+
+        Paths are normalized so a non-canonical ``container_path`` config (e.g. a
+        trailing slash) still matches the canonical path captured in
+        ``skill_context`` (#3938). Returns ``None`` if the registry can't load —
+        both the slash and in-context sources then bind nothing for that call
+        (fail closed). This is a deliberate availability-for-security trade-off:
+        a transient registry read failure mid-run drops the injection for that
+        call rather than trusting stale caller-supplied data.
+        """
+        try:
+            storage = self._storage()
+            skills = storage.load_skills(enabled_only=False)
+            container_root = storage.get_container_root()
+        except Exception:
+            logger.exception("Failed to load skills while resolving secret bindings")
+            return None
+        return {posixpath.normpath(skill.get_container_file_path(container_root)): skill for skill in skills}
+
+    def _resolve_registry_skill(self, registry: dict[str, Skill], path: object, *, require_autonomous: bool) -> Skill | None:
+        """Resolve a container path to a live registry skill eligible for secret
+        binding, or ``None``.
+
+        Match strictly by normalized container file path — never by name. A
+        by-name fallback would be a confused deputy: DeerFlow lets a custom skill
+        shadow a same-named public/legacy one (load_skills de-dupes by name,
+        custom wins), so a reference to public/foo could bind the custom foo's
+        secrets. A path that does not resolve simply binds nothing (the safe
+        direction), which also fails closed on a caller-forged path (#3938).
+
+        Gates: the skill must be enabled, declare secrets, and be allowlisted for
+        this agent. ``require_autonomous`` additionally enforces the
+        ``secrets-autonomous`` opt-out for the in-context path; the slash path
+        passes ``False`` because explicit activation is the ceremony that opt-out
+        is meant to preserve.
+        """
+        if not isinstance(path, str) or not path:
+            return None
+        skill = registry.get(posixpath.normpath(path))
+        if skill is None or not skill.enabled or not skill.required_secrets:
+            return None
+        if require_autonomous and not skill.secrets_autonomous:
+            return None
+        if self._available_skills is not None and skill.name not in self._available_skills:
+            return None
+        return skill
+
+    def _in_context_secret_sources(self, request: ModelRequest, registry: dict[str, Skill]) -> list[tuple[str, tuple[SecretRequirement, ...]]]:
+        """Map ``ThreadState.skill_context`` entries to declared-secret sources.
+
+        Entries are references to skills the model actually loaded in this
+        thread. Each is re-validated against the live registry so a skill that
+        was disabled, uninstalled, opted out, or removed from the agent's
+        allowlist after being read stops binding immediately.
+        """
+        state = getattr(request, "state", None) or {}
+        try:
+            entries = state.get("skill_context") or []
+        except AttributeError:
+            return []
+
+        sources: list[tuple[str, tuple[SecretRequirement, ...]]] = []
+        seen: set[str] = set()
+        for entry in entries:
+            if not isinstance(entry, dict):
+                continue
+            skill = self._resolve_registry_skill(registry, entry.get("path"), require_autonomous=True)
+            if skill is None or skill.name in seen:
+                continue
+            seen.add(skill.name)
+            sources.append((skill.name, tuple(skill.required_secrets)))
+        return sources
+
+    @staticmethod
+    def _record_secret_binding(context: dict, audit_state: dict, *, hook: str) -> None:
+        journal = context.get("__run_journal")
+        if journal is None:
+            return
+        try:
+            journal.record_middleware(
+                "skill_secrets",
+                name="SkillActivationMiddleware",
+                hook=hook,
+                action="bind_secrets",
+                changes=audit_state,
+            )
+        except Exception:
+            logger.debug("Failed to record skill secret binding audit event", exc_info=True)
 
     @staticmethod
     def _make_activation_message(target: HumanMessage, activation_content: str) -> HumanMessage:
@@ -337,9 +498,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
         request: ModelRequest,
         handler: Callable[[ModelRequest], ModelResponse],
     ) -> ModelResponse | AIMessage:
-        prepared = self._prepare_model_request(request, hook="wrap_model_call")
-        if prepared is None:
-            return handler(request)
+        prepared = self._handle_model_request(request, hook="wrap_model_call")
         if isinstance(prepared, AIMessage):
             return prepared
         return handler(prepared)
@@ -350,9 +509,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
         request: ModelRequest,
         handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
     ) -> ModelResponse | AIMessage:
-        prepared = await asyncio.to_thread(self._prepare_model_request, request, hook="awrap_model_call")
-        if prepared is None:
-            return await handler(request)
+        prepared = await asyncio.to_thread(self._handle_model_request, request, hook="awrap_model_call")
         if isinstance(prepared, AIMessage):
             return prepared
         return await handler(prepared)

backend/packages/harness/deerflow/runtime/secret_context.py

@@ -51,9 +51,24 @@ def read_active_secrets(context: Any) -> dict[str, str]:
     return _string_pairs(context.get(ACTIVE_SECRETS_CONTEXT_KEY))
 
 
+# Private run-context keys the skill-activation middleware uses to carry secret
+# bindings across a run. Only ``secrets`` / ``__active_skill_secrets`` hold
+# values; the binding-source and audit keys hold names only. All are listed so
+# the redaction allowlist stays a complete guard even if a future edit starts
+# storing a value under one of the name-only keys.
+_SLASH_SECRET_SOURCE_KEY = "__slash_skill_secret_source"
+_SECRETS_BINDING_AUDIT_KEY = "__skill_secrets_binding_audit"
+
 # Run-context keys whose values are request-scoped secrets and must be stripped
 # before a context mapping is serialized anywhere observable (traces, logs).
-REDACTED_CONTEXT_KEYS = frozenset({SECRETS_CONTEXT_KEY, ACTIVE_SECRETS_CONTEXT_KEY})
+REDACTED_CONTEXT_KEYS = frozenset(
+    {
+        SECRETS_CONTEXT_KEY,
+        ACTIVE_SECRETS_CONTEXT_KEY,
+        _SLASH_SECRET_SOURCE_KEY,
+        _SECRETS_BINDING_AUDIT_KEY,
+    }
+)
 
 
 def redact_secret_context_keys(context: Any) -> Any:

backend/packages/harness/deerflow/skills/parser.py

@@ -103,6 +103,22 @@ def parse_required_secrets(raw: object, skill_file: Path) -> tuple[SecretRequire
     return tuple(secrets)
 
 
+def parse_secrets_autonomous(raw: object, skill_file: Path) -> bool:
+    """Parse the optional ``secrets-autonomous`` frontmatter field (issue #3914).
+
+    ``True`` (the default) lets declared secrets bind while the skill is
+    in-context via an autonomous model load; ``False`` restricts binding to
+    explicit ``/slash`` activation. A malformed (non-boolean) value fails
+    closed to ``False`` — the safer, less-injection direction.
+    """
+    if raw is None:
+        return True
+    if isinstance(raw, bool):
+        return raw
+    logger.warning("Ignoring malformed secrets-autonomous value in %s: %r (autonomous binding disabled)", skill_file, raw)
+    return False
+
+
 def parse_skill_file(skill_file: Path, category: SkillCategory, relative_path: Path | None = None) -> Skill | None:
     """Parse a SKILL.md file and extract metadata.
 
@@ -170,6 +186,8 @@ def parse_skill_file(skill_file: Path, category: SkillCategory, relative_path: P
             logger.error("Invalid required-secrets in %s: %s", skill_file, exc)
             return None
 
+        secrets_autonomous = parse_secrets_autonomous(metadata.get("secrets-autonomous"), skill_file)
+
         return Skill(
             name=name,
             description=description,
@@ -181,6 +199,7 @@ def parse_skill_file(skill_file: Path, category: SkillCategory, relative_path: P
             allowed_tools=allowed_tools,
             enabled=True,  # Actual state comes from the extensions config file.
             required_secrets=required_secrets,
+            secrets_autonomous=secrets_autonomous,
         )
 
     except Exception:

backend/packages/harness/deerflow/skills/types.py

@@ -49,6 +49,10 @@ class Skill:
     allowed_tools: tuple[str, ...] | None = None
     enabled: bool = False  # Whether this skill is enabled
     required_secrets: tuple[SecretRequirement, ...] = field(default_factory=tuple)
+    # Whether declared secrets may bind when the skill is in-context via an
+    # autonomous model load (skill_context), or only on explicit /slash
+    # activation. Frontmatter: ``secrets-autonomous`` (default true).
+    secrets_autonomous: bool = True
 
     @property
     def skill_path(self) -> str:

backend/tests/test_skill_request_scoped_secrets.py

@@ -287,6 +287,22 @@ class TestSecretCarrier:
         ctx = _build_runtime_context("t", "r", {"secrets": {"ERP_TOKEN": "v"}})
         assert ctx["secrets"] == {"ERP_TOKEN": "v"}
 
+    def test_build_run_config_strips_caller_dunder_context_keys(self):
+        """Security (#3938): the harness writes private ``__``-prefixed keys into
+        ``runtime.context`` (binding sources, active-secret set, run journal). A
+        caller must not be able to seed them via ``config.context`` and forge
+        internal state — they are stripped at the gateway boundary."""
+        from app.gateway.services import build_run_config
+
+        config = build_run_config(
+            "thread-1",
+            {"context": {"secrets": {"ERP_TOKEN": "v"}, "__slash_skill_secret_source": {"path": "x"}, "__active_skill_secrets": {"ADMIN": "stolen"}}},
+            None,
+        )
+        assert config["context"]["secrets"] == {"ERP_TOKEN": "v"}
+        assert "__slash_skill_secret_source" not in config["context"]
+        assert "__active_skill_secrets" not in config["context"]
+
     def test_extract_request_secrets_filters_non_string_pairs(self):
         from deerflow.runtime.secret_context import extract_request_secrets
 
@@ -300,7 +316,7 @@ class TestSecretCarrier:
         assert extract_request_secrets(None) == {}
 
 
-def _make_secret_skill(tmp_path: Path, name: str, required_secrets):
+def _make_secret_skill(tmp_path: Path, name: str, required_secrets, *, enabled: bool = True, secrets_autonomous: bool = True):
     skill_dir = tmp_path / name
     skill_dir.mkdir()
     skill_file = skill_dir / "SKILL.md"
@@ -313,8 +329,9 @@ def _make_secret_skill(tmp_path: Path, name: str, required_secrets):
         skill_file=skill_file,
         relative_path=Path(name),
         category=SkillCategory.CUSTOM,
-        enabled=True,
-        required_secrets=required_secrets,
+        enabled=enabled,
+        required_secrets=tuple(required_secrets),
+        secrets_autonomous=secrets_autonomous,
     )
 
 
@@ -529,6 +546,308 @@ class TestActivationBindsSecrets:
         assert read_active_secrets(context2) == {}
 
 
+def _skill_context_entry(skill) -> dict:
+    return {
+        "name": skill.name,
+        "path": f"/mnt/skills/{skill.category}/{skill.name}/SKILL.md",
+        "description": skill.description,
+        "loaded_at": 0,
+    }
+
+
+class TestInContextBindsSecrets:
+    """Binding point A+ (issue #3914 gap 1): a skill the model loaded earlier in
+    the thread (tracked by ``ThreadState.skill_context``) keeps receiving its
+    declared secrets on later turns — without a fresh ``/slash`` — as long as
+    the caller supplies the values on the current request. Authorization stays
+    three-gated regardless of activation style: skill enabled by the operator,
+    values supplied per-request by the caller, names declared in frontmatter.
+    """
+
+    def _run_call(self, tmp_path, monkeypatch, skills, *, context, skill_context=None, message="continue the report", available_skills=None, middleware=None, container_root="/mnt/skills"):
+        from deerflow.agents.middlewares import skill_activation_middleware as mw
+        from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
+
+        storage = SimpleNamespace(
+            load_skills=lambda *, enabled_only: list(skills),
+            get_container_root=lambda: container_root,
+            get_skills_root_path=lambda: tmp_path,
+        )
+        monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: storage)
+        mw_inst = middleware or SkillActivationMiddleware(available_skills=available_skills)
+        mw_inst.wrap_model_call(
+            ModelRequest(
+                model=object(),
+                messages=[HumanMessage(content=message, id="m1")],
+                state={"messages": [], "skill_context": skill_context or []},
+                runtime=SimpleNamespace(context=context),
+            ),
+            lambda r: AIMessage(content="ok"),
+        )
+        return mw_inst
+
+    def test_in_context_skill_binds_secrets_without_slash(self, tmp_path, monkeypatch):
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-123", "UNRELATED": "x"}}
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
+
+        assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
+
+    def test_binding_clears_when_skill_evicted_from_context(self, tmp_path, monkeypatch):
+        """Long-lived binding follows skill_context membership exactly: once the
+        entry is evicted (capacity) the injection disappears on the next call."""
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        mw_inst = self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
+        assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
+
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[], middleware=mw_inst)
+        assert read_active_secrets(context) == {}
+
+    def test_disabled_skill_in_context_not_bound(self, tmp_path, monkeypatch):
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")], enabled=False)
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
+
+        assert read_active_secrets(context) == {}
+
+    def test_skill_outside_agent_allowlist_not_bound(self, tmp_path, monkeypatch):
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        self._run_call(
+            tmp_path,
+            monkeypatch,
+            [skill],
+            context=context,
+            skill_context=[_skill_context_entry(skill)],
+            available_skills={"some-other-skill"},
+        )
+
+        assert read_active_secrets(context) == {}
+
+    def test_secrets_autonomous_false_blocks_in_context_but_not_slash(self, tmp_path, monkeypatch):
+        """The per-skill opt-out keeps explicit-activation ceremony available for
+        high-sensitivity skills: in-context binding is refused, slash still works."""
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")], secrets_autonomous=False)
+
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
+        assert read_active_secrets(context) == {}
+
+        slash_context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        self._run_call(tmp_path, monkeypatch, [skill], context=slash_context, message="/erp-report go")
+        assert read_active_secrets(slash_context) == {"ERP_TOKEN": "tok-123"}
+
+    def test_slash_and_in_context_sources_merge(self, tmp_path, monkeypatch):
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        loaded = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        slashed = _make_secret_skill(tmp_path, "crm-sync", [SecretRequirement("CRM_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-erp", "CRM_TOKEN": "tok-crm"}}
+        self._run_call(
+            tmp_path,
+            monkeypatch,
+            [loaded, slashed],
+            context=context,
+            skill_context=[_skill_context_entry(loaded)],
+            message="/crm-sync push the numbers",
+        )
+
+        assert read_active_secrets(context) == {"ERP_TOKEN": "tok-erp", "CRM_TOKEN": "tok-crm"}
+
+    def test_forged_slash_source_cannot_bypass_gates(self, tmp_path, monkeypatch):
+        """Security (#3938): `runtime.context` is caller-mergeable, so a client can
+        forge `__slash_skill_secret_source`. The slash source is re-validated
+        against the live registry (enabled + allowlist), so a forged source naming
+        a non-existent skill binds nothing — no gate bypass."""
+        from deerflow.runtime.secret_context import _SLASH_SECRET_SOURCE_KEY, read_active_secrets
+
+        context = {
+            "secrets": {"ADMIN_TOKEN": "stolen"},
+            _SLASH_SECRET_SOURCE_KEY: {"path": "/mnt/skills/custom/attacker/SKILL.md", "skill_name": "attacker", "requirements": [["ADMIN_TOKEN", False]]},
+        }
+        self._run_call(tmp_path, monkeypatch, [], context=context, message="no slash here")
+        assert read_active_secrets(context) == {}
+
+    def test_forged_slash_source_ignores_caller_requirements_and_allowlist(self, tmp_path, monkeypatch):
+        """Even if a forged path resolves to a real skill, the caller's forged
+        requirements are ignored (only the registry skill's own declared secrets
+        bind) and the allowlist still applies."""
+        from deerflow.runtime.secret_context import _SLASH_SECRET_SOURCE_KEY, read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {
+            "secrets": {"ADMIN_TOKEN": "stolen", "ERP_TOKEN": "ok"},
+            _SLASH_SECRET_SOURCE_KEY: {"path": "/mnt/skills/custom/erp-report/SKILL.md", "requirements": [["ADMIN_TOKEN", False]]},
+        }
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, available_skills={"other"})
+        assert read_active_secrets(context) == {}
+
+    def test_malformed_slash_source_does_not_crash(self, tmp_path, monkeypatch):
+        """Robustness (#3938): a forged malformed slash source must fail closed
+        (bind nothing), never raise and 500 the run."""
+        from deerflow.runtime.secret_context import _SLASH_SECRET_SOURCE_KEY, read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        for bad in ({"requirements": [["X"]]}, {"requirements": "abc"}, {"path": 123}, "not-a-dict", {"path": ["a"]}, {}):
+            context = {"secrets": {"ERP_TOKEN": "v"}, _SLASH_SECRET_SOURCE_KEY: bad}
+            self._run_call(tmp_path, monkeypatch, [skill], context=context, message="x")
+            assert read_active_secrets(context) == {}, f"bad={bad!r}"
+
+    def test_trailing_slash_container_root_still_binds(self, tmp_path, monkeypatch):
+        """Latent bug (#3938): a non-canonical container_path (trailing slash) must
+        not silently disable in-context binding — paths are normalized both sides."""
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        entry = {"name": "erp-report", "path": "/mnt/skills/custom/erp-report/SKILL.md", "description": "d", "loaded_at": 0}
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[entry], container_root="/mnt/skills/")
+        assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
+
+    def test_shadowing_name_does_not_bind_unread_skill(self, tmp_path, monkeypatch):
+        """Confused-deputy guard: a custom skill may shadow a same-named public
+        one (load_skills de-dupes by name, custom wins). A thread that read the
+        PUBLIC foo (no declared secrets) must NOT bind the CUSTOM foo's declared
+        secret — matching is by exact container path, never by name."""
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        # Registry exposes only the custom foo (name de-dup, custom wins); the
+        # model read the public foo, whose path differs.
+        custom_foo = _make_secret_skill(tmp_path, "foo", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        entry = {
+            "name": "foo",
+            "path": "/mnt/skills/public/foo/SKILL.md",  # the PUBLIC one the model actually read
+            "description": "d",
+            "loaded_at": 0,
+        }
+        self._run_call(tmp_path, monkeypatch, [custom_foo], context=context, skill_context=[entry])
+
+        assert read_active_secrets(context) == {}
+
+    def test_stale_path_does_not_fall_back_to_name(self, tmp_path, monkeypatch):
+        """A skill_context path that no longer resolves must not degrade to a
+        name match — it simply does not bind."""
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        entry = {
+            "name": "erp-report",
+            "path": "/mnt/skills/custom/erp-report-OLD-PATH/SKILL.md",
+            "description": "d",
+            "loaded_at": 0,
+        }
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[entry])
+
+        assert read_active_secrets(context) == {}
+
+    def test_no_caller_secrets_means_no_binding(self, tmp_path, monkeypatch):
+        """The supply gate: without caller-provided values on THIS request there
+        is nothing to inject, no matter what is in skill_context."""
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {}}
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
+
+        assert read_active_secrets(context) == {}
+
+    def test_binding_change_recorded_in_audit_journal_names_only(self, tmp_path, monkeypatch):
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        journal = MagicMock()
+        context = {"secrets": {"ERP_TOKEN": "tok-secret-value"}, "__run_journal": journal}
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
+
+        bind_calls = [call for call in journal.record_middleware.call_args_list if call.kwargs.get("action") == "bind_secrets"]
+        assert len(bind_calls) == 1
+        changes = bind_calls[0].kwargs["changes"]
+        assert changes["skills"] == ["erp-report"]
+        assert changes["secrets"] == ["ERP_TOKEN"]
+        # Values must never reach the audit journal.
+        assert "tok-secret-value" not in str(bind_calls[0])
+
+    def test_slash_binding_persists_across_model_calls_in_same_run(self, tmp_path, monkeypatch):
+        """#3861 semantics preserved under per-call recompute: after the single
+        activation call, the tool loop issues more model calls without a fresh
+        slash — the binding must survive on the shared run context."""
+        from deerflow.runtime.secret_context import read_active_secrets
+
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        context = {"secrets": {"ERP_TOKEN": "tok-123"}}
+        mw_inst = self._run_call(tmp_path, monkeypatch, [skill], context=context, message="/erp-report go")
+        assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
+
+        # Later model call in the SAME run (same context object): no slash in the
+        # latest message, skill never entered skill_context (slash injects the
+        # body directly, no read_file happens).
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, message="tool loop continues", middleware=mw_inst)
+        assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
+
+    def test_unchanged_binding_not_re_recorded(self, tmp_path, monkeypatch):
+        skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
+        journal = MagicMock()
+        context = {"secrets": {"ERP_TOKEN": "tok-1"}, "__run_journal": journal}
+        mw_inst = self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
+        self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)], middleware=mw_inst)
+
+        bind_calls = [call for call in journal.record_middleware.call_args_list if call.kwargs.get("action") == "bind_secrets"]
+        assert len(bind_calls) == 1
+
+
+class TestSecretsAutonomousParsing:
+    """Frontmatter ``secrets-autonomous`` controls in-context (autonomous) binding."""
+
+    def _parse(self, tmp_path, frontmatter_extra: str):
+        from deerflow.skills.parser import parse_skill_file
+        from deerflow.skills.types import SkillCategory
+
+        skill_dir = tmp_path / "erp-report"
+        skill_dir.mkdir()
+        skill_file = skill_dir / "SKILL.md"
+        skill_file.write_text(
+            f"""---
+name: erp-report
+description: Pull an ERP report.
+required-secrets:
+  - ERP_TOKEN
+{frontmatter_extra}---
+
+Body.
+""",
+            encoding="utf-8",
+        )
+        return parse_skill_file(skill_file, SkillCategory.CUSTOM)
+
+    def test_defaults_to_true(self, tmp_path):
+        skill = self._parse(tmp_path, "")
+        assert skill is not None
+        assert skill.secrets_autonomous is True
+
+    def test_explicit_false(self, tmp_path):
+        skill = self._parse(tmp_path, "secrets-autonomous: false\n")
+        assert skill is not None
+        assert skill.secrets_autonomous is False
+
+    def test_malformed_value_fails_closed(self, tmp_path, caplog):
+        """A non-boolean value disables autonomous binding (the safer direction)
+        instead of silently enabling it."""
+        skill = self._parse(tmp_path, 'secrets-autonomous: "yes please"\n')
+        assert skill is not None
+        assert skill.secrets_autonomous is False
+
+
 class TestBashToolInjectsActiveSecrets:
     """The bash tool forwards the per-run injection set to execute_command(env=...)."""