fix(security): add DEER_FLOW_CREDENTIALS_KEY and wire the channel credentials cipher (#6611)

* fix(security): add DEER_FLOW_CREDENTIALS_KEY for at-rest credential encryption

There was no at-rest encryption key. ChannelCredentialCipher had no
production caller: the channel service, the scheduled-task notification
outbox and the channel-connections router all built
ChannelConnectionRepository without a cipher, so store_credentials refused
(RuntimeError) and get_credentials always returned None.

- New harness module deerflow.config.credentials_key: env-only
  DEER_FLOW_CREDENTIALS_KEY (comma-separated Fernet keys; first encrypts,
  all decrypt via MultiFernet), CredentialsCipher with encrypt_text /
  decrypt_text / rotate_text, a fernet:v2: prefix with legacy fernet:v1:
  values still readable, and get_credentials_cipher() as the process-wide
  entry point. Malformed keys are refused without echoing key material.
  Without the variable a single instance generates
  {base_dir}/.credentials_key. Never derived from AUTH_JWT_SECRET.
- New deerflow.config.secret_file.read_or_create_secret_file: O_EXCL
  create (0600) plus read-back with a bounded settle window, shared by
  .credentials_key, .jwt_secret (fixes two replicas cold-starting on a
  shared volume keeping different JWT secrets) and the managed-model key.
- ChannelCredentialCipher now builds on CredentialsCipher (same
  constructors); the Gateway loads the cipher off the event loop at
  startup when channel_connections is enabled and passes it to every
  ChannelConnectionRepository. Undecryptable values stay "missing + warn".
- Startup gate: a malformed key stops startup in any topology; a declared
  multi-instance deployment with channel_connections enabled and no key
  is refused. Plain multi-worker shares the generated file and is allowed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(deploy): generate and inject DEER_FLOW_CREDENTIALS_KEY in Helm, compose and deploy.sh

- Helm: the app Secret generates a Fernet key (randBytes 32 translated to
  the urlsafe alphabet) and preserves it across upgrades via lookup like
  AUTH_JWT_SECRET; the gateway reads it through an optional secretKeyRef
  so a user-managed existingAppSecret without it still boots (the Gateway
  gate refuses only multi-instance deployments that store credentials).
  README, NOTES and values document backup, rotation and existingAppSecret.
- scripts/deploy.sh resolves the key shell -> .env -> persisted file ->
  generated, persisting it as $DEER_FLOW_HOME/.credentials_key (0600,
  noclobber/O_EXCL with read-back): the same file the Gateway generates in
  its runtime home, so either side's key is the one both keep.
- Both compose files pass DEER_FLOW_CREDENTIALS_KEY=${...:-} to the gateway.
- .env.example entry with a generation hint; README and CHANGELOG notes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(security): make abandoned secret-file replacement single-winner

A rename is last-writer-wins, so replicas that restart together and all
find the same abandoned empty secret file could each replace it and keep
a different value. The process that exclusively creates
<name>.replacing now owns the replacement, re-checks the file is still
empty, and publishes by renaming the claim over it; peers wait for that
value, bounded so a crashed replacer cannot block startup forever.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* docs(changelog): link DEER_FLOW_CREDENTIALS_KEY entries to #6611

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(security): name the setting that turns off a credentials-key consumer

The multi-instance refusal ended with "or disable
channel_connections.enabled=true"; it now says "or set
channel_connections.enabled=false".

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(security): never expose a secret file before its value is written

O_EXCL published the name before its content, so a creator that stalled
mid-write left an empty file a peer replaced after the settle window; the
resumed creator then returned a value nobody persisted, or deleted the
peer's valid file when its own write failed. Creators now write and sync
a private temporary file and hard-link it to the name, which fails if
the name exists; a failed creator only removes its own temporary file.
Filesystems without hard links publish through the same single-winner
<name>.replacing claim that replaces abandoned empty files, so creation
and replacement are serialized there.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(models): recover an empty managed-model key when no catalog exists

An interrupted first save could leave an empty key file; every later save
then passed empty bytes to Fernet and the endpoint returned 503 until the
file was removed by hand. An empty key with no catalog encrypted nothing,
so it now goes through the secret-file recovery path; next to an existing
catalog it is still refused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(docker): pass the dotenv DEER_FLOW_CREDENTIALS_KEY through make docker-start

scripts/docker.sh runs Compose from docker/ without --env-file, so the dev
gateway's DEER_FLOW_CREDENTIALS_KEY=${DEER_FLOW_CREDENTIALS_KEY:-} entry
interpolated to empty and overrode the key the checkout .env supplied via
env_file. Export it from the dotenv like AUTH_TRUSTED_PROXIES.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

.env.example

@@ -107,6 +107,14 @@ UNBROWSE_API_KEY=your-unbrowse-api-key
 # manually when Gateway workers also run outside the bundled deploy script.
 # DEER_FLOW_INTERNAL_AUTH_TOKEN=your-shared-internal-token
 
+# At-rest encryption key for stored credentials (a Fernet key). Comma-separate
+# several to rotate: the first encrypts, all decrypt. Unset, a single instance
+# generates DEER_FLOW_HOME/.credentials_key (`make up` persists the same file);
+# instances that do not share DEER_FLOW_HOME must all set the same value.
+# Back it up: losing it makes stored credentials unreadable. Generate:
+#   python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
+# DEER_FLOW_CREDENTIALS_KEY=your-fernet-key
+
 # Provisioner API key for sandbox authentication (required when using provisioner/K8s sandbox mode).
 # The same value must be set on the provisioner container and in config.yaml sandbox.provisioner_api_key.
 # Generate: openssl rand -hex 32

CHANGELOG.md

@@ -4388,6 +4388,22 @@ This release closes that milestone with **439 merged pull requests**.
   truncated at that character. These providers now split on `"
 "` only, as the shared parser already documents and as
   LocalSandbox, the AIO backend and E2B already do. ([#6595])
+- **security:** Add `DEER_FLOW_CREDENTIALS_KEY`, an env-only at-rest
+  encryption key for stored credentials. Per-connection IM channel credentials
+  had an encryption path that no production code wired up, so they could be
+  neither stored nor read and Slack always used the deployment bot token. The
+  Gateway now loads the key at startup (comma-separated Fernet keys: the first
+  encrypts, all decrypt, for rotation; values carry a `fernet:v2:` prefix and
+  earlier `fernet:v1:` values stay readable) and passes it to every channel
+  connection repository; undecryptable values are treated as missing. Unset, a
+  single instance generates `{DEER_FLOW_HOME}/.credentials_key`; a declared
+  multi-instance deployment with `channel_connections` enabled refuses to start
+  without the key, and a malformed key is refused without being echoed. The
+  Helm chart generates the key into its app Secret and preserves it across
+  upgrades, `make up` persists it next to the runtime home, and both compose
+  files pass it to the Gateway. `.jwt_secret` (and the managed-model key) are
+  now created exclusively and read back, so replicas cold-starting on a shared
+  volume no longer keep different session-signing secrets. ([#6611])
 
 ### Documentation
 
@@ -9517,3 +9533,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6588]: https://github.com/bytedance/deer-flow/pull/6588
 [#6590]: https://github.com/bytedance/deer-flow/pull/6590
 [#6595]: https://github.com/bytedance/deer-flow/pull/6595
+[#6611]: https://github.com/bytedance/deer-flow/pull/6611

CHANGELOG_zh.md

@@ -3381,6 +3381,17 @@
   的匹配行会在该字符处被截断。现在这三个提供者只按 `"
 "` 切分，与共享解析器
   既有的约定以及 LocalSandbox、AIO 后端、E2B 的行为一致。([#6595])
+- **安全：** 新增仅从环境变量读取的静态凭据加密密钥 `DEER_FLOW_CREDENTIALS_KEY`。
+  按连接存储的 IM 渠道凭据此前虽有加密路径，但没有任何生产代码接入，既无法写入
+  也无法读取，Slack 因而始终使用部署级 bot token。现在 Gateway 在启动时加载该密钥
+  （逗号分隔的 Fernet 密钥：第一个用于加密，全部用于解密，以支持轮换；密文带
+  `fernet:v2:` 前缀，旧的 `fernet:v1:` 密文仍可读取），并传给所有渠道连接仓库；
+  无法解密的值按缺失处理。未设置时单实例会生成 `{DEER_FLOW_HOME}/.credentials_key`；
+  声明为多实例且启用 `channel_connections` 的部署在缺少密钥时拒绝启动，格式错误的
+  密钥同样会被拒绝且不会回显。Helm chart 将密钥生成到 app Secret 并在升级时保留，
+  `make up` 将其持久化到运行时目录，两个 compose 文件都会把它传给 Gateway。
+  `.jwt_secret`（以及托管模型密钥）现在以独占方式创建并回读，共享卷上同时冷启动的
+  副本不再各自保留不同的会话签名密钥。 ([#6611])
 
 ### 文档
 
@@ -7653,3 +7664,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6588]: https://github.com/bytedance/deer-flow/pull/6588
 [#6590]: https://github.com/bytedance/deer-flow/pull/6590
 [#6595]: https://github.com/bytedance/deer-flow/pull/6595
+[#6611]: https://github.com/bytedance/deer-flow/pull/6611

README.md

@@ -533,7 +533,7 @@ Browser login uses `HttpOnly` session cookies. The login page offers a "keep me
 DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser-facing scheme and origin behind a proxy. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header. Configure the outer trusted proxy to replace or strip client-supplied forwarding headers before traffic reaches DeerFlow.
 
 > [!IMPORTANT]
-> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its own multi-worker coordination.
+> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its own multi-worker coordination.
 >
 > In single-process JSONL deployments, cancelling an admitted event-store mutation waits for its background file I/O, rollback, and bookkeeping to settle before releasing the thread write lock. This prevents an older cancelled write from recreating deleted records or rolling back a later successful write. Cancellation can therefore wait on slow storage; it does not stop an in-flight filesystem operation. Callers still waiting to acquire the lock can cancel without starting a mutation. A batch spanning multiple threads drains its current thread group before propagating cancellation; subsequent thread groups do not start.
 >

backend/app/channels/service.py

@@ -95,7 +95,8 @@ def _merge_channel_connection_runtime_config(channels_config: dict[str, Any], ap
     merge_runtime_channel_configs(channels_config, connection_config)
 
 
-def _make_connection_repo(connection_config: ChannelConnectionsConfig | None):
+def _make_connection_repo(connection_config: ChannelConnectionsConfig | None, *, credentials_cipher: Any | None = None):
+    """Build the connection repository; ``credentials_cipher`` encrypts per-connection credentials."""
     if connection_config is None or not getattr(connection_config, "enabled", False):
         return None
 
@@ -110,7 +111,7 @@ def _make_connection_repo(connection_config: ChannelConnectionsConfig | None):
     if session_factory is None:
         logger.warning("Channel connections are enabled but database persistence is not available")
         return None
-    return ChannelConnectionRepository(session_factory)
+    return ChannelConnectionRepository(session_factory, cipher=credentials_cipher)
 
 
 class ChannelService:
@@ -177,13 +178,16 @@ class ChannelService:
         app_config: AppConfig | None = None,
         *,
         get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
+        credentials_cipher: Any | None = None,
     ) -> ChannelService:
         """Create a ChannelService from the application config.
 
         ``get_stream_bridge`` is threaded straight through to the
         ``ChannelManager`` (see its docstring); it is optional so direct
         callers (including most tests) that don't need follow-up-buffer
-        auto-draining can omit it.
+        auto-draining can omit it. ``credentials_cipher`` is the Gateway's
+        ``DEER_FLOW_CREDENTIALS_KEY`` cipher for per-connection credentials;
+        without it stored credentials stay unavailable.
         """
         if app_config is None:
             from deerflow.config.app_config import get_app_config
@@ -200,7 +204,7 @@ class ChannelService:
         require_bound_identity = bool(connections_enabled and getattr(connection_config, "require_bound_identity", True))
         return cls(
             channels_config=channels_config,
-            connection_repo=_make_connection_repo(connection_config),
+            connection_repo=_make_connection_repo(connection_config, credentials_cipher=credentials_cipher),
             require_bound_identity=require_bound_identity,
             app_config=app_config,
             get_stream_bridge=get_stream_bridge,
@@ -663,6 +667,7 @@ async def start_channel_service(
     app_config: AppConfig | None = None,
     *,
     get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
+    credentials_cipher: Any | None = None,
 ) -> ChannelService:
     """Create and start the global ChannelService from app config.
 
@@ -679,7 +684,7 @@ async def start_channel_service(
     # from_app_config resolves the channel store path (realpath) and reads the
     # runtime config files; keep that disk IO off the event loop.
     # asyncio.to_thread forwards both args and kwargs to the target callable.
-    service = await asyncio.to_thread(ChannelService.from_app_config, app_config, get_stream_bridge=get_stream_bridge)
+    service = await asyncio.to_thread(ChannelService.from_app_config, app_config, get_stream_bridge=get_stream_bridge, credentials_cipher=credentials_cipher)
     _channel_service = service
 
     async def rollback_failed_start() -> None:

backend/app/gateway/app.py

@@ -437,7 +437,29 @@ async def _shutdown_subagent_batch_service(app: FastAPI) -> None:
         set_subagent_batch_submitter(None)
 
 
-def _scheduled_task_notification_repos(startup_config: AppConfig):
+async def _load_credentials_cipher(startup_config: AppConfig):
+    """Load the at-rest credentials cipher when an enabled feature stores credentials, else ``None``.
+
+    Without ``DEER_FLOW_CREDENTIALS_KEY`` the first load creates
+    ``{base_dir}/.credentials_key``, so it runs in a worker thread. The startup
+    gate has already refused a malformed environment key; a key file that
+    cannot be read or created leaves stored credentials unavailable (the
+    behaviour before the key existed) instead of failing the Gateway.
+    """
+    from app.gateway.deps import credentials_key_consumers
+
+    if not credentials_key_consumers(startup_config):
+        return None
+    from deerflow.config import credentials_key
+
+    try:
+        return await asyncio.to_thread(credentials_key.get_credentials_cipher)
+    except Exception:
+        logger.exception("Failed to load the credentials encryption key; stored channel connection credentials stay unavailable")
+        return None
+
+
+def _scheduled_task_notification_repos(startup_config: AppConfig, credentials_cipher=None):
     """Return ``(connection_repo, notification_repo)`` for the scheduled-run outbox (issue #4254).
 
     Both are None unless channel connections are enabled, because the completion
@@ -454,7 +476,7 @@ def _scheduled_task_notification_repos(startup_config: AppConfig):
     session_factory = get_session_factory()
     if session_factory is None:
         return None, None
-    return ChannelConnectionRepository(session_factory), NotificationDeliveryRepository(session_factory)
+    return ChannelConnectionRepository(session_factory, cipher=credentials_cipher), NotificationDeliveryRepository(session_factory)
 
 
 async def _start_scheduled_task_notification_delivery(app: FastAPI, startup_config: AppConfig, notification_repo, connection_repo) -> None:
@@ -680,12 +702,18 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.exception("Failed to start summary generation workers")
 
+        # At-rest credentials cipher (DEER_FLOW_CREDENTIALS_KEY) shared by every
+        # ChannelConnectionRepository this process builds; None when no enabled
+        # feature stores credentials.
+        credentials_cipher = await _load_credentials_cipher(startup_config)
+        app.state.credentials_cipher = credentials_cipher
+
         # Enqueue side of the scheduled-run notification outbox (issue #4254).
         # It only needs the durable table, so it is wired with the scheduler;
         # the delivery worker starts after the channel service, further down.
         notification_connection_repo = scheduled_notification_repo = None
         try:
-            notification_connection_repo, scheduled_notification_repo = _scheduled_task_notification_repos(startup_config)
+            notification_connection_repo, scheduled_notification_repo = _scheduled_task_notification_repos(startup_config, credentials_cipher)
         except Exception:
             logger.exception("Failed to prepare the scheduled-task notification outbox")
 
@@ -736,6 +764,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             channel_service = await start_channel_service(
                 startup_config,
                 get_stream_bridge=lambda: getattr(app.state, "stream_bridge", None),
+                credentials_cipher=credentials_cipher,
             )
             logger.info("Channel service started: %s", channel_service.get_status())
         except Exception:

backend/app/gateway/auth/config.py

@@ -33,29 +33,22 @@ _auth_config: AuthConfig | None = None
 
 
 def _load_or_create_secret() -> str:
-    """Load persisted JWT secret from ``{base_dir}/.jwt_secret``, or generate and persist a new one."""
+    """Load persisted JWT secret from ``{base_dir}/.jwt_secret``, or generate and persist a new one.
+
+    The file is created exclusively and read back, so processes that share the
+    runtime home and cold-start together (uvicorn workers, replicas on a shared
+    volume) converge on one secret instead of each keeping its own.
+    """
     from deerflow.config.paths import get_paths
+    from deerflow.config.secret_file import read_or_create_secret_file
 
     paths = get_paths()
     secret_file = paths.base_dir / _SECRET_FILE
 
     try:
-        if secret_file.exists():
-            secret = secret_file.read_text(encoding="utf-8").strip()
-            if secret:
-                return secret
+        return read_or_create_secret_file(secret_file, lambda: secrets.token_urlsafe(32))
     except OSError as exc:
-        raise RuntimeError(f"Failed to read JWT secret from {secret_file}. Set AUTH_JWT_SECRET explicitly or fix DEER_FLOW_HOME/base directory permissions so DeerFlow can read its persisted auth secret.") from exc
-
-    secret = secrets.token_urlsafe(32)
-    try:
-        secret_file.parent.mkdir(parents=True, exist_ok=True)
-        fd = os.open(secret_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
-        with os.fdopen(fd, "w", encoding="utf-8") as fh:
-            fh.write(secret)
-    except OSError as exc:
-        raise RuntimeError(f"Failed to persist JWT secret to {secret_file}. Set AUTH_JWT_SECRET explicitly or fix DEER_FLOW_HOME/base directory permissions so DeerFlow can store a stable auth secret.") from exc
-    return secret
+        raise RuntimeError(f"Failed to read or persist the JWT secret at {secret_file}. Set AUTH_JWT_SECRET explicitly or fix DEER_FLOW_HOME/base directory permissions so DeerFlow can keep a stable auth secret.") from exc
 
 
 def get_auth_config() -> AuthConfig:

backend/app/gateway/deps.py

@@ -31,6 +31,7 @@ from langgraph.types import Checkpointer
 from app.gateway.auth.errors import AuthErrorCode
 from deerflow.community.browser_automation.session import browser_multi_worker_error
 from deerflow.config.app_config import AppConfig, get_app_config
+from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, CREDENTIALS_KEY_FILENAME, GENERATE_KEY_COMMAND, CredentialsKeyError, parse_credentials_keys
 from deerflow.config.deployment_config import multi_instance_declaration
 from deerflow.persistence.feedback import FeedbackRepository
 from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, RunContext, RunManager, StreamBridge
@@ -228,6 +229,56 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
         )
 
 
+def credentials_key_consumers(config: AppConfig) -> list[str]:
+    """Return the enabled features that store data under ``DEER_FLOW_CREDENTIALS_KEY``.
+
+    Each entry is the setting that enabled the feature, as refusal text spells
+    it. The Gateway loads the key only when this list is non-empty, and the
+    multi-instance gate below refuses a missing key only for these features, so
+    a deployment that stores no credentials keeps booting without one.
+    """
+    consumers: list[str] = []
+    connections = getattr(config, "channel_connections", None)
+    if connections is not None and getattr(connections, "enabled", False):
+        # ChannelConnectionRepository encrypts per-connection credentials.
+        consumers.append("channel_connections.enabled=true")
+    return consumers
+
+
+def _enforce_credentials_key(config: AppConfig) -> None:
+    """Refuse a malformed credentials key, or a missing one where instances cannot share the generated file.
+
+    A set ``DEER_FLOW_CREDENTIALS_KEY`` must parse (any topology): starting with
+    a key that cannot decrypt anything would silently hide stored credentials.
+    Without it each instance generates ``{base_dir}/.credentials_key``. Workers
+    of one process tree share that file and converge on one key through its
+    exclusive create, so only the explicit multi-instance declaration -- Pods or
+    hosts that need not share a runtime home -- requires the variable, and only
+    when a feature from :func:`credentials_key_consumers` is enabled.
+    """
+    raw = os.environ.get(CREDENTIALS_KEY_ENV_VAR, "")
+    if raw.strip():
+        try:
+            parse_credentials_keys(raw)
+        except CredentialsKeyError as exc:
+            raise SystemExit(str(exc)) from None
+        return
+    declaration = multi_instance_declaration(config)
+    if declaration is None:
+        return
+    consumers = credentials_key_consumers(config)
+    if not consumers:
+        return
+    enabled = ", ".join(consumers)
+    disabled = ", ".join(consumer.removesuffix("=true") + "=false" for consumer in consumers)
+    raise SystemExit(
+        f"{declaration.knob} with {enabled} requires {CREDENTIALS_KEY_ENV_VAR}: without it every instance generates its own "
+        f"{{base_dir}}/{CREDENTIALS_KEY_FILENAME}, and credentials one instance stores cannot be decrypted on its peers. "
+        f"Generate one key with {GENERATE_KEY_COMMAND} and set the same value on every instance (the Helm chart and "
+        f"scripts/deploy.sh generate and inject it automatically), {declaration.rollback}, or set {disabled}."
+    )
+
+
 def _validate_agent_storage(config: AppConfig) -> None:
     """Fail fast on an agent-storage backend the database cannot support.
 
@@ -600,6 +651,9 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
     # SQLite write-locks cannot support concurrent multi-process access.
     # ------------------------------------------------------------------
     _enforce_postgres_for_multi_worker(startup_config)
+    # Reject a malformed DEER_FLOW_CREDENTIALS_KEY, or a missing one where
+    # declared instances store credentials but cannot share the generated file.
+    _enforce_credentials_key(startup_config)
     # Reject agent_storage.backend='db' on a non-durable database, and warn on
     # node-divergent file storage under multi-worker Postgres.
     _validate_agent_storage(startup_config)

backend/app/gateway/routers/channel_connections.py

@@ -229,7 +229,8 @@ def _get_repository(request: Request, config: ChannelConnectionsConfig) -> Chann
     if sf is None:
         raise HTTPException(status_code=503, detail="Channel connection persistence is not available")
 
-    repo = ChannelConnectionRepository(sf)
+    # The lifespan loads the DEER_FLOW_CREDENTIALS_KEY cipher off the event loop.
+    repo = ChannelConnectionRepository(sf, cipher=getattr(request.app.state, "credentials_cipher", None))
     request.app.state.channel_connection_repo = repo
     return repo
 

backend/docs/CONFIGURATION.md

@@ -592,8 +592,27 @@ Notes:
 - IM chat-to-thread bindings are shared through the database too: with `database.backend: sqlite` or `postgres` the `ChannelManager` keeps the `channel_name:chat_id[:topic_id]` → thread mapping of unbound IM conversations in the `channel_thread_bindings` table (migration `0038_channel_thread_bindings`), so a conversation created on one instance continues on the same thread when its next message lands on another. On the first start after upgrading, an existing `{base_dir}/channels/store.json` is imported once — only into an empty table, with `INSERT … ON CONFLICT DO NOTHING` so concurrently starting instances cannot duplicate a binding — and renamed `store.json.migrated`; a populated table leaves the file untouched. `memory` keeps the per-process JSON file.
 - The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, and the WeChat QR-login guard, which otherwise only look at the worker count.
 - With the default DeerMem backend, the declaration also warns when the derived SQLite retrieval index sits inside the shared `storage_path` (the default `{storage_path}/.retrieval`). Set `memory.backend_config.retrieval_index_path` to an instance-local directory (relative values resolve against `storage_path`): SQLite WAL is unsupported on network filesystems, and the index is rebuilt from the Markdown facts at startup and re-synced per user scope after a peer writes.
+- A declared multi-instance deployment that enables a feature storing credentials (today `channel_connections.enabled: true`) also needs one shared `DEER_FLOW_CREDENTIALS_KEY`; see [Credentials encryption key](#credentials-encryption-key-deer_flow_credentials_key).
 - Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it.
 
+### Credentials encryption key (`DEER_FLOW_CREDENTIALS_KEY`)
+
+DeerFlow encrypts credentials it stores at rest (today: per-connection IM channel credentials in the `channel_credentials` table) with a deployment key read **only** from the environment — there is no `config.yaml` key, and it is never derived from `AUTH_JWT_SECRET`.
+
+```bash
+# Generate a key (urlsafe base64 of 32 random bytes, a Fernet key):
+python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
+export DEER_FLOW_CREDENTIALS_KEY=<key>
+```
+
+- **Rotation:** the value may be a comma-separated list. The first key encrypts new values; every listed key decrypts. Prepend the new key (`new,old`), restart, and keep the old key listed until the values it encrypted have been rewritten. Stored values carry a `fernet:v2:` prefix; `fernet:v1:` values written by earlier builds stay readable under the same key.
+- **Single instance:** when the variable is unset or blank, the Gateway generates `{DEER_FLOW_HOME}/.credentials_key` (mode `0600`) on first use and reuses it. The file is created exclusively, so uvicorn workers sharing the runtime home converge on one key. It is only created when a feature that stores credentials is enabled.
+- **Multi-instance:** instances that do not share a runtime home would each generate their own file and could not decrypt what their peers stored. Startup therefore refuses a declared multi-instance deployment (`deployment.multi_instance: true` / `DEER_FLOW_MULTI_INSTANCE=1`) that enables `channel_connections` without the variable. Multiple workers of one process tree (`GATEWAY_WORKERS > 1` without the declaration) share the file and are not refused. Deployments that store no credentials keep booting without a key.
+- **Malformed keys** (wrong length, standard rather than urlsafe base64, any bad entry in the list) stop startup with an error that never repeats the key material.
+- **Backup:** losing the key makes every stored credential unreadable. Unreadable values are treated as missing (logged as a warning), never as a crash — Slack, for example, falls back to the deployment bot token.
+- **Deployment tooling:** the Helm chart generates the key into its app Secret and preserves it across upgrades; `make up` (`scripts/deploy.sh`) honors a shell or `.env` value and otherwise generates and persists `$DEER_FLOW_HOME/.credentials_key` — the same file the Gateway would generate, so whichever side created it first wins.
+- The persisted `.jwt_secret` (used when `AUTH_JWT_SECRET` is unset) is created the same exclusive way, so replicas cold-starting on a shared volume no longer keep different session-signing secrets.
+
 ### Agent Storage
 
 Custom agent **definitions** (`config.yaml` + `SOUL.md`) are stored per-user on
@@ -1555,6 +1574,7 @@ models:
 - `DEER_FLOW_CONFIG_PATH` - Custom config file path
 - `DEER_FLOW_EXTENSIONS_CONFIG_PATH` - Custom extensions config file path
 - `DEER_FLOW_HOME` - Runtime state directory (defaults to `.deer-flow` under the project root)
+- `DEER_FLOW_CREDENTIALS_KEY` - At-rest encryption key(s) for stored credentials; comma-separate to rotate. See [Credentials encryption key](#credentials-encryption-key-deer_flow_credentials_key)
 - `DEER_FLOW_SKILLS_PATH` - Skills directory when `skills.path` is omitted
 - `GATEWAY_ENABLE_DOCS` - Set to `false` to disable Swagger UI (`/docs`), ReDoc (`/redoc`), and OpenAPI schema (`/openapi.json`) endpoints (default: `true`)
 

backend/docs/IM_CHANNEL_CONNECTIONS.md

@@ -539,7 +539,7 @@ Connection records live in SQL tables under `deerflow.persistence.channel_connec
 - `channel_connections`: owner user, provider identity, workspace/guild/team, status, metadata.
 - `channel_oauth_states`: one-time connect codes and Telegram deep-link state.
 - `channel_conversations`: connection-scoped IM conversation to DeerFlow thread mapping.
-- `channel_credentials`: reserved for future provider-token flows, not used by the local/private binding flow.
+- `channel_credentials`: per-connection provider tokens for future provider-token flows, encrypted with `DEER_FLOW_CREDENTIALS_KEY` (see [CONFIGURATION.md](CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)); not written by the local/private binding flow.
 
 Incoming messages that resolve to a connection carry `connection_id`, `owner_user_id`, and `workspace_id`. `ChannelManager` uses `owner_user_id` as the DeerFlow run user id and preserves the raw platform user id as `channel_user_id`.
 
@@ -561,8 +561,11 @@ the channel state directory.
   responses mask password fields, and mutating runtime/channel-worker APIs
   require an admin user.
 - Stored per-connection credentials use the `channel_credentials` encryption
-  path. If stored credential material cannot be decrypted, DeerFlow treats it
-  as unavailable instead of using corrupt secrets.
+  path, keyed by `DEER_FLOW_CREDENTIALS_KEY` (or the generated
+  `{DEER_FLOW_HOME}/.credentials_key` on a single instance). If stored
+  credential material cannot be decrypted — a lost or changed key — DeerFlow
+  treats it as unavailable instead of using corrupt secrets. A declared
+  multi-instance deployment with channel connections enabled must set the key.
 - The local plaintext runtime credential fallback is documented above; prefer
   deployment-managed environment/config secrets for non-local deployments until
   a dedicated secret backend is configured.

backend/packages/harness/deerflow/config/AGENTS.md

@@ -124,6 +124,26 @@ cache, including AppConfig's middleware snapshot, until a readable revision retu
 
 **Shared Reset Markers** (`shared_reset_marker.py`): caches derived from the extensions config can go stale without a byte of it changing (a remote MCP server's `tools/list`, a skill installed or edited on the shared volume by another process). `SharedResetMarker(suffix)` owns one hidden JSON file beside the resolved config, `.<config name>.<suffix>.json`; `publish()` replaces it atomically with a random generation under `extensions_config_write_lock` + `extensions_config_file_lock` (`publish_locked()` when the caller already holds both), chaining `previous_generation` and an optional `user_id`. `SharedResetMarkerTracker.poll()` re-reads the marker at most once per second (monotonic clock, never blocking behind a concurrent poller), adopts the current state silently on its first poll, and reports a `SharedResetChange`: `user_ids={user}` only when exactly one publication carrying that user happened since the last poll, otherwise `None` (retire everything), including deleted/unreadable markers and config-path switches. `note_own_publication()` lets the writer skip the redundant self-invalidation. `resolve_shared_config_path()` maps a missing explicit path to `None` (no shared directory: callers answer `scope=process`). Consumers: `mcp/cache.py` (`mcp-cache-reset`) and `agents/lead_agent/prompt.py` (`skills-cache-reset`). Tests: `tests/test_shared_reset_marker.py`.
 
+**Credentials key** (`credentials_key.py`): `DEER_FLOW_CREDENTIALS_KEY` is env-only
+(no `config.yaml` key, never derived from `AUTH_JWT_SECRET`): comma-separated
+Fernet keys, first encrypts, all decrypt (`MultiFernet`). `get_credentials_cipher()`
+is the process-wide entry point for consumers that store secrets at rest; it
+returns a `CredentialsCipher` (`encrypt_text`/`decrypt_text`/`rotate_text`,
+`fernet:v2:` prefix, legacy `fernet:v1:` readable). `decrypt_text` raises
+`CredentialsDecryptError` (an `InvalidToken`); callers treat that as missing data
+and never crash. Unset, it generates `{base_dir}/.credentials_key` -- file I/O,
+so async callers use `asyncio.to_thread`. Error text never carries key material.
+A new consumer must also join `app.gateway.deps.credentials_key_consumers`, which
+drives both Gateway loading and the multi-instance refusal.
+
+**Secret files** (`secret_file.py`): `read_or_create_secret_file` is the one way
+to create a shared secret file (`.credentials_key`, `.jwt_secret`, the managed
+model key): write a `0600` temp file, then hard-link it exclusively and read
+back, so the name is never empty or partial and concurrent creators converge.
+No hard links (SMB) or an abandoned empty file (older releases): single-winner
+`<name>.replacing` claim plus rename. Never overwrites a non-empty invalid file.
+Tests: `tests/test_secret_file.py`, `tests/test_credentials_key.py`.
+
 ### Config Schema
 
 **`config.yaml`** key sections:

backend/packages/harness/deerflow/config/credentials_key.py

@@ -0,0 +1,165 @@
+"""At-rest encryption key for stored credentials (``DEER_FLOW_CREDENTIALS_KEY``).
+
+The key is environment-only: one or more comma-separated Fernet keys
+(urlsafe base64 of 32 random bytes). The first key encrypts and every key
+decrypts, so rotation is "prepend a new key, re-encrypt with
+:meth:`CredentialsCipher.rotate_text`, then drop the old key". It is never
+derived from ``AUTH_JWT_SECRET`` or any other secret.
+
+Without the variable a single instance generates ``{base_dir}/.credentials_key``
+once (exclusive create, mode ``0600``), so uvicorn workers sharing the runtime
+home converge on one key. Instances that do not share a runtime home cannot
+rely on that file; the Gateway refuses a declared multi-instance deployment
+whose enabled features store credentials without the variable
+(``app.gateway.deps._enforce_credentials_key``).
+
+Losing the key makes every value encrypted with it unreadable; consumers treat
+such values as missing rather than failing.
+
+Ciphertext format: ``fernet:v2:<Fernet token>``. ``fernet:v1:`` (and bare
+tokens) were written by the earlier ``ChannelCredentialCipher``; they are still
+decrypted with the configured keys.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+import threading
+from collections.abc import Sequence
+from pathlib import Path
+
+from cryptography.fernet import Fernet, InvalidToken, MultiFernet
+
+from deerflow.config.secret_file import InvalidSecretFileError, read_or_create_secret_file
+
+logger = logging.getLogger(__name__)
+
+CREDENTIALS_KEY_ENV_VAR = "DEER_FLOW_CREDENTIALS_KEY"
+CREDENTIALS_KEY_FILENAME = ".credentials_key"
+CIPHERTEXT_PREFIX = "fernet:v2:"
+LEGACY_CIPHERTEXT_PREFIX = "fernet:v1:"
+GENERATE_KEY_COMMAND = 'python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"'
+
+# urlsafe base64 of exactly 32 bytes: 43 alphabet characters plus one "=" pad.
+_FERNET_KEY_PATTERN = re.compile(r"[A-Za-z0-9_-]{43}=")
+_KEY_FILE_SETTLE_SECONDS = 1.0
+
+
+class CredentialsKeyError(ValueError):
+    """The configured key material is unusable. The message never contains key material."""
+
+
+class CredentialsDecryptError(InvalidToken):
+    """A stored value cannot be decrypted with any configured key (wrong or lost key, or corrupt data)."""
+
+
+def parse_credentials_keys(raw: str, *, source: str = CREDENTIALS_KEY_ENV_VAR) -> list[str]:
+    """Split and validate a comma-separated key list; blank entries are ignored.
+
+    ``source`` names where the value came from in error messages.
+    """
+    keys = [part.strip() for part in raw.split(",") if part.strip()]
+    if not keys:
+        raise CredentialsKeyError(f"{source} holds no key. Generate one with: {GENERATE_KEY_COMMAND}")
+    for index, key in enumerate(keys, start=1):
+        if not _is_fernet_key(key):
+            position = f"entry {index} of {len(keys)}" if len(keys) > 1 else "the value"
+            raise CredentialsKeyError(f"{source}: {position} is not a Fernet key (urlsafe base64 of 32 bytes, 44 characters). Generate one with: {GENERATE_KEY_COMMAND}")
+    return keys
+
+
+def _is_fernet_key(key: str) -> bool:
+    if not _FERNET_KEY_PATTERN.fullmatch(key):
+        return False
+    try:
+        Fernet(key)
+    except (TypeError, ValueError):
+        return False
+    return True
+
+
+class CredentialsCipher:
+    """Text encryption under the deployment credentials key(s).
+
+    Built on :class:`cryptography.fernet.MultiFernet`: the first key encrypts,
+    every key decrypts. Values carry the ``fernet:v2:`` prefix.
+    """
+
+    def __init__(self, fernets: Sequence[Fernet]) -> None:
+        if not fernets:
+            raise CredentialsKeyError("at least one credentials key is required")
+        self._multi = MultiFernet(list(fernets))
+
+    @classmethod
+    def from_keys(cls, keys: Sequence[str | bytes]) -> CredentialsCipher:
+        return cls([Fernet(key) for key in keys])
+
+    def encrypt_text(self, value: str) -> str:
+        return CIPHERTEXT_PREFIX + self._multi.encrypt(value.encode("utf-8")).decode("ascii")
+
+    def decrypt_text(self, value: str) -> str:
+        """Decrypt a ``fernet:v2:`` value, or a legacy ``fernet:v1:`` / bare token.
+
+        Raises :class:`CredentialsDecryptError` (an ``InvalidToken``) when no
+        configured key can decrypt it.
+        """
+        token = value.removeprefix(CIPHERTEXT_PREFIX) if value.startswith(CIPHERTEXT_PREFIX) else value.removeprefix(LEGACY_CIPHERTEXT_PREFIX)
+        try:
+            return self._multi.decrypt(token.encode("ascii")).decode("utf-8")
+        except (InvalidToken, UnicodeError):
+            raise CredentialsDecryptError("stored value cannot be decrypted with the configured credentials key(s)") from None
+
+    def rotate_text(self, value: str) -> str:
+        """Re-encrypt ``value`` under the primary (first) key, for re-encryption after rotation."""
+        return self.encrypt_text(self.decrypt_text(value))
+
+
+def load_credentials_cipher(*, base_dir: Path | None = None) -> CredentialsCipher:
+    """Build the deployment cipher from ``DEER_FLOW_CREDENTIALS_KEY``, else the generated key file.
+
+    Performs file I/O when the variable is unset; async callers must run it in
+    a worker thread. Raises :class:`CredentialsKeyError` for malformed key
+    material in either source.
+    """
+    raw = os.environ.get(CREDENTIALS_KEY_ENV_VAR, "")
+    if raw.strip():
+        return CredentialsCipher.from_keys(parse_credentials_keys(raw))
+
+    if base_dir is None:
+        from deerflow.config.paths import get_paths
+
+        base_dir = get_paths().base_dir
+    key_file = Path(base_dir) / CREDENTIALS_KEY_FILENAME
+    try:
+        value = read_or_create_secret_file(
+            key_file,
+            lambda: Fernet.generate_key().decode("ascii"),
+            validate=lambda candidate: parse_credentials_keys(candidate, source=str(key_file)),
+            settle_seconds=_KEY_FILE_SETTLE_SECONDS,
+        )
+    except InvalidSecretFileError:
+        raise CredentialsKeyError(f"{key_file} does not hold a valid credentials key. Restore it from a backup (stored credentials are unreadable without it), or set {CREDENTIALS_KEY_ENV_VAR}.") from None
+    logger.info("%s is not set; using the credentials key in %s. Back this file up: losing it makes stored credentials unreadable.", CREDENTIALS_KEY_ENV_VAR, key_file)
+    return CredentialsCipher.from_keys(parse_credentials_keys(value, source=str(key_file)))
+
+
+_cipher: CredentialsCipher | None = None
+_cipher_lock = threading.Lock()
+
+
+def get_credentials_cipher() -> CredentialsCipher:
+    """Return the process-wide cipher, loading it on first use (may perform file I/O)."""
+    global _cipher
+    with _cipher_lock:
+        if _cipher is None:
+            _cipher = load_credentials_cipher()
+        return _cipher
+
+
+def reset_credentials_cipher() -> None:
+    """Forget the process-wide cipher so the next call reloads it (tests, key rotation on restart)."""
+    global _cipher
+    with _cipher_lock:
+        _cipher = None

backend/packages/harness/deerflow/config/managed_models.py

@@ -22,6 +22,7 @@ from deerflow.config.file_signature import get_config_signature
 from deerflow.config.managed_model_providers import resolve_managed_model_provider
 from deerflow.config.model_config import ModelConfig
 from deerflow.config.runtime_paths import runtime_home
+from deerflow.config.secret_file import read_or_create_secret_file
 
 if TYPE_CHECKING:
     from deerflow.config.app_config import AppConfig
@@ -81,10 +82,14 @@ class ManagedModelStore:
     def _cipher(self, *, create: bool = False):
         from cryptography.fernet import Fernet
 
-        if not self.key_path.exists():
+        # An empty key file is an older release's interrupted first save: it
+        # encrypted nothing, so it is recoverable exactly when no catalog exists.
+        if not self.key_path.exists() or not self.key_path.read_bytes().strip():
             if not create or self.path.exists():
                 raise ValueError("Managed model encryption key is missing; restore it from backup")
-            self._write(self.key_path, Fernet.generate_key())
+            # Exclusive create (replacing an abandoned empty file): a peer sharing
+            # the runtime home that published first keeps its key.
+            read_or_create_secret_file(self.key_path, lambda: Fernet.generate_key().decode("ascii"))
         return Fernet(self.key_path.read_bytes())
 
     def list(self) -> list[ManagedModel]:

backend/packages/harness/deerflow/config/secret_file.py

@@ -0,0 +1,185 @@
+"""Exclusive-create helper for small secret files shared by concurrent processes.
+
+Gateway processes that share a runtime home -- uvicorn workers of one host, or
+replicas on a shared volume -- cold-start together and may all find a secret
+file missing. A read-then-truncating write lets each keep a different value,
+so ``read_or_create_secret_file`` publishes exclusively instead: exactly one
+creator wins and every other process reads the winner's value back.
+
+A creator writes and syncs the whole value into a private temporary file first
+and then hard-links it to the final name, which fails if the name exists. The
+name therefore never appears empty or half-written, however long a creator
+stalls, and a creator that fails only ever removes its own temporary file.
+Filesystems without hard links (SMB shares) publish through the replacement
+claim described below instead.
+
+An empty or incomplete file can still come from an older release (which
+created the name before writing it) or from external tampering. The helper
+re-reads it for a bounded settle window; a non-empty file that never validates
+is refused and left untouched, because it may be the only copy of a key that
+protects stored data. An empty one is abandoned -- nothing can have been
+signed or encrypted with an empty secret -- and is replaced. A rename is
+last-writer-wins, so replacement is single-winner: the process that exclusively
+creates ``<name>.replacing`` re-checks the name under that claim, writes the
+new value into the claim and renames it over the name, which publishes the
+value and releases the claim in one step; every other process waits for that
+value, for a bounded time.
+"""
+
+from __future__ import annotations
+
+import errno
+import logging
+import os
+import tempfile
+import time
+from collections.abc import Callable
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+_POLL_INTERVAL_SECONDS = 0.05
+# How long to wait on a peer's replacement claim before declaring it abandoned.
+_CLAIM_WAIT_SECONDS = 10.0
+# os.link() errors meaning the filesystem cannot hard-link at all.
+_NO_HARD_LINK_ERRNOS = frozenset({errno.EPERM, errno.EOPNOTSUPP, errno.ENOTSUP, errno.ENOSYS})
+
+
+class InvalidSecretFileError(ValueError):
+    """The secret file holds content that never validated; its content is never included."""
+
+
+def read_or_create_secret_file(
+    path: Path,
+    generate: Callable[[], str],
+    *,
+    validate: Callable[[str], object] | None = None,
+    settle_seconds: float = 1.0,
+) -> str:
+    """Return the secret stored at ``path``, creating it exclusively (mode ``0600``) when absent.
+
+    ``generate`` is called only when this process attempts the create, and its
+    value is returned only when this process wins; otherwise the value another
+    process published is returned. ``validate`` raises ``ValueError`` for an
+    incomplete or malformed value (whitespace is stripped before both checks).
+    ``OSError`` from the filesystem propagates.
+    """
+    path = Path(path)
+    deadline: float | None = None
+    claim_deadline: float | None = None
+    while True:
+        value = _read(path)
+        if value is None:
+            if _create_exclusive(path, generate):
+                # Read back what we published, or what a peer published first.
+                continue
+        elif value and _is_valid(value, validate):
+            return value
+        else:
+            if deadline is None:
+                deadline = time.monotonic() + settle_seconds
+            if time.monotonic() < deadline:
+                time.sleep(_POLL_INTERVAL_SECONDS)
+                continue
+            if value:
+                raise InvalidSecretFileError(f"{path} does not hold a valid secret; restore it from a backup or remove it")
+            if _publish_under_claim(path, generate, replace_empty=True):
+                continue
+        # A peer holds the replacement claim: wait for its value, bounded so a
+        # replacer that crashed mid-claim cannot block startup forever.
+        if claim_deadline is None:
+            claim_deadline = time.monotonic() + max(_CLAIM_WAIT_SECONDS, settle_seconds)
+        if time.monotonic() >= claim_deadline:
+            raise InvalidSecretFileError(f"{path} is missing or empty and {_claim_path(path)} was left by an interrupted creation; remove both so a new secret can be generated")
+        time.sleep(_POLL_INTERVAL_SECONDS)
+
+
+def _read(path: Path) -> str | None:
+    try:
+        return path.read_text(encoding="utf-8").strip()
+    except FileNotFoundError:
+        return None
+
+
+def _is_valid(value: str, validate: Callable[[str], object] | None) -> bool:
+    if validate is None:
+        return True
+    try:
+        validate(value)
+    except ValueError:
+        return False
+    return True
+
+
+def _write_all(fd: int, data: bytes) -> None:
+    view = memoryview(data)
+    while view:
+        written = os.write(fd, view)
+        view = view[written:]
+    os.fsync(fd)
+
+
+def _create_exclusive(path: Path, generate: Callable[[], str]) -> bool:
+    """Publish a generated value at the missing ``path`` without ever exposing it partially written.
+
+    Returns ``False`` only when the filesystem cannot hard-link and a peer holds
+    the replacement claim; otherwise the caller re-reads ``path``.
+    """
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
+    try:
+        try:
+            _write_all(fd, generate().encode("utf-8"))
+        finally:
+            os.close(fd)
+        try:
+            os.link(temporary, path)
+        except FileExistsError:
+            pass  # A peer published first.
+        except OSError as exc:
+            if exc.errno not in _NO_HARD_LINK_ERRNOS:
+                raise
+            logger.debug("%s cannot hard-link (%s); publishing through the replacement claim", path.parent, exc)
+            return _publish_under_claim(path, generate, replace_empty=False)
+        return True
+    finally:
+        Path(temporary).unlink(missing_ok=True)
+
+
+def _claim_path(path: Path) -> Path:
+    return path.with_name(f"{path.name}.replacing")
+
+
+def _publish_under_claim(path: Path, generate: Callable[[], str], *, replace_empty: bool) -> bool:
+    """Publish a generated value at ``path`` while holding the exclusive ``<name>.replacing`` claim.
+
+    ``replace_empty`` replaces an abandoned empty file; otherwise only a missing
+    name is filled (creation on filesystems without hard links). Returns
+    ``False`` only when a peer holds the claim. Holding it excludes every other
+    claimant until our rename publishes the value (and removes the claim), and
+    ``path`` is re-read under the claim, so a process that decided before a
+    peer's value landed backs off instead of overwriting a value peers may
+    already have returned. A failure removes only the claim we created.
+    """
+    claim = _claim_path(path)
+    try:
+        fd = os.open(claim, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0), 0o600)
+    except FileExistsError:
+        return False
+    published = False
+    try:
+        try:
+            if _read(path) != ("" if replace_empty else None):
+                # A peer's value landed (or the name changed state) before our claim.
+                return True
+            if replace_empty:
+                logger.warning("Replacing the empty secret file %s left by an interrupted creation", path)
+            _write_all(fd, generate().encode("utf-8"))
+        finally:
+            os.close(fd)
+        os.replace(claim, path)
+        published = True
+        return True
+    finally:
+        if not published:
+            claim.unlink(missing_ok=True)

backend/packages/harness/deerflow/persistence/channel_connections/sql.py

@@ -15,6 +15,7 @@ from sqlalchemy import delete, func, select, text, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
+from deerflow.config.credentials_key import CredentialsCipher
 from deerflow.persistence.channel_connections.model import (
     ChannelConnectionRow,
     ChannelConversationRow,
@@ -32,27 +33,29 @@ logger = logging.getLogger(__name__)
 _UPSERT_MAX_ATTEMPTS = 3
 
 
-class ChannelCredentialCipher:
-    """Encrypts provider credentials before they are persisted."""
+class ChannelCredentialCipher(CredentialsCipher):
+    """Historical constructors for the channel-credentials cipher.
+
+    Production passes the deployment cipher from
+    ``deerflow.config.credentials_key`` (``DEER_FLOW_CREDENTIALS_KEY``); this
+    subclass only keeps the earlier API: a single ``Fernet`` and ``from_key``'s
+    SHA-256 passphrase derivation. New values carry ``fernet:v2:``; the
+    ``fernet:v1:`` values it wrote before remain readable under the same key.
+    """
 
     def __init__(self, fernet: Fernet) -> None:
-        self._fernet = fernet
+        super().__init__([fernet])
 
     @classmethod
     def from_key(cls, key: str) -> ChannelCredentialCipher:
         digest = hashlib.sha256(key.encode("utf-8")).digest()
         return cls(Fernet(base64.urlsafe_b64encode(digest)))
 
-    def encrypt_text(self, value: str | None) -> str | None:
-        if value is None:
-            return None
-        return "fernet:v1:" + self._fernet.encrypt(value.encode("utf-8")).decode("ascii")
+    def encrypt_text(self, value: str | None) -> str | None:  # type: ignore[override]  # historical None passthrough
+        return None if value is None else super().encrypt_text(value)
 
-    def decrypt_text(self, value: str | None) -> str | None:
-        if value is None:
-            return None
-        token = value.removeprefix("fernet:v1:")
-        return self._fernet.decrypt(token.encode("ascii")).decode("utf-8")
+    def decrypt_text(self, value: str | None) -> str | None:  # type: ignore[override]  # historical None passthrough
+        return None if value is None else super().decrypt_text(value)
 
 
 class ChannelConnectionRepository:
@@ -62,7 +65,7 @@ class ChannelConnectionRepository:
         self,
         session_factory: async_sessionmaker[AsyncSession],
         *,
-        cipher: ChannelCredentialCipher | None = None,
+        cipher: CredentialsCipher | None = None,
     ) -> None:
         self.session_factory = session_factory
         self._cipher = cipher
@@ -93,6 +96,11 @@ class ChannelConnectionRepository:
             raise RuntimeError("channel connection encryption key is required")
         return self._cipher.encrypt_text(value)
 
+    def _decrypt_optional_secret(self, value: str | None) -> str | None:
+        if value is None or self._cipher is None:
+            return None
+        return self._cipher.decrypt_text(value)
+
     @staticmethod
     def _connection_to_dict(row: ChannelConnectionRow) -> dict[str, Any]:
         data = row.to_dict()
@@ -245,12 +253,12 @@ class ChannelConnectionRepository:
             if row is None:
                 row = ChannelCredentialRow(connection_id=connection_id)
                 session.add(row)
-            row.encrypted_access_token = self._cipher.encrypt_text(access_token)
-            row.encrypted_refresh_token = self._cipher.encrypt_text(refresh_token)
+            row.encrypted_access_token = self._encrypt_optional_secret(access_token)
+            row.encrypted_refresh_token = self._encrypt_optional_secret(refresh_token)
             row.token_type = token_type
             row.expires_at = expires_at
             row.refresh_expires_at = refresh_expires_at
-            row.encrypted_extra_json = self._cipher.encrypt_text(json.dumps(extra or {}, ensure_ascii=False))
+            row.encrypted_extra_json = self._encrypt_optional_secret(json.dumps(extra or {}, ensure_ascii=False))
             row.version = (row.version or 0) + 1
             await session.commit()
 
@@ -262,11 +270,11 @@ class ChannelConnectionRepository:
             if row is None:
                 return None
             try:
-                extra_raw = self._cipher.decrypt_text(row.encrypted_extra_json)
+                extra_raw = self._decrypt_optional_secret(row.encrypted_extra_json)
                 return {
                     "connection_id": row.connection_id,
-                    "access_token": self._cipher.decrypt_text(row.encrypted_access_token),
-                    "refresh_token": self._cipher.decrypt_text(row.encrypted_refresh_token),
+                    "access_token": self._decrypt_optional_secret(row.encrypted_access_token),
+                    "refresh_token": self._decrypt_optional_secret(row.encrypted_refresh_token),
                     "token_type": row.token_type,
                     "expires_at": self._coerce_datetime(row.expires_at),
                     "refresh_expires_at": self._coerce_datetime(row.refresh_expires_at),

backend/tests/blocking_io/test_credentials_key_offloop.py

@@ -0,0 +1,36 @@
+"""Regression anchor: the Gateway loads the credentials key off the event loop.
+
+``app.gateway.app._load_credentials_cipher`` runs inside the lifespan. Without
+``DEER_FLOW_CREDENTIALS_KEY`` it creates ``{base_dir}/.credentials_key`` (mkdir,
+exclusive create, fsync, read-back), all of which must go through
+``asyncio.to_thread``; the strict Blockbuster gate fails this test otherwise.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+# Imported at collection time: importing the Gateway app reads package
+# metadata, which is not the startup path under test.
+from app.gateway.app import _load_credentials_cipher
+from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, CREDENTIALS_KEY_FILENAME, reset_credentials_cipher
+from deerflow.config.paths import Paths
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_credentials_key_generation_does_not_block_the_event_loop(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    home = tmp_path / "missing" / "home"
+    monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False)
+    monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=home))
+    reset_credentials_cipher()
+    try:
+        cipher = await _load_credentials_cipher(SimpleNamespace(channel_connections=SimpleNamespace(enabled=True)))
+    finally:
+        reset_credentials_cipher()
+
+    assert cipher is not None
+    assert (home / CREDENTIALS_KEY_FILENAME).exists()

backend/tests/test_auth_config.py

@@ -88,3 +88,65 @@ def test_auth_config_empty_secret_file_generates_new(tmp_path):
             assert (tmp_path / ".jwt_secret").read_text().strip() == config.jwt_secret
     finally:
         cfg._auth_config = old
+
+
+def test_jwt_secret_creation_keeps_a_peer_replica_value(tmp_path, monkeypatch):
+    """Two replicas cold-starting on a shared home volume must sign with one key.
+
+    Deterministic interleaving: the peer creates ``.jwt_secret`` after this
+    process found none but before it writes. A truncating write would replace
+    the peer's value and both replicas would keep different secrets.
+    """
+    from deerflow.config.paths import Paths
+
+    secret_file = tmp_path / ".jwt_secret"
+
+    def peer_wins_the_race(_nbytes):
+        secret_file.write_text("peer-replica-secret-value-0123456789", encoding="utf-8")
+        return "this-replica-secret-value-0123456789"
+
+    monkeypatch.setattr(cfg.secrets, "token_urlsafe", peer_wins_the_race)
+    with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=tmp_path)):
+        secret = cfg._load_or_create_secret()
+
+    assert secret == "peer-replica-secret-value-0123456789"
+    assert secret_file.read_text(encoding="utf-8") == "peer-replica-secret-value-0123456789"
+
+
+def test_concurrent_jwt_secret_creators_converge(tmp_path):
+    import threading
+
+    from deerflow.config.paths import Paths
+
+    workers = 8
+    barrier = threading.Barrier(workers)
+    results: list[str] = []
+    lock = threading.Lock()
+
+    def run():
+        barrier.wait(timeout=10)
+        value = cfg._load_or_create_secret()
+        with lock:
+            results.append(value)
+
+    with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=tmp_path)):
+        threads = [threading.Thread(target=run) for _ in range(workers)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=30)
+
+    assert len(results) == workers
+    assert len(set(results)) == 1
+    assert (tmp_path / ".jwt_secret").read_text(encoding="utf-8") == results[0]
+    if os.name == "posix":
+        assert (tmp_path / ".jwt_secret").stat().st_mode & 0o777 == 0o600
+
+
+def test_unusable_jwt_secret_location_keeps_the_actionable_error(tmp_path):
+    from deerflow.config.paths import Paths
+
+    not_a_directory = tmp_path / "home"
+    not_a_directory.write_text("", encoding="utf-8")
+    with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=not_a_directory)), pytest.raises(RuntimeError, match="AUTH_JWT_SECRET"):
+        cfg._load_or_create_secret()

backend/tests/test_channel_credentials_wiring.py

@@ -0,0 +1,261 @@
+"""Per-connection channel credentials use the deployment credentials key.
+
+Before ``DEER_FLOW_CREDENTIALS_KEY`` no production construction site passed a
+cipher to ``ChannelConnectionRepository``: ``store_credentials`` refused
+(``RuntimeError``) and ``get_credentials`` always answered ``None``, so Slack
+replies always fell back to the deployment bot token. The Gateway now loads
+the cipher once at startup (off the event loop) for the features that need it
+and hands it to the channel service, the scheduled-task notification outbox
+and the channel-connections router.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from cryptography.fernet import Fernet
+
+from app.channels.message_bus import MessageBus, OutboundMessage
+from deerflow.config.credentials_key import (
+    CIPHERTEXT_PREFIX,
+    CREDENTIALS_KEY_ENV_VAR,
+    CREDENTIALS_KEY_FILENAME,
+    CredentialsCipher,
+    load_credentials_cipher,
+    reset_credentials_cipher,
+)
+from deerflow.config.paths import Paths
+
+KEY_OLD = Fernet.generate_key().decode("ascii")
+KEY_NEW = Fernet.generate_key().decode("ascii")
+
+
+@pytest.fixture(autouse=True)
+def _isolated_cipher(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
+    monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False)
+    monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path / "home"))
+    reset_credentials_cipher()
+    yield
+    reset_credentials_cipher()
+
+
+def _startup_config(*, channel_connections: bool) -> SimpleNamespace:
+    return SimpleNamespace(channel_connections=SimpleNamespace(enabled=channel_connections))
+
+
+@pytest.fixture
+async def session_factory(tmp_path: Path):
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+
+    await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'channels.db'}", sqlite_dir=str(tmp_path))
+    try:
+        yield get_session_factory()
+    finally:
+        await close_engine()
+
+
+async def _connection(repo) -> str:
+    connection = await repo.upsert_connection(owner_user_id="alice", provider="slack", external_account_id="U-alice", workspace_id="T1")
+    return connection["id"]
+
+
+# ---------------------------------------------------------------------------
+# Startup loading
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_no_key_is_loaded_or_generated_when_no_feature_needs_it(tmp_path: Path) -> None:
+    from app.gateway.app import _load_credentials_cipher
+
+    assert await _load_credentials_cipher(_startup_config(channel_connections=False)) is None
+    assert not (tmp_path / "home" / CREDENTIALS_KEY_FILENAME).exists()
+
+
+@pytest.mark.anyio
+async def test_channel_connections_load_the_generated_key_file(tmp_path: Path) -> None:
+    from app.gateway.app import _load_credentials_cipher
+
+    cipher = await _load_credentials_cipher(_startup_config(channel_connections=True))
+
+    assert isinstance(cipher, CredentialsCipher)
+    key_file = tmp_path / "home" / CREDENTIALS_KEY_FILENAME
+    assert key_file.exists()
+    assert CredentialsCipher.from_keys([key_file.read_text(encoding="utf-8").strip()]).decrypt_text(cipher.encrypt_text("x")) == "x"
+
+
+@pytest.mark.anyio
+async def test_channel_connections_prefer_the_environment_key(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    from app.gateway.app import _load_credentials_cipher
+
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, KEY_NEW)
+
+    cipher = await _load_credentials_cipher(_startup_config(channel_connections=True))
+
+    assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(cipher.encrypt_text("x")) == "x"
+    assert not (tmp_path / "home" / CREDENTIALS_KEY_FILENAME).exists()
+
+
+@pytest.mark.anyio
+async def test_an_unloadable_key_degrades_to_no_stored_credentials(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
+    """A broken key file must not take the Gateway down: credentials stay unavailable, as before the key existed."""
+    from app.gateway import app as gateway_app
+
+    def broken_loader():
+        raise OSError("read-only file system")
+
+    monkeypatch.setattr("deerflow.config.credentials_key.get_credentials_cipher", broken_loader)
+    with caplog.at_level(logging.ERROR, logger=gateway_app.logger.name):
+        assert await gateway_app._load_credentials_cipher(_startup_config(channel_connections=True)) is None
+    assert any("credentials" in record.getMessage().lower() for record in caplog.records)
+
+
+# ---------------------------------------------------------------------------
+# Construction sites
+# ---------------------------------------------------------------------------
+
+
+def test_the_channel_service_repository_gets_the_cipher(monkeypatch: pytest.MonkeyPatch) -> None:
+    from app.channels.service import _make_connection_repo
+
+    sentinel_factory = MagicMock(name="session_factory")
+    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: sentinel_factory)
+    cipher = CredentialsCipher.from_keys([KEY_NEW])
+
+    repo = _make_connection_repo(SimpleNamespace(enabled=True), credentials_cipher=cipher)
+
+    assert repo is not None
+    assert repo.session_factory is sentinel_factory
+    assert repo._cipher is cipher
+    assert _make_connection_repo(SimpleNamespace(enabled=True))._cipher is None, "callers that pass no cipher keep the old behaviour"
+
+
+def test_the_notification_outbox_repository_gets_the_cipher(monkeypatch: pytest.MonkeyPatch) -> None:
+    from app.gateway.app import _scheduled_task_notification_repos
+
+    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: MagicMock(name="session_factory"))
+    cipher = CredentialsCipher.from_keys([KEY_NEW])
+
+    connection_repo, notification_repo = _scheduled_task_notification_repos(_startup_config(channel_connections=True), cipher)
+
+    assert connection_repo._cipher is cipher
+    assert notification_repo is not None
+
+
+def test_the_router_repository_gets_the_cipher_from_app_state(monkeypatch: pytest.MonkeyPatch) -> None:
+    from app.gateway.routers import channel_connections
+
+    monkeypatch.setattr(channel_connections, "get_session_factory", lambda: MagicMock(name="session_factory"))
+    cipher = CredentialsCipher.from_keys([KEY_NEW])
+    request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(credentials_cipher=cipher)))
+
+    repo = channel_connections._get_repository(request, MagicMock())
+
+    assert repo._cipher is cipher
+    assert request.app.state.channel_connection_repo is repo
+
+
+def test_start_channel_service_threads_the_cipher_through(monkeypatch: pytest.MonkeyPatch) -> None:
+    import anyio
+
+    import app.channels.service as service_module
+
+    captured: dict[str, object] = {}
+
+    class _FakeService:
+        async def start(self):
+            return None
+
+    def fake_from_app_config(app_config=None, *, get_stream_bridge=None, credentials_cipher=None):
+        captured["credentials_cipher"] = credentials_cipher
+        return _FakeService()
+
+    cipher = CredentialsCipher.from_keys([KEY_NEW])
+    monkeypatch.setattr(service_module.ChannelService, "from_app_config", staticmethod(fake_from_app_config))
+    monkeypatch.setattr(service_module, "_channel_service", None)
+
+    anyio.run(lambda: service_module.start_channel_service(credentials_cipher=cipher))
+
+    assert captured["credentials_cipher"] is cipher
+
+
+# ---------------------------------------------------------------------------
+# Repository behaviour under the deployment key
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_credentials_round_trip_under_the_deployment_key(session_factory, tmp_path: Path) -> None:
+    from sqlalchemy import select
+
+    from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialRow
+
+    repo = ChannelConnectionRepository(session_factory, cipher=load_credentials_cipher(base_dir=tmp_path))
+    connection_id = await _connection(repo)
+
+    await repo.store_credentials(connection_id, access_token="xoxb-connection-token", extra={"bot_user_id": "B1"})
+
+    async with session_factory() as session:
+        row = (await session.execute(select(ChannelCredentialRow))).scalar_one()
+    assert row.encrypted_access_token.startswith(CIPHERTEXT_PREFIX)
+    assert "xoxb-connection-token" not in row.encrypted_access_token
+
+    restarted = ChannelConnectionRepository(session_factory, cipher=load_credentials_cipher(base_dir=tmp_path))
+    credentials = await restarted.get_credentials(connection_id)
+    assert credentials is not None
+    assert credentials["access_token"] == "xoxb-connection-token"
+    assert credentials["extra"] == {"bot_user_id": "B1"}
+
+
+@pytest.mark.anyio
+async def test_rotation_keeps_rows_written_under_the_previous_key_readable(session_factory) -> None:
+    from deerflow.persistence.channel_connections import ChannelConnectionRepository
+
+    before = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_OLD]))
+    connection_id = await _connection(before)
+    await before.store_credentials(connection_id, access_token="written-before-rotation")
+
+    after = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_NEW, KEY_OLD]))
+
+    assert (await after.get_credentials(connection_id))["access_token"] == "written-before-rotation"
+
+
+@pytest.mark.anyio
+async def test_a_changed_key_treats_stored_credentials_as_missing(session_factory, caplog: pytest.LogCaptureFixture) -> None:
+    from deerflow.persistence.channel_connections import ChannelConnectionRepository
+
+    before = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_OLD]))
+    connection_id = await _connection(before)
+    await before.store_credentials(connection_id, access_token="unreadable-after-key-loss")
+
+    after = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_NEW]))
+    with caplog.at_level(logging.WARNING, logger="deerflow.persistence.channel_connections.sql"):
+        assert await after.get_credentials(connection_id) is None
+    assert any("Unable to decrypt channel connection credentials" in record.getMessage() for record in caplog.records)
+    assert not any("unreadable-after-key-loss" in record.getMessage() for record in caplog.records)
+
+
+@pytest.mark.anyio
+async def test_slack_falls_back_to_the_deployment_token_without_stored_credentials(session_factory) -> None:
+    from app.channels.slack import SlackChannel
+    from deerflow.persistence.channel_connections import ChannelConnectionRepository
+
+    repo = ChannelConnectionRepository(session_factory, cipher=CredentialsCipher.from_keys([KEY_NEW]))
+    connection_id = await _connection(repo)
+    operator_client = MagicMock(name="operator_client")
+    factory = MagicMock(name="web_client_factory")
+    channel = SlackChannel(bus=MessageBus(), config={"connection_repo": repo, "web_client_factory": factory})
+    channel._web_client = operator_client
+    message = OutboundMessage(channel_name="slack", chat_id="C1", thread_id="t1", text="hi", connection_id=connection_id)
+
+    assert await channel._get_web_client_for_message(message) is operator_client
+    factory.assert_not_called()
+
+    await repo.store_credentials(connection_id, access_token="xoxb-connection-token")
+    connection_client = await channel._get_web_client_for_message(message)
+    factory.assert_called_once_with(token="xoxb-connection-token")
+    assert connection_client is factory.return_value

backend/tests/test_channels.py

@@ -6513,7 +6513,7 @@ class TestGithubFollowupBuffer:
             def get_status(self):
                 return {}
 
-        def fake_from_app_config(app_config=None, *, get_stream_bridge=None):
+        def fake_from_app_config(app_config=None, *, get_stream_bridge=None, credentials_cipher=None):
             captured["get_stream_bridge"] = get_stream_bridge
             return _FakeService()
 

backend/tests/test_compose_credentials_key.py

@@ -0,0 +1,104 @@
+"""Both compose stacks pass ``DEER_FLOW_CREDENTIALS_KEY`` through to the Gateway.
+
+``scripts/deploy.sh`` resolves the key (shell, ``.env``, the persisted
+``$DEER_FLOW_HOME/.credentials_key``, or a freshly generated one) and Compose
+interpolates it into the gateway service. ``:-`` keeps a stack started without
+the deploy script quiet: an empty value makes the Gateway fall back to the same
+``.credentials_key`` file under its runtime home, which is the host's
+``DEER_FLOW_HOME`` bind mount.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+import yaml
+from support.compose import DOCKER, requires_docker_compose
+from support.shell import find_script_bash
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+COMPOSE_PATHS = {
+    "prod": REPO_ROOT / "docker" / "docker-compose.yaml",
+    "dev": REPO_ROOT / "docker" / "docker-compose-dev.yaml",
+}
+EXPECTED_ENTRY = "DEER_FLOW_CREDENTIALS_KEY=${DEER_FLOW_CREDENTIALS_KEY:-}"
+BASH_EXECUTABLE = find_script_bash()
+DOTENV_KEY = "u6Tp1Gk0bq8m6C2YkS9oWw7xH3fLr5eN1aZdQjVtXc4="
+SHELL_KEY = "Zq3Lw8Rn2Tb6Yv1Kc5Hm9Pd4Sx7Ga0Je3Uf6Ni2Ro8M="
+
+
+@pytest.mark.parametrize("variant", sorted(COMPOSE_PATHS))
+def test_gateway_receives_the_credentials_key(variant: str) -> None:
+    services = yaml.safe_load(COMPOSE_PATHS[variant].read_text(encoding="utf-8"))["services"]
+
+    assert EXPECTED_ENTRY in services["gateway"]["environment"]
+    assert "DEER_FLOW_HOME=/app/backend/.deer-flow" in services["gateway"]["environment"], "the fallback key file lives in the runtime home"
+
+
+@pytest.mark.parametrize("variant", sorted(COMPOSE_PATHS))
+def test_only_the_gateway_receives_the_credentials_key(variant: str) -> None:
+    services = yaml.safe_load(COMPOSE_PATHS[variant].read_text(encoding="utf-8"))["services"]
+    for name, service in services.items():
+        if name == "gateway":
+            continue
+        environment = service.get("environment") or []
+        entries = environment if isinstance(environment, list) else [f"{key}={value}" for key, value in environment.items()]
+        assert not any(str(entry).startswith("DEER_FLOW_CREDENTIALS_KEY") for entry in entries), name
+
+
+def _render_through_docker_start(tmp_path: Path, env_file: str, shell_value: str | None) -> dict[str, str]:
+    """Render the dev gateway environment the way ``make docker-start`` interpolates it.
+
+    ``scripts/docker.sh start`` runs Compose from ``docker/`` without
+    ``--env-file``; a ``docker`` shell function swaps ``up ...`` for ``config``.
+    """
+    docker_dir = tmp_path / "docker"
+    shutil.copytree(REPO_ROOT / "docker", docker_dir)
+    (tmp_path / "frontend").mkdir()
+    (tmp_path / "frontend" / ".env").write_text("", encoding="utf-8")
+    (tmp_path / ".env").write_text(env_file, encoding="utf-8")
+    (tmp_path / "config.yaml").write_text("sandbox:\n  use: deerflow.sandbox.local:LocalSandboxProvider\n", encoding="utf-8")
+    (tmp_path / "extensions_config.json").write_text("{}", encoding="utf-8")
+    rendered = tmp_path / "rendered.json"
+    env = {key: value for key, value in os.environ.items() if not key.startswith(("DEER_FLOW_", "COMPOSE_", "AUTH_"))}
+    if shell_value is not None:
+        env["DEER_FLOW_CREDENTIALS_KEY"] = shell_value
+    script = f"""
+source '{REPO_ROOT / "scripts" / "docker.sh"}'
+PROJECT_ROOT='{tmp_path}'
+DOCKER_DIR='{docker_dir}'
+require_compose_version() {{ :; }}
+docker() {{
+    local args=()
+    for arg in "$@"; do
+        [ "$arg" = up ] && break
+        args+=("$arg")
+    done
+    command '{DOCKER}' "${{args[@]}}" config --format json > '{rendered}'
+}}
+start
+"""
+    subprocess.run([BASH_EXECUTABLE, "-c", script], env=env, capture_output=True, text=True, timeout=120, check=True)
+    return json.loads(rendered.read_text(encoding="utf-8"))["services"]["gateway"]["environment"]
+
+
+@requires_docker_compose
+@pytest.mark.skipif(BASH_EXECUTABLE is None, reason="bash is required to run scripts/docker.sh")
+@pytest.mark.parametrize(
+    ("env_file", "shell_value", "expected"),
+    [
+        ("", None, ""),
+        (f"DEER_FLOW_CREDENTIALS_KEY={DOTENV_KEY}\n", None, DOTENV_KEY),
+        (f"DEER_FLOW_CREDENTIALS_KEY='{DOTENV_KEY}'\r\n", None, DOTENV_KEY),
+        (f"DEER_FLOW_CREDENTIALS_KEY={DOTENV_KEY}\n", SHELL_KEY, SHELL_KEY),
+    ],
+    ids=["unset", "dotenv", "quoted-crlf", "shell-export-wins"],
+)
+def test_make_docker_start_keeps_the_dotenv_credentials_key(tmp_path, env_file: str, shell_value: str | None, expected: str):
+    """The dev launcher must hand the checkout .env key to interpolation; an empty override would replace it."""
+    assert _render_through_docker_start(tmp_path, env_file, shell_value)["DEER_FLOW_CREDENTIALS_KEY"] == expected

backend/tests/test_credentials_key.py

@@ -0,0 +1,254 @@
+"""At-rest credentials key ``DEER_FLOW_CREDENTIALS_KEY`` and the cipher built from it.
+
+The key is env-only: one or more comma-separated urlsafe-base64 Fernet keys,
+the first encrypting and every one decrypting, so an operator rotates by
+prepending a new key. Without it a single instance generates
+``{base_dir}/.credentials_key`` once. It is never derived from
+``AUTH_JWT_SECRET``.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.fernet import Fernet, InvalidToken
+
+from deerflow.config import credentials_key as ck
+from deerflow.config.credentials_key import (
+    CIPHERTEXT_PREFIX,
+    CREDENTIALS_KEY_ENV_VAR,
+    CREDENTIALS_KEY_FILENAME,
+    GENERATE_KEY_COMMAND,
+    CredentialsCipher,
+    CredentialsDecryptError,
+    CredentialsKeyError,
+    get_credentials_cipher,
+    load_credentials_cipher,
+    parse_credentials_keys,
+    reset_credentials_cipher,
+)
+from deerflow.persistence.channel_connections import ChannelCredentialCipher
+
+KEY_OLD = Fernet.generate_key().decode("ascii")
+KEY_NEW = Fernet.generate_key().decode("ascii")
+
+
+@pytest.fixture(autouse=True)
+def _isolated_key_env(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False)
+    reset_credentials_cipher()
+    yield
+    reset_credentials_cipher()
+
+
+def _std_base64_key() -> str:
+    """A 32-byte key spelled with the standard alphabet (``+``/``/``), as Helm's randBytes emits it."""
+    raw = bytes([0xFB, 0xFF] * 16)
+    value = base64.b64encode(raw).decode("ascii")
+    assert "+" in value or "/" in value
+    return value
+
+
+# ---------------------------------------------------------------------------
+# Parsing
+# ---------------------------------------------------------------------------
+
+
+def test_parse_accepts_a_single_key() -> None:
+    assert parse_credentials_keys(KEY_NEW) == [KEY_NEW]
+
+
+def test_parse_keeps_rotation_order_and_ignores_blank_entries() -> None:
+    assert parse_credentials_keys(f" {KEY_NEW} , {KEY_OLD} ,") == [KEY_NEW, KEY_OLD]
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [
+        "",
+        " , ",
+        "not-a-fernet-key",
+        KEY_NEW[:-1],
+        KEY_NEW + "A",
+        _std_base64_key(),
+        "é" * 44,
+        f"{KEY_NEW},short-second-key",
+    ],
+    ids=["empty", "only-separators", "garbage", "truncated", "too-long", "standard-alphabet", "non-ascii", "second-entry-bad"],
+)
+def test_parse_rejects_malformed_keys_without_echoing_them(raw: str) -> None:
+    with pytest.raises(CredentialsKeyError) as exc_info:
+        parse_credentials_keys(raw)
+
+    message = str(exc_info.value)
+    assert CREDENTIALS_KEY_ENV_VAR in message
+    assert GENERATE_KEY_COMMAND in message, "the refusal must say how to generate a valid key"
+    for entry in filter(None, (part.strip() for part in raw.split(","))):
+        assert entry not in message, "key material must never reach logs or refusal text"
+    assert exc_info.value.__cause__ is None
+
+
+# ---------------------------------------------------------------------------
+# Cipher
+# ---------------------------------------------------------------------------
+
+
+def test_cipher_writes_a_versioned_prefix_and_round_trips() -> None:
+    cipher = CredentialsCipher.from_keys([KEY_NEW])
+
+    token = cipher.encrypt_text("xoxb-secret")
+
+    assert token.startswith(CIPHERTEXT_PREFIX)
+    assert CIPHERTEXT_PREFIX == "fernet:v2:"
+    assert "xoxb-secret" not in token
+    assert cipher.decrypt_text(token) == "xoxb-secret"
+
+
+def test_the_first_key_encrypts_and_every_key_decrypts() -> None:
+    old_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("written-before-rotation")
+    rotated = CredentialsCipher.from_keys([KEY_NEW, KEY_OLD])
+
+    assert rotated.decrypt_text(old_token) == "written-before-rotation"
+
+    new_token = rotated.encrypt_text("written-after-rotation")
+    assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(new_token) == "written-after-rotation"
+    with pytest.raises(CredentialsDecryptError):
+        CredentialsCipher.from_keys([KEY_OLD]).decrypt_text(new_token)
+
+
+def test_rotate_re_encrypts_under_the_primary_key() -> None:
+    old_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("long-lived-token")
+    rotated = CredentialsCipher.from_keys([KEY_NEW, KEY_OLD])
+
+    re_encrypted = rotated.rotate_text(old_token)
+
+    assert re_encrypted.startswith(CIPHERTEXT_PREFIX)
+    assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(re_encrypted) == "long-lived-token"
+
+
+def test_an_undecryptable_value_raises_an_invalid_token_subclass() -> None:
+    token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("secret")
+    cipher = CredentialsCipher.from_keys([KEY_NEW])
+
+    with pytest.raises(CredentialsDecryptError) as exc_info:
+        cipher.decrypt_text(token)
+    assert isinstance(exc_info.value, InvalidToken), "existing `except InvalidToken` call sites keep working"
+
+    for garbage in ("fernet:v2:not-a-token", "fernet:v2:é", "plain text"):
+        with pytest.raises(CredentialsDecryptError):
+            cipher.decrypt_text(garbage)
+
+
+def test_legacy_v1_values_written_by_the_passphrase_cipher_stay_readable() -> None:
+    """Exactly what ``ChannelCredentialCipher.from_key`` wrote before the v2 format."""
+    derived = Fernet(base64.urlsafe_b64encode(hashlib.sha256(b"legacy-passphrase").digest()))
+    legacy_value = "fernet:v1:" + derived.encrypt(b"xoxb-legacy").decode("ascii")
+
+    cipher = ChannelCredentialCipher.from_key("legacy-passphrase")
+
+    assert cipher.decrypt_text(legacy_value) == "xoxb-legacy"
+    assert cipher.decrypt_text(legacy_value.removeprefix("fernet:v1:")) == "xoxb-legacy", "the old reader accepted a bare token"
+    assert cipher.encrypt_text("new-value").startswith(CIPHERTEXT_PREFIX)
+    assert isinstance(cipher, CredentialsCipher)
+
+
+def test_v1_values_under_a_raw_key_are_readable_by_the_deployment_cipher() -> None:
+    """``ChannelCredentialCipher(Fernet(key))`` wrote ``fernet:v1:`` under a raw key; the same key in the env reads it."""
+    legacy_value = "fernet:v1:" + Fernet(KEY_OLD).encrypt(b"value").decode("ascii")
+
+    assert ChannelCredentialCipher(Fernet(KEY_OLD)).decrypt_text(legacy_value) == "value"
+    assert CredentialsCipher.from_keys([KEY_NEW, KEY_OLD]).decrypt_text(legacy_value) == "value"
+
+
+# ---------------------------------------------------------------------------
+# Loading
+# ---------------------------------------------------------------------------
+
+
+def test_the_environment_key_list_wins_and_no_file_is_written(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, f"{KEY_NEW},{KEY_OLD}")
+    old_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("old")
+
+    cipher = load_credentials_cipher(base_dir=tmp_path)
+
+    assert cipher.decrypt_text(old_token) == "old"
+    assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(cipher.encrypt_text("new")) == "new"
+    assert not (tmp_path / CREDENTIALS_KEY_FILENAME).exists()
+
+
+def test_a_single_instance_generates_the_key_file_once(tmp_path: Path) -> None:
+    first = load_credentials_cipher(base_dir=tmp_path)
+    key_file = tmp_path / CREDENTIALS_KEY_FILENAME
+
+    assert key_file.exists()
+    persisted = key_file.read_text(encoding="utf-8").strip()
+    assert parse_credentials_keys(persisted) == [persisted]
+    if os.name == "posix":
+        assert key_file.stat().st_mode & 0o777 == 0o600
+
+    token = first.encrypt_text("survives a restart")
+    assert load_credentials_cipher(base_dir=tmp_path).decrypt_text(token) == "survives a restart"
+    assert key_file.read_text(encoding="utf-8").strip() == persisted
+
+
+def test_a_blank_environment_value_means_unset(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """Compose renders ``${DEER_FLOW_CREDENTIALS_KEY:-}`` as an empty string."""
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, "  ")
+
+    load_credentials_cipher(base_dir=tmp_path)
+
+    assert (tmp_path / CREDENTIALS_KEY_FILENAME).exists()
+
+
+def test_a_malformed_environment_key_is_refused_before_touching_disk(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, "definitely-not-a-key")
+
+    with pytest.raises(CredentialsKeyError):
+        load_credentials_cipher(base_dir=tmp_path)
+
+    assert not (tmp_path / CREDENTIALS_KEY_FILENAME).exists()
+
+
+def test_a_corrupt_key_file_is_refused_and_left_alone(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """The file may be the only copy of the key that encrypted stored credentials."""
+    monkeypatch.setattr(ck, "_KEY_FILE_SETTLE_SECONDS", 0.05)
+    key_file = tmp_path / CREDENTIALS_KEY_FILENAME
+    key_file.write_text("corrupted-key-material", encoding="utf-8")
+
+    with pytest.raises(CredentialsKeyError) as exc_info:
+        load_credentials_cipher(base_dir=tmp_path)
+
+    assert str(key_file) in str(exc_info.value)
+    assert "corrupted-key-material" not in str(exc_info.value)
+    assert key_file.read_text(encoding="utf-8") == "corrupted-key-material"
+
+
+def test_the_key_is_never_derived_from_the_jwt_secret(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("AUTH_JWT_SECRET", KEY_OLD)
+    jwt_derived_token = CredentialsCipher.from_keys([KEY_OLD]).encrypt_text("secret")
+
+    cipher = load_credentials_cipher(base_dir=tmp_path)
+
+    with pytest.raises(CredentialsDecryptError):
+        cipher.decrypt_text(jwt_derived_token)
+    assert (tmp_path / CREDENTIALS_KEY_FILENAME).read_text(encoding="utf-8").strip() != KEY_OLD
+
+
+def test_the_process_cipher_is_loaded_once_until_reset(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    from deerflow.config.paths import Paths
+
+    monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, KEY_OLD)
+
+    first = get_credentials_cipher()
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, KEY_NEW)
+    assert get_credentials_cipher() is first
+
+    reset_credentials_cipher()
+    reloaded = get_credentials_cipher()
+    assert reloaded is not first
+    assert CredentialsCipher.from_keys([KEY_NEW]).decrypt_text(reloaded.encrypt_text("x")) == "x"

backend/tests/test_deploy_dotenv_secrets.py

@@ -31,12 +31,16 @@ REPO_ROOT = Path(__file__).resolve().parents[2]
 BASH = find_script_bash()
 pytestmark = pytest.mark.skipif(BASH is None, reason="repo shell-script tests need Git Bash on Windows")
 
-SECRETS = ("BETTER_AUTH_SECRET", "DEER_FLOW_INTERNAL_AUTH_TOKEN")
+SECRETS = ("BETTER_AUTH_SECRET", "DEER_FLOW_INTERNAL_AUTH_TOKEN", "DEER_FLOW_CREDENTIALS_KEY")
 PERSISTED_FILE = {
     "BETTER_AUTH_SECRET": ".better-auth-secret",
     "DEER_FLOW_INTERNAL_AUTH_TOKEN": ".internal-auth-token",
+    # Same name the Gateway itself generates under its runtime home, which is
+    # this directory's bind mount: one key whichever side created it first.
+    "DEER_FLOW_CREDENTIALS_KEY": ".credentials_key",
 }
-GENERATED = re.compile(r"set:[A-Za-z0-9_\-]{32,}")
+# A Fernet key ends in one "=" of base64 padding.
+GENERATED = re.compile(r"set:[A-Za-z0-9_\-]{32,}=?")
 
 # The fake docker answers `compose ... config` the way Compose renders the
 # script's stub project: it reads the `${KEY}` reference off stdin, looks the
@@ -65,6 +69,7 @@ esac
 {
   printf 'BETTER_AUTH_SECRET=%s\\n' "${BETTER_AUTH_SECRET+set:}${BETTER_AUTH_SECRET:-}"
   printf 'DEER_FLOW_INTERNAL_AUTH_TOKEN=%s\\n' "${DEER_FLOW_INTERNAL_AUTH_TOKEN+set:}${DEER_FLOW_INTERNAL_AUTH_TOKEN:-}"
+  printf 'DEER_FLOW_CREDENTIALS_KEY=%s\\n' "${DEER_FLOW_CREDENTIALS_KEY+set:}${DEER_FLOW_CREDENTIALS_KEY:-}"
 } > "$CAPTURE_SECRETS"
 for arg in "$@"; do printf "%s\\n" "$arg"; done > "$CAPTURE_DOCKER_ARGS"
 exit 0
@@ -164,7 +169,8 @@ def test_deploy_asks_compose_to_interpolate_the_secret_like_the_real_project(tmp
     # The stub project on stdin is what gets interpolated: it must reference
     # the secret and nothing from the real compose file.
     assert config_args[config_args.index("-f") + 1] == "-"
-    assert "${DEER_FLOW_INTERNAL_AUTH_TOKEN}" in config_stdin
+    # Each probe replaces the captured stub; the last secret resolved is the last one probed.
+    assert f"${{{SECRETS[-1]}}}" in config_stdin
     assert "docker-compose.yaml" not in config_stdin
 
 
@@ -384,3 +390,36 @@ def test_real_compose_empty_shell_export_still_gets_a_generated_secret(tmp_path,
     _, observed, _, _, _, _ = _run_deploy_build(tmp_path, worktree, real_docker=real_docker, shell_env={key: ""})
 
     assert GENERATED.fullmatch(observed[key]), observed[key]
+
+
+# ── DEER_FLOW_CREDENTIALS_KEY specifics ─────────────────────────────────────
+
+
+def test_deploy_generates_a_valid_fernet_credentials_key(tmp_path):
+    """The Gateway refuses a malformed key, so the generated one must parse as Fernet."""
+    from cryptography.fernet import Fernet
+
+    from deerflow.config.credentials_key import parse_credentials_keys
+
+    worktree = _worktree(tmp_path)
+
+    _, observed, _, _, _, home = _run_deploy_build(tmp_path, worktree, compose_environment={})
+
+    key = observed["DEER_FLOW_CREDENTIALS_KEY"].removeprefix("set:")
+    Fernet(key)
+    assert parse_credentials_keys(key) == [key]
+    assert (home / ".credentials_key").read_text(encoding="utf-8").strip() == key
+
+
+def test_deploy_reuses_a_key_file_the_gateway_already_generated(tmp_path):
+    """A stack first started without deploy.sh already has the Gateway's own key file: keep it."""
+    worktree = _worktree(tmp_path)
+    home = tmp_path / "deer-flow-home"
+    home.mkdir()
+    (home / ".credentials_key").write_text("gateway-generated-key", encoding="utf-8")
+
+    result, observed, _, _, _, _ = _run_deploy_build(tmp_path, worktree, compose_environment={})
+
+    assert observed["DEER_FLOW_CREDENTIALS_KEY"] == "set:gateway-generated-key"
+    assert (home / ".credentials_key").read_text(encoding="utf-8") == "gateway-generated-key"
+    assert "DEER_FLOW_CREDENTIALS_KEY loaded from" in result.stdout

backend/tests/test_deploy_home_writability.py

@@ -51,6 +51,7 @@ def deploy_fixture(tmp_path: Path):
         # Keep the preflight independent of secret generation and optional extras.
         BETTER_AUTH_SECRET="test-better-auth-secret",
         DEER_FLOW_INTERNAL_AUTH_TOKEN="test-internal-auth-token",
+        DEER_FLOW_CREDENTIALS_KEY="test-credentials-key",
         UV_EXTRAS="redis",
     )
     return worktree, env, home, capture
@@ -76,8 +77,12 @@ def unwritable_home(deploy_fixture):
 
 
 @pytest.fixture(
-    params=[("BETTER_AUTH_SECRET", ".better-auth-secret"), ("DEER_FLOW_INTERNAL_AUTH_TOKEN", ".internal-auth-token")],
-    ids=["better-auth", "internal-auth"],
+    params=[
+        ("BETTER_AUTH_SECRET", ".better-auth-secret"),
+        ("DEER_FLOW_INTERNAL_AUTH_TOKEN", ".internal-auth-token"),
+        ("DEER_FLOW_CREDENTIALS_KEY", ".credentials_key"),
+    ],
+    ids=["better-auth", "internal-auth", "credentials-key"],
 )
 def persisted_secret(request, deploy_fixture):
     _, _, home, _ = deploy_fixture
@@ -150,6 +155,7 @@ def test_deploy_down_allows_unwritable_home(deploy_fixture, unwritable_home):
     _, env, _, capture = deploy_fixture
     env.pop("BETTER_AUTH_SECRET")
     env.pop("DEER_FLOW_INTERNAL_AUTH_TOKEN")
+    env.pop("DEER_FLOW_CREDENTIALS_KEY")
 
     result = _run_deploy(deploy_fixture, "down")
 
@@ -226,6 +232,7 @@ def test_deploy_down_allows_unreadable_persisted_secret(deploy_fixture, unreadab
     _, env, home, capture = deploy_fixture
     env.pop("BETTER_AUTH_SECRET")
     env.pop("DEER_FLOW_INTERNAL_AUTH_TOKEN")
+    env.pop("DEER_FLOW_CREDENTIALS_KEY")
 
     result = _run_deploy(deploy_fixture, "down")
 

backend/tests/test_docker_sandbox_mode_detection.py

@@ -572,6 +572,7 @@ def test_aio_deploy_socket_preflight_allows_windows_when_docker_reachable(tmp_pa
     env["DEER_FLOW_DOCKER_SOCKET"] = "/var/run/docker.sock"
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(
@@ -597,6 +598,7 @@ def test_aio_deploy_socket_preflight_rejects_missing_socket_on_posix(tmp_path):
     env["DEER_FLOW_DOCKER_SOCKET"] = "/nonexistent/docker.sock"
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(
@@ -632,6 +634,7 @@ def test_aio_deploy_socket_preflight_rejects_missing_custom_socket_on_windows(tm
     env["DEER_FLOW_DOCKER_SOCKET"] = "/nonexistent/docker.sock"
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(
@@ -668,6 +671,7 @@ def test_aio_deploy_socket_preflight_rejects_windows_when_docker_unreachable(tmp
     env["DEER_FLOW_DOCKER_SOCKET"] = "/var/run/docker.sock"
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(
@@ -704,6 +708,7 @@ def test_aio_deploy_socket_unsets_default_on_windows(tmp_path):
     env["DEER_FLOW_DOCKER_SOCKET"] = "/var/run/docker.sock"
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(
@@ -740,6 +745,7 @@ def test_aio_deploy_socket_preserves_unset_default_on_windows(tmp_path):
     env.pop("DEER_FLOW_DOCKER_SOCKET", None)
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(
@@ -779,6 +785,7 @@ def test_aio_deploy_socket_preserves_custom_socket_on_windows(tmp_path):
     env["DEER_FLOW_DOCKER_SOCKET"] = "/custom/docker.sock"
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(
@@ -816,6 +823,7 @@ def test_aio_deploy_socket_reads_from_dotenv(tmp_path):
     env.pop("DEER_FLOW_DOCKER_SOCKET", None)
     env["BETTER_AUTH_SECRET"] = "test-secret"
     env["DEER_FLOW_INTERNAL_AUTH_TOKEN"] = "test-token"
+    env["DEER_FLOW_CREDENTIALS_KEY"] = "test-credentials-key"
     env["UV_EXTRAS"] = "redis"
 
     result = subprocess.run(

backend/tests/test_helm_credentials_key.py

@@ -0,0 +1,106 @@
+"""The Helm chart generates and injects ``DEER_FLOW_CREDENTIALS_KEY``.
+
+Every gateway Pod must encrypt and decrypt stored credentials with one key, so
+the chart generates a Fernet key into the app Secret once, preserves it across
+upgrades through ``lookup`` like ``AUTH_JWT_SECRET``, and injects it into the
+gateway. Sprig's ``randBytes`` emits standard base64; Fernet keys use the
+urlsafe alphabet, so the template translates ``+``/``/`` to ``-``/``_``.
+
+The ``helm template`` tests skip when helm is not installed; CI's runner has it.
+"""
+
+from __future__ import annotations
+
+import base64
+import re
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+import yaml
+from cryptography.fernet import Fernet
+
+from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, parse_credentials_keys
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+CHART = REPO_ROOT / "deploy" / "helm" / "deer-flow"
+APP_SECRET_TEMPLATE = CHART / "templates" / "secret-app.yaml"
+README = CHART / "README.md"
+NOTES = CHART / "templates" / "NOTES.txt"
+VALUES = CHART / "values.yaml"
+
+_FERNET_KEY = re.compile(r"[A-Za-z0-9_-]{43}=")
+
+
+def _render_chart(*settings: str) -> list[dict]:
+    helm = shutil.which("helm")
+    if helm is None:
+        pytest.skip("helm is unavailable")
+    command = [helm, "template", "deer-flow", str(CHART)]
+    for setting in settings:
+        command.extend(["--set", setting])
+    rendered = subprocess.run(command, check=True, capture_output=True, text=True).stdout
+    return [document for document in yaml.safe_load_all(rendered) if isinstance(document, dict)]
+
+
+def _app_secret(documents: list[dict]) -> dict:
+    return next(document for document in documents if document.get("kind") == "Secret" and document["metadata"]["name"].endswith("-app"))
+
+
+def _gateway(documents: list[dict]) -> dict:
+    deployment = next(document for document in documents if document.get("kind") == "Deployment" and document["metadata"]["name"].endswith("-gateway"))
+    return next(container for container in deployment["spec"]["template"]["spec"]["containers"] if container["name"] == "gateway")
+
+
+def _env(container: dict) -> dict[str, dict]:
+    return {item["name"]: item for item in container["env"]}
+
+
+def test_rendered_app_secret_carries_a_valid_fernet_key() -> None:
+    key = _app_secret(_render_chart())["stringData"][CREDENTIALS_KEY_ENV_VAR]
+
+    assert _FERNET_KEY.fullmatch(key), "must be urlsafe base64 of 32 bytes"
+    assert len(base64.urlsafe_b64decode(key)) == 32
+    Fernet(key)
+    assert parse_credentials_keys(key) == [key], "the Gateway's own parser must accept the generated key"
+
+
+def test_each_fresh_render_generates_an_independent_key() -> None:
+    """randBytes, not a constant: two installs must not share a key."""
+    first = _app_secret(_render_chart())["stringData"][CREDENTIALS_KEY_ENV_VAR]
+    second = _app_secret(_render_chart())["stringData"][CREDENTIALS_KEY_ENV_VAR]
+    assert first != second
+
+
+def test_app_secret_template_preserves_the_key_across_upgrades() -> None:
+    """``helm template`` never sees ``lookup`` results, so pin the preservation in the source."""
+    template = APP_SECRET_TEMPLATE.read_text(encoding="utf-8")
+    assert f'index $prev.data "{CREDENTIALS_KEY_ENV_VAR}"' in template, "a regenerated key would make every stored credential unreadable"
+    assert f"{CREDENTIALS_KEY_ENV_VAR}: {{{{ $credentialsKey | quote }}}}" in template
+    assert 'randBytes 32 | replace "+" "-" | replace "/" "_"' in template
+
+
+@pytest.mark.parametrize("replicas", ["1", "3"])
+def test_gateway_reads_the_key_from_the_app_secret(replicas: str) -> None:
+    gateway = _gateway(_render_chart(f"gateway.replicas={replicas}"))
+    reference = _env(gateway)[CREDENTIALS_KEY_ENV_VAR]["valueFrom"]["secretKeyRef"]
+
+    assert reference["name"].endswith("-app")
+    assert reference["key"] == CREDENTIALS_KEY_ENV_VAR
+    # Optional so a user-managed existingAppSecret without the key still boots: the
+    # Gateway's own startup gate refuses a multi-instance deployment that stores
+    # credentials without it, and leaves the others alone.
+    assert reference["optional"] is True
+
+
+def test_existing_app_secret_is_honored_for_the_key() -> None:
+    documents = _render_chart("existingAppSecret=my-app-secret")
+    assert not any(document.get("kind") == "Secret" and document["metadata"]["name"].endswith("-app") for document in documents)
+    assert _env(_gateway(documents))[CREDENTIALS_KEY_ENV_VAR]["valueFrom"]["secretKeyRef"]["name"] == "my-app-secret"
+
+
+def test_docs_mention_the_key() -> None:
+    for document in (README, NOTES, VALUES):
+        assert CREDENTIALS_KEY_ENV_VAR in document.read_text(encoding="utf-8"), document.name
+    assert ".credentials_key" in README.read_text(encoding="utf-8"), "the README must say how to keep an auto-generated key"

backend/tests/test_managed_models.py

@@ -60,6 +60,55 @@ def test_missing_encryption_key_never_replaced(store):
     assert not store.key_path.exists()
 
 
+def test_first_save_keeps_a_key_a_peer_created_concurrently(store, monkeypatch):
+    """A peer that publishes the key file first must win; replacing it would orphan the peer's catalog."""
+    from cryptography.fernet import Fernet
+
+    peer_key = Fernet.generate_key()
+    original_generate = Fernet.generate_key
+
+    def peer_wins_the_race():
+        store.key_path.parent.mkdir(parents=True, exist_ok=True)
+        store.key_path.write_bytes(peer_key)
+        return original_generate()
+
+    monkeypatch.setattr(Fernet, "generate_key", staticmethod(peer_wins_the_race))
+    store.save(profile(), expected_revision=None)
+
+    assert store.key_path.read_bytes() == peer_key
+    assert Fernet(peer_key).decrypt(store.path.read_bytes())
+
+
+def test_first_save_recovers_an_empty_key_left_by_an_interrupted_creation(store):
+    """An empty key file with no catalog encrypted no data, so the next save must replace it, not 503 forever."""
+    from cryptography.fernet import Fernet
+
+    store.key_path.parent.mkdir(parents=True, exist_ok=True)
+    store.key_path.write_bytes(b"")
+
+    store.save(profile(), expected_revision=None)
+
+    key = store.key_path.read_bytes()
+    assert key.strip()
+    assert Fernet(key).decrypt(store.path.read_bytes())
+    assert ManagedModelStore().list()[0].api_key.get_secret_value() == "test-secret"
+
+
+def test_an_empty_key_next_to_an_existing_catalog_is_never_replaced(store):
+    """The catalog was encrypted with the lost key; a new key would silently orphan it."""
+    store.save(profile(), expected_revision=None)
+    catalog = store.path.read_bytes()
+    store.key_path.write_bytes(b"")
+
+    with pytest.raises(ValueError):
+        store.save(profile(), expected_revision=None)
+    with pytest.raises(ValueError, match="restore it from backup"):
+        store._cipher(create=True)
+
+    assert store.key_path.read_bytes() == b""
+    assert store.path.read_bytes() == catalog
+
+
 @pytest.mark.parametrize("url", ["file:///tmp/test", "https://user:pass@example.com/v1", "https://example.com/v1?key=abc", "https://example.com/#fragment"])
 def test_endpoint_validation(url):
     with pytest.raises(ValidationError):

backend/tests/test_multi_worker_postgres_gate.py

@@ -18,8 +18,17 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 from fastapi import FastAPI
 
-from app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_agent_storage, _validate_login_throttle_storage, _validate_memory_retrieval_index, langgraph_runtime
+from app.gateway.deps import (
+    _enforce_credentials_key,
+    _enforce_postgres_for_multi_worker,
+    _validate_agent_storage,
+    _validate_login_throttle_storage,
+    _validate_memory_retrieval_index,
+    credentials_key_consumers,
+    langgraph_runtime,
+)
 from app.gateway.routers.browser import _browser_tools_enabled
+from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, GENERATE_KEY_COMMAND
 from deerflow.config.database_config import DatabaseConfig
 from deerflow.config.deployment_config import MULTI_INSTANCE_ENV_VAR, DeploymentConfig, multi_instance_declaration
 from deerflow.config.run_ownership_config import RunOwnershipConfig
@@ -75,6 +84,7 @@ def isolated_worker_env(monkeypatch):
     monkeypatch.delenv("WEB_CONCURRENCY", raising=False)
     monkeypatch.delenv(MULTI_INSTANCE_ENV_VAR, raising=False)
     monkeypatch.delenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", raising=False)
+    monkeypatch.delenv(CREDENTIALS_KEY_ENV_VAR, raising=False)
 
 
 # ---------------------------------------------------------------------------
@@ -846,3 +856,100 @@ def test_retrieval_index_warning_only_covers_the_bundled_fts5_index(caplog, memo
     with caplog.at_level("WARNING"):
         _validate_memory_retrieval_index(config)
     assert _retrieval_index_warnings(caplog) == []
+
+
+# ---------------------------------------------------------------------------
+# At-rest credentials key (DEER_FLOW_CREDENTIALS_KEY)
+# ---------------------------------------------------------------------------
+
+_VALID_KEY = "QUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUE="
+
+
+def _with_channel_connections(config, *, enabled: bool = True):
+    config.channel_connections = SimpleNamespace(enabled=enabled)
+    return config
+
+
+def test_credentials_key_consumers_name_the_enabled_features():
+    assert credentials_key_consumers(_cluster_ready()) == [], "a config without the section uses no key"
+    assert credentials_key_consumers(_with_channel_connections(_cluster_ready(), enabled=False)) == []
+    assert credentials_key_consumers(_with_channel_connections(_cluster_ready())) == ["channel_connections.enabled=true"]
+
+
+def test_declared_multi_instance_with_channel_connections_requires_the_credentials_key():
+    """Each instance would otherwise generate its own key file and could not read its peers' credentials."""
+    with pytest.raises(SystemExit) as exc_info:
+        _enforce_credentials_key(_with_channel_connections(_cluster_ready(deployment_multi_instance=True)))
+    msg = str(exc_info.value)
+    assert CREDENTIALS_KEY_ENV_VAR in msg
+    assert "deployment.multi_instance=true" in msg
+    assert "channel_connections.enabled=true" in msg
+    assert GENERATE_KEY_COMMAND in msg, "the refusal must say how to generate a key"
+    assert "Helm" in msg
+    assert "deployment.multi_instance=false" in msg, "must name the rollback knob"
+    assert "or set channel_connections.enabled=false" in msg, "must name the setting that turns the consumer off"
+
+
+def test_env_declared_multi_instance_names_the_env_knob(monkeypatch):
+    monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1")
+    with pytest.raises(SystemExit) as exc_info:
+        _enforce_credentials_key(_with_channel_connections(_cluster_ready()))
+    msg = str(exc_info.value)
+    assert f"{MULTI_INSTANCE_ENV_VAR}=1" in msg
+    assert f"unset {MULTI_INSTANCE_ENV_VAR}" in msg
+
+
+def test_declared_multi_instance_accepts_a_shared_credentials_key(monkeypatch):
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, _VALID_KEY)
+    _enforce_credentials_key(_with_channel_connections(_cluster_ready(deployment_multi_instance=True)))
+
+
+@pytest.mark.parametrize("channel_connections", [None, False], ids=["section-absent", "disabled"])
+def test_declared_multi_instance_without_key_consumers_keeps_booting(channel_connections):
+    """Existing multi-instance deployments that store no credentials must not need the key."""
+    config = _cluster_ready(deployment_multi_instance=True)
+    if channel_connections is not None:
+        _with_channel_connections(config, enabled=channel_connections)
+    _enforce_credentials_key(config)
+
+
+def test_multi_worker_without_a_declaration_shares_the_generated_key_file(monkeypatch):
+    """Workers of one process tree share base_dir, where the exclusive-create key file converges."""
+    monkeypatch.setenv("GATEWAY_WORKERS", "4")
+    _enforce_credentials_key(_with_channel_connections(_cluster_ready()))
+
+
+def test_single_instance_without_a_credentials_key_keeps_booting():
+    _enforce_credentials_key(_with_channel_connections(_config_with_backend("sqlite")))
+
+
+@pytest.mark.parametrize("declared", [False, True], ids=["single-instance", "multi-instance"])
+@pytest.mark.parametrize("value", ["not-a-fernet-key", f"{_VALID_KEY},second-entry-is-garbage"], ids=["garbage", "bad-rotation-entry"])
+def test_a_malformed_credentials_key_is_refused_without_echoing_it(monkeypatch, declared, value):
+    monkeypatch.setenv(CREDENTIALS_KEY_ENV_VAR, value)
+    with pytest.raises(SystemExit) as exc_info:
+        _enforce_credentials_key(_with_channel_connections(_cluster_ready(deployment_multi_instance=declared)))
+    msg = str(exc_info.value)
+    assert CREDENTIALS_KEY_ENV_VAR in msg
+    assert "not-a-fernet-key" not in msg and "second-entry-is-garbage" not in msg and _VALID_KEY not in msg
+
+
+@pytest.mark.asyncio
+async def test_langgraph_runtime_enforces_the_credentials_key_before_persistence_setup():
+    init_engine_from_config = AsyncMock(name="init_engine_from_config")
+
+    @asynccontextmanager
+    async def _noop_stream_bridge(_config):
+        yield MagicMock()
+
+    with (
+        patch("deerflow.persistence.engine.init_engine_from_config", init_engine_from_config),
+        patch("deerflow.runtime.make_stream_bridge", side_effect=_noop_stream_bridge) as make_stream_bridge,
+    ):
+        startup_config = _with_channel_connections(_cluster_ready(deployment_multi_instance=True))
+        with pytest.raises(SystemExit, match=CREDENTIALS_KEY_ENV_VAR):
+            async with langgraph_runtime(FastAPI(), startup_config):
+                pass
+
+    init_engine_from_config.assert_not_called()
+    make_stream_bridge.assert_not_called()

backend/tests/test_secret_file.py

@@ -0,0 +1,351 @@
+"""Exclusive-create helper for small secret files shared by concurrent processes.
+
+Gateway replicas that share a runtime home (a Helm ``ReadWriteMany`` volume, a
+compose bind mount, or several uvicorn workers of one host) cold-start at the
+same time and each find no secret file yet. A read-then-``O_TRUNC`` write lets
+every one of them keep a different value -- for ``.jwt_secret`` that is two
+replicas signing sessions with different keys, for ``.credentials_key`` it is
+credentials one replica cannot decrypt. ``read_or_create_secret_file`` creates
+with ``O_EXCL`` and reads the winner's value back instead.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import threading
+from pathlib import Path
+
+import pytest
+
+from deerflow.config.secret_file import InvalidSecretFileError, read_or_create_secret_file
+
+
+def test_creates_a_missing_file_with_owner_only_permissions(tmp_path: Path) -> None:
+    path = tmp_path / "nested" / ".secret"
+
+    value = read_or_create_secret_file(path, lambda: "generated-value")
+
+    assert value == "generated-value"
+    assert path.read_text(encoding="utf-8") == "generated-value"
+    if os.name == "posix":
+        assert path.stat().st_mode & 0o777 == 0o600
+
+
+def test_reuses_an_existing_value_without_generating(tmp_path: Path) -> None:
+    path = tmp_path / ".secret"
+    path.write_text("existing-value\n", encoding="utf-8")
+
+    def generate() -> str:
+        raise AssertionError("an existing secret must never be regenerated")
+
+    assert read_or_create_secret_file(path, generate) == "existing-value"
+    assert path.read_text(encoding="utf-8") == "existing-value\n"
+
+
+def test_a_peer_that_publishes_first_wins(tmp_path: Path) -> None:
+    """Deterministic race: the peer creates the file between our existence check and our create.
+
+    A read-then-truncating write would overwrite the peer's value and return
+    our own, leaving the two processes on different secrets.
+    """
+    path = tmp_path / ".secret"
+
+    def generate() -> str:
+        path.write_text("peer-value", encoding="utf-8")
+        return "our-value"
+
+    assert read_or_create_secret_file(path, generate) == "peer-value"
+    assert path.read_text(encoding="utf-8") == "peer-value"
+
+
+def test_concurrent_creators_converge_on_one_value(tmp_path: Path) -> None:
+    path = tmp_path / ".secret"
+    workers = 16
+    barrier = threading.Barrier(workers)
+    results: list[str] = []
+    errors: list[BaseException] = []
+    lock = threading.Lock()
+
+    def run(index: int) -> None:
+        try:
+            barrier.wait(timeout=10)
+            value = read_or_create_secret_file(path, lambda: f"value-{index}")
+            with lock:
+                results.append(value)
+        except BaseException as exc:  # noqa: BLE001 - surface every worker failure
+            with lock:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=run, args=(index,)) for index in range(workers)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=30)
+
+    assert not errors
+    assert len(results) == workers
+    assert len(set(results)) == 1, f"creators diverged: {sorted(set(results))}"
+    assert path.read_text(encoding="utf-8") == results[0]
+
+
+def test_waits_for_a_peer_that_has_created_but_not_yet_written(tmp_path: Path) -> None:
+    """Older releases published the name before its content: an empty file may be one of them mid-write."""
+    path = tmp_path / ".secret"
+    path.touch()
+    writer = threading.Timer(0.1, lambda: path.write_text("late-value", encoding="utf-8"))
+    writer.start()
+    try:
+        value = read_or_create_secret_file(path, lambda: "our-value", settle_seconds=5)
+    finally:
+        writer.join()
+
+    assert value == "late-value"
+    assert path.read_text(encoding="utf-8") == "late-value"
+
+
+def test_retries_a_partially_written_value_that_fails_validation(tmp_path: Path) -> None:
+    path = tmp_path / ".secret"
+    path.write_text("0123", encoding="utf-8")
+
+    def validate(value: str) -> None:
+        if len(value) != 10:
+            raise ValueError("incomplete")
+
+    writer = threading.Timer(0.1, lambda: path.write_text("0123456789", encoding="utf-8"))
+    writer.start()
+    try:
+        value = read_or_create_secret_file(path, lambda: "abcdefghij", validate=validate, settle_seconds=5)
+    finally:
+        writer.join()
+
+    assert value == "0123456789"
+
+
+def test_an_abandoned_empty_file_is_replaced(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
+    """A crash between create and write (or an older release's truncating write) leaves an empty file.
+
+    Nothing can have been signed or encrypted with an empty secret, so once the
+    settle window passes it is replaced instead of failing startup forever.
+    """
+    path = tmp_path / ".secret"
+    path.touch()
+
+    with caplog.at_level(logging.WARNING, logger="deerflow.config.secret_file"):
+        value = read_or_create_secret_file(path, lambda: "fresh-value", settle_seconds=0.05)
+
+    assert value == "fresh-value"
+    assert path.read_text(encoding="utf-8") == "fresh-value"
+    assert any(str(path) in record.getMessage() for record in caplog.records)
+    if os.name == "posix":
+        assert path.stat().st_mode & 0o777 == 0o600
+
+
+def test_concurrent_replacers_of_an_abandoned_empty_file_converge(tmp_path: Path) -> None:
+    """Replicas restarting together after a crash all find the same abandoned empty file.
+
+    The replacement is a last-writer-wins rename, so a replacer must not trust
+    its own value until peers that decided to replace in the same window have
+    written theirs.
+    """
+    path = tmp_path / ".secret"
+    path.touch()
+    workers = 8
+    barrier = threading.Barrier(workers)
+    results: list[str] = []
+    errors: list[BaseException] = []
+    lock = threading.Lock()
+
+    def run(index: int) -> None:
+        def generate() -> str:
+            # Stagger the replacements so an early replacer would otherwise
+            # read back its own value before a later one overwrites it.
+            threading.Event().wait(0.01 * index)
+            return f"value-{index}"
+
+        try:
+            barrier.wait(timeout=10)
+            value = read_or_create_secret_file(path, generate, settle_seconds=0.2)
+            with lock:
+                results.append(value)
+        except BaseException as exc:  # noqa: BLE001 - surface every worker failure
+            with lock:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=run, args=(index,)) for index in range(workers)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=30)
+
+    assert not errors
+    assert len(results) == workers
+    assert len(set(results)) == 1, f"replacers diverged: {sorted(set(results))}"
+    assert path.read_text(encoding="utf-8") == results[0]
+
+
+def test_waits_for_a_peer_that_holds_the_replacement_claim(tmp_path: Path) -> None:
+    path = tmp_path / ".secret"
+    path.touch()
+    claim = tmp_path / ".secret.replacing"
+    claim.write_text("peer-value", encoding="utf-8")
+    publisher = threading.Timer(0.3, lambda: os.replace(claim, path))
+    publisher.start()
+    try:
+        value = read_or_create_secret_file(path, lambda: "our-value", settle_seconds=0.05)
+    finally:
+        publisher.join()
+
+    assert value == "peer-value"
+    assert not claim.exists()
+
+
+def test_a_claim_left_by_a_crashed_replacer_is_refused_after_a_bounded_wait(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("deerflow.config.secret_file._CLAIM_WAIT_SECONDS", 0.2)
+    path = tmp_path / ".secret"
+    path.touch()
+    claim = tmp_path / ".secret.replacing"
+    claim.touch()
+
+    with pytest.raises(InvalidSecretFileError, match="interrupted creation"):
+        read_or_create_secret_file(path, lambda: "our-value", settle_seconds=0.05)
+
+    assert path.read_text(encoding="utf-8") == ""
+
+
+def test_a_late_replacer_does_not_overwrite_a_published_replacement(tmp_path: Path) -> None:
+    """A process that saw the empty file before a peer's replacement landed must back off."""
+    path = tmp_path / ".secret"
+    path.write_text("peer-value", encoding="utf-8")
+    from deerflow.config.secret_file import _publish_under_claim
+
+    assert _publish_under_claim(path, lambda: "late-value", replace_empty=True) is True
+    assert path.read_text(encoding="utf-8") == "peer-value"
+    assert not (tmp_path / ".secret.replacing").exists()
+
+
+def test_a_persistently_invalid_file_is_refused_without_echoing_it(tmp_path: Path) -> None:
+    """A non-empty value may be the only copy of a key that encrypted data: never overwrite it."""
+    path = tmp_path / ".secret"
+    path.write_text("operator-typo-in-the-key", encoding="utf-8")
+
+    def validate(value: str) -> None:
+        raise ValueError(f"bad value {value}")
+
+    with pytest.raises(InvalidSecretFileError) as exc_info:
+        read_or_create_secret_file(path, lambda: "replacement", validate=validate, settle_seconds=0.05)
+
+    assert str(path) in str(exc_info.value)
+    assert "operator-typo-in-the-key" not in str(exc_info.value)
+    assert exc_info.value.__cause__ is None, "the validator's message may quote the value"
+    assert path.read_text(encoding="utf-8") == "operator-typo-in-the-key"
+
+
+def test_a_failed_write_leaves_no_empty_file_for_peers_to_wait_on(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    path = tmp_path / ".secret"
+
+    def broken_write(_fd: int, _data: bytes) -> None:
+        raise OSError("disk full")
+
+    monkeypatch.setattr("deerflow.config.secret_file._write_all", broken_write)
+
+    with pytest.raises(OSError, match="disk full"):
+        read_or_create_secret_file(path, lambda: "value")
+
+    assert not path.exists()
+
+
+def _pause_one_creator(monkeypatch: pytest.MonkeyPatch, thread_name: str, *, fail: bool = False) -> tuple[threading.Event, threading.Event]:
+    """Hold ``thread_name`` inside its secret write until released; the other threads write normally."""
+    import deerflow.config.secret_file as secret_file
+
+    entered = threading.Event()
+    release = threading.Event()
+    original_write_all = secret_file._write_all
+
+    def write_all(fd: int, data: bytes) -> None:
+        if threading.current_thread().name == thread_name:
+            entered.set()
+            assert release.wait(timeout=10)
+            if fail:
+                raise OSError("disk full")
+        original_write_all(fd, data)
+
+    monkeypatch.setattr(secret_file, "_write_all", write_all)
+    return entered, release
+
+
+@pytest.mark.parametrize("fail", [False, True], ids=["paused-then-writes", "paused-then-fails"])
+def test_a_paused_creator_never_exposes_a_file_a_peer_could_replace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fail: bool) -> None:
+    """A creator stalled mid-write must not leave a visible empty file behind.
+
+    Otherwise a peer treats it as abandoned after the settle window and
+    publishes its own value; the resumed creator then returns a value nobody
+    persisted, or (when its write fails) deletes the peer's valid file.
+    """
+    path = tmp_path / ".secret"
+    entered, release = _pause_one_creator(monkeypatch, "first", fail=fail)
+    results: dict[str, str | BaseException] = {}
+
+    def run(name: str) -> None:
+        try:
+            results[name] = read_or_create_secret_file(path, lambda: f"{name}-value", settle_seconds=0.05)
+        except BaseException as exc:  # noqa: BLE001 - asserted below
+            results[name] = exc
+
+    first = threading.Thread(target=run, args=("first",), name="first")
+    first.start()
+    assert entered.wait(timeout=10)
+    peer = threading.Thread(target=run, args=("peer",), name="peer")
+    peer.start()
+    peer.join(timeout=10)
+    release.set()
+    first.join(timeout=10)
+
+    assert results["peer"] == "peer-value"
+    assert path.read_text(encoding="utf-8") == "peer-value", "the peer's published value must survive"
+    if fail:
+        assert isinstance(results["first"], OSError)
+    else:
+        assert results["first"] == "peer-value", "the resumed creator must adopt the persisted value"
+
+
+def test_creation_without_hard_links_still_converges(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Filesystems without hard links (SMB shares) publish through the replacement claim instead."""
+    import errno
+
+    def no_hard_links(*_args, **_kwargs):
+        raise OSError(errno.EPERM, "Operation not permitted")
+
+    monkeypatch.setattr(os, "link", no_hard_links)
+    path = tmp_path / ".secret"
+    workers = 8
+    barrier = threading.Barrier(workers)
+    results: list[str] = []
+    lock = threading.Lock()
+
+    def run(index: int) -> None:
+        barrier.wait(timeout=10)
+        value = read_or_create_secret_file(path, lambda: f"value-{index}", settle_seconds=0.05)
+        with lock:
+            results.append(value)
+
+    threads = [threading.Thread(target=run, args=(index,)) for index in range(workers)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=30)
+
+    assert len(results) == workers
+    assert len(set(results)) == 1, f"creators diverged: {sorted(set(results))}"
+    assert path.read_text(encoding="utf-8") == results[0]
+    assert not (tmp_path / ".secret.replacing").exists()
+    if os.name == "posix":
+        assert path.stat().st_mode & 0o777 == 0o600
+
+
+def test_no_temporary_files_are_left_behind(tmp_path: Path) -> None:
+    path = tmp_path / ".secret"
+    read_or_create_secret_file(path, lambda: "value")
+
+    assert sorted(entry.name for entry in tmp_path.iterdir()) == [".secret"]

deploy/helm/deer-flow/README.md

@@ -379,6 +379,25 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002
   JWT=$(kubectl -n deer-flow exec deploy/deer-flow-gateway -- cat /app/backend/.deer-flow/.jwt_secret)
   kubectl -n deer-flow patch secret deer-flow-app -p "{\"stringData\":{\"AUTH_JWT_SECRET\":\"$JWT\"}}"
   ```
+- **Credentials key.** The app Secret also holds `DEER_FLOW_CREDENTIALS_KEY`,
+  the Fernet key the gateway encrypts stored credentials with (per-connection
+  IM channel credentials today). The chart generates it once (32 random bytes,
+  urlsafe base64) and preserves it across upgrades via `lookup`; every gateway
+  Pod reads the same value. **Back it up**: deleting the Secret or the release
+  generates a new key, and credentials stored under the old one become
+  unreadable (they are treated as missing, never as an error). To rotate,
+  prepend a new key — `new,old` — and keep the old one listed until stored
+  values have been rewritten. An `existingAppSecret` should carry it; the env
+  entry is optional, so without it a single Pod falls back to an auto-generated
+  `.credentials_key` on the home volume, while a multi-instance gateway with
+  `channel_connections.enabled` refuses to start. If you add the key to an
+  `existingAppSecret` later, copy that file's value so stored credentials stay
+  readable:
+
+  ```bash
+  KEY=$(kubectl -n deer-flow exec deploy/deer-flow-gateway -- cat /app/backend/.deer-flow/.credentials_key)
+  kubectl -n deer-flow patch secret my-app-secret -p "{\"stringData\":{\"DEER_FLOW_CREDENTIALS_KEY\":\"$KEY\"}}"
+  ```
 - **Scheduled task recovery.** If a deployment explicitly enables
   `scheduler.multi_instance: true`, it must use shared Postgres,
   `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`.
@@ -493,7 +512,7 @@ but not file mode — so a PVC written by an earlier **root** run (e.g. a cluste
 that ran the gateway as root before enabling this hardening, or a backup restore
 of root-owned files) will keep files like `.jwt_secret` at `0600 root:root`. The
 non-root gateway (uid 1000) then can't read them and crashes on the first auth
-request with `RuntimeError: Failed to read JWT secret from .../​.jwt_secret`.
+request with `RuntimeError: Failed to read or persist the JWT secret at .../​.jwt_secret`.
 
 **Fresh installs are unaffected** — uid 1000 creates every file as `1000:1000`.
 

deploy/helm/deer-flow/templates/NOTES.txt

@@ -74,10 +74,12 @@ Provisioner / sandbox notes:
 Generated secrets (persisted across upgrades):
 
   • BETTER_AUTH_SECRET, DEER_FLOW_INTERNAL_AUTH_TOKEN, AUTH_JWT_SECRET (the
-    session-cookie signing key shared by every gateway Pod) and
+    session-cookie signing key shared by every gateway Pod),
     PROVISIONER_API_KEY (the key the gateway presents to the sandbox
-    provisioner, which answers 401 to every /api/* request without it) are
-    stored in the Secret `{{ include "deer-flow.appSecret" . }}`.
+    provisioner, which answers 401 to every /api/* request without it) and
+    DEER_FLOW_CREDENTIALS_KEY (the at-rest key for stored credentials; back it
+    up, losing it makes them unreadable) are stored in the Secret
+    `{{ include "deer-flow.appSecret" . }}`.
 
 Database ({{ if .Values.postgresql.enabled }}bundled postgres{{ else }}external postgres{{ end }}):
 

deploy/helm/deer-flow/templates/_helpers.tpl

@@ -71,7 +71,8 @@ imagePullSecrets:
 {{- end -}}
 
 {{/* Name of the Secret holding the app secrets (BETTER_AUTH_SECRET,
-     DEER_FLOW_INTERNAL_AUTH_TOKEN, AUTH_JWT_SECRET, PROVISIONER_API_KEY).
+     DEER_FLOW_INTERNAL_AUTH_TOKEN, AUTH_JWT_SECRET, PROVISIONER_API_KEY,
+     DEER_FLOW_CREDENTIALS_KEY).
      `existingAppSecret` points every consumer (gateway, frontend,
      provisioner) at a user-managed Secret instead and skips generating one;
      it must carry PROVISIONER_API_KEY while the provisioner is enabled and

deploy/helm/deer-flow/templates/gateway-deployment.yaml

@@ -150,6 +150,18 @@ spec:
                   name: {{ include "deer-flow.appSecret" . }}
                   key: AUTH_JWT_SECRET
                   optional: {{ ne (include "deer-flow.gatewayMultiInstance" .) "true" }}
+            # At-rest key for stored credentials, shared by every Pod so one
+            # Pod can decrypt what another stored. Optional so a user-managed
+            # existingAppSecret without it still boots: a single Pod then falls
+            # back to .credentials_key on the home volume, and the Gateway's own
+            # startup gate refuses a multi-instance deployment that stores
+            # credentials (channel_connections) without the key.
+            - name: DEER_FLOW_CREDENTIALS_KEY
+              valueFrom:
+                secretKeyRef:
+                  name: {{ include "deer-flow.appSecret" . }}
+                  key: DEER_FLOW_CREDENTIALS_KEY
+                  optional: true
             {{- if .Values.provisioner.enabled }}
             # Shared gateway -> bundled-provisioner key. config.yaml references
             # it as sandbox.provisioner_api_key: $PROVISIONER_API_KEY, and the

deploy/helm/deer-flow/templates/secret-app.yaml

@@ -4,11 +4,13 @@
 {{- $internalToken := "" -}}
 {{- $jwtSecret := "" -}}
 {{- $provisionerKey := "" -}}
+{{- $credentialsKey := "" -}}
 {{- if $prev -}}
 {{- $betterAuth = index $prev.data "BETTER_AUTH_SECRET" | default "" | b64dec -}}
 {{- $internalToken = index $prev.data "DEER_FLOW_INTERNAL_AUTH_TOKEN" | default "" | b64dec -}}
 {{- $jwtSecret = index $prev.data "AUTH_JWT_SECRET" | default "" | b64dec -}}
 {{- $provisionerKey = index $prev.data "PROVISIONER_API_KEY" | default "" | b64dec -}}
+{{- $credentialsKey = index $prev.data "DEER_FLOW_CREDENTIALS_KEY" | default "" | b64dec -}}
 {{- end -}}
 {{- if not $betterAuth -}}{{- $betterAuth = randAlphaNum 48 | lower -}}{{- end -}}
 {{- if not $internalToken -}}{{- $internalToken = randAlphaNum 40 -}}{{- end -}}
@@ -22,6 +24,11 @@
        value. Generated once and preserved across upgrades so a rollout never
        leaves the two sides on different keys. */ -}}
 {{- if not $provisionerKey -}}{{- $provisionerKey = randAlphaNum 48 -}}{{- end -}}
+{{- /* At-rest key for stored credentials (a Fernet key: urlsafe base64 of 32
+       random bytes). randBytes emits standard base64, so translate its
+       alphabet. Generated once and preserved across upgrades: a new key would
+       make every credential the gateway Pods stored unreadable. */ -}}
+{{- if not $credentialsKey -}}{{- $credentialsKey = randBytes 32 | replace "+" "-" | replace "/" "_" -}}{{- end -}}
 apiVersion: v1
 kind: Secret
 metadata:
@@ -35,4 +42,5 @@ stringData:
   DEER_FLOW_INTERNAL_AUTH_TOKEN: {{ $internalToken | quote }}
   AUTH_JWT_SECRET: {{ $jwtSecret | quote }}
   PROVISIONER_API_KEY: {{ $provisionerKey | quote }}
+  DEER_FLOW_CREDENTIALS_KEY: {{ $credentialsKey | quote }}
 {{- end -}}

deploy/helm/deer-flow/values.yaml

@@ -338,7 +338,10 @@ existingSecret: ""
 #    DEER_FLOW_INTERNAL_AUTH_TOKEN, PROVISIONER_API_KEY (the shared gateway ->
 #    sandbox-provisioner key, required while `provisioner.enabled`) and, for
 #    more than one gateway instance, AUTH_JWT_SECRET (the session-cookie
-#    signing key every Pod must share).
+#    signing key every Pod must share). It should also hold
+#    DEER_FLOW_CREDENTIALS_KEY (the at-rest key for stored credentials, a
+#    Fernet key); a multi-instance gateway with channel_connections enabled
+#    refuses to start without it.
 existingAppSecret: ""
 
 # -- Ingress in front of nginx (port 2026). nginx preserves all internal routing.

docker/docker-compose-dev.yaml

@@ -239,6 +239,8 @@ services:
       - DEER_FLOW_CHANNELS_LANGGRAPH_URL=${DEER_FLOW_CHANNELS_LANGGRAPH_URL:-http://gateway:8001/api}
       - DEER_FLOW_CHANNELS_GATEWAY_URL=${DEER_FLOW_CHANNELS_GATEWAY_URL:-http://gateway:8001}
       - DEER_FLOW_INTERNAL_AUTH_TOKEN=${DEER_FLOW_INTERNAL_AUTH_TOKEN:-}
+      # At-rest key for stored credentials; empty falls back to .credentials_key in DEER_FLOW_HOME.
+      - DEER_FLOW_CREDENTIALS_KEY=${DEER_FLOW_CREDENTIALS_KEY:-}
       - DEER_FLOW_HOST_BASE_DIR=${DEER_FLOW_ROOT}/backend/.deer-flow
       - DEER_FLOW_SANDBOX_HOST=host.docker.internal
       # The login lockout counts failures per client IP, but every browser

docker/docker-compose.yaml

@@ -24,6 +24,7 @@
 #   DEER_FLOW_REPO_ROOT              — repo root (used for skills host path in DooD)
 #   BETTER_AUTH_SECRET               — required for frontend auth/session security
 #   DEER_FLOW_INTERNAL_AUTH_TOKEN    — shared internal Gateway auth token for multi-worker IM channels
+#   DEER_FLOW_CREDENTIALS_KEY        — at-rest Fernet key for stored credentials (deploy.sh persists it as $DEER_FLOW_HOME/.credentials_key)
 #
 # LangSmith tracing is disabled by default (LANGSMITH_TRACING=false).
 # Set LANGSMITH_TRACING=true and LANGSMITH_API_KEY in .env to enable it.
@@ -165,6 +166,9 @@ services:
       - DEER_FLOW_CHANNELS_LANGGRAPH_URL=${DEER_FLOW_CHANNELS_LANGGRAPH_URL:-http://gateway:8001/api}
       - DEER_FLOW_CHANNELS_GATEWAY_URL=${DEER_FLOW_CHANNELS_GATEWAY_URL:-http://gateway:8001}
       - DEER_FLOW_INTERNAL_AUTH_TOKEN=${DEER_FLOW_INTERNAL_AUTH_TOKEN}
+      # At-rest key for stored credentials (scripts/deploy.sh resolves or generates it).
+      # Empty falls back to .credentials_key in the runtime home mounted above.
+      - DEER_FLOW_CREDENTIALS_KEY=${DEER_FLOW_CREDENTIALS_KEY:-}
       # DooD path/network translation
       - DEER_FLOW_HOST_BASE_DIR=${DEER_FLOW_HOME}
       - DEER_FLOW_SANDBOX_HOST=host.docker.internal

scripts/AGENTS.md

@@ -97,9 +97,10 @@ the `env_file` value); shell exports still win. Pinned by
 
 `deploy.sh` never sources the repo-root `.env`; Compose reads it via
 `--env-file`, and shell exports outrank that file during interpolation (an
-exported-but-empty variable still wins). So `BETTER_AUTH_SECRET` and
-`DEER_FLOW_INTERNAL_AUTH_TOKEN` resolve shell → `.env` → persisted file under
-`DEER_FLOW_HOME` → freshly generated, and a `.env`-provided value is left
+exported-but-empty variable still wins). So `BETTER_AUTH_SECRET`,
+`DEER_FLOW_INTERNAL_AUTH_TOKEN` and `DEER_FLOW_CREDENTIALS_KEY` resolve shell →
+`.env` → persisted file under `DEER_FLOW_HOME` → freshly generated, and a
+`.env`-provided value is left
 unexported so Compose parses it itself. Whether `.env` provides one is
 Compose's answer, not a `KEY=VALUE` grep: Compose also accepts `KEY: VALUE`
 lines and interpolates `${VAR}` inside values, so the script renders a stub
@@ -112,6 +113,9 @@ failing probe stops the script rather than guessing. `read_dotenv_value`
 stays for the end-of-run summary only. Do not export a value the script read
 from `.env`: that shadows Compose's own dotenv parsing and re-creates the bug
 where `make up` replaced the operator's secret with a generated one.
+The credentials key (a Fernet key) persists as `.credentials_key`, the file the
+Gateway itself generates in that runtime home, via a noclobber (`O_EXCL`)
+create that reads a concurrent winner back.
 `backend/tests/test_deploy_dotenv_secrets.py` pins the order and the probe;
 its real-Compose cases run against the installed `docker` CLI and against any
 standalone binaries listed in `DEER_FLOW_TEST_COMPOSE_BINARIES`.

scripts/deploy.sh

@@ -298,6 +298,53 @@ elif [ "$CMD" != "down" ] && [ -z "$DEER_FLOW_INTERNAL_AUTH_TOKEN" ]; then
     fi
 fi
 
+# ── DEER_FLOW_CREDENTIALS_KEY ────────────────────────────────────────────────
+# At-rest Fernet key the Gateway encrypts stored credentials with. Losing it
+# makes those credentials unreadable, so back the file up. It is persisted
+# under the name the Gateway itself generates in its runtime home (the
+# $DEER_FLOW_HOME bind mount), so a key either side created first is the one
+# both keep. noclobber turns the redirection into an exclusive (O_EXCL)
+# create: when a concurrent run wins, its key is read back instead of ours.
+
+_credentials_key_file="$DEER_FLOW_HOME/.credentials_key"
+if [ "$CMD" != "down" ] && [ -z "$DEER_FLOW_CREDENTIALS_KEY" ] && dotenv_provides_secret DEER_FLOW_CREDENTIALS_KEY; then
+    echo -e "${GREEN}✓ DEER_FLOW_CREDENTIALS_KEY loaded from $ENV_FILE${NC}"
+elif [ "$CMD" != "down" ] && [ -z "$DEER_FLOW_CREDENTIALS_KEY" ]; then
+    _generated_credentials_key=""
+    if [ ! -f "$_credentials_key_file" ]; then
+        # urlsafe base64 of 32 random bytes, the Fernet key format.
+        if command -v python3 > /dev/null 2>&1 && \
+            _generated_credentials_key="$(python3 -c 'import sys; sys.version_info >= (3, 6) or sys.exit(1); import base64, os; print(base64.urlsafe_b64encode(os.urandom(32)).decode())' 2>/dev/null)"; then
+            true
+        elif command -v python > /dev/null 2>&1 && \
+            _generated_credentials_key="$(python -c 'import sys; sys.version_info >= (3, 6) or sys.exit(1); import base64, os; print(base64.urlsafe_b64encode(os.urandom(32)).decode())' 2>/dev/null)"; then
+            true
+        elif command -v openssl > /dev/null 2>&1 && \
+            _generated_credentials_key="$(openssl rand -base64 32 | tr '+/' '-_')"; then
+            true
+        else
+            echo -e "${RED}✗ Cannot generate DEER_FLOW_CREDENTIALS_KEY: python3, python, and openssl are all unavailable.${NC}" >&2
+            echo -e "${RED}  Set DEER_FLOW_CREDENTIALS_KEY manually before running make up.${NC}" >&2
+            exit 1
+        fi
+        if ! ( umask 077; set -o noclobber; printf '%s\n' "$_generated_credentials_key" > "$_credentials_key_file" ) 2>/dev/null; then
+            _generated_credentials_key=""
+        fi
+    fi
+    if [ -n "$_generated_credentials_key" ]; then
+        export DEER_FLOW_CREDENTIALS_KEY="$_generated_credentials_key"
+        echo -e "${GREEN}✓ DEER_FLOW_CREDENTIALS_KEY generated → $_credentials_key_file (back it up)${NC}"
+    else
+        if [ ! -r "$_credentials_key_file" ]; then
+            fail_home_permission "$_credentials_key_file" readable
+        fi
+        export DEER_FLOW_CREDENTIALS_KEY
+        DEER_FLOW_CREDENTIALS_KEY="$(cat "$_credentials_key_file")"
+        echo -e "${GREEN}✓ DEER_FLOW_CREDENTIALS_KEY loaded from $_credentials_key_file${NC}"
+    fi
+    unset _generated_credentials_key
+fi
+
 # ── UV_EXTRAS auto-detection ─────────────────────────────────────────────────
 # The production Dockerfile accepts UV_EXTRAS as a single build-arg token and
 # adds the --extra prefix itself. Convert the detector's uv flag string

scripts/docker.sh

@@ -150,9 +150,10 @@ ensure_env_files() {
 # start runs Compose from docker/ without --env-file, so ${VAR} interpolation in
 # docker-compose-dev.yaml sees the shell only, never the checkout .env. Export
 # the .env values that interpolation consumes; a value already set in the shell
-# (even empty) wins, as it does for Compose. AUTH_TRUSTED_PROXIES belongs here
-# because the Gateway's `environment:` entry, which defaults to nginx, outranks
-# the same key loaded through env_file and would replace the operator's list.
+# (even empty) wins, as it does for Compose. AUTH_TRUSTED_PROXIES and
+# DEER_FLOW_CREDENTIALS_KEY belong here because the Gateway's `environment:`
+# entries (defaulting to nginx and to empty) outrank the same keys loaded through
+# env_file and would replace the operator's values.
 load_proxy_env_from_dotenv() {
     local env_file="$PROJECT_ROOT/.env"
     local var
@@ -163,7 +164,7 @@ load_proxy_env_from_dotenv() {
         return
     fi
 
-    for var in HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY http_proxy https_proxy all_proxy no_proxy AUTH_TRUSTED_PROXIES; do
+    for var in HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY http_proxy https_proxy all_proxy no_proxy AUTH_TRUSTED_PROXIES DEER_FLOW_CREDENTIALS_KEY; do
         if [ -z "${!var+x}" ]; then
             line="$(grep -E "^[[:space:]]*${var}=" "$env_file" | tail -n 1 || true)"
             if [ -n "$line" ]; then