feat(channels): expose IM channel_user_id to sandbox commands as DEERFLOW_CHANNEL_USER_ID (#3926)

* feat(channels): expose channel_user_id to sandbox commands as DEERFLOW_CHANNEL_USER_ID

IM-channel skills need the sender's platform identity (Feishu open_id,
Slack Uxxx, ...). The channel manager already writes channel_user_id into
body.context, but the Gateway whitelist dropped it. Forward it into the
runtime context only (never configurable, which is checkpointed), and have
bash_tool export it as a fixed env var through a shell-quoted command
prefix.

The identity deliberately does not ride execute_command(env=...): that
channel is reserved for request-scoped secrets, and a non-empty env
switches AioSandbox onto the bash.exec path (fresh session per call,
image >= 1.9.3 required), which would have broken every IM bash command
on older sandbox images and abandoned persistent-shell semantics on new
ones. A command-string export keeps the legacy path, stays visible in
audit logs (it is an identifier, not a secret), and gives per-call
correctness in group chats where one thread and sandbox are shared by
senders with different platform ids.

Skipped on the Windows local sandbox, whose PowerShell/cmd.exe fallback
has no POSIX export.

Part of #3914

* feat(channels): propagate channel_user_id to subagents; cap value length

Review findings from the pre-PR verification pass:

- Subagent delegation dropped the sender identity: task_tool now captures
  channel_user_id from the parent runtime context and the executor
  forwards it into the subagent's context, mirroring the guardrail
  attribution fields (user_role/oauth_*/run_id). Without this, bash
  commands delegated via task lost the group-chat sender's id.
- body.context is client-writable on web requests, so values over 256
  chars are ignored instead of bloating every command string sent to the
  sandbox.

* fix(channels): set-or-unset channel_user_id so identity is per-call regardless of AIO session persistence

Review (willem-bd): the identity export could leak across senders in a
shared group-chat AIO sandbox. The AIO no-env path reuses a persistent
shell session (the class-lock reason, #1433), and the 256-char/type guard
made some commands carry no prefix — so a dropped-id command could resolve
the id a previous sender exported.

Make per-call correctness independent of session semantics: an IM-channel
command (channel_user_id present in context) now always carries an explicit
prefix — export VAR=<quoted> for a valid id, or unset VAR for an unusable
one (empty / non-str / over the cap). Non-IM runs (no key) are untouched.
A prefix unset has none of the '& ; unset' suffix hazard raised earlier.

Verified on a real AIO 1.11.0 container: the no-id shell path auto-creates
a session per call (does not persist today), but an explicit shared session
DOES persist (export stale-A -> readback [stale-A]); the unset prefix
clears it (-> []). So the fix holds even on an image whose no-id path
persists. Regression tests cover the dropped-id group-chat window and the
non-IM passthrough.

Part of #3914

* test(channels): align channel_user_id task test with new Command return shape

The merge from main changed task_tool to return a Command(update=...)
instead of a plain string; update the assertion to extract the tool
message via the existing _task_tool_message helper, matching the sibling
tests. Fixes the CI backend-unit-tests failure introduced by the merge.

backend/AGENTS.md

@@ -446,7 +446,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 **Message Flow**:
 1. External platform -> Channel impl -> `MessageBus.publish_inbound()`
 2. `ChannelManager._dispatch_loop()` consumes from queue
-3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`
+3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`. The Gateway forwards `channel_user_id` from `body.context` into the runtime context only (never `configurable`, which is checkpointed), and `bash_tool` exposes it to sandbox commands as the fixed env var `DEERFLOW_CHANNEL_USER_ID` — via a shell-quoted command-string prefix, NOT the `execute_command(env=...)` channel, which is reserved for request-scoped secrets and would switch `AioSandbox` onto the `bash.exec` path (image >= 1.9.3, fresh session per call). Per-call injection keeps group-chat identity correct (one thread/sandbox, many senders) **without depending on the AIO shell's session semantics**: every IM-channel command carries an explicit `export VAR=<id>; ` (valid id) or `unset VAR; ` (empty / non-str / over the 256-char cap, since `body.context` is client-writable). The AIO no-env path reuses a persistent shell session (the reason for the class lock, #1433), so a bare command could otherwise resolve a stale id an earlier sender exported; the `unset` closes the window the length/type guard would open (a dropped id would inherit the previous sender's value). Non-IM runs (no `channel_user_id` in context) are left untouched. Not injected on the Windows local sandbox (its PowerShell/cmd.exe fallback has no `export`/`unset`). Propagates across `task` delegation: `task_tool` captures the dispatching turn's id and the subagent executor forwards it into the subagent's runtime context, same as the guardrail attribution fields. The var is informational, never authorization-grade: any bash command can overwrite it (and web clients can set `body.context.channel_user_id`), so skills must not treat it as authenticated identity. Tests: `tests/test_channel_user_id_env.py`
 4. For chat: look up/create thread through Gateway's LangGraph-compatible API
 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`)
 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound

backend/app/gateway/services.py

@@ -194,6 +194,11 @@ def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, An
                 runtime_context.setdefault(key, context[key])
     if "user_id" in context and isinstance(runtime_context, dict):
         runtime_context.setdefault("user_id", context["user_id"])
+    # The raw platform user id from IM channels (Feishu open_id, Slack Uxxx, ...)
+    # follows the same runtime-context-only rule as user_id: tools may read it,
+    # but it never enters ``configurable`` (checkpointed with the thread).
+    if "channel_user_id" in context and isinstance(runtime_context, dict):
+        runtime_context.setdefault("channel_user_id", context["channel_user_id"])
 
 
 def inject_authenticated_user_context(config: dict[str, Any], request: Request) -> None:

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

@@ -1544,6 +1544,54 @@ def _truncate_ls_output(output: str, max_chars: int) -> str:
     return f"{output[:kept]}{marker}"
 
 
+# Fixed env var exposing the IM-channel platform user id (Feishu open_id,
+# Slack Uxxx, ...) to sandbox commands, so skills can act on the current end
+# user's channel identity (#3914). An identifier, not a secret.
+CHANNEL_USER_ID_ENV = "DEERFLOW_CHANNEL_USER_ID"
+
+_CHANNEL_USER_ID_CONTEXT_KEY = "channel_user_id"
+
+# body.context is client-writable on web requests, so bound the value: real
+# platform ids are tens of chars; anything past this is hostile or corrupt and
+# must not bloat every command string sent to the sandbox.
+_CHANNEL_USER_ID_MAX_LEN = 256
+
+
+def _is_windows() -> bool:
+    return os.name == "nt"
+
+
+def _channel_identity_prefix(runtime: Runtime) -> str | None:
+    """Build the command prefix that sets or clears the channel-user-id env var.
+
+    Returns ``None`` for a non-IM run (no ``channel_user_id`` key in context) so
+    the command is left untouched. For an IM run the prefix is always emitted:
+
+    - valid id (non-empty str within the length cap) → ``export VAR=<quoted>; ``
+    - unusable id (empty / non-str / over the cap) → ``unset VAR; ``
+
+    The id deliberately rides the command string instead of the
+    ``execute_command(env=...)`` channel: a non-empty ``env`` switches
+    ``AioSandbox`` to the ``bash.exec`` API (fresh session per call, image
+    >= 1.9.3 required), which is reserved for request-scoped secrets. Emitting an
+    explicit ``export``-or-``unset`` on every IM command makes per-call identity
+    correct **without depending on the AIO shell's session semantics**: the AIO
+    no-env path reuses a persistent shell session (the reason for the class lock,
+    #1433), so a bare command could otherwise resolve a stale value exported by
+    an earlier sender in a shared group-chat sandbox. The ``unset`` closes the
+    window the length/type guard would otherwise open — a sender whose id is
+    dropped inherits the previous sender's value. Values are identifiers, not
+    secrets, so keeping them in the audit-visible command string is fine.
+    """
+    context = getattr(runtime, "context", None)
+    if not isinstance(context, dict) or _CHANNEL_USER_ID_CONTEXT_KEY not in context:
+        return None
+    channel_user_id = context.get(_CHANNEL_USER_ID_CONTEXT_KEY)
+    if isinstance(channel_user_id, str) and 0 < len(channel_user_id) <= _CHANNEL_USER_ID_MAX_LEN:
+        return f"export {CHANNEL_USER_ID_ENV}={shlex.quote(channel_user_id)}; "
+    return f"unset {CHANNEL_USER_ID_ENV}; "
+
+
 @tool("bash", parse_docstring=True)
 def bash_tool(runtime: Runtime, description: str, command: str) -> str:
     """Execute a bash command in a Linux environment.
@@ -1566,6 +1614,7 @@ def bash_tool(runtime: Runtime, description: str, command: str) -> str:
         # Request-scoped secrets resolved for the active skill (#3861); injected as
         # per-call env into the subprocess, never placed in the command string.
         injected_env = read_active_secrets(getattr(runtime, "context", None)) or None
+        identity_prefix = _channel_identity_prefix(runtime)
         if is_local_sandbox(runtime):
             if not is_host_bash_allowed():
                 return f"Error: {LOCAL_HOST_BASH_DISABLED_MESSAGE}"
@@ -1574,6 +1623,10 @@ def bash_tool(runtime: Runtime, description: str, command: str) -> str:
             validate_local_bash_command_paths(command, thread_data)
             command = replace_virtual_paths_in_command(command, thread_data)
             command = _apply_cwd_prefix(command, thread_data)
+            # POSIX-only: the Windows local sandbox may execute via
+            # PowerShell/cmd.exe where `export` is not valid syntax.
+            if identity_prefix and not _is_windows():
+                command = identity_prefix + command
             try:
                 from deerflow.config.app_config import get_app_config
 
@@ -1589,6 +1642,8 @@ def bash_tool(runtime: Runtime, description: str, command: str) -> str:
                 max_chars,
             )
         ensure_thread_directories_exist(runtime)
+        if identity_prefix:
+            command = identity_prefix + command
         try:
             from deerflow.config.app_config import get_app_config
 

backend/packages/harness/deerflow/subagents/executor.py

@@ -295,6 +295,7 @@ class SubagentExecutor:
         oauth_provider: str | None = None,
         oauth_id: str | None = None,
         run_id: str | None = None,
+        channel_user_id: str | None = None,
         deerflow_trace_id: str | None = None,
     ):
         """Initialize the executor.
@@ -342,6 +343,10 @@ class SubagentExecutor:
         self.oauth_provider = oauth_provider
         self.oauth_id = oauth_id
         self.run_id = run_id
+        # IM-channel sender identity captured at task_tool dispatch: group
+        # chats share one thread across senders, so delegated bash commands
+        # must export the dispatching turn's id, not none at all.
+        self.channel_user_id = channel_user_id
         self.deerflow_trace_id = deerflow_trace_id
 
         self._base_tools = _filter_tools(
@@ -604,6 +609,8 @@ class SubagentExecutor:
             context["oauth_provider"] = self.oauth_provider
             context["oauth_id"] = self.oauth_id
             context["run_id"] = self.run_id
+            if self.channel_user_id:
+                context["channel_user_id"] = self.channel_user_id
             if self.deerflow_trace_id:
                 context[DEERFLOW_TRACE_METADATA_KEY] = self.deerflow_trace_id
             context["is_subagent"] = True

backend/packages/harness/deerflow/tools/builtins/task_tool.py

@@ -327,6 +327,9 @@ async def task_tool(
     oauth_provider = parent_context.get("oauth_provider")
     oauth_id = parent_context.get("oauth_id")
     run_id = parent_context.get("run_id")
+    # IM-channel sender identity: group chats share one thread across senders,
+    # so delegated bash commands need the dispatching turn's channel_user_id.
+    channel_user_id = parent_context.get("channel_user_id")
     deerflow_trace_id = normalize_trace_id(parent_context.get(DEERFLOW_TRACE_METADATA_KEY)) or normalize_trace_id(metadata.get(DEERFLOW_TRACE_METADATA_KEY)) or get_current_trace_id()
 
     parent_available_skills = metadata.get("available_skills")
@@ -371,6 +374,7 @@ async def task_tool(
         "oauth_provider": oauth_provider,
         "oauth_id": oauth_id,
         "run_id": run_id,
+        "channel_user_id": channel_user_id,
         "deerflow_trace_id": deerflow_trace_id,
     }
     if resolved_app_config is not None:

backend/tests/test_channel_user_id_env.py

@@ -0,0 +1,200 @@
+"""Tests for exposing the IM-channel platform user id to sandbox commands (#3914).
+
+Two halves:
+- Gateway: ``merge_run_context_overrides`` forwards ``channel_user_id`` from
+  ``body.context`` into ``config['context']`` (runtime context) only — never
+  into ``configurable`` (which is checkpointed).
+- Sandbox: ``bash_tool`` exposes the id as the fixed env var
+  ``DEERFLOW_CHANNEL_USER_ID`` via an ``export`` prefix on the command string.
+  It must NOT ride the ``env=`` parameter: on ``AioSandbox`` a non-empty env
+  switches execution to the ``bash.exec`` API, which requires image >= 1.9.3
+  and abandons the persistent shell session — that channel is reserved for
+  request-scoped secrets.
+"""
+
+from types import SimpleNamespace
+
+from deerflow.sandbox.tools import (
+    CHANNEL_USER_ID_ENV,
+    _channel_identity_prefix,
+    bash_tool,
+)
+
+_THREAD_DATA = {
+    "workspace_path": "/tmp/deer-flow/threads/t1/user-data/workspace",
+    "uploads_path": "/tmp/deer-flow/threads/t1/user-data/uploads",
+    "outputs_path": "/tmp/deer-flow/threads/t1/user-data/outputs",
+}
+
+
+def _aio_runtime(context: dict) -> SimpleNamespace:
+    return SimpleNamespace(
+        state={"sandbox": {"sandbox_id": "aio-sandbox-1"}, "thread_data": _THREAD_DATA.copy()},
+        context=context,
+    )
+
+
+class _CapturingSandbox:
+    def __init__(self, output: str = "ok"):
+        self.calls: list[dict] = []
+        self._output = output
+
+    def execute_command(self, command: str, env=None, timeout=None) -> str:
+        self.calls.append({"command": command, "env": env})
+        return self._output
+
+
+def _run_bash(monkeypatch, runtime, command: str = "echo hi") -> _CapturingSandbox:
+    sandbox = _CapturingSandbox()
+    monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
+    monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None)
+    bash_tool.func(runtime=runtime, description="test", command=command)
+    return sandbox
+
+
+class TestMergeRunContextOverridesChannelUserId:
+    def test_channel_user_id_propagates_to_runtime_context_only(self):
+        from app.gateway.services import build_run_config, merge_run_context_overrides
+
+        config = build_run_config("thread-1", None, None)
+        merge_run_context_overrides(config, {"channel_user_id": "ou_feishu_123"})
+
+        assert config["context"]["channel_user_id"] == "ou_feishu_123"
+        # Never into configurable: that mapping is checkpointed with the thread.
+        assert "channel_user_id" not in config["configurable"]
+
+    def test_existing_runtime_context_value_wins(self):
+        """setdefault semantics: a server-side value stamped earlier must not be
+        overridden by the client-supplied body.context."""
+        from app.gateway.services import build_run_config, merge_run_context_overrides
+
+        config = build_run_config("thread-1", None, None)
+        config.setdefault("context", {})["channel_user_id"] = "server-stamped"
+        merge_run_context_overrides(config, {"channel_user_id": "client-supplied"})
+
+        assert config["context"]["channel_user_id"] == "server-stamped"
+
+    def test_absent_channel_user_id_adds_nothing(self):
+        from app.gateway.services import build_run_config, merge_run_context_overrides
+
+        config = build_run_config("thread-1", None, None)
+        merge_run_context_overrides(config, {"model_name": "gpt"})
+
+        assert "channel_user_id" not in config.get("context", {})
+
+
+class TestBashToolChannelIdentityPrefix:
+    def test_identity_exported_and_env_stays_none(self, monkeypatch):
+        """The id rides the command string; env must stay None so AioSandbox
+        keeps the legacy persistent-shell path (regression guard for the
+        #3921/#3922 bash.exec capability gap)."""
+        sandbox = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "ou_feishu_123"}))
+
+        assert len(sandbox.calls) == 1
+        assert sandbox.calls[0]["command"] == f"export {CHANNEL_USER_ID_ENV}=ou_feishu_123; echo hi"
+        assert sandbox.calls[0]["env"] is None
+
+    def test_no_channel_user_id_leaves_command_unchanged(self, monkeypatch):
+        sandbox = _run_bash(monkeypatch, _aio_runtime({"thread_id": "t1"}))
+
+        assert sandbox.calls[0]["command"] == "echo hi"
+        assert sandbox.calls[0]["env"] is None
+
+    def test_per_call_identity_follows_current_context(self, monkeypatch):
+        """Group chats share one thread/sandbox: each message's run carries that
+        sender's id, so consecutive commands must each export their own value."""
+        first = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "sender-a"}))
+        second = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "sender-b"}))
+
+        assert "sender-a" in first.calls[0]["command"]
+        assert "sender-b" in second.calls[0]["command"]
+
+    def test_value_is_shell_quoted(self, monkeypatch):
+        """A hostile platform id must not be able to inject shell syntax."""
+        sandbox = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "x'; rm -rf /tmp/y; '"}))
+
+        command = sandbox.calls[0]["command"]
+        assert command.endswith("; echo hi")
+        # shlex.quote wraps the value; the raw injection payload must not appear
+        # as executable syntax outside the quoted region.
+        assert "export " + CHANNEL_USER_ID_ENV + "='x'\"'\"'; rm -rf /tmp/y; '\"'\"''; echo hi" == command
+
+    def test_secrets_and_identity_compose(self, monkeypatch):
+        """Active skill secrets keep the env= channel; the identity keeps the
+        command-string channel. They must not mix."""
+        runtime = _aio_runtime(
+            {
+                "channel_user_id": "ou_1",
+                "__active_skill_secrets": {"ERP_TOKEN": "secret-value"},
+            }
+        )
+        sandbox = _run_bash(monkeypatch, runtime)
+
+        call = sandbox.calls[0]
+        assert call["env"] == {"ERP_TOKEN": "secret-value"}
+        assert call["command"].startswith(f"export {CHANNEL_USER_ID_ENV}=ou_1; ")
+        assert "secret-value" not in call["command"]
+
+    def test_non_im_run_leaves_command_untouched(self):
+        """No channel_user_id key at all → non-IM run → prefix is None so the
+        command (the vast majority: Web/API/subagent) is unchanged."""
+        assert _channel_identity_prefix(SimpleNamespace(context={"thread_id": "t1"})) is None
+        assert _channel_identity_prefix(SimpleNamespace(context={})) is None
+        assert _channel_identity_prefix(SimpleNamespace(context=None)) is None
+
+    def test_unusable_value_emits_unset_not_none(self, monkeypatch):
+        """An IM run whose id is unusable (empty / non-str / over the cap) must
+        emit ``unset`` — not skip the prefix. Skipping would let a bare command
+        resolve a stale value left in the AIO persistent shell by an earlier
+        sender (willem-bd's group-chat leak window)."""
+        for bad in ("", 123, "x" * 5000, None):
+            prefix = _channel_identity_prefix(SimpleNamespace(context={"channel_user_id": bad}))
+            assert prefix == f"unset {CHANNEL_USER_ID_ENV}; ", f"value={bad!r}"
+
+    def test_group_chat_dropped_id_clears_previous_sender(self, monkeypatch):
+        """Sender A (valid) then sender B (over-cap id, dropped): B's command must
+        carry ``unset`` so it cannot inherit A's exported id in a shared
+        persistent-shell sandbox — per-call correctness independent of session
+        persistence."""
+        a = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "sender-a"}))
+        b = _run_bash(monkeypatch, _aio_runtime({"channel_user_id": "b" * 5000}))
+
+        assert a.calls[0]["command"] == f"export {CHANNEL_USER_ID_ENV}=sender-a; echo hi"
+        assert b.calls[0]["command"] == f"unset {CHANNEL_USER_ID_ENV}; echo hi"
+        assert b.calls[0]["env"] is None
+
+    def test_windows_local_sandbox_skips_prefix(self, monkeypatch):
+        """On Windows the local sandbox may execute via PowerShell/cmd.exe where
+        POSIX ``export`` is not valid syntax — skip injection rather than break
+        every IM-channel command."""
+        runtime = SimpleNamespace(
+            state={"sandbox": {"sandbox_id": "local"}, "thread_data": _THREAD_DATA.copy()},
+            context={"channel_user_id": "ou_1", "thread_id": "t1"},
+        )
+        sandbox = _CapturingSandbox()
+        monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
+        monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None)
+        monkeypatch.setattr("deerflow.sandbox.tools.is_host_bash_allowed", lambda: True)
+        monkeypatch.setattr("deerflow.sandbox.tools._is_windows", lambda: True)
+
+        bash_tool.func(runtime=runtime, description="test", command="echo hi")
+
+        assert len(sandbox.calls) == 1
+        assert "export" not in sandbox.calls[0]["command"]
+
+    def test_posix_local_sandbox_gets_prefix(self, monkeypatch):
+        runtime = SimpleNamespace(
+            state={"sandbox": {"sandbox_id": "local"}, "thread_data": _THREAD_DATA.copy()},
+            context={"channel_user_id": "ou_1", "thread_id": "t1"},
+        )
+        sandbox = _CapturingSandbox()
+        monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
+        monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None)
+        monkeypatch.setattr("deerflow.sandbox.tools.is_host_bash_allowed", lambda: True)
+
+        bash_tool.func(runtime=runtime, description="test", command="echo hi")
+
+        assert len(sandbox.calls) == 1
+        command = sandbox.calls[0]["command"]
+        assert command.startswith(f"export {CHANNEL_USER_ID_ENV}=ou_1; ")
+        assert command.endswith("echo hi")

backend/tests/test_subagent_executor.py

@@ -2417,6 +2417,43 @@ class TestSubagentGuardrailAttribution:
         assert context.get("run_id") == "run-42"
         assert context.get("is_subagent") is True
 
+    @pytest.mark.anyio
+    async def test_aexecute_propagates_channel_user_id_to_subagent_context(
+        self,
+        classes,
+        executor_module,
+        monkeypatch,
+    ):
+        """The IM-channel sender identity captured at task_tool must reach the
+        subagent's ``astream`` context so delegated bash commands export the
+        dispatching turn's ``DEERFLOW_CHANNEL_USER_ID`` (group chats share one
+        thread across senders)."""
+        SubagentExecutor = classes["SubagentExecutor"]
+        SubagentConfig = classes["SubagentConfig"]
+        executor = SubagentExecutor(
+            config=SubagentConfig(
+                name="general-purpose",
+                description="Channel identity test agent",
+                system_prompt="You are a channel identity test agent.",
+                max_turns=5,
+                timeout_seconds=30,
+            ),
+            tools=[],
+            parent_model="test-model",
+            thread_id="thread-channel-1",
+            trace_id="trace-channel-1",
+            channel_user_id="ou_group_sender_1",
+        )
+        fake_agent = _FakeStreamAgent()
+        monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
+        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+
+        await executor._aexecute("do something")
+
+        context = fake_agent.captured_context
+        assert context is not None
+        assert context.get("channel_user_id") == "ou_group_sender_1"
+
     @pytest.mark.anyio
     async def test_aexecute_context_defaults_to_none_when_attribution_absent(
         self,

backend/tests/test_task_tool_core_logic.py

@@ -186,6 +186,47 @@ def test_task_tool_returns_error_for_unknown_subagent(monkeypatch):
     assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == "Unknown subagent type 'general-purpose'. Available: general-purpose"
 
 
+def test_task_tool_forwards_channel_user_id_to_executor(monkeypatch):
+    """The IM-channel sender identity must survive delegation: in group chats
+    one thread serves many senders, so a subagent's bash commands need the
+    dispatching turn's channel_user_id (same propagation rule as user_role /
+    oauth attribution)."""
+    runtime = _make_runtime()
+    runtime.context["channel_user_id"] = "ou_group_sender_1"
+    captured = {}
+
+    class DummyExecutor:
+        def __init__(self, **kwargs):
+            captured["executor_kwargs"] = kwargs
+
+        def execute_async(self, prompt, task_id=None):
+            return task_id or "generated-task-id"
+
+    monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
+    monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor)
+    monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
+    monkeypatch.setattr(
+        task_tool_module,
+        "get_background_task_result",
+        lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"),
+    )
+    monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _event: None)
+    monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
+    monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
+
+    output = _run_task_tool(
+        runtime=runtime,
+        description="运行子任务",
+        prompt="collect diagnostics",
+        subagent_type="general-purpose",
+        tool_call_id="tc-channel-id",
+    )
+
+    message = _task_tool_message(output)
+    assert message.content == "Task Succeeded. Result: done"
+    assert captured["executor_kwargs"]["channel_user_id"] == "ou_group_sender_1"
+
+
 def test_task_tool_rejects_bash_subagent_when_host_bash_disabled(monkeypatch):
     monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
     monkeypatch.setattr(task_tool_module, "is_host_bash_allowed", lambda: False)