feat(lark): sidecar credential broker for sandbox lark-cli (Pattern B) (#4501)

* feat(lark): sidecar credential broker for sandbox lark-cli (Pattern B)

Removes the plaintext Lark credential mounts (appSecret + OAuth tokens)
from the sandbox container. A long-running broker sidecar owns lark-cli
and the per-user config/data dirs and serves the command surface over
Pod loopback; the sandbox gets only a forwarding shim on PATH, so the
raw credential files never exist in the sandbox filesystem.

- lark_broker.py: stdlib-only loopback broker (argv passthrough with
  shell=False, server-injected credential env, bounded I/O) + shim
  script constant + install-shim mode.
- docker/lark-cli-broker: init(install-shim) + serve image.
- provisioner: LARK_CLI_BROKER_IMAGE + provision_lark_cli_broker →
  shim init container + lark-cli-broker sidecar (config/data mounted
  sidecar-only); credentials dropped from the sandbox container;
  /api/capabilities reports lark_cli_broker_image. Broker supersedes
  the Pattern A init-container binary when both are configured.
- gateway: lark_cli_env_overlay(broker=True) omits config/data env;
  sandbox_lark_broker_active() TTL-cached mode resolver; broker added
  to sandbox_runtime_mode / readiness and the settings UI.

Opt-in and off by default (empty LARK_CLI_BROKER_IMAGE ⇒ no change).

Closes #4338

* fix(lark): address Pattern B broker review findings (#4501)

Follow-up to the sidecar credential broker addressing the PR #4501 review:

- shim: split the on-PATH lark-cli into a /bin/sh launcher + Python shim body
  so broker mode fails loudly (exit 127, actionable message) instead of ENOEXEC
  when the sandbox image ships no python3; interpreter pinnable via
  DEERFLOW_LARK_BROKER_PYTHON. Launcher bakes in the shim's absolute path since
  $0 is the bare command name when run off PATH.
- broker: drop the dead cwd payload field (broker can't see the sandbox FS) and
  document the command-surface-only / no-file-IO limitation.
- broker: return a structured 500 JSON on unexpected exec errors so the shim
  gets a meaningful message, not an opaque transport failure; set a handler
  socket timeout to bound slow/stuck connections.
- broker: add an opt-in DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS denylist that
  refuses secret-dumping subcommands before spawning the binary, forwarded from
  the provisioner sidecar.
- gateway: tighten the per-bash-call broker probe timeout (1.5s) and cache
  negatives longer (300s) so non-broker remote-provisioner users don't pay a
  latency hit; guard the mode cache with a lock; drop the dead
  _probe_provisioner_lark_cli_init_image wrapper.
- docs: remove the broken design-doc link from the broker README.

Adds tests for launcher python resolution, cwd omission, denylist enforcement,
500-on-error, hot-path probe timeout + negative caching, and provisioner
denylist-env wiring.

backend/AGENTS.md

@@ -710,7 +710,7 @@ E2B output sync records remote file versions and actual host file metadata in a
   - `skills/describe.py` — `build_describe_skill_tool(catalog)` builds the `describe_skill` tool as a closure; `build_skill_search_setup(skills, enabled, ...)` produces a `SkillSearchSetup(describe_skill_tool, skill_names)` that is wired into both the LangGraph agent factory (`agent.py`) and the embedded client (`client.py`).
 - **Slash activation**: `/skill-name task` loads that enabled skill's `SKILL.md` for the current model call only. The resolver rejects leading whitespace, missing separators, reserved channel commands (`/new`, `/help`, `/bootstrap`, `/status`, `/models`, `/memory`, `/goal`), disabled skills, and skills outside a custom agent's whitelist.
 - **Installation**: `POST /api/skills/install` extracts .skill ZIP archive to custom/ directory
-- **Managed integrations**: Lark/Feishu CLI support installs one global official `lark-*` pack as read-only `SkillCategory.INTEGRATION` entries under `/mnt/skills/integrations/lark-cli/...`; enabled flags, app configuration, and OAuth data remain per-user. Install resolves the newest `larksuite/cli` release from GitHub (`releases/latest`) at install time (falling back to a bottom-line pinned version if the lookup fails) rather than hard-coding the pack version; integrity relies on the official host + structural archive guards + a recorded hash of the effective installed tree after shared guidance injection (not a pinned archive-byte SHA, which GitHub does not keep stable). The Gateway image still installs a pinned `@larksuite/cli` binary, so `get_lark_integration_status` surfaces `latest_available_version` and `runtime_version_mismatch` for the UI. AIO installs additionally verify and publish official Linux amd64/arm64 binaries under `{DEER_FLOW_HOME}/integrations/lark-cli/sandbox-cli`, mounted read-only at `/mnt/integrations/lark-cli/runtime`; `/mnt/integrations/lark-cli/config` (app credentials, incl. the long-lived `appSecret`) is mounted **read-only** into the sandbox and `/mnt/integrations/lark-cli/data` (refreshable OAuth tokens) stays writable, both mapping to owner-only per-user credential directories. **Sandbox trust boundary:** those two dirs are still *readable* by arbitrary sandbox processes (the agent's `bash` tool, or code reached via prompt-injection in a tool result), so the app secret and tokens are exposed to sandbox-side code even though they never reach the browser — the read-only config mount only prevents in-sandbox tampering, not read/exfiltration. The sidecar credential-broker follow-up (issue #4338) is the planned fix that removes these plaintext mounts from sandbox execution. `DEER_FLOW_LARK_CLI_SANDBOX_RUNTIME_DIR` supplies a validated, symlink-free pre-staged runtime for air-gapped deployments. For the remote provisioner (K8s), the runtime is instead provisioned by an optional init container + shared `emptyDir` (Pattern A): set `LARK_CLI_INIT_IMAGE` on the provisioner (see `docker/lark-cli-init/`) and the Gateway sends `provision_lark_cli_runtime` on sandbox create once the pack is installed, so remote installs skip the Gateway-side GitHub download entirely. `get_lark_integration_status(check_runtime=True)` surfaces `sandbox_runtime_mode` (`none` / `gateway-download` / `init-container`) and `sandbox_runtime_ready` (init-container mode reads the provisioner `GET /api/capabilities`) so a green UI can't hide a chat-time `lark-cli: command not found`. Cheap status probes are explicitly not live-verified; users authorize or reconnect through the browser device-flow endpoints instead of running terminal commands.
+- **Managed integrations**: Lark/Feishu CLI support installs one global official `lark-*` pack as read-only `SkillCategory.INTEGRATION` entries under `/mnt/skills/integrations/lark-cli/...`; enabled flags, app configuration, and OAuth data remain per-user. Install resolves the newest `larksuite/cli` release from GitHub (`releases/latest`) at install time (falling back to a bottom-line pinned version if the lookup fails) rather than hard-coding the pack version; integrity relies on the official host + structural archive guards + a recorded hash of the effective installed tree after shared guidance injection (not a pinned archive-byte SHA, which GitHub does not keep stable). The Gateway image still installs a pinned `@larksuite/cli` binary, so `get_lark_integration_status` surfaces `latest_available_version` and `runtime_version_mismatch` for the UI. AIO installs additionally verify and publish official Linux amd64/arm64 binaries under `{DEER_FLOW_HOME}/integrations/lark-cli/sandbox-cli`, mounted read-only at `/mnt/integrations/lark-cli/runtime`; `/mnt/integrations/lark-cli/config` (app credentials, incl. the long-lived `appSecret`) is mounted **read-only** into the sandbox and `/mnt/integrations/lark-cli/data` (refreshable OAuth tokens) stays writable, both mapping to owner-only per-user credential directories. **Sandbox trust boundary:** those two dirs are still *readable* by arbitrary sandbox processes (the agent's `bash` tool, or code reached via prompt-injection in a tool result), so the app secret and tokens are exposed to sandbox-side code even though they never reach the browser — the read-only config mount only prevents in-sandbox tampering, not read/exfiltration. The sidecar credential-broker (Pattern B, issue #4338) is the fix that removes these plaintext mounts from sandbox execution: set `LARK_CLI_BROKER_IMAGE` on the provisioner (see `docker/lark-cli-broker/`) and the Gateway sends `provision_lark_cli_broker` on sandbox create. The provisioner then runs a `lark-cli-broker` sidecar that owns the per-user `config`/`data` (mounted into the **sidecar only**, at `/var/lark/{config,data}`) and serves the `lark-cli` command surface on Pod loopback (`http://127.0.0.1:8788`); a shim init container (`install-shim`) writes a forwarding `lark-cli` into the shared runtime `emptyDir`, so the sandbox gets `DEERFLOW_LARK_BROKER_URL` + a shim on PATH but **no** credential files. The on-PATH `bin/lark-cli` is a `/bin/sh` launcher that resolves a Python 3 interpreter and execs the Python shim body (`bin/lark-cli-shim.py`) beside it, so broker mode does not silently ENOEXEC on a sandbox image without a `#!/usr/bin/env python3`-resolvable interpreter — it fails loudly (exit 127, actionable message) and can be pinned with `DEERFLOW_LARK_BROKER_PYTHON`. The broker runs `lark-cli` in the sidecar's cwd and cannot see the sandbox filesystem, so cwd is intentionally **not** forwarded and file-I/O subcommands relative to the sandbox cwd are unsupported (command surface only). An optional `DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS` denylist (comma-separated command prefixes, forwarded from the provisioner) lets the broker refuse secret-dumping subcommands before spawning the binary. `lark_cli_env_overlay(broker=True)` therefore omits `LARKSUITE_CLI_CONFIG_DIR`/`DATA_DIR`; `sandbox_lark_broker_active()` (TTL-cached provisioner `/api/capabilities` probe, tight timeout + longer negative caching on the bash hot path) selects broker vs. binary mode for both the bash env overlay and status. `DEER_FLOW_LARK_CLI_SANDBOX_RUNTIME_DIR` supplies a validated, symlink-free pre-staged runtime for air-gapped deployments. For the remote provisioner (K8s), the runtime binary is otherwise provisioned by an optional init container + shared `emptyDir` (Pattern A): set `LARK_CLI_INIT_IMAGE` on the provisioner (see `docker/lark-cli-init/`) and the Gateway sends `provision_lark_cli_runtime` on sandbox create once the pack is installed, so remote installs skip the Gateway-side GitHub download entirely. Broker (Pattern B) supersedes the init-container binary (Pattern A) when both images are configured. `get_lark_integration_status(check_runtime=True)` surfaces `sandbox_runtime_mode` (`none` / `gateway-download` / `init-container` / `broker`) and `sandbox_runtime_ready` (remote modes read the provisioner `GET /api/capabilities`: `lark_cli_init_image` / `lark_cli_broker_image`) so a green UI can't hide a chat-time `lark-cli: command not found`. Cheap status probes are explicitly not live-verified; users authorize or reconnect through the browser device-flow endpoints instead of running terminal commands.
 - **SkillScan**: `packages/harness/deerflow/skills/skillscan/` is the native deterministic scanner for `.skill` archives and agent-managed skill writes. It runs offline before the LLM scanner, emits structured findings (`rule_id`, `severity`, `file`, `line`, `message`, `remediation`, redacted `evidence` — category/analyzer are encoded in the `rule_id` prefix), blocks `CRITICAL`, and passes warning findings into `scan_skill_content()`. `scan_archive_preflight()` / `scan_skill_dir()` are pure sync functions (dispatch off the event loop); `enforce_static_scan()` applies the blocking policy and the `skill_scan.enabled` kill switch. The Python instance-client signal deliberately follows only a one-level, same-scope evidence chain (PR #4265 review): a proven imported constructor bound to a simple name, optional name-to-name alias propagation, rebinding invalidation, and a constructor-supported outbound method or context-manager use; bare canonical-looking names never fall back to module identity. Nested scopes never inherit client handles and inherit only constructor aliases proven stable by a binding-only enclosing-scope prepass. Comprehensions, walrus-bearing statements, annotations, executable expressions inside complex binding targets, unsupported operations, and ambiguous flows produce no finding from this signal; skipped constructs invalidate all names they may bind, while representative false negatives are pinned by `test_python_declared_false_negatives_stay_unreported`. Compound bodies are walked from isolated copies so wrapping code in `if True:` is not a bypass, while copied scope entries, binding-only prepasses, and AST visits consume a deterministic work budget and the walk stops after its first sink. Budget or recursion exhaustion skips only this best-effort signal and retains deterministic findings already collected for the file. Do not add Semgrep/OpenGrep or YAML rule-engine dependencies to the core path; Phase 1 rule specs live in Python constants next to their analyzers in `skillscan/orchestrator.py`.
 - **Skill Review Core**: `packages/harness/deerflow/skills/review/` provides read-only package snapshots, deterministic facts, resource/eval analysis, report rendering, and the CLI (`python -m deerflow.skills.review.cli`). It reuses the shared frontmatter helper and SkillScan; it must not import `app.*`, execute target scripts, install dependencies, or call networks. JSON contracts live in `contracts/skill_review/`. The `review_skill_package` built-in tool labels results with `review_subject_entry` and never `skill_context_entry`, so reviewing a target does not activate it, bind its `required-secrets`, or apply its `allowed-tools`. Its model-visible `ToolMessage.content` is a compact JSON payload with untrusted control tags neutralized; the full raw review payload, including Markdown renders, stays in `ToolMessage.artifact`. CI should run the CLI with `--fail-on error --fail-on-incomplete` so blocker/error findings and truncated/not-assessed packages fail the gate. The public `skills/public/skill-reviewer` skill owns semantic readiness review and suggestions only; mutation and runtime experiments remain owned by `skill-creator`.
 

backend/app/gateway/routers/integrations.py

@@ -78,7 +78,7 @@ class LarkIntegrationStatusResponse(BaseModel):
     install_path: str = Field(..., description="Host path of the managed Lark skill pack")
     cli: LarkCliProbeResponse
     auth: LarkAuthProbeResponse
-    sandbox_runtime_mode: str = Field("none", description="How lark-cli is provisioned into the sandbox: none, gateway-download, or init-container")
+    sandbox_runtime_mode: str = Field("none", description="How lark-cli is provisioned into the sandbox: none, gateway-download, init-container, or broker")
     sandbox_runtime_ready: bool = Field(False, description="Whether the sandbox lark-cli runtime is provisioned and usable at chat time")
     sandbox_runtime_detail: str | None = Field(None, description="Human-readable reason when the sandbox runtime is not ready")
 

backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py

@@ -959,6 +959,25 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             logger.warning(f"Could not determine Lark integration state: {e}")
             return False
 
+    @staticmethod
+    def _lark_broker_active(user_id: str | None = None) -> bool:
+        """Whether this user's sandbox should use the lark-cli broker (Pattern B).
+
+        True only when the Lark pack is installed AND the remote provisioner
+        reports a configured broker image. When true, the provisioner keeps the
+        credentials in a sidecar and the sandbox gets only a shim, so the
+        Gateway-side credential-mount overlay must not run either.
+        """
+        try:
+            if not AioSandboxProvider._lark_integration_active(user_id):
+                return False
+            from deerflow.integrations.lark_cli import sandbox_lark_broker_active
+
+            return sandbox_lark_broker_active()
+        except Exception as e:  # pragma: no cover - defensive
+            logger.warning(f"Could not determine Lark broker state: {e}")
+            return False
+
     @staticmethod
     def _get_lark_cli_runtime_mounts(*, user_id: str | None = None) -> list[tuple[str, str, bool]]:
         """Mount the per-user lark-cli config/data dirs used by Settings auth.
@@ -1915,6 +1934,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         effective_user_id = self._effective_acquire_user_id(user_id)
         extra_mounts = self._get_extra_mounts(thread_id, user_id=effective_user_id)
         provision_lark_cli_runtime = self._lark_integration_active(effective_user_id)
+        provision_lark_cli_broker = self._lark_broker_active(effective_user_id)
 
         # Enforce replicas: only warm-pool containers count toward eviction budget.
         # Active sandboxes are in use by live threads and must not be forcibly stopped.
@@ -1929,6 +1949,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             extra_mounts=extra_mounts or None,
             user_id=effective_user_id,
             provision_lark_cli_runtime=provision_lark_cli_runtime,
+            provision_lark_cli_broker=provision_lark_cli_broker,
         )
 
         # Wait for sandbox to be ready
@@ -1943,6 +1964,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         effective_user_id = self._effective_acquire_user_id(user_id)
         extra_mounts = await asyncio.to_thread(self._get_extra_mounts, thread_id, user_id=effective_user_id)
         provision_lark_cli_runtime = await asyncio.to_thread(self._lark_integration_active, effective_user_id)
+        provision_lark_cli_broker = await asyncio.to_thread(self._lark_broker_active, effective_user_id)
 
         # Enforce replicas: only warm-pool containers count toward eviction budget.
         # Active sandboxes are in use by live threads and must not be forcibly stopped.
@@ -1958,6 +1980,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             extra_mounts=extra_mounts or None,
             user_id=effective_user_id,
             provision_lark_cli_runtime=provision_lark_cli_runtime,
+            provision_lark_cli_broker=provision_lark_cli_broker,
         )
 
         # Wait for sandbox to be ready without blocking the event loop.

backend/packages/harness/deerflow/community/aio_sandbox/backend.py

@@ -110,6 +110,7 @@ class SandboxBackend(ABC):
         *,
         user_id: str | None = None,
         provision_lark_cli_runtime: bool = False,
+        provision_lark_cli_broker: bool = False,
     ) -> SandboxInfo:
         """Create/provision a new sandbox.
 
@@ -122,6 +123,10 @@ class SandboxBackend(ABC):
             provision_lark_cli_runtime: Ask the backend to provision the sandbox
                 lark-cli runtime via its native mechanism (e.g. the provisioner's
                 init container + emptyDir). Backends that can't do this ignore it.
+            provision_lark_cli_broker: Ask the backend to provision a lark-cli
+                broker sidecar (Pattern B, issue #4338) so credentials stay out of
+                the sandbox. Supersedes ``provision_lark_cli_runtime`` when the
+                backend supports it; backends that can't do this ignore it.
 
         Returns:
             SandboxInfo with connection details.

backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py

@@ -272,6 +272,7 @@ class LocalContainerBackend(SandboxBackend):
         *,
         user_id: str | None = None,
         provision_lark_cli_runtime: bool = False,
+        provision_lark_cli_broker: bool = False,
     ) -> SandboxInfo:
         """Start a new container and return its connection info.
 
@@ -283,6 +284,8 @@ class LocalContainerBackend(SandboxBackend):
                 interface compatibility with remote backends.
             provision_lark_cli_runtime: Ignored — the local backend provisions the
                 lark-cli runtime via the Gateway-download bind mount in extra_mounts.
+            provision_lark_cli_broker: Ignored — the local backend has no sandbox
+                boundary to protect, so it keeps the credential-mount overlay.
 
         Returns:
             SandboxInfo with container details.
@@ -290,7 +293,7 @@ class LocalContainerBackend(SandboxBackend):
         Raises:
             RuntimeError: If the container fails to start.
         """
-        del user_id, provision_lark_cli_runtime
+        del user_id, provision_lark_cli_runtime, provision_lark_cli_broker
         container_name = f"{self._container_prefix}-{sandbox_id}"
 
         # Retry loop: if Docker rejects the port (e.g. a stale container still

backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py

@@ -39,28 +39,41 @@ _PROVISIONER_EXTRA_MOUNT_PATHS = {
 }
 
 _LARK_CLI_RUNTIME_CONTAINER_PATH = "/mnt/integrations/lark-cli/runtime"
+_LARK_CLI_CONFIG_CONTAINER_PATH = "/mnt/integrations/lark-cli/config"
+_LARK_CLI_DATA_CONTAINER_PATH = "/mnt/integrations/lark-cli/data"
 
 
 def _provisioner_extra_mounts_payload(
     extra_mounts: list[tuple[str, str, bool]] | None,
     *,
     provision_lark_cli_runtime: bool = False,
+    provision_lark_cli_broker: bool = False,
 ) -> list[dict[str, object]]:
     """Return only extra mounts the provisioner knows how to recreate safely.
 
     When ``provision_lark_cli_runtime`` is set, the provisioner supplies the
     lark-cli runtime via an init container + emptyDir, so the runtime extra mount
     is dropped here to avoid a colliding hostPath/PVC mount at the same path. The
-    per-user config/data credential mounts are always forwarded.
+    per-user config/data credential mounts are still forwarded (they are mounted
+    into the sandbox in Pattern A).
+
+    When ``provision_lark_cli_broker`` is set (Pattern B, issue #4338), the
+    provisioner runs a broker sidecar that holds the credentials, so the
+    config/data mounts are **forwarded** (the provisioner wires them into the
+    sidecar, not the sandbox) while the runtime mount is dropped. Nothing changes
+    in this payload beyond keeping config/data available for the provisioner to
+    place; the runtime entry is dropped in both modes.
     """
     if not extra_mounts:
         return []
 
+    drop_runtime = provision_lark_cli_runtime or provision_lark_cli_broker
+
     payload: list[dict[str, object]] = []
     for host_path, container_path, read_only in extra_mounts:
         if container_path not in _PROVISIONER_EXTRA_MOUNT_PATHS:
             continue
-        if provision_lark_cli_runtime and container_path == _LARK_CLI_RUNTIME_CONTAINER_PATH:
+        if drop_runtime and container_path == _LARK_CLI_RUNTIME_CONTAINER_PATH:
             continue
         payload.append(
             {
@@ -115,6 +128,7 @@ class RemoteSandboxBackend(SandboxBackend):
         *,
         user_id: str | None = None,
         provision_lark_cli_runtime: bool = False,
+        provision_lark_cli_broker: bool = False,
     ) -> SandboxInfo:
         """Create a sandbox Pod + Service via the provisioner.
 
@@ -127,6 +141,7 @@ class RemoteSandboxBackend(SandboxBackend):
             extra_mounts,
             user_id=user_id,
             provision_lark_cli_runtime=provision_lark_cli_runtime,
+            provision_lark_cli_broker=provision_lark_cli_broker,
         )
 
     def destroy(self, info: SandboxInfo) -> None:
@@ -199,6 +214,7 @@ class RemoteSandboxBackend(SandboxBackend):
         *,
         user_id: str | None = None,
         provision_lark_cli_runtime: bool = False,
+        provision_lark_cli_broker: bool = False,
     ) -> SandboxInfo:
         """POST /api/sandboxes → create Pod + Service."""
         effective_user_id = user_id or get_effective_user_id()
@@ -209,10 +225,12 @@ class RemoteSandboxBackend(SandboxBackend):
             "user_id": effective_user_id,
             "include_legacy_skills": include_legacy_skills,
             "provision_lark_cli_runtime": provision_lark_cli_runtime,
+            "provision_lark_cli_broker": provision_lark_cli_broker,
         }
         provisioner_extra_mounts = _provisioner_extra_mounts_payload(
             extra_mounts,
             provision_lark_cli_runtime=provision_lark_cli_runtime,
+            provision_lark_cli_broker=provision_lark_cli_broker,
         )
         if provisioner_extra_mounts:
             payload["extra_mounts"] = provisioner_extra_mounts

backend/packages/harness/deerflow/integrations/lark_broker.py

@@ -0,0 +1,457 @@
+"""Lark CLI sandbox credential broker (Pattern B, issue #4338).
+
+Pattern A (PR #3971) provisions the ``lark-cli`` *binary* into the sandbox but
+still mounts the per-user credential directories (``config`` with the long-lived
+``appSecret`` and ``data`` with OAuth tokens) into the sandbox container, where
+the agent's ``bash`` tool can read them.
+
+This module implements the broker half of Pattern B: a long-lived process that
+owns ``lark-cli`` + the credentials and exposes only the *command surface* over
+loopback. The sandbox gets a tiny ``lark-cli`` shim on ``PATH`` that forwards
+argv/stdin to the broker, so the raw credential files never exist in the sandbox
+filesystem.
+
+Everything here is Python-3-stdlib only so the same module can run inside the
+minimal broker sidecar image without extra dependencies.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+import os
+import shutil
+import subprocess
+import sys
+import threading
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# ── Loopback wire contract ────────────────────────────────────────────────
+
+# The sandbox and the broker sidecar share the Pod network namespace, so the
+# shim reaches the broker on loopback. The port is fixed and injected into the
+# sandbox as DEERFLOW_LARK_BROKER_URL.
+LARK_BROKER_DEFAULT_HOST = "127.0.0.1"
+LARK_BROKER_DEFAULT_PORT = 8788
+LARK_BROKER_URL_ENV = "DEERFLOW_LARK_BROKER_URL"
+LARK_BROKER_EXEC_PATH = "/v1/exec"
+LARK_BROKER_HEALTH_PATH = "/v1/health"
+
+# Guards. Bounded so a compromised sandbox cannot exhaust the broker.
+LARK_BROKER_MAX_REQUEST_BYTES = 1 * 1024 * 1024
+LARK_BROKER_MAX_OUTPUT_BYTES = 4 * 1024 * 1024
+LARK_BROKER_DEFAULT_TIMEOUT_SECONDS = 120
+LARK_BROKER_MAX_CONCURRENCY = 8
+# Per-connection socket timeout. ThreadingHTTPServer spawns a thread per
+# connection, so without this a sandbox could declare a large Content-Length and
+# never send the body, parking a thread forever. Bounds the read so a slow/stuck
+# client releases its thread. (Loopback-only, so this only guards the sandbox
+# against tying up its own broker.)
+LARK_BROKER_SOCKET_TIMEOUT_SECONDS = 30
+
+# Optional env knob (comma-separated) for the subcommand denylist below.
+LARK_BROKER_DENY_SUBCOMMANDS_ENV = "DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS"
+
+# The runtime layout the sandbox sees is two files in ``bin/``:
+#
+#   bin/lark-cli          the POSIX-sh *launcher* (below) — the executable on PATH
+#   bin/lark-cli-shim.py  the Python *shim* body (below) it execs
+#
+# Pattern A's launcher is pure ``#!/bin/sh`` (it just execs the arch-dispatched
+# binary) and therefore has no runtime deps. The broker shim has to speak HTTP,
+# so it is Python — but shipping it as a bare ``#!/usr/bin/env python3`` script
+# would make every ``lark-cli`` call ENOEXEC/exit-127 on a sandbox image without
+# ``python3`` on PATH, and because broker mode is opt-in that could slip past CI
+# and only surface for an operator. The ``/bin/sh`` launcher resolves a Python 3
+# interpreter itself (using only shell built-ins, so it still works when PATH is
+# empty and the interpreter is pinned) and, if none exists, fails *loudly* with
+# an actionable message instead of an opaque ENOEXEC. ``DEERFLOW_LARK_BROKER_PYTHON``
+# pins a specific interpreter for images that ship Python under a non-standard
+# name.
+#
+# The launcher's path to the shim body is baked in at install time rather than
+# derived from ``$0``: when the sandbox runs ``lark-cli`` off PATH, ``$0`` is the
+# bare command name with no directory, so a sibling lookup would fail. The
+# install dir is a stable, shared mount, so the absolute path is valid in both
+# the init container that writes it and the sandbox that reads it.
+#
+# Both scripts are kept here as the single source of truth and mirrored by the
+# broker image build via ``install_shim``, exactly like ``LARK_CLI_SANDBOX_LAUNCHER_SCRIPT``
+# for Pattern A, so the image copies can never drift from the Gateway's.
+LARK_BROKER_PYTHON_ENV = "DEERFLOW_LARK_BROKER_PYTHON"
+LARK_CLI_BROKER_SHIM_FILENAME = "lark-cli-shim.py"
+_LARK_CLI_BROKER_SHIM_PATH_PLACEHOLDER = "@@LARK_CLI_BROKER_SHIM_PATH@@"
+
+LARK_CLI_BROKER_LAUNCHER_TEMPLATE = (
+    "#!/bin/sh\n"
+    "# DeerFlow lark-cli broker launcher (Pattern B). Resolves a Python 3\n"
+    "# interpreter and execs the forwarding shim. Fails loudly (not with an opaque\n"
+    "# ENOEXEC) when the sandbox image ships no python3. Uses only shell built-ins\n"
+    "# so it still works when PATH is empty and DEERFLOW_LARK_BROKER_PYTHON pins\n"
+    "# the interpreter.\n"
+    "set -eu\n"
+    'shim="' + _LARK_CLI_BROKER_SHIM_PATH_PLACEHOLDER + '"\n'
+    'if [ -n "${DEERFLOW_LARK_BROKER_PYTHON:-}" ]; then\n'
+    '  exec "$DEERFLOW_LARK_BROKER_PYTHON" "$shim" "$@"\n'
+    "fi\n"
+    "for _py in python3 python; do\n"
+    '  if command -v "$_py" >/dev/null 2>&1; then\n'
+    '    exec "$_py" "$shim" "$@"\n'
+    "  fi\n"
+    "done\n"
+    'echo "lark-cli: broker mode needs a Python 3 interpreter but none was found;" >&2\n'
+    'echo "          set DEERFLOW_LARK_BROKER_PYTHON to a python3 path in the sandbox image." >&2\n'
+    "exit 127\n"
+)
+
+
+def render_launcher_script(shim_path: str) -> str:
+    """Render the ``/bin/sh`` launcher with the shim body's absolute path baked in."""
+    return LARK_CLI_BROKER_LAUNCHER_TEMPLATE.replace(_LARK_CLI_BROKER_SHIM_PATH_PLACEHOLDER, shim_path)
+
+
+# The shim reads argv/stdin, POSTs to the broker, and replays the broker's
+# stdout/stderr/exit code. On any transport failure it fails loudly and non-zero
+# so a broker outage never looks like a successful lark-cli run. It is invoked as
+# ``<python> lark-cli-shim.py <args...>`` by the launcher above (so it does not
+# rely on its own shebang being resolvable), and stdin/argv pass straight through.
+LARK_CLI_BROKER_SHIM_SCRIPT = r'''#!/usr/bin/env python3
+"""DeerFlow lark-cli broker shim (Pattern B). Forwards argv/stdin to the broker.
+
+Note: the broker runs lark-cli in the *sidecar's* working directory and cannot
+see the sandbox filesystem, so cwd is intentionally not forwarded. Subcommands
+that read/write files relative to the sandbox cwd are unsupported in broker mode.
+"""
+import base64
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+
+BROKER_URL = os.environ.get("DEERFLOW_LARK_BROKER_URL", "http://127.0.0.1:8788")
+
+
+def _fail(message, code=127):
+    sys.stderr.write("lark-cli: " + message + "\n")
+    sys.exit(code)
+
+
+def main():
+    try:
+        stdin_bytes = b"" if sys.stdin is None or sys.stdin.isatty() else sys.stdin.buffer.read()
+    except Exception:
+        stdin_bytes = b""
+    payload = json.dumps(
+        {
+            "args": sys.argv[1:],
+            "stdin_b64": base64.b64encode(stdin_bytes).decode("ascii"),
+        }
+    ).encode("utf-8")
+    req = urllib.request.Request(
+        BROKER_URL.rstrip("/") + "/v1/exec",
+        data=payload,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=600) as resp:
+            body = json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        detail = ""
+        try:
+            detail = json.loads(exc.read().decode("utf-8")).get("error", "")
+        except Exception:
+            detail = ""
+        _fail("broker rejected request (HTTP %d%s)" % (exc.code, ": " + detail if detail else ""))
+    except (urllib.error.URLError, OSError) as exc:
+        _fail("broker unreachable at %s (%s)" % (BROKER_URL, exc))
+    except Exception as exc:  # noqa: BLE001
+        _fail("broker call failed (%s)" % exc)
+    sys.stdout.buffer.write(base64.b64decode(body.get("stdout_b64", "")))
+    sys.stdout.buffer.flush()
+    sys.stderr.buffer.write(base64.b64decode(body.get("stderr_b64", "")))
+    sys.stderr.buffer.flush()
+    sys.exit(int(body.get("exit_code", 1)))
+
+
+if __name__ == "__main__":
+    main()
+'''
+
+
+# ── Broker server ─────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class BrokerConfig:
+    """Runtime configuration for the broker sidecar."""
+
+    lark_cli_path: str
+    config_dir: str
+    data_dir: str
+    host: str = LARK_BROKER_DEFAULT_HOST
+    port: int = LARK_BROKER_DEFAULT_PORT
+    timeout_seconds: int = LARK_BROKER_DEFAULT_TIMEOUT_SECONDS
+    # Opt-in denylist of ``lark-cli`` subcommand paths the broker refuses to run
+    # (issue #4338 hardening). Each entry is a space-joined command prefix, e.g.
+    # "config show" or "auth token", matched against the leading non-flag tokens
+    # of the request. Narrows the command surface a prompt-injected agent can
+    # reach — the broker already removes the credential *files*, but the full
+    # command surface stays reachable unless a secret-dumping subcommand is denied
+    # here. Empty by default (no behavior change).
+    deny_subcommands: tuple[tuple[str, ...], ...] = ()
+
+    def credential_env(self) -> dict[str, str]:
+        """Env the broker injects into every lark-cli invocation.
+
+        The client never supplies these — the broker owns the credential paths,
+        so a sandbox process cannot point lark-cli at a different profile.
+        """
+        return {
+            "LARKSUITE_CLI_CONFIG_DIR": self.config_dir,
+            "LARKSUITE_CLI_DATA_DIR": self.data_dir,
+            "LARKSUITE_CLI_NO_UPDATE_NOTIFIER": "1",
+            "LARKSUITE_CLI_NO_SKILLS_NOTIFIER": "1",
+        }
+
+
+def parse_deny_subcommands(raw: str | None) -> tuple[tuple[str, ...], ...]:
+    """Parse the comma-separated denylist env into command-prefix tuples.
+
+    ``"config show, auth token"`` → ``(("config", "show"), ("auth", "token"))``.
+    Blank/whitespace-only entries are dropped.
+    """
+    if not raw:
+        return ()
+    prefixes: list[tuple[str, ...]] = []
+    for entry in raw.split(","):
+        tokens = tuple(entry.split())
+        if tokens:
+            prefixes.append(tokens)
+    return tuple(prefixes)
+
+
+def _denied_subcommand(deny: tuple[tuple[str, ...], ...], args: list[str]) -> tuple[str, ...] | None:
+    """Return the matched denylist prefix if ``args`` is a denied subcommand.
+
+    Matches against the leading non-flag tokens (options and their values are
+    skipped) so ``config --json show`` is still caught by a ``config show`` rule.
+    """
+    if not deny:
+        return None
+    positional = [token for token in args if not token.startswith("-")]
+    for prefix in deny:
+        if positional[: len(prefix)] == list(prefix):
+            return prefix
+    return None
+
+
+@dataclass(frozen=True)
+class ExecResult:
+    exit_code: int
+    stdout: bytes
+    stderr: bytes
+    truncated: bool
+
+
+def run_lark_cli(config: BrokerConfig, args: list[str], stdin: bytes) -> ExecResult:
+    """Run a single ``lark-cli`` invocation with broker-owned credentials.
+
+    ``args`` is passed as an argv list with ``shell=False`` so a sandbox-supplied
+    argument can never be shell-interpreted into a second command. A configured
+    ``deny_subcommands`` prefix is refused before the binary is ever spawned.
+    """
+    denied = _denied_subcommand(config.deny_subcommands, args)
+    if denied is not None:
+        message = f"lark-cli: subcommand '{' '.join(denied)}' is disabled in broker mode\n"
+        return ExecResult(126, b"", message.encode("utf-8"), False)
+    env = {**os.environ, **config.credential_env()}
+    try:
+        completed = subprocess.run(  # noqa: S603 - argv list, shell=False, fixed binary
+            [config.lark_cli_path, *args],
+            input=stdin,
+            capture_output=True,
+            timeout=config.timeout_seconds,
+            check=False,
+            env=env,
+        )
+    except subprocess.TimeoutExpired:
+        return ExecResult(124, b"", b"lark-cli: broker timed out\n", False)
+    except FileNotFoundError:
+        return ExecResult(127, b"", b"lark-cli: binary not found in broker\n", False)
+
+    stdout, out_trunc = _cap(completed.stdout or b"")
+    stderr, err_trunc = _cap(completed.stderr or b"")
+    return ExecResult(completed.returncode, stdout, stderr, out_trunc or err_trunc)
+
+
+def _cap(data: bytes) -> tuple[bytes, bool]:
+    if len(data) <= LARK_BROKER_MAX_OUTPUT_BYTES:
+        return data, False
+    return data[:LARK_BROKER_MAX_OUTPUT_BYTES], True
+
+
+def make_handler(config: BrokerConfig) -> type[BaseHTTPRequestHandler]:
+    """Build a request handler bound to ``config``.
+
+    A bounded semaphore caps concurrency so a flood of sandbox calls cannot spawn
+    unbounded ``lark-cli`` subprocesses.
+    """
+    semaphore = threading.BoundedSemaphore(LARK_BROKER_MAX_CONCURRENCY)
+
+    class Handler(BaseHTTPRequestHandler):
+        # Bound per-connection reads so a client that declares a large
+        # Content-Length and never sends the body cannot park its thread forever
+        # (ThreadingHTTPServer is one-thread-per-connection). stdlib reads this
+        # to arm socket timeouts.
+        timeout = LARK_BROKER_SOCKET_TIMEOUT_SECONDS
+
+        # Quiet: default BaseHTTPRequestHandler logs to stderr per request.
+        def log_message(self, *_args: Any) -> None:  # noqa: D401
+            return
+
+        def _send_json(self, status: int, body: dict[str, Any]) -> None:
+            payload = json.dumps(body).encode("utf-8")
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+
+        def do_GET(self) -> None:  # noqa: N802 - stdlib API
+            if self.path.rstrip("/") == LARK_BROKER_HEALTH_PATH:
+                self._send_json(200, {"ok": True})
+                return
+            self._send_json(404, {"error": "not found"})
+
+        def do_POST(self) -> None:  # noqa: N802 - stdlib API
+            if self.path.rstrip("/") != LARK_BROKER_EXEC_PATH:
+                self._send_json(404, {"error": "not found"})
+                return
+            try:
+                length = int(self.headers.get("Content-Length", "0"))
+            except ValueError:
+                self._send_json(400, {"error": "bad content-length"})
+                return
+            if length <= 0 or length > LARK_BROKER_MAX_REQUEST_BYTES:
+                self._send_json(413, {"error": "request too large"})
+                return
+            try:
+                request = json.loads(self.rfile.read(length).decode("utf-8"))
+                args = request["args"]
+                if not isinstance(args, list) or not all(isinstance(a, str) for a in args):
+                    raise ValueError("args must be a list of strings")
+                stdin = base64.b64decode(request.get("stdin_b64", "") or "")
+            except Exception:  # noqa: BLE001 - untrusted client input
+                self._send_json(400, {"error": "invalid request"})
+                return
+
+            if not semaphore.acquire(blocking=False):
+                self._send_json(503, {"error": "broker busy"})
+                return
+            try:
+                result = run_lark_cli(config, args, stdin)
+            except Exception:  # noqa: BLE001 - keep the wire contract uniform
+                # run_lark_cli already maps the expected failures (timeout,
+                # missing binary) to ExecResults; anything else (OSError,
+                # PermissionError, …) would otherwise close the connection with
+                # no body and surface as an opaque transport error in the shim.
+                # Return a structured 500 so the shim reports a meaningful error.
+                logger.exception("lark-cli exec failed unexpectedly")
+                self._send_json(500, {"error": "broker exec failed"})
+                return
+            finally:
+                semaphore.release()
+
+            self._send_json(
+                200,
+                {
+                    "exit_code": result.exit_code,
+                    "stdout_b64": base64.b64encode(result.stdout).decode("ascii"),
+                    "stderr_b64": base64.b64encode(result.stderr).decode("ascii"),
+                    "truncated": result.truncated,
+                },
+            )
+
+    return Handler
+
+
+def serve(config: BrokerConfig) -> ThreadingHTTPServer:
+    """Start the broker HTTP server bound to loopback and return it."""
+    if not shutil.which(config.lark_cli_path) and not os.path.isfile(config.lark_cli_path):
+        logger.warning("lark-cli not found at %s; broker will report 127 for exec", config.lark_cli_path)
+    server = ThreadingHTTPServer((config.host, config.port), make_handler(config))
+    logger.info("lark-cli broker listening on %s:%d", config.host, config.port)
+    return server
+
+
+def install_shim(dest_dir: str, *, version: str | None = None) -> str:
+    """Write the launcher + shim + runtime marker into the sandbox runtime dir.
+
+    Called by the broker image's ``install-shim`` init-container mode. Produces
+    the same ``bin/lark-cli`` + ``.deerflow-lark-cli-runtime.json`` layout Pattern
+    A stages, but marked ``kind="shim"`` so the runtime validator knows the
+    ``linux-*`` binaries are intentionally absent (the sidecar holds the real
+    binary).
+
+    Two files are written into ``bin/``: the executable-on-PATH ``lark-cli`` is a
+    ``/bin/sh`` *launcher* that resolves a Python 3 interpreter and execs the
+    ``lark-cli-shim.py`` *body* next to it. Splitting them keeps broker mode from
+    silently failing with ENOEXEC on a sandbox image that has no ``python3`` (the
+    launcher fails loudly with an actionable message instead). Both come from the
+    in-process constants so the image copy can never drift from the Gateway's.
+    """
+    dest = os.path.abspath(dest_dir)
+    bin_dir = os.path.join(dest, "bin")
+    os.makedirs(bin_dir, exist_ok=True)
+    shim_body = os.path.join(bin_dir, LARK_CLI_BROKER_SHIM_FILENAME)
+    with open(shim_body, "w", encoding="utf-8") as handle:
+        handle.write(LARK_CLI_BROKER_SHIM_SCRIPT)
+    os.chmod(shim_body, 0o755)
+    launcher = os.path.join(bin_dir, "lark-cli")
+    with open(launcher, "w", encoding="utf-8") as handle:
+        handle.write(render_launcher_script(shim_body))
+    os.chmod(launcher, 0o755)
+    marker = os.path.join(dest, ".deerflow-lark-cli-runtime.json")
+    with open(marker, "w", encoding="utf-8") as handle:
+        json.dump({"version": version or "unknown", "kind": "shim"}, handle)
+    return launcher
+
+
+def _config_from_env() -> BrokerConfig:
+    return BrokerConfig(
+        lark_cli_path=os.environ.get("DEERFLOW_LARK_BROKER_CLI", "lark-cli"),
+        config_dir=os.environ.get("LARKSUITE_CLI_CONFIG_DIR", "/var/lark/config"),
+        data_dir=os.environ.get("LARKSUITE_CLI_DATA_DIR", "/var/lark/data"),
+        host=os.environ.get("DEERFLOW_LARK_BROKER_HOST", LARK_BROKER_DEFAULT_HOST),
+        port=int(os.environ.get("DEERFLOW_LARK_BROKER_PORT", str(LARK_BROKER_DEFAULT_PORT))),
+        timeout_seconds=int(os.environ.get("DEERFLOW_LARK_BROKER_TIMEOUT", str(LARK_BROKER_DEFAULT_TIMEOUT_SECONDS))),
+        deny_subcommands=parse_deny_subcommands(os.environ.get(LARK_BROKER_DENY_SUBCOMMANDS_ENV)),
+    )
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+    argv = sys.argv[1:]
+    if argv and argv[0] == "install-shim":
+        dest = argv[1] if len(argv) > 1 else os.environ.get("LARK_CLI_RUNTIME_DEST", "/mnt/integrations/lark-cli/runtime")
+        launcher = install_shim(dest, version=os.environ.get("LARK_CLI_VERSION"))
+        logger.info("Installed lark-cli broker shim at %s", launcher)
+        return
+    server = serve(_config_from_env())
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.shutdown()
+
+
+if __name__ == "__main__":
+    main()

backend/packages/harness/deerflow/integrations/lark_cli.py

@@ -73,6 +73,7 @@ except ImportError:  # pragma: no cover - Windows fallback
 
 from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import Paths, get_paths
+from deerflow.integrations.lark_broker import LARK_BROKER_URL_ENV
 from deerflow.skills.installer import is_executable_binary_prefix, is_symlink_member, is_unsafe_zip_member
 from deerflow.skills.parser import parse_skill_file
 from deerflow.skills.permissions import make_skill_tree_sandbox_readable
@@ -108,6 +109,11 @@ LARK_CLI_SANDBOX_RUNTIME_DIR = "/mnt/integrations/lark-cli/runtime"
 LARK_CLI_LINUX_ARCHES = ("amd64", "arm64")
 LARK_CLI_RUNTIME_MANIFEST_FILE = ".deerflow-lark-cli-runtime.json"
 
+# Pattern B (issue #4338): loopback URL the sandbox shim uses to reach the broker
+# sidecar. LARK_BROKER_URL_ENV is imported from the broker module so the shim,
+# server, and Gateway overlay share one source of truth.
+LARK_BROKER_SANDBOX_URL = "http://127.0.0.1:8788"
+
 # Arch-dispatch launcher for the sandbox runtime layout. Shared by the Gateway
 # writer (`_write_lark_cli_sandbox_launcher`) and the `docker/lark-cli-init`
 # init image so the two producers of `bin/lark-cli` never drift.
@@ -522,13 +528,23 @@ def _lark_cli_managed_path() -> str | None:
     return None
 
 
-def lark_cli_env_overlay(user_id: str, *, sandbox_paths: bool = False) -> dict[str, str]:
+def lark_cli_env_overlay(user_id: str, *, sandbox_paths: bool = False, broker: bool = False) -> dict[str, str]:
     """Environment overlay for lark-cli using DeerFlow-managed credentials.
 
     The directories are per-user so a local trusted-mode login cannot bleed
-    across accounts. Auth Proxy support can later replace these directories for
-    sandbox execution without changing the status API contract.
+    across accounts.
+
+    When ``broker`` is set (Pattern B, issue #4338), the sandbox talks to a
+    broker sidecar that owns the credentials, so the overlay carries only the
+    broker URL and the runtime PATH — never ``LARKSUITE_CLI_CONFIG_DIR`` /
+    ``DATA_DIR``. This keeps the plaintext app secret / OAuth tokens out of the
+    sandbox filesystem entirely. ``broker`` implies ``sandbox_paths``.
     """
+    if broker:
+        return {
+            "PATH": f"{LARK_CLI_SANDBOX_RUNTIME_DIR}/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+            LARK_BROKER_URL_ENV: LARK_BROKER_SANDBOX_URL,
+        }
     if sandbox_paths:
         config_dir: Path | str = LARK_CLI_SANDBOX_CONFIG_DIR
         data_dir: Path | str = LARK_CLI_SANDBOX_DATA_DIR
@@ -660,7 +676,12 @@ def _resolve_sandbox_runtime_readiness(
     - ``gateway-download``: local AIO — the Gateway stages a ``sandbox-cli`` dir
       and bind-mounts it; ready when that dir validates.
     - ``init-container``: remote provisioner — a lark-cli init image provisions
-      the runtime; ready when the provisioner reports the init image is configured.
+      the runtime binary + plaintext credential mounts (Pattern A); ready when
+      the provisioner reports the init image is configured.
+    - ``broker``: remote provisioner — a lark-cli broker sidecar holds the
+      credentials and the sandbox gets only a shim (Pattern B, issue #4338);
+      ready when the provisioner reports the broker image is configured. Broker
+      supersedes ``init-container`` when both are available.
 
     ``probe`` gates the (best-effort, short-timeout) provisioner capability call.
     """
@@ -670,12 +691,16 @@ def _resolve_sandbox_runtime_readiness(
     if _uses_remote_provisioner(config):
         if not probe:
             return "init-container", False, None
-        init_image = _probe_provisioner_lark_cli_init_image(config)
-        if init_image is None:
-            return "init-container", False, "Could not reach the provisioner to confirm the lark-cli init image."
-        if not init_image:
-            return "init-container", False, "The provisioner has no lark-cli init image configured (LARK_CLI_INIT_IMAGE)."
-        return "init-container", True, None
+        caps = _probe_provisioner_capabilities(config)
+        if caps is None:
+            return "init-container", False, "Could not reach the provisioner to confirm the lark-cli runtime image."
+        # Pattern B (broker) supersedes Pattern A (init-container binary) when
+        # the provisioner has a broker image configured.
+        if caps["lark_cli_broker_image"]:
+            return "broker", True, None
+        if caps["lark_cli_init_image"]:
+            return "init-container", True, None
+        return "init-container", False, "The provisioner has no lark-cli runtime image configured (LARK_CLI_INIT_IMAGE / LARK_CLI_BROKER_IMAGE)."
 
     # Local AIO: Gateway-download runtime dir.
     runtime_dir = lark_cli_managed_sandbox_dir()
@@ -686,6 +711,61 @@ def _resolve_sandbox_runtime_readiness(
     return "gateway-download", True, None
 
 
+LARK_BROKER_MODE_TTL_SECONDS = 60
+# Negative results (broker not active) are cached longer than positive ones: a
+# non-broker remote-provisioner deployment stays non-broker for the life of the
+# process far more often than it flips on, so this keeps the hot bash path from
+# re-probing every minute. A positive result still refreshes on the shorter TTL.
+LARK_BROKER_MODE_NEGATIVE_TTL_SECONDS = 300
+# Tight probe budget on the per-bash-call hot path: unlike the Settings status
+# probe (5s, user is waiting on a page), this runs inline before a sandbox
+# lark-cli command, so a slow/unreachable provisioner must not add seconds of
+# latency to every first-call-per-TTL for non-broker deployments.
+LARK_BROKER_MODE_PROBE_TIMEOUT_SECONDS = 1.5
+# Guards the cache attribute on sandbox_lark_broker_active against concurrent
+# bash invocations so the correctness story doesn't rely on idempotent races.
+_LARK_BROKER_MODE_CACHE_LOCK = threading.Lock()
+
+
+def sandbox_lark_broker_active(config: AppConfig | None = None) -> bool:
+    """Whether sandbox ``lark-cli`` runs in broker mode (Pattern B).
+
+    Cached with a short TTL because it is consulted on every ``lark-cli`` bash
+    call and reads the provisioner capability over HTTP. Broker mode requires a
+    remote provisioner that reports a configured broker image; any other config
+    (local AIO, init-container binary mode, unreachable provisioner) is False, so
+    the caller falls back to the credential-mount overlay.
+
+    The probe uses a tight timeout and negatives are cached longer than positives
+    so a non-broker remote-provisioner deployment does not pay a latency penalty
+    on the bash hot path.
+    """
+    if config is None:
+        try:
+            from deerflow.config.app_config import get_app_config
+
+            config = get_app_config()
+        except Exception:  # noqa: BLE001 - degrade to non-broker overlay
+            return False
+
+    now = time.monotonic()
+    with _LARK_BROKER_MODE_CACHE_LOCK:
+        cached = getattr(sandbox_lark_broker_active, "_cache", None)
+        if cached is not None:
+            ts, value = cached
+            ttl = LARK_BROKER_MODE_TTL_SECONDS if value else LARK_BROKER_MODE_NEGATIVE_TTL_SECONDS
+            if now - ts < ttl:
+                return value
+
+    active = False
+    if _uses_aio_sandbox(config) and _uses_remote_provisioner(config):
+        caps = _probe_provisioner_capabilities(config, timeout=LARK_BROKER_MODE_PROBE_TIMEOUT_SECONDS)
+        active = bool(caps and caps["lark_cli_broker_image"])
+    with _LARK_BROKER_MODE_CACHE_LOCK:
+        sandbox_lark_broker_active._cache = (now, active)  # type: ignore[attr-defined]
+    return active
+
+
 def get_lark_integration_status(
     user_id: str,
     config: AppConfig,
@@ -862,12 +942,14 @@ def _uses_remote_provisioner(config: AppConfig) -> bool:
     return bool(_sandbox_config_value(config, "provisioner_url"))
 
 
-def _probe_provisioner_lark_cli_init_image(config: AppConfig) -> bool | None:
-    """Best-effort read of the provisioner's lark-cli init-image capability.
+def _probe_provisioner_capabilities(config: AppConfig, *, timeout: float = 5.0) -> dict[str, bool] | None:
+    """Best-effort read of the provisioner's lark-cli capabilities.
 
-    Returns True/False when the provisioner answers, or None when it can't be
-    reached. Used only to surface a sandbox-runtime readiness signal; failures
-    degrade to "not ready" rather than raising.
+    Returns the capability dict when the provisioner answers, or None when it
+    can't be reached. Used both for the status readiness signal and to select
+    broker vs. binary mode on the bash hot path; failures degrade to "not
+    ready"/"not broker" rather than raising. ``timeout`` is caller-tunable so the
+    per-bash-call probe can use a tighter budget than the Settings status probe.
     """
     base = _sandbox_config_value(config, "provisioner_url")
     if not base:
@@ -877,9 +959,14 @@ def _probe_provisioner_lark_cli_init_image(config: AppConfig) -> bool | None:
     url = f"{base.rstrip('/')}/api/capabilities"
     try:
         request = urllib.request.Request(url, headers={"User-Agent": "deer-flow", **headers})
-        with urllib.request.urlopen(request, timeout=5) as response:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
             payload = json.loads(response.read().decode("utf-8"))
-        return bool(payload.get("lark_cli_init_image"))
+        if not isinstance(payload, dict):
+            return None
+        return {
+            "lark_cli_init_image": bool(payload.get("lark_cli_init_image")),
+            "lark_cli_broker_image": bool(payload.get("lark_cli_broker_image")),
+        }
     except Exception:
         return None
 

backend/packages/harness/deerflow/sandbox/tools.py

@@ -1738,13 +1738,18 @@ def _lark_cli_env_from_runtime(runtime: Runtime, command: str, *, sandbox_paths:
     unrelated unauthenticated profile. Keep this scoped to commands that
     actually call ``lark-cli`` so ordinary bash calls do not switch AIO into the
     env-bearing execution path.
+
+    In broker mode (Pattern B, issue #4338) a sidecar owns the credentials, so
+    the overlay carries only the broker URL + runtime PATH — the config/data
+    directories are never injected into the sandbox.
     """
     if not _LARK_CLI_COMMAND_RE.search(command):
         return None
     try:
-        from deerflow.integrations.lark_cli import lark_cli_env_overlay
+        from deerflow.integrations.lark_cli import lark_cli_env_overlay, sandbox_lark_broker_active
 
-        return lark_cli_env_overlay(resolve_runtime_user_id(runtime), sandbox_paths=sandbox_paths)
+        broker = sandbox_paths and sandbox_lark_broker_active()
+        return lark_cli_env_overlay(resolve_runtime_user_id(runtime), sandbox_paths=sandbox_paths, broker=broker)
     except Exception:
         logger.warning("Could not build Lark CLI env overlay; running command without managed auth", exc_info=True)
         return None

backend/tests/test_aio_sandbox_provider.py

@@ -518,6 +518,7 @@ def test_remote_backend_create_forwards_effective_user_id(monkeypatch):
         "user_id": "user-7",
         "include_legacy_skills": True,
         "provision_lark_cli_runtime": False,
+        "provision_lark_cli_broker": False,
     }
 
 
@@ -562,18 +563,52 @@ def test_create_sandbox_requests_runtime_when_lark_installed(tmp_path, monkeypat
 
     captured: dict = {}
 
-    def _create(thread_id, sandbox_id, *, extra_mounts=None, user_id=None, provision_lark_cli_runtime=False):
+    def _create(thread_id, sandbox_id, *, extra_mounts=None, user_id=None, provision_lark_cli_runtime=False, provision_lark_cli_broker=False):
         captured["provision_lark_cli_runtime"] = provision_lark_cli_runtime
+        captured["provision_lark_cli_broker"] = provision_lark_cli_broker
         return aio_mod.SandboxInfo(sandbox_id=sandbox_id, sandbox_url="http://sandbox")
 
     provider._backend = SimpleNamespace(create=_create, destroy=MagicMock(), discover=MagicMock(return_value=None))
     monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready", lambda *_a, **_k: True)
     monkeypatch.setattr(provider, "_get_extra_mounts", lambda *_a, **_k: [])
     monkeypatch.setattr(aio_mod.AioSandboxProvider, "_lark_integration_active", staticmethod(lambda user_id=None: True))
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_lark_broker_active", staticmethod(lambda user_id=None: False))
     monkeypatch.setattr(provider, "_register_created_sandbox", lambda *a, **k: "sandbox-lark")
 
     provider._create_sandbox("thread-lark", "sandbox-lark", user_id="alice")
     assert captured["provision_lark_cli_runtime"] is True
+    assert captured["provision_lark_cli_broker"] is False
+
+
+def test_create_sandbox_requests_broker_when_active(tmp_path, monkeypatch):
+    """Broker mode (Pattern B) is requested when the provisioner reports it."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._config = {"replicas": 3}
+    provider._thread_locks = {}
+    provider._warm_pool = {}
+    provider._sandbox_infos = {}
+    provider._thread_sandboxes = {}
+    provider._last_activity = {}
+    provider._lock = aio_mod.threading.Lock()
+
+    captured: dict = {}
+
+    def _create(thread_id, sandbox_id, *, extra_mounts=None, user_id=None, provision_lark_cli_runtime=False, provision_lark_cli_broker=False):
+        captured["provision_lark_cli_runtime"] = provision_lark_cli_runtime
+        captured["provision_lark_cli_broker"] = provision_lark_cli_broker
+        return aio_mod.SandboxInfo(sandbox_id=sandbox_id, sandbox_url="http://sandbox")
+
+    provider._backend = SimpleNamespace(create=_create, destroy=MagicMock(), discover=MagicMock(return_value=None))
+    monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready", lambda *_a, **_k: True)
+    monkeypatch.setattr(provider, "_get_extra_mounts", lambda *_a, **_k: [])
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_lark_integration_active", staticmethod(lambda user_id=None: True))
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_lark_broker_active", staticmethod(lambda user_id=None: True))
+    monkeypatch.setattr(provider, "_register_created_sandbox", lambda *a, **k: "sandbox-broker")
+
+    provider._create_sandbox("thread-broker", "sandbox-broker", user_id="alice")
+    assert captured["provision_lark_cli_runtime"] is True
+    assert captured["provision_lark_cli_broker"] is True
 
 
 def test_create_sandbox_skips_runtime_when_lark_absent(tmp_path, monkeypatch):
@@ -590,18 +625,21 @@ def test_create_sandbox_skips_runtime_when_lark_absent(tmp_path, monkeypatch):
 
     captured: dict = {}
 
-    def _create(thread_id, sandbox_id, *, extra_mounts=None, user_id=None, provision_lark_cli_runtime=False):
+    def _create(thread_id, sandbox_id, *, extra_mounts=None, user_id=None, provision_lark_cli_runtime=False, provision_lark_cli_broker=False):
         captured["provision_lark_cli_runtime"] = provision_lark_cli_runtime
+        captured["provision_lark_cli_broker"] = provision_lark_cli_broker
         return aio_mod.SandboxInfo(sandbox_id=sandbox_id, sandbox_url="http://sandbox")
 
     provider._backend = SimpleNamespace(create=_create, destroy=MagicMock(), discover=MagicMock(return_value=None))
     monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready", lambda *_a, **_k: True)
     monkeypatch.setattr(provider, "_get_extra_mounts", lambda *_a, **_k: [])
     monkeypatch.setattr(aio_mod.AioSandboxProvider, "_lark_integration_active", staticmethod(lambda user_id=None: False))
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_lark_broker_active", staticmethod(lambda user_id=None: False))
     monkeypatch.setattr(provider, "_register_created_sandbox", lambda *a, **k: "sandbox-nolark")
 
     provider._create_sandbox("thread-nolark", "sandbox-nolark", user_id="alice")
     assert captured["provision_lark_cli_runtime"] is False
+    assert captured["provision_lark_cli_broker"] is False
 
 
 # ── Sandbox client teardown (#2872) ──────────────────────────────────────────

backend/tests/test_lark_broker.py

@@ -0,0 +1,357 @@
+"""Tests for the Lark CLI sandbox credential broker (Pattern B, issue #4338)."""
+
+from __future__ import annotations
+
+import base64
+import json
+import os
+import subprocess
+import sys
+import textwrap
+import urllib.request
+from http.client import HTTPConnection
+from pathlib import Path
+
+import pytest
+
+from deerflow.integrations import lark_broker
+from deerflow.integrations.lark_broker import BrokerConfig, run_lark_cli, serve
+
+
+def _fake_lark_cli(tmp_path: Path) -> str:
+    """A stub 'lark-cli' that echoes argv, stdin, and the credential env.
+
+    Lets tests assert argv fidelity, stdin round-trip, and that the broker (not
+    the caller) controls LARKSUITE_CLI_CONFIG_DIR / DATA_DIR.
+    """
+    script = tmp_path / "fake-lark-cli"
+    script.write_text(
+        textwrap.dedent(
+            """\
+            #!/usr/bin/env python3
+            import json, os, sys
+            sys.stderr.write("ERR:" + " ".join(sys.argv[1:]) + "\\n")
+            print(json.dumps({
+                "argv": sys.argv[1:],
+                "stdin": sys.stdin.read(),
+                "config_dir": os.environ.get("LARKSUITE_CLI_CONFIG_DIR"),
+                "data_dir": os.environ.get("LARKSUITE_CLI_DATA_DIR"),
+            }))
+            sys.exit(7 if "--boom" in sys.argv else 0)
+            """
+        ),
+        encoding="utf-8",
+    )
+    script.chmod(0o755)
+    return str(script)
+
+
+def _config(tmp_path: Path, port: int = 0) -> BrokerConfig:
+    return BrokerConfig(
+        lark_cli_path=_fake_lark_cli(tmp_path),
+        config_dir="/broker/only/config",
+        data_dir="/broker/only/data",
+        port=port,
+    )
+
+
+# ── run_lark_cli (in-process, no server) ───────────────────────────────────
+
+
+def test_run_lark_cli_forwards_argv_stdin_and_credential_env(tmp_path: Path) -> None:
+    config = _config(tmp_path)
+    result = run_lark_cli(config, ["auth", "status", "--json"], b"piped-input")
+
+    assert result.exit_code == 0
+    payload = json.loads(result.stdout.decode())
+    assert payload["argv"] == ["auth", "status", "--json"]
+    assert payload["stdin"] == "piped-input"
+    # The broker, not the caller, owns the credential paths.
+    assert payload["config_dir"] == "/broker/only/config"
+    assert payload["data_dir"] == "/broker/only/data"
+    assert result.stderr == b"ERR:auth status --json\n"
+
+
+def test_run_lark_cli_propagates_exit_code(tmp_path: Path) -> None:
+    result = run_lark_cli(_config(tmp_path), ["do", "--boom"], b"")
+    assert result.exit_code == 7
+
+
+def test_run_lark_cli_never_shell_interprets_args(tmp_path: Path) -> None:
+    # A shell metacharacter must reach the binary as one literal arg, not run a
+    # second command (shell=False, argv list).
+    result = run_lark_cli(_config(tmp_path), ["value; touch /tmp/pwned"], b"")
+    payload = json.loads(result.stdout.decode())
+    assert payload["argv"] == ["value; touch /tmp/pwned"]
+    assert not Path("/tmp/pwned").exists()
+
+
+def test_run_lark_cli_missing_binary_returns_127(tmp_path: Path) -> None:
+    config = BrokerConfig(lark_cli_path=str(tmp_path / "nope"), config_dir="c", data_dir="d")
+    result = run_lark_cli(config, ["x"], b"")
+    assert result.exit_code == 127
+
+
+# ── HTTP server ────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def broker_server(tmp_path: Path):
+    server = serve(_config(tmp_path, port=0))
+    import threading
+
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    host, port = server.server_address[0], server.server_address[1]
+    try:
+        yield host, port
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+
+def _post_exec(host: str, port: int, body: dict) -> tuple[int, dict]:
+    conn = HTTPConnection(host, port, timeout=10)
+    data = json.dumps(body).encode()
+    conn.request("POST", "/v1/exec", body=data, headers={"Content-Type": "application/json"})
+    resp = conn.getresponse()
+    parsed = json.loads(resp.read().decode())
+    conn.close()
+    return resp.status, parsed
+
+
+def test_exec_endpoint_round_trips(broker_server) -> None:
+    host, port = broker_server
+    status, body = _post_exec(host, port, {"args": ["ping"], "stdin_b64": base64.b64encode(b"hi").decode()})
+    assert status == 200
+    assert body["exit_code"] == 0
+    payload = json.loads(base64.b64decode(body["stdout_b64"]).decode())
+    assert payload["argv"] == ["ping"]
+    assert payload["stdin"] == "hi"
+
+
+def test_exec_endpoint_ignores_client_supplied_credential_paths(broker_server) -> None:
+    host, port = broker_server
+    # Even if a malicious client tries to smuggle env-like args, the broker sets
+    # the credential dirs itself; the stub reports the broker-owned values.
+    status, body = _post_exec(host, port, {"args": ["whoami"]})
+    assert status == 200
+    payload = json.loads(base64.b64decode(body["stdout_b64"]).decode())
+    assert payload["config_dir"] == "/broker/only/config"
+    assert payload["data_dir"] == "/broker/only/data"
+
+
+def test_exec_endpoint_rejects_invalid_body(broker_server) -> None:
+    host, port = broker_server
+    status, body = _post_exec(host, port, {"args": "not-a-list"})
+    assert status == 400
+
+
+def test_health_endpoint(broker_server) -> None:
+    host, port = broker_server
+    with urllib.request.urlopen(f"http://{host}:{port}/v1/health", timeout=5) as resp:
+        assert json.loads(resp.read().decode())["ok"] is True
+
+
+def test_exec_endpoint_returns_500_json_on_unexpected_error(broker_server, monkeypatch) -> None:
+    """An unexpected exec error must return a structured 500, not close the
+    connection with no body (which the shim would see as an opaque transport
+    failure)."""
+    host, port = broker_server
+
+    def _boom(*_args, **_kwargs):
+        raise PermissionError("simulated exec failure")
+
+    monkeypatch.setattr(lark_broker, "run_lark_cli", _boom)
+    status, body = _post_exec(host, port, {"args": ["auth", "status"]})
+    assert status == 500
+    assert body["error"]
+
+
+# ── Shim script ─────────────────────────────────────────────────────────────
+
+
+def test_shim_forwards_and_replays_exit_code(broker_server, tmp_path: Path) -> None:
+    host, port = broker_server
+    shim = tmp_path / "lark-cli"
+    shim.write_text(lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT, encoding="utf-8")
+    shim.chmod(0o755)
+
+    completed = subprocess.run(
+        [sys.executable, str(shim), "do", "--boom"],
+        input=b"",
+        capture_output=True,
+        env={**os.environ, lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}"},
+        timeout=30,
+    )
+    assert completed.returncode == 7
+    assert b"ERR:do --boom" in completed.stderr
+
+
+def test_shim_fails_loudly_when_broker_unreachable(tmp_path: Path) -> None:
+    shim = tmp_path / "lark-cli"
+    shim.write_text(lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT, encoding="utf-8")
+    shim.chmod(0o755)
+    completed = subprocess.run(
+        [sys.executable, str(shim), "auth", "status"],
+        input=b"",
+        capture_output=True,
+        # Nothing is listening here.
+        env={**os.environ, lark_broker.LARK_BROKER_URL_ENV: "http://127.0.0.1:1"},
+        timeout=30,
+    )
+    assert completed.returncode != 0
+    assert b"broker unreachable" in completed.stderr
+
+
+def test_install_shim_writes_runtime_layout(tmp_path: Path) -> None:
+    """install-shim mode stages the same bin/lark-cli + marker layout Pattern A
+    uses, marked kind=shim so the runtime validator tolerates absent binaries.
+
+    bin/lark-cli is now the POSIX-sh launcher and the Python body lives beside it
+    as bin/lark-cli-shim.py, so broker mode does not hard-depend on a resolvable
+    #!/usr/bin/env python3 shebang.
+    """
+    dest = tmp_path / "runtime"
+    launcher = lark_broker.install_shim(str(dest), version="v1.0.65")
+
+    assert Path(launcher) == dest / "bin" / "lark-cli"
+    launcher_text = (dest / "bin" / "lark-cli").read_text(encoding="utf-8")
+    assert launcher_text.startswith("#!/bin/sh")
+    # The shim body's absolute path is baked into the launcher (not derived from
+    # $0, which is the bare command name when run off PATH).
+    shim_body = dest / "bin" / lark_broker.LARK_CLI_BROKER_SHIM_FILENAME
+    assert str(shim_body) in launcher_text
+    assert lark_broker._LARK_CLI_BROKER_SHIM_PATH_PLACEHOLDER not in launcher_text
+    assert os.access(dest / "bin" / "lark-cli", os.X_OK)
+    assert shim_body.read_text(encoding="utf-8") == lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT
+    assert os.access(shim_body, os.X_OK)
+    marker = json.loads((dest / ".deerflow-lark-cli-runtime.json").read_text())
+    assert marker == {"version": "v1.0.65", "kind": "shim"}
+
+
+def test_launcher_resolves_python_and_forwards(broker_server, tmp_path: Path) -> None:
+    """The /bin/sh launcher finds python3 on PATH and execs the shim body."""
+    host, port = broker_server
+    dest = tmp_path / "runtime"
+    lark_broker.install_shim(str(dest), version="v1.0.65")
+    launcher = dest / "bin" / "lark-cli"
+
+    # A PATH that has the python from this test runner so the launcher resolves it.
+    py_dir = str(Path(sys.executable).parent)
+    completed = subprocess.run(
+        [str(launcher), "do", "--boom"],
+        input=b"",
+        capture_output=True,
+        env={
+            "PATH": py_dir + os.pathsep + "/usr/bin:/bin",
+            lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}",
+        },
+        timeout=30,
+    )
+    assert completed.returncode == 7
+    assert b"ERR:do --boom" in completed.stderr
+
+
+def test_launcher_can_pin_interpreter_via_env(broker_server, tmp_path: Path) -> None:
+    """DEERFLOW_LARK_BROKER_PYTHON pins the interpreter for images with no python3
+    on PATH (the launcher must not silently ENOEXEC)."""
+    host, port = broker_server
+    dest = tmp_path / "runtime"
+    lark_broker.install_shim(str(dest), version="v1.0.65")
+    launcher = dest / "bin" / "lark-cli"
+
+    completed = subprocess.run(
+        [str(launcher), "ping"],
+        input=b"",
+        capture_output=True,
+        # Deliberately no python on PATH; the pin is the only way to resolve it.
+        env={
+            "PATH": "/nonexistent",
+            lark_broker.LARK_BROKER_PYTHON_ENV: sys.executable,
+            lark_broker.LARK_BROKER_URL_ENV: f"http://{host}:{port}",
+        },
+        timeout=30,
+    )
+    assert completed.returncode == 0
+
+
+def test_launcher_fails_loudly_without_python(tmp_path: Path) -> None:
+    """With no python interpreter resolvable, the launcher exits 127 with an
+    actionable message rather than an opaque ENOEXEC."""
+    dest = tmp_path / "runtime"
+    lark_broker.install_shim(str(dest), version="v1.0.65")
+    launcher = dest / "bin" / "lark-cli"
+
+    completed = subprocess.run(
+        [str(launcher), "auth", "status"],
+        input=b"",
+        capture_output=True,
+        env={"PATH": "/nonexistent"},
+        timeout=30,
+    )
+    assert completed.returncode == 127
+    assert b"Python 3 interpreter" in completed.stderr
+
+
+# ── cwd is not forwarded (command surface only, no sandbox file I/O) ─────────
+
+
+def test_exec_payload_omits_cwd(tmp_path: Path) -> None:
+    """The shim must not forward cwd: the broker can't see the sandbox FS, so a
+    dead cwd field would imply support that does not exist."""
+    assert '"cwd"' not in lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT
+    assert "os.getcwd()" not in lark_broker.LARK_CLI_BROKER_SHIM_SCRIPT
+
+
+# ── subcommand denylist (issue #4338 hardening) ──────────────────────────────
+
+
+def test_parse_deny_subcommands() -> None:
+    assert lark_broker.parse_deny_subcommands(None) == ()
+    assert lark_broker.parse_deny_subcommands("") == ()
+    assert lark_broker.parse_deny_subcommands("config show, auth token") == (
+        ("config", "show"),
+        ("auth", "token"),
+    )
+    # Blank entries are dropped.
+    assert lark_broker.parse_deny_subcommands("config show, ,") == (("config", "show"),)
+
+
+def test_denied_subcommand_is_refused_before_spawning_binary(tmp_path: Path) -> None:
+    config = BrokerConfig(
+        lark_cli_path=_fake_lark_cli(tmp_path),
+        config_dir="/broker/only/config",
+        data_dir="/broker/only/data",
+        deny_subcommands=(("config", "show"),),
+    )
+    result = run_lark_cli(config, ["config", "show", "--json"], b"")
+    assert result.exit_code == 126
+    assert b"disabled in broker mode" in result.stderr
+    # The stub echoes argv on stdout; a refused call never runs it.
+    assert result.stdout == b""
+
+
+def test_denied_subcommand_matches_through_leading_flags(tmp_path: Path) -> None:
+    config = BrokerConfig(
+        lark_cli_path=_fake_lark_cli(tmp_path),
+        config_dir="c",
+        data_dir="d",
+        deny_subcommands=(("config", "show"),),
+    )
+    # Options interleaved with the subcommand path are skipped when matching.
+    result = run_lark_cli(config, ["--json", "config", "show"], b"")
+    assert result.exit_code == 126
+
+
+def test_allowed_subcommand_still_runs_with_denylist(tmp_path: Path) -> None:
+    config = BrokerConfig(
+        lark_cli_path=_fake_lark_cli(tmp_path),
+        config_dir="c",
+        data_dir="d",
+        deny_subcommands=(("config", "show"),),
+    )
+    result = run_lark_cli(config, ["auth", "status"], b"")
+    assert result.exit_code == 0
+    payload = json.loads(result.stdout.decode())
+    assert payload["argv"] == ["auth", "status"]

backend/tests/test_lark_cli_integration.py

@@ -419,20 +419,34 @@ def test_status_runtime_mode_init_container_ready(monkeypatch, tmp_path) -> None
         use="deerflow.community.aio_sandbox:AioSandboxProvider",
         provisioner_url="http://provisioner:8002",
     )
-    monkeypatch.setattr(lark_cli, "_probe_provisioner_lark_cli_init_image", lambda _config: True)
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", lambda _config: {"lark_cli_init_image": True, "lark_cli_broker_image": False})
     mode, ready, detail = lark_cli._resolve_sandbox_runtime_readiness(config, probe=True)
     assert mode == "init-container"
     assert ready is True
     assert detail is None
 
 
+def test_status_runtime_mode_broker_supersedes_init_container(monkeypatch, tmp_path) -> None:
+    config = _config(tmp_path / "skills")
+    config.sandbox = SimpleNamespace(
+        use="deerflow.community.aio_sandbox:AioSandboxProvider",
+        provisioner_url="http://provisioner:8002",
+    )
+    # Broker (Pattern B) wins even when the init image is also configured.
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", lambda _config: {"lark_cli_init_image": True, "lark_cli_broker_image": True})
+    mode, ready, detail = lark_cli._resolve_sandbox_runtime_readiness(config, probe=True)
+    assert mode == "broker"
+    assert ready is True
+    assert detail is None
+
+
 def test_status_runtime_mode_init_container_not_configured(monkeypatch, tmp_path) -> None:
     config = _config(tmp_path / "skills")
     config.sandbox = SimpleNamespace(
         use="deerflow.community.aio_sandbox:AioSandboxProvider",
         provisioner_url="http://provisioner:8002",
     )
-    monkeypatch.setattr(lark_cli, "_probe_provisioner_lark_cli_init_image", lambda _config: False)
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", lambda _config: {"lark_cli_init_image": False, "lark_cli_broker_image": False})
     mode, ready, detail = lark_cli._resolve_sandbox_runtime_readiness(config, probe=True)
     assert mode == "init-container"
     assert ready is False
@@ -445,7 +459,7 @@ def test_status_runtime_mode_init_container_unreachable(monkeypatch, tmp_path) -
         use="deerflow.community.aio_sandbox:AioSandboxProvider",
         provisioner_url="http://provisioner:8002",
     )
-    monkeypatch.setattr(lark_cli, "_probe_provisioner_lark_cli_init_image", lambda _config: None)
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", lambda _config: None)
     mode, ready, detail = lark_cli._resolve_sandbox_runtime_readiness(config, probe=True)
     assert mode == "init-container"
     assert ready is False
@@ -462,13 +476,82 @@ def test_status_runtime_probe_skipped_when_not_requested(monkeypatch, tmp_path)
     def _fail(_config):  # pragma: no cover - must not be called
         raise AssertionError("provisioner should not be probed when probe=False")
 
-    monkeypatch.setattr(lark_cli, "_probe_provisioner_lark_cli_init_image", _fail)
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", _fail)
     mode, ready, detail = lark_cli._resolve_sandbox_runtime_readiness(config, probe=False)
     assert mode == "init-container"
     assert ready is False
     assert detail is None
 
 
+def _reset_broker_mode_cache() -> None:
+    if hasattr(lark_cli.sandbox_lark_broker_active, "_cache"):
+        del lark_cli.sandbox_lark_broker_active._cache
+
+
+def test_sandbox_lark_broker_active_uses_tight_hot_path_timeout(monkeypatch, tmp_path) -> None:
+    """The per-bash-call broker probe must use the tight hot-path timeout, not the
+    5s Settings-status budget, so non-broker remote-provisioner users don't pay a
+    multi-second latency hit on the first lark-cli call per TTL."""
+    _reset_broker_mode_cache()
+    config = _config(tmp_path / "skills")
+    config.sandbox = SimpleNamespace(
+        use="deerflow.community.aio_sandbox:AioSandboxProvider",
+        provisioner_url="http://provisioner:8002",
+    )
+    seen: dict[str, float] = {}
+
+    def _capture(_config, *, timeout):
+        seen["timeout"] = timeout
+        return {"lark_cli_init_image": False, "lark_cli_broker_image": True}
+
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", _capture)
+    try:
+        assert lark_cli.sandbox_lark_broker_active(config) is True
+        assert seen["timeout"] == lark_cli.LARK_BROKER_MODE_PROBE_TIMEOUT_SECONDS
+    finally:
+        _reset_broker_mode_cache()
+
+
+def test_sandbox_lark_broker_active_caches_negative_result(monkeypatch, tmp_path) -> None:
+    """A non-broker result is cached (longer TTL) so the hot path stops probing."""
+    _reset_broker_mode_cache()
+    config = _config(tmp_path / "skills")
+    config.sandbox = SimpleNamespace(
+        use="deerflow.community.aio_sandbox:AioSandboxProvider",
+        provisioner_url="http://provisioner:8002",
+    )
+    calls = {"n": 0}
+
+    def _probe(_config, *, timeout):
+        calls["n"] += 1
+        return {"lark_cli_init_image": True, "lark_cli_broker_image": False}
+
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", _probe)
+    try:
+        assert lark_cli.sandbox_lark_broker_active(config) is False
+        assert lark_cli.sandbox_lark_broker_active(config) is False
+        # Second call served from cache — the provisioner is probed only once.
+        assert calls["n"] == 1
+    finally:
+        _reset_broker_mode_cache()
+
+
+def test_sandbox_lark_broker_active_false_without_remote_provisioner(monkeypatch, tmp_path) -> None:
+    """Local AIO (no provisioner URL) never probes and is never broker mode."""
+    _reset_broker_mode_cache()
+    config = _config(tmp_path / "skills")
+    config.sandbox = SimpleNamespace(use="deerflow.community.aio_sandbox:AioSandboxProvider")
+
+    def _fail(_config, *, timeout):  # pragma: no cover - must not be called
+        raise AssertionError("no provisioner should be probed without a provisioner_url")
+
+    monkeypatch.setattr(lark_cli, "_probe_provisioner_capabilities", _fail)
+    try:
+        assert lark_cli.sandbox_lark_broker_active(config) is False
+    finally:
+        _reset_broker_mode_cache()
+
+
 def test_install_lark_integration_is_idempotent_across_reinstalls(monkeypatch, tmp_path):
     reset_skill_storage()
     _patch_paths(monkeypatch, tmp_path / "home")

backend/tests/test_provisioner_pvc_volumes.py

@@ -561,3 +561,146 @@ class TestLarkCliInitContainer:
         assert runtime_mounts[0].name == provisioner_module.LARK_CLI_RUNTIME_VOLUME_NAME
         mount_paths = {m.mount_path for m in pod.spec.containers[0].volume_mounts}
         assert "/mnt/integrations/lark-cli/config" in mount_paths
+
+
+class TestLarkCliBrokerSidecar:
+    """Broker sidecar provisioning of the sandbox lark-cli runtime (Pattern B)."""
+
+    @staticmethod
+    def _credential_mounts(provisioner_module):
+        return [
+            provisioner_module.ExtraMount(
+                host_path="/state/users/alice/integrations/lark-cli/config",
+                container_path="/mnt/integrations/lark-cli/config",
+                read_only=True,
+            ),
+            provisioner_module.ExtraMount(
+                host_path="/state/users/alice/integrations/lark-cli/data",
+                container_path="/mnt/integrations/lark-cli/data",
+                read_only=False,
+            ),
+        ]
+
+    def test_no_sidecar_when_broker_image_unset(self, provisioner_module):
+        provisioner_module.SKILLS_PVC_NAME = ""
+        provisioner_module.USERDATA_PVC_NAME = ""
+        provisioner_module.LARK_CLI_BROKER_IMAGE = ""
+        pod = provisioner_module._build_pod(
+            "sandbox-1",
+            "thread-1",
+            provision_lark_cli_broker=True,
+        )
+        container_names = {c.name for c in pod.spec.containers}
+        assert "lark-cli-broker" not in container_names
+
+    def test_no_sidecar_when_flag_disabled(self, provisioner_module):
+        provisioner_module.SKILLS_PVC_NAME = ""
+        provisioner_module.USERDATA_PVC_NAME = ""
+        provisioner_module.LARK_CLI_BROKER_IMAGE = "deer-flow/lark-cli-broker:v1.0.65"
+        pod = provisioner_module._build_pod(
+            "sandbox-1",
+            "thread-1",
+            provision_lark_cli_broker=False,
+        )
+        container_names = {c.name for c in pod.spec.containers}
+        assert "lark-cli-broker" not in container_names
+
+    def test_broker_sidecar_and_shim_when_enabled(self, provisioner_module):
+        provisioner_module.SKILLS_PVC_NAME = ""
+        provisioner_module.USERDATA_PVC_NAME = ""
+        provisioner_module.DEER_FLOW_HOST_BASE_DIR = "/state"
+        provisioner_module.LARK_CLI_BROKER_IMAGE = "deer-flow/lark-cli-broker:v1.0.65"
+        pod = provisioner_module._build_pod(
+            "sandbox-1",
+            "thread-1",
+            user_id="alice",
+            extra_mounts=self._credential_mounts(provisioner_module),
+            provision_lark_cli_broker=True,
+        )
+
+        # Shim init container from the broker image.
+        assert pod.spec.init_containers is not None
+        assert len(pod.spec.init_containers) == 1
+        init = pod.spec.init_containers[0]
+        assert init.name == "lark-cli-shim-init"
+        assert init.image == "deer-flow/lark-cli-broker:v1.0.65"
+        assert init.args == ["install-shim", provisioner_module.LARK_CLI_RUNTIME_CONTAINER_PATH]
+
+        # Broker sidecar alongside the sandbox container.
+        sidecars = [c for c in pod.spec.containers if c.name == "lark-cli-broker"]
+        assert len(sidecars) == 1
+        sidecar = sidecars[0]
+        assert sidecar.image == "deer-flow/lark-cli-broker:v1.0.65"
+        assert sidecar.args == ["serve"]
+        # Credentials mounted into the sidecar only.
+        sidecar_paths = {m.mount_path for m in sidecar.volume_mounts}
+        assert provisioner_module.LARK_BROKER_SIDECAR_CONFIG_PATH in sidecar_paths
+        assert provisioner_module.LARK_BROKER_SIDECAR_DATA_PATH in sidecar_paths
+
+        # Sandbox container: runtime shim mount + broker URL env, NO config/data.
+        sandbox = pod.spec.containers[0]
+        sandbox_paths = {m.mount_path for m in sandbox.volume_mounts}
+        assert provisioner_module.LARK_CLI_RUNTIME_CONTAINER_PATH in sandbox_paths
+        assert "/mnt/integrations/lark-cli/config" not in sandbox_paths
+        assert "/mnt/integrations/lark-cli/data" not in sandbox_paths
+        env = {e.name: e.value for e in (sandbox.env or [])}
+        assert env.get("DEERFLOW_LARK_BROKER_URL") == provisioner_module.LARK_BROKER_URL
+
+    def test_broker_supersedes_init_container(self, provisioner_module):
+        """Both images set + both flags on → broker wins (shim init, sidecar)."""
+        provisioner_module.SKILLS_PVC_NAME = ""
+        provisioner_module.USERDATA_PVC_NAME = ""
+        provisioner_module.DEER_FLOW_HOST_BASE_DIR = "/state"
+        provisioner_module.LARK_CLI_INIT_IMAGE = "deer-flow/lark-cli-init:v1.0.65"
+        provisioner_module.LARK_CLI_BROKER_IMAGE = "deer-flow/lark-cli-broker:v1.0.65"
+        pod = provisioner_module._build_pod(
+            "sandbox-1",
+            "thread-1",
+            user_id="alice",
+            extra_mounts=self._credential_mounts(provisioner_module),
+            provision_lark_cli_runtime=True,
+            provision_lark_cli_broker=True,
+        )
+        assert pod.spec.init_containers[0].name == "lark-cli-shim-init"
+        assert any(c.name == "lark-cli-broker" for c in pod.spec.containers)
+
+    def test_broker_sidecar_forwards_subcommand_denylist_when_configured(self, provisioner_module):
+        """When DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS is set on the provisioner,
+        it is forwarded to the broker sidecar so it can refuse secret-dump
+        subcommands (issue #4338 hardening)."""
+        provisioner_module.SKILLS_PVC_NAME = ""
+        provisioner_module.USERDATA_PVC_NAME = ""
+        provisioner_module.DEER_FLOW_HOST_BASE_DIR = "/state"
+        provisioner_module.LARK_CLI_BROKER_IMAGE = "deer-flow/lark-cli-broker:v1.0.65"
+        provisioner_module.LARK_CLI_BROKER_DENY_SUBCOMMANDS = "config show, auth token"
+        try:
+            pod = provisioner_module._build_pod(
+                "sandbox-1",
+                "thread-1",
+                user_id="alice",
+                extra_mounts=self._credential_mounts(provisioner_module),
+                provision_lark_cli_broker=True,
+            )
+            sidecar = next(c for c in pod.spec.containers if c.name == "lark-cli-broker")
+            env = {e.name: e.value for e in (sidecar.env or [])}
+            assert env.get("DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS") == "config show, auth token"
+        finally:
+            provisioner_module.LARK_CLI_BROKER_DENY_SUBCOMMANDS = ""
+
+    def test_broker_sidecar_omits_denylist_env_when_unset(self, provisioner_module):
+        """Empty denylist ⇒ no env var (nothing blocked, no behavior change)."""
+        provisioner_module.SKILLS_PVC_NAME = ""
+        provisioner_module.USERDATA_PVC_NAME = ""
+        provisioner_module.DEER_FLOW_HOST_BASE_DIR = "/state"
+        provisioner_module.LARK_CLI_BROKER_IMAGE = "deer-flow/lark-cli-broker:v1.0.65"
+        provisioner_module.LARK_CLI_BROKER_DENY_SUBCOMMANDS = ""
+        pod = provisioner_module._build_pod(
+            "sandbox-1",
+            "thread-1",
+            user_id="alice",
+            extra_mounts=self._credential_mounts(provisioner_module),
+            provision_lark_cli_broker=True,
+        )
+        sidecar = next(c for c in pod.spec.containers if c.name == "lark-cli-broker")
+        env = {e.name for e in (sidecar.env or [])}
+        assert "DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS" not in env

backend/tests/test_remote_sandbox_backend.py

@@ -165,12 +165,13 @@ def test_create_delegates_to_provisioner_create(monkeypatch, expected_user_id):
     backend = RemoteSandboxBackend("http://provisioner:8002")
     expected = SandboxInfo(sandbox_id="abc123", sandbox_url="http://k3s:31001")
 
-    def mock_create(thread_id: str, sandbox_id: str, extra_mounts=None, *, user_id=None, provision_lark_cli_runtime=False):
+    def mock_create(thread_id: str, sandbox_id: str, extra_mounts=None, *, user_id=None, provision_lark_cli_runtime=False, provision_lark_cli_broker=False):
         assert thread_id == "thread-1"
         assert sandbox_id == "abc123"
         assert extra_mounts == [("/host", "/container", False)]
         assert user_id == expected_user_id
         assert provision_lark_cli_runtime is True
+        assert provision_lark_cli_broker is False
         return expected
 
     monkeypatch.setattr(backend, "_provisioner_create", mock_create)
@@ -197,6 +198,7 @@ def test_provisioner_create_returns_sandbox_info(monkeypatch):
             "user_id": "test-user-autouse",
             "include_legacy_skills": True,
             "provision_lark_cli_runtime": False,
+            "provision_lark_cli_broker": False,
         }
         assert timeout == 30
         return _StubResponse(payload={"sandbox_id": "abc123", "sandbox_url": "http://k3s:31001"})
@@ -317,6 +319,7 @@ def test_provisioner_create_accepts_anonymous_thread_id(monkeypatch):
             "user_id": "test-user-autouse",
             "include_legacy_skills": False,
             "provision_lark_cli_runtime": False,
+            "provision_lark_cli_broker": False,
         }
         assert timeout == 30
         return _StubResponse(payload={"sandbox_id": "anon123", "sandbox_url": "http://k3s:31002"})

docker/docker-compose-dev.yaml

@@ -50,6 +50,9 @@ services:
       # Set to a published tag (e.g. deer-flow/lark-cli-init:v1.0.65) to provision
       # the sandbox lark-cli runtime via an init container + emptyDir.
       - LARK_CLI_INIT_IMAGE=${LARK_CLI_INIT_IMAGE:-}
+      # Optional lark-cli broker image (Pattern B, issue #4338). Empty ⇒ broker
+      # off. Supersedes LARK_CLI_INIT_IMAGE when both are set.
+      - LARK_CLI_BROKER_IMAGE=${LARK_CLI_BROKER_IMAGE:-}
       # Host paths for K8s HostPath volumes (must be absolute paths accessible by K8s node)
       # On Docker Desktop/OrbStack, use your actual host paths like /Users/username/...
       # Set these in your shell before running docker-compose:

docker/docker-compose.yaml

@@ -162,6 +162,11 @@ services:
       # Set to a published tag (e.g. deer-flow/lark-cli-init:v1.0.65) to provision
       # the sandbox lark-cli runtime via an init container + emptyDir.
       - LARK_CLI_INIT_IMAGE=${LARK_CLI_INIT_IMAGE:-}
+      # Optional lark-cli broker image (Pattern B, issue #4338). When set, the
+      # sandbox gets a shim + broker sidecar that holds the credentials, so the
+      # plaintext config/data are never mounted into the sandbox. Supersedes
+      # LARK_CLI_INIT_IMAGE when both are set. Empty ⇒ broker off.
+      - LARK_CLI_BROKER_IMAGE=${LARK_CLI_BROKER_IMAGE:-}
       - SKILLS_HOST_PATH=${DEER_FLOW_REPO_ROOT}/skills
       - THREADS_HOST_PATH=${DEER_FLOW_HOME}/threads
       - DEER_FLOW_HOST_BASE_DIR=${DEER_FLOW_HOME}

docker/lark-cli-broker/Dockerfile

@@ -0,0 +1,79 @@
+# syntax=docker/dockerfile:1
+#
+# DeerFlow "lark-cli broker" image (Pattern B, issue #4338).
+#
+# Purpose: hold `lark-cli` + the per-user Lark credentials in a long-running
+# sidecar and expose only the command surface over loopback, so the plaintext
+# app secret / OAuth tokens never get mounted into the sandbox container.
+#
+# The one image runs in two modes (dispatched by the first CLI arg):
+#
+#   install-shim <dest>   init-container mode: write the Python shim +
+#                         runtime marker into the shared emptyDir at <dest>
+#                         (default /mnt/integrations/lark-cli/runtime), so the
+#                         sandbox finds `bin/lark-cli` on PATH — same layout the
+#                         Pattern A init image produces, marked kind=shim.
+#
+#   serve  (default CMD)  sidecar mode: run the broker HTTP server on loopback
+#                         with the real `lark-cli` and credential env pointing at
+#                         the sidecar-only /var/lark/{config,data} mounts.
+#
+# At BUILD time (network available) this image downloads and SHA-256-verifies the
+# official `larksuite/cli` Linux release binaries (via the shared build-runtime.sh
+# from docker/lark-cli-init) into /opt/lark-cli, so the sidecar has a real binary.
+#
+# The shim itself is written from the in-process constant
+# LARK_CLI_BROKER_SHIM_SCRIPT (deerflow.integrations.lark_broker), so the image's
+# shim can never drift from the Gateway's copy.
+#
+# Build (defaults to the pinned version below):
+#   docker build -t deer-flow/lark-cli-broker:v1.0.65 \
+#     --build-arg LARK_CLI_VERSION=v1.0.65 \
+#     -f docker/lark-cli-broker/Dockerfile .
+#
+# NOTE: build context is the repo root (the module lives under backend/).
+
+FROM debian:bookworm-slim AS builder
+
+ARG APT_MIRROR
+ARG LARK_CLI_VERSION=v1.0.65
+
+RUN if [ -n "${APT_MIRROR}" ]; then \
+      sed -i "s|deb.debian.org|${APT_MIRROR}|g" /etc/apt/sources.list.d/debian.sources 2>/dev/null || true; \
+      sed -i "s|deb.debian.org|${APT_MIRROR}|g" /etc/apt/sources.list 2>/dev/null || true; \
+    fi
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ca-certificates \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Reuse the Pattern A build script so the real sidecar binary is staged with the
+# identical download + SHA-256 verification path.
+COPY docker/lark-cli-init/build-runtime.sh /usr/local/bin/build-runtime.sh
+RUN chmod +x /usr/local/bin/build-runtime.sh \
+    && LARK_CLI_VERSION="${LARK_CLI_VERSION}" /usr/local/bin/build-runtime.sh /opt/lark-cli
+
+# ── Final image: python3 + the staged real binary + the broker module ────────
+FROM python:3.12-slim
+
+ARG LARK_CLI_VERSION=v1.0.65
+ENV LARK_CLI_VERSION=${LARK_CLI_VERSION}
+
+COPY --from=builder /opt/lark-cli /opt/lark-cli
+# Single stdlib-only module drives both install-shim and serve.
+COPY backend/packages/harness/deerflow/integrations/lark_broker.py /opt/broker/lark_broker.py
+COPY docker/lark-cli-broker/entrypoint.sh /usr/local/bin/lark-cli-broker
+RUN chmod +x /usr/local/bin/lark-cli-broker
+
+# serve mode: the sidecar runs the real arch-dispatch launcher and reads the
+# credential dirs the provisioner mounts into the sidecar only.
+ENV DEERFLOW_LARK_BROKER_CLI=/opt/lark-cli/bin/lark-cli \
+    LARKSUITE_CLI_CONFIG_DIR=/var/lark/config \
+    LARKSUITE_CLI_DATA_DIR=/var/lark/data \
+    DEERFLOW_LARK_BROKER_HOST=127.0.0.1 \
+    DEERFLOW_LARK_BROKER_PORT=8788 \
+    LARK_CLI_RUNTIME_DEST=/mnt/integrations/lark-cli/runtime
+
+ENTRYPOINT ["/usr/local/bin/lark-cli-broker"]
+CMD ["serve"]

docker/lark-cli-broker/README.md

@@ -0,0 +1,111 @@
+# lark-cli broker image (Pattern B)
+
+This image implements **Pattern B** (issue #4338): instead of mounting the
+per-user Lark credential directories into the sandbox (Pattern A still does), a
+long-running **sidecar** holds `lark-cli` + the credentials and serves only the
+command surface over loopback. The sandbox gets a tiny `lark-cli` **shim** on
+`PATH` that forwards argv/stdin to the sidecar.
+
+Result: the raw `appSecret` / OAuth token files **never exist in the sandbox
+filesystem**, so a compromised or prompt-injected agent can no longer
+`cat`/exfiltrate them — while any authorized `lark-cli` subcommand still runs.
+
+## Two modes, one image
+
+Dispatched by the first CLI argument:
+
+- `install-shim <dest>` — **init container**: writes the launcher + Python shim +
+  `.deerflow-lark-cli-runtime.json` (`kind: "shim"`) into the shared `emptyDir`
+  at `<dest>` (default `/mnt/integrations/lark-cli/runtime`), then exits `0`. The
+  sandbox then finds `bin/lark-cli` exactly where
+  `lark_cli_env_overlay(sandbox_paths=True)` points `PATH` — same layout the
+  Pattern A init image produces.
+- `serve` (default `CMD`) — **sidecar**: runs the broker HTTP server on
+  `127.0.0.1:8788` with the real `lark-cli` and the credential env pointing at
+  the sidecar-only `/var/lark/{config,data}` mounts.
+
+The executable on `PATH` (`bin/lark-cli`) is a `/bin/sh` **launcher** that
+resolves a Python 3 interpreter and execs the **shim body** (`bin/lark-cli-shim.py`)
+beside it (by its baked-in absolute path, since `$0` is the bare command name
+when run off `PATH`); both are written from the in-process
+`LARK_CLI_BROKER_LAUNCHER_TEMPLATE` / `LARK_CLI_BROKER_SHIM_SCRIPT`
+(`deerflow.integrations.lark_broker`), so the image's copies can never drift from
+the Gateway's. Splitting the sh launcher from the Python body means broker mode
+does **not** hard-depend on `python3` resolving via a `#!/usr/bin/env python3`
+shebang: if no `python3`/`python` is on the sandbox `PATH`, the launcher exits
+`127` with an actionable message (set `DEERFLOW_LARK_BROKER_PYTHON` to a known
+interpreter path) instead of an opaque ENOEXEC. The stock `all-in-one-sandbox`
+image ships Python 3, so the default path needs no configuration.
+
+## Build
+
+Build context is the **repo root** (the broker module lives under `backend/`):
+
+```bash
+docker build -t deer-flow/lark-cli-broker:v1.0.65 \
+  --build-arg LARK_CLI_VERSION=v1.0.65 \
+  -f docker/lark-cli-broker/Dockerfile .
+```
+
+The tag should encode the lark-cli version so it can be bumped independently of
+the upstream `all-in-one-sandbox` image.
+
+## Wiring it into the provisioner
+
+Broker mode is **opt-in** and off by default. Enable it by publishing this image
+and pointing the provisioner at it:
+
+- Set `LARK_CLI_BROKER_IMAGE` on the provisioner to the published tag. Empty ⇒
+  broker off (Pattern A / legacy path, no behavior change).
+- When set, and the Gateway sends `provision_lark_cli_broker` on sandbox create,
+  the provisioner adds:
+  - a `lark-cli-runtime` `emptyDir` shared by an init container and the sandbox;
+  - a `lark-cli-shim-init` init container (`install-shim`) that stages the shim;
+  - a `lark-cli-broker` **sidecar** (`serve`) with the per-user `config` (RO) /
+    `data` (RW) credential mounts — **into the sidecar only**;
+  - the sandbox container gets the runtime RO mount + `DEERFLOW_LARK_BROKER_URL`
+    and **no** `config`/`data` mounts.
+- Broker mode **supersedes** Pattern A when both are configured.
+- The provisioner reports it via `GET /api/capabilities`
+  (`{"lark_cli_broker_image": true|false}`), which the Gateway surfaces as the
+  Lark integration sandbox-runtime readiness signal in
+  `/api/integrations/lark/status` (`sandbox_runtime_mode: "broker"`).
+
+> Publishing note: this repository currently ships only backend/frontend images.
+> Publishing a `lark-cli-broker` tag is a fast-follow; until then the feature
+> stays behind the empty-default `LARK_CLI_BROKER_IMAGE`.
+
+## Broker HTTP contract (loopback)
+
+- `POST /v1/exec` — body `{"args": [...], "stdin_b64": "..."}`; response
+  `{"exit_code", "stdout_b64", "stderr_b64", "truncated"}`. `args` is run with
+  `shell=False`, so a sandbox-supplied argument can never be shell-injected. The
+  broker injects the credential env itself; the client cannot override it.
+  Unexpected broker-side errors return a `500 {"error": ...}` so the shim always
+  gets a structured response rather than an opaque transport failure.
+- `GET /v1/health` — `{"ok": true}`.
+
+Bound to loopback only. In K8s the sandbox and sidecar share the Pod network
+namespace, so `127.0.0.1` reaches the sidecar and nothing outside the Pod can.
+
+### No file I/O relative to the sandbox cwd
+
+The broker runs `lark-cli` in the **sidecar's** working directory and cannot see
+the sandbox filesystem, so the sandbox's cwd is intentionally **not** forwarded.
+`lark-cli` subcommands that read or write files by a path relative to the
+sandbox cwd (e.g. uploading a local file) are therefore unsupported in broker
+mode — this is a command-surface-only bridge, not a filesystem bridge. Absolute
+paths still refer to the sidecar's filesystem, not the sandbox's.
+
+### Optional subcommand denylist (hardening)
+
+The broker removes the credential *files* from the sandbox, but the full
+`lark-cli` command surface stays reachable, so any subcommand that prints/exports
+tokens could still exfiltrate them. Set `DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS`
+on the sidecar to a comma-separated list of command prefixes the broker should
+refuse (matched against the leading non-flag tokens), e.g.
+`DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS="config show, auth token"`. Denied calls
+return exit `126` with a `subcommand ... is disabled` message and never spawn the
+binary. Empty by default (no behavior change); confirm the deployed `lark-cli`
+version's subcommand surface has no trivial secret-dump command before enabling
+broker mode in production.

docker/lark-cli-broker/entrypoint.sh

@@ -0,0 +1,29 @@
+#!/bin/sh
+# DeerFlow lark-cli broker entrypoint (Pattern B, issue #4338).
+#
+# Dispatches to one of two modes on the single stdlib-only module:
+#
+#   install-shim [dest]   write the shim + runtime marker into the shared
+#                         emptyDir (init-container mode), then exit 0.
+#   serve                 run the broker HTTP server on loopback (sidecar mode).
+#
+# The module is copied to /opt/broker/lark_broker.py and run as a script; it has
+# no third-party dependencies, so no harness install is needed in this image.
+set -eu
+
+MODULE="/opt/broker/lark_broker.py"
+
+case "${1:-serve}" in
+  install-shim)
+    shift
+    exec python3 "${MODULE}" install-shim "$@"
+    ;;
+  serve)
+    exec python3 "${MODULE}"
+    ;;
+  *)
+    echo "Unknown lark-cli-broker mode: ${1:-}" >&2
+    echo "Usage: lark-cli-broker [install-shim <dest> | serve]" >&2
+    exit 2
+    ;;
+esac

docker/provisioner/README.md

@@ -146,6 +146,7 @@ The provisioner is configured via environment variables (set in [docker-compose-
 | `K8S_NAMESPACE` | `deer-flow` | Kubernetes namespace for sandbox resources |
 | `SANDBOX_IMAGE` | `enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest` | AIO-compatible container image for sandbox Pods |
 | `LARK_CLI_INIT_IMAGE` | empty (feature off) | Optional lark-cli init image (Pattern A). When set, sandbox Pods requesting the lark-cli runtime get an init container + shared `emptyDir` that provisions `lark-cli`, instead of a hostPath/PVC runtime mount. See [`docker/lark-cli-init`](../lark-cli-init/README.md) |
+| `LARK_CLI_BROKER_IMAGE` | empty (feature off) | Optional lark-cli broker image (Pattern B, issue #4338). When set, sandbox Pods requesting the broker get a shim init container + a `lark-cli-broker` sidecar that holds the credentials; the plaintext `config`/`data` are mounted into the **sidecar only**, never the sandbox. Supersedes `LARK_CLI_INIT_IMAGE` when both are set. See [`docker/lark-cli-broker`](../lark-cli-broker/README.md) |
 | `SKILLS_HOST_PATH` | - | **Host machine** path to skills directory (must be absolute) |
 | `THREADS_HOST_PATH` | - | **Host machine** path to threads data directory (must be absolute) |
 | `SKILLS_PVC_NAME` | empty (use hostPath) | PVC name for skills volume; when set, sandbox Pods use PVC instead of hostPath |

docker/provisioner/app.py

@@ -67,6 +67,24 @@ SANDBOX_IMAGE = os.environ.get(
 LARK_CLI_INIT_IMAGE = os.environ.get("LARK_CLI_INIT_IMAGE", "")
 LARK_CLI_RUNTIME_CONTAINER_PATH = "/mnt/integrations/lark-cli/runtime"
 LARK_CLI_RUNTIME_VOLUME_NAME = "lark-cli-runtime"
+# Optional "lark-cli broker" image (Pattern B, issue #4338). When set, sandbox
+# Pods requesting the broker get an init container that stages a shim + a
+# long-running broker sidecar that holds the credentials, instead of mounting the
+# plaintext config/data credential dirs into the sandbox container. Empty ⇒ broker
+# off (Pattern A / legacy behavior). Broker supersedes Pattern A when both are set.
+LARK_CLI_BROKER_IMAGE = os.environ.get("LARK_CLI_BROKER_IMAGE", "")
+# Optional comma-separated lark-cli subcommand denylist forwarded to the broker
+# sidecar (issue #4338 hardening). Empty ⇒ no subcommand is blocked. See the
+# broker README's "subcommand denylist" section.
+LARK_CLI_BROKER_DENY_SUBCOMMANDS = os.environ.get("DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS", "")
+LARK_CLI_CONFIG_CONTAINER_PATH = "/mnt/integrations/lark-cli/config"
+LARK_CLI_DATA_CONTAINER_PATH = "/mnt/integrations/lark-cli/data"
+# Where the broker sidecar reads the per-user credentials (sidecar-only paths).
+LARK_BROKER_SIDECAR_CONFIG_PATH = "/var/lark/config"
+LARK_BROKER_SIDECAR_DATA_PATH = "/var/lark/data"
+LARK_BROKER_CONFIG_VOLUME_NAME = "lark-cli-config"
+LARK_BROKER_DATA_VOLUME_NAME = "lark-cli-data"
+LARK_BROKER_URL = "http://127.0.0.1:8788"
 SKILLS_HOST_PATH = os.environ.get("SKILLS_HOST_PATH", "/skills")
 THREADS_HOST_PATH = os.environ.get("THREADS_HOST_PATH", "/.deer-flow/threads")
 DEER_FLOW_HOST_BASE_DIR = os.environ.get("DEER_FLOW_HOST_BASE_DIR", "/.deer-flow")
@@ -199,24 +217,54 @@ def _lark_cli_runtime_enabled(provision_lark_cli_runtime: bool) -> bool:
     return bool(LARK_CLI_INIT_IMAGE) and provision_lark_cli_runtime
 
 
+def _lark_cli_broker_enabled(provision_lark_cli_broker: bool) -> bool:
+    """Whether to provision the lark-cli broker sidecar (Pattern B)."""
+    return bool(LARK_CLI_BROKER_IMAGE) and provision_lark_cli_broker
+
+
 def _runtime_provided_extra_mounts(
     extra_mounts: list["ExtraMount"] | None,
     *,
     provision_lark_cli_runtime: bool,
+    provision_lark_cli_broker: bool = False,
 ) -> list["ExtraMount"]:
-    """Drop the lark-cli runtime extra mount when the init container supersedes it.
+    """Drop lark-cli extra mounts the init container / broker sidecar supersede.
 
-    The init container + emptyDir provides ``/mnt/integrations/lark-cli/runtime``,
-    so a hostPath/PVC mount at the same path would collide. The per-user
-    ``config`` / ``data`` credential mounts are left untouched.
+    Pattern A (init container + emptyDir) provides
+    ``/mnt/integrations/lark-cli/runtime``, so a hostPath/PVC mount at the same
+    path would collide — it is dropped, leaving the per-user ``config`` / ``data``
+    credential mounts intact.
+
+    Pattern B (broker sidecar) additionally moves the ``config`` / ``data``
+    credential mounts off the *sandbox* container and into the sidecar, so those
+    are dropped here too — the sandbox never sees plaintext credentials.
     """
-    if not extra_mounts or not _lark_cli_runtime_enabled(provision_lark_cli_runtime):
+    dropped: set[str] = set()
+    if _lark_cli_broker_enabled(provision_lark_cli_broker):
+        dropped = {
+            LARK_CLI_RUNTIME_CONTAINER_PATH,
+            LARK_CLI_CONFIG_CONTAINER_PATH,
+            LARK_CLI_DATA_CONTAINER_PATH,
+        }
+    elif _lark_cli_runtime_enabled(provision_lark_cli_runtime):
+        dropped = {LARK_CLI_RUNTIME_CONTAINER_PATH}
+    if not extra_mounts or not dropped:
         return list(extra_mounts or [])
-    return [
-        mount
-        for mount in extra_mounts
-        if posixpath.normpath(mount.container_path) != LARK_CLI_RUNTIME_CONTAINER_PATH
-    ]
+    return [mount for mount in extra_mounts if posixpath.normpath(mount.container_path) not in dropped]
+
+
+def _lark_broker_credential_mounts(extra_mounts: list["ExtraMount"] | None) -> dict[str, "ExtraMount"]:
+    """Extract the config/data credential mounts the broker sidecar needs.
+
+    Keyed by container path so the caller can wire each into the sidecar's fixed
+    ``/var/lark/{config,data}`` paths.
+    """
+    result: dict[str, ExtraMount] = {}
+    for mount in _validated_extra_mounts(extra_mounts):
+        normalized = posixpath.normpath(mount.container_path)
+        if normalized in (LARK_CLI_CONFIG_CONTAINER_PATH, LARK_CLI_DATA_CONTAINER_PATH):
+            result[normalized] = mount
+    return result
 
 
 def _extra_mount_pvc_sub_path(host_path: str) -> str:
@@ -355,6 +403,12 @@ class CreateSandboxRequest(BaseModel):
     # lark-cli runtime via an init container + emptyDir instead of a runtime
     # hostPath/PVC extra mount.
     provision_lark_cli_runtime: bool = False
+    # When true (and LARK_CLI_BROKER_IMAGE is configured), provision a lark-cli
+    # broker sidecar (Pattern B, issue #4338): a shim in the sandbox forwards to
+    # the sidecar, which holds the credentials — so the plaintext config/data are
+    # mounted into the sidecar only, never the sandbox. Supersedes the runtime
+    # binary + credential mounts when enabled.
+    provision_lark_cli_broker: bool = False
 
 
 class SandboxResponse(BaseModel):
@@ -430,6 +484,7 @@ def _build_volumes(
     include_legacy_skills: bool = False,
     extra_mounts: list[ExtraMount] | None = None,
     provision_lark_cli_runtime: bool = False,
+    provision_lark_cli_broker: bool = False,
 ) -> list[k8s_client.V1Volume]:
     """Build volume list: PVC when configured, otherwise hostPath.
 
@@ -521,16 +576,48 @@ def _build_volumes(
             _runtime_provided_extra_mounts(
                 extra_mounts,
                 provision_lark_cli_runtime=provision_lark_cli_runtime,
+                provision_lark_cli_broker=provision_lark_cli_broker,
             )
         )
     )
-    if _lark_cli_runtime_enabled(provision_lark_cli_runtime):
+    # The runtime emptyDir is shared by the init container (writer) and the
+    # sandbox container (reader) in both Pattern A and Pattern B (shim).
+    if _lark_cli_runtime_enabled(provision_lark_cli_runtime) or _lark_cli_broker_enabled(provision_lark_cli_broker):
         volumes.append(
             k8s_client.V1Volume(
                 name=LARK_CLI_RUNTIME_VOLUME_NAME,
                 empty_dir=k8s_client.V1EmptyDirVolumeSource(),
             )
         )
+    # Pattern B: the config/data credential volumes go to the broker sidecar only.
+    if _lark_cli_broker_enabled(provision_lark_cli_broker):
+        credential_mounts = _lark_broker_credential_mounts(extra_mounts)
+        for container_path, volume_name in (
+            (LARK_CLI_CONFIG_CONTAINER_PATH, LARK_BROKER_CONFIG_VOLUME_NAME),
+            (LARK_CLI_DATA_CONTAINER_PATH, LARK_BROKER_DATA_VOLUME_NAME),
+        ):
+            mount = credential_mounts.get(container_path)
+            if mount is None:
+                continue
+            if USERDATA_PVC_NAME:
+                volumes.append(
+                    k8s_client.V1Volume(
+                        name=volume_name,
+                        persistent_volume_claim=k8s_client.V1PersistentVolumeClaimVolumeSource(
+                            claim_name=USERDATA_PVC_NAME,
+                        ),
+                    )
+                )
+            else:
+                volumes.append(
+                    k8s_client.V1Volume(
+                        name=volume_name,
+                        host_path=k8s_client.V1HostPathVolumeSource(
+                            path=mount.host_path,
+                            type="Directory" if mount.read_only else "DirectoryOrCreate",
+                        ),
+                    )
+                )
     return volumes
 
 
@@ -541,6 +628,7 @@ def _build_volume_mounts(
     include_legacy_skills: bool = False,
     extra_mounts: list[ExtraMount] | None = None,
     provision_lark_cli_runtime: bool = False,
+    provision_lark_cli_broker: bool = False,
 ) -> list[k8s_client.V1VolumeMount]:
     """Build volume mount list, mirroring three-way skills layout.
 
@@ -600,10 +688,12 @@ def _build_volume_mounts(
             _runtime_provided_extra_mounts(
                 extra_mounts,
                 provision_lark_cli_runtime=provision_lark_cli_runtime,
+                provision_lark_cli_broker=provision_lark_cli_broker,
             )
         )
     )
-    if _lark_cli_runtime_enabled(provision_lark_cli_runtime):
+    # Sandbox reads the runtime dir (real binary in Pattern A, shim in Pattern B).
+    if _lark_cli_runtime_enabled(provision_lark_cli_runtime) or _lark_cli_broker_enabled(provision_lark_cli_broker):
         mounts.append(
             k8s_client.V1VolumeMount(
                 name=LARK_CLI_RUNTIME_VOLUME_NAME,
@@ -617,8 +707,32 @@ def _build_volume_mounts(
 
 def _build_lark_cli_init_containers(
     provision_lark_cli_runtime: bool,
+    provision_lark_cli_broker: bool = False,
 ) -> list[k8s_client.V1Container]:
-    """Init container that copies the lark-cli runtime into the shared emptyDir."""
+    """Init container that stages the lark-cli runtime into the shared emptyDir.
+
+    Pattern B (broker) supersedes Pattern A: the broker image's ``install-shim``
+    mode writes the forwarding shim; Pattern A's init image copies the real
+    binary layout.
+    """
+    runtime_mount = k8s_client.V1VolumeMount(
+        name=LARK_CLI_RUNTIME_VOLUME_NAME,
+        mount_path=LARK_CLI_RUNTIME_CONTAINER_PATH,
+        read_only=False,
+    )
+    secure = k8s_client.V1SecurityContext(privileged=False, allow_privilege_escalation=False)
+    if _lark_cli_broker_enabled(provision_lark_cli_broker):
+        return [
+            k8s_client.V1Container(
+                name="lark-cli-shim-init",
+                image=LARK_CLI_BROKER_IMAGE,
+                image_pull_policy="IfNotPresent",
+                args=["install-shim", LARK_CLI_RUNTIME_CONTAINER_PATH],
+                env=[k8s_client.V1EnvVar(name="LARK_CLI_RUNTIME_DEST", value=LARK_CLI_RUNTIME_CONTAINER_PATH)],
+                volume_mounts=[runtime_mount],
+                security_context=secure,
+            )
+        ]
     if not _lark_cli_runtime_enabled(provision_lark_cli_runtime):
         return []
     return [
@@ -632,13 +746,62 @@ def _build_lark_cli_init_containers(
                     value=LARK_CLI_RUNTIME_CONTAINER_PATH,
                 )
             ],
-            volume_mounts=[
-                k8s_client.V1VolumeMount(
-                    name=LARK_CLI_RUNTIME_VOLUME_NAME,
-                    mount_path=LARK_CLI_RUNTIME_CONTAINER_PATH,
-                    read_only=False,
-                )
-            ],
+            volume_mounts=[runtime_mount],
+            security_context=secure,
+        )
+    ]
+
+
+def _build_lark_cli_broker_sidecars(
+    provision_lark_cli_broker: bool,
+    extra_mounts: list[ExtraMount] | None,
+) -> list[k8s_client.V1Container]:
+    """Broker sidecar that holds lark-cli + the per-user credentials (Pattern B).
+
+    The config/data credential dirs are mounted **only** here (never on the
+    sandbox container), so the plaintext app secret / OAuth tokens stay out of
+    the sandbox filesystem. The broker serves the command surface on loopback.
+    """
+    if not _lark_cli_broker_enabled(provision_lark_cli_broker):
+        return []
+    credential_mounts = _lark_broker_credential_mounts(extra_mounts)
+    volume_mounts: list[k8s_client.V1VolumeMount] = []
+    for container_path, volume_name, sidecar_path in (
+        (LARK_CLI_CONFIG_CONTAINER_PATH, LARK_BROKER_CONFIG_VOLUME_NAME, LARK_BROKER_SIDECAR_CONFIG_PATH),
+        (LARK_CLI_DATA_CONTAINER_PATH, LARK_BROKER_DATA_VOLUME_NAME, LARK_BROKER_SIDECAR_DATA_PATH),
+    ):
+        mount = credential_mounts.get(container_path)
+        if mount is None:
+            continue
+        sidecar_mount = k8s_client.V1VolumeMount(
+            name=volume_name,
+            mount_path=sidecar_path,
+            read_only=mount.read_only,
+        )
+        if USERDATA_PVC_NAME:
+            sidecar_mount.sub_path = _extra_mount_pvc_sub_path(mount.host_path)
+        volume_mounts.append(sidecar_mount)
+    broker_env = [
+        k8s_client.V1EnvVar(name="LARKSUITE_CLI_CONFIG_DIR", value=LARK_BROKER_SIDECAR_CONFIG_PATH),
+        k8s_client.V1EnvVar(name="LARKSUITE_CLI_DATA_DIR", value=LARK_BROKER_SIDECAR_DATA_PATH),
+    ]
+    # Forward the optional subcommand denylist so the broker refuses secret-dump
+    # subcommands (issue #4338 hardening); omitted when unset ⇒ nothing blocked.
+    if LARK_CLI_BROKER_DENY_SUBCOMMANDS:
+        broker_env.append(
+            k8s_client.V1EnvVar(
+                name="DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS",
+                value=LARK_CLI_BROKER_DENY_SUBCOMMANDS,
+            )
+        )
+    return [
+        k8s_client.V1Container(
+            name="lark-cli-broker",
+            image=LARK_CLI_BROKER_IMAGE,
+            image_pull_policy="IfNotPresent",
+            args=["serve"],
+            env=broker_env,
+            volume_mounts=volume_mounts,
             security_context=k8s_client.V1SecurityContext(
                 privileged=False,
                 allow_privilege_escalation=False,
@@ -655,10 +818,11 @@ def _build_pod(
     include_legacy_skills: bool = False,
     extra_mounts: list[ExtraMount] | None = None,
     provision_lark_cli_runtime: bool = False,
+    provision_lark_cli_broker: bool = False,
 ) -> k8s_client.V1Pod:
     """Construct a Pod manifest for a single sandbox."""
     init_containers = (
-        _build_lark_cli_init_containers(provision_lark_cli_runtime) or None
+        _build_lark_cli_init_containers(provision_lark_cli_runtime, provision_lark_cli_broker) or None
     )
     return k8s_client.V1Pod(
         metadata=k8s_client.V1ObjectMeta(
@@ -677,6 +841,11 @@ def _build_pod(
                     name="sandbox",
                     image=SANDBOX_IMAGE,
                     image_pull_policy="IfNotPresent",
+                    env=(
+                        [k8s_client.V1EnvVar(name="DEERFLOW_LARK_BROKER_URL", value=LARK_BROKER_URL)]
+                        if _lark_cli_broker_enabled(provision_lark_cli_broker)
+                        else None
+                    ),
                     ports=[
                         k8s_client.V1ContainerPort(
                             name="http",
@@ -722,12 +891,14 @@ def _build_pod(
                         include_legacy_skills=include_legacy_skills,
                         extra_mounts=extra_mounts,
                         provision_lark_cli_runtime=provision_lark_cli_runtime,
+                        provision_lark_cli_broker=provision_lark_cli_broker,
                     ),
                     security_context=k8s_client.V1SecurityContext(
                         privileged=False,
                         allow_privilege_escalation=True,
                     ),
-                )
+                ),
+                *_build_lark_cli_broker_sidecars(provision_lark_cli_broker, extra_mounts),
             ],
             init_containers=init_containers,
             volumes=_build_volumes(
@@ -736,6 +907,7 @@ def _build_pod(
                 include_legacy_skills=include_legacy_skills,
                 extra_mounts=extra_mounts,
                 provision_lark_cli_runtime=provision_lark_cli_runtime,
+                provision_lark_cli_broker=provision_lark_cli_broker,
             ),
             restart_policy="Always",
         ),
@@ -825,11 +997,15 @@ async def health():
 async def capabilities():
     """Report provisioner-side capabilities the Gateway cannot infer statically.
 
-    ``lark_cli_init_image`` reflects whether a lark-cli init image is configured,
-    which the Gateway surfaces as the Lark integration sandbox-runtime readiness
-    signal so a green UI can't hide a chat-time ``command not found``.
+    ``lark_cli_init_image`` / ``lark_cli_broker_image`` reflect whether a lark-cli
+    init image (Pattern A) / broker image (Pattern B) is configured, which the
+    Gateway surfaces as the Lark integration sandbox-runtime readiness signal so a
+    green UI can't hide a chat-time ``command not found``.
     """
-    return {"lark_cli_init_image": bool(LARK_CLI_INIT_IMAGE)}
+    return {
+        "lark_cli_init_image": bool(LARK_CLI_INIT_IMAGE),
+        "lark_cli_broker_image": bool(LARK_CLI_BROKER_IMAGE),
+    }
 
 
 @app.post("/api/sandboxes", response_model=SandboxResponse)
@@ -844,14 +1020,16 @@ def create_sandbox(req: CreateSandboxRequest):
     user_id = req.user_id
     include_legacy_skills = req.include_legacy_skills
     provision_lark_cli_runtime = req.provision_lark_cli_runtime
+    provision_lark_cli_broker = req.provision_lark_cli_broker
 
     logger.info(
-        "Received request to create sandbox '%s' for thread '%s' user '%s' include_legacy_skills=%s provision_lark_cli_runtime=%s",
+        "Received request to create sandbox '%s' for thread '%s' user '%s' include_legacy_skills=%s provision_lark_cli_runtime=%s provision_lark_cli_broker=%s",
         sandbox_id,
         thread_id,
         user_id,
         include_legacy_skills,
         _lark_cli_runtime_enabled(provision_lark_cli_runtime),
+        _lark_cli_broker_enabled(provision_lark_cli_broker),
     )
 
     # ── Fast path: sandbox already exists ────────────────────────────
@@ -874,6 +1052,7 @@ def create_sandbox(req: CreateSandboxRequest):
                 include_legacy_skills=include_legacy_skills,
                 extra_mounts=req.extra_mounts,
                 provision_lark_cli_runtime=provision_lark_cli_runtime,
+                provision_lark_cli_broker=provision_lark_cli_broker,
             ),
         )
         logger.info(f"Created Pod {_pod_name(sandbox_id)}")

frontend/src/components/workspace/settings/integrations-settings-page.tsx

@@ -575,11 +575,13 @@ function LarkIntegrationCard() {
                   ok={data.sandbox_runtime_ready}
                   value={
                     data.sandbox_runtime_ready
-                      ? data.sandbox_runtime_mode === "init-container"
-                        ? t.settings.integrations.lark
-                            .sandboxRuntimeInitContainer
-                        : t.settings.integrations.lark
-                            .sandboxRuntimeGatewayDownload
+                      ? data.sandbox_runtime_mode === "broker"
+                        ? t.settings.integrations.lark.sandboxRuntimeBroker
+                        : data.sandbox_runtime_mode === "init-container"
+                          ? t.settings.integrations.lark
+                              .sandboxRuntimeInitContainer
+                          : t.settings.integrations.lark
+                              .sandboxRuntimeGatewayDownload
                       : (data.sandbox_runtime_detail ??
                         t.settings.integrations.lark.sandboxRuntimeNotReady)
                   }

frontend/src/core/i18n/locales/en-US.ts

@@ -855,6 +855,7 @@ export const enUS: Translations = {
         auth: "Auth",
         sandboxRuntime: "Sandbox runtime",
         sandboxRuntimeInitContainer: "Provisioned by init container",
+        sandboxRuntimeBroker: "Provisioned by broker sidecar",
         sandboxRuntimeGatewayDownload: "Provisioned by Gateway",
         sandboxRuntimeNotReady:
           "Not ready — lark-cli may be missing at chat time",

frontend/src/core/i18n/locales/types.ts

@@ -726,6 +726,7 @@ export interface Translations {
         auth: string;
         sandboxRuntime: string;
         sandboxRuntimeInitContainer: string;
+        sandboxRuntimeBroker: string;
         sandboxRuntimeGatewayDownload: string;
         sandboxRuntimeNotReady: string;
         notInstalled: string;

frontend/src/core/i18n/locales/zh-CN.ts

@@ -822,6 +822,7 @@ export const zhCN: Translations = {
         auth: "授权",
         sandboxRuntime: "沙箱运行时",
         sandboxRuntimeInitContainer: "由 init container 提供",
+        sandboxRuntimeBroker: "由 broker sidecar 提供",
         sandboxRuntimeGatewayDownload: "由 Gateway 提供",
         sandboxRuntimeNotReady: "未就绪 —— 对话时 lark-cli 可能不可用",
         notInstalled: "尚未安装",

frontend/src/core/integrations/lark/types.ts

@@ -20,7 +20,8 @@ export interface LarkAuthProbe {
 export type LarkSandboxRuntimeMode =
   | "none"
   | "gateway-download"
-  | "init-container";
+  | "init-container"
+  | "broker";
 
 export interface LarkIntegrationStatus {
   installed: boolean;